kunthawat/ALwrity
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Alwrity copilotkit integration - 0.5.7

Browse Source
This commit is contained in:
ajaysi
2025-08-28 20:46:42 +05:30
parent f76381030b
commit 1fa2067301
17 changed files with 7820 additions and 201 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
backend/.onboarding_progress.json
View File

@@ -50,14 +50,14 @@
"title": "Complete Setup",
"description": "Finalize and complete onboarding",
"status": "completed",
"completed_at": "2025-07-31T12:18:48.982697",
"completed_at": "2025-08-28T13:33:52.944161",
"data": {},
"validation_errors": []
}
],
"current_step": 6,
"started_at": "2025-07-30T18:45:53.838059",
"last_updated": "2025-07-31T12:18:48.992288",
"last_updated": "2025-08-28T13:33:52.958699",
"is_completed": true,
"completed_at": "2025-07-31T12:18:48.992276"
"completed_at": "2025-08-28T13:33:52.958699"
}

71
backend/api/content_planning/strategy_copilot.py Normal file
View File

@@ -0,0 +1,71 @@
from fastapi import APIRouter, HTTPException, Depends
from sqlalchemy.orm import Session
from typing import Dict, Any, List
from services.database import get_db
from services.strategy_copilot_service import StrategyCopilotService
router = APIRouter(prefix="/api/content-planning/strategy", tags=["strategy-copilot"])
@router.post("/generate-category-data")
async def generate_category_data(
request: Dict[str, Any],
db: Session = Depends(get_db)
):
"""Generate data for a specific category based on user description."""
try:
service = StrategyCopilotService(db)
result = await service.generate_category_data(
category=request["category"],
user_description=request["userDescription"],
current_form_data=request["currentFormData"]
)
return {"success": True, "data": result}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@router.post("/validate-field")
async def validate_field(
request: Dict[str, Any],
db: Session = Depends(get_db)
):
"""Validate a specific strategy field."""
try:
service = StrategyCopilotService(db)
result = await service.validate_field(
field_id=request["fieldId"],
value=request["value"]
)
return result
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@router.post("/analyze")
async def analyze_strategy(
request: Dict[str, Any],
db: Session = Depends(get_db)
):
"""Analyze complete strategy for completeness and coherence."""
try:
service = StrategyCopilotService(db)
result = await service.analyze_strategy(
form_data=request["formData"]
)
return result
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@router.post("/generate-suggestions")
async def generate_suggestions(
request: Dict[str, Any],
db: Session = Depends(get_db)
):
"""Generate suggestions for a specific field."""
try:
service = StrategyCopilotService(db)
result = await service.generate_field_suggestions(
field_id=request["fieldId"],
current_form_data=request["currentFormData"]
)
return result
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))

8
backend/app.py
View File

@@ -53,6 +53,9 @@ from api.component_logic import router as component_logic_router
from api.content_planning.api.router import router as content_planning_router
from api.user_data import router as user_data_router
# Import strategy copilot endpoints
from api.content_planning.strategy_copilot import router as strategy_copilot_router
# Import database service
from services.database import init_database, close_database
@@ -76,9 +79,7 @@ from api.seo_dashboard import (
app = FastAPI(
title="ALwrity Backend API",
description="Backend API for ALwrity - AI-powered content creation platform",
version="2.0.0",
docs_url="/api/docs",
redoc_url="/api/redoc"
version="1.0.0"
)
# Add CORS middleware
@@ -365,6 +366,7 @@ app.include_router(component_logic_router)
# Include content planning router
app.include_router(content_planning_router)
app.include_router(user_data_router)
app.include_router(strategy_copilot_router)
# SEO Dashboard endpoints
@app.get("/api/seo-dashboard/data")

2
backend/requirements.txt
View File

@@ -9,6 +9,8 @@ tenacity>=8.2.3
# Database dependencies
sqlalchemy>=2.0.25
copilotkit
# AI/ML dependencies - using more flexible versions
openai>=1.3.0
anthropic>=0.7.0

2
backend/services/llm_providers/gemini_provider.py
View File

@@ -169,7 +169,7 @@ def gemini_text_response(prompt, temperature, top_p, n, max_tokens, system_promp
# FIXME: Expose model_name in main_config
try:
response = client.models.generate_content(
model='gemini-2.5-pro',
model='gemini-2.0-flash-lite',
contents=prompt,
config=types.GenerateContentConfig(
system_instruction=system_prompt,

389
backend/services/strategy_copilot_service.py Normal file
View File

@@ -0,0 +1,389 @@
from typing import Dict, Any, List, Optional
from sqlalchemy.orm import Session
from loguru import logger
from services.onboarding_data_service import OnboardingDataService
from services.user_data_service import UserDataService
from services.llm_providers.gemini_provider import gemini_text_response, gemini_structured_json_response
class StrategyCopilotService:
"""Service for CopilotKit strategy assistance using Gemini."""
def __init__(self, db: Session):
self.db = db
self.onboarding_service = OnboardingDataService()
self.user_data_service = UserDataService(db)
async def generate_category_data(
self,
category: str,
user_description: str,
current_form_data: Dict[str, Any]
) -> Dict[str, Any]:
"""Generate data for a specific category."""
try:
# Get user onboarding data
user_id = 1 # TODO: Get from auth context
onboarding_data = self.onboarding_service.get_personalized_ai_inputs(user_id)
# Build prompt for category generation
prompt = self._build_category_generation_prompt(
category, user_description, current_form_data, onboarding_data
)
# Generate response using Gemini
response = gemini_text_response(
prompt=prompt,
temperature=0.3,
top_p=0.9,
n=1,
max_tokens=2048,
system_prompt="You are ALwrity's Strategy Assistant. Generate appropriate values for strategy fields."
)
# Parse and validate response
generated_data = self._parse_category_response(response, category)
return generated_data
except Exception as e:
logger.error(f"Error generating category data: {str(e)}")
raise
async def validate_field(self, field_id: str, value: Any) -> Dict[str, Any]:
"""Validate a specific strategy field."""
try:
# Get field definition
field_definition = self._get_field_definition(field_id)
# Build validation prompt
prompt = self._build_validation_prompt(field_definition, value)
# Generate validation response using Gemini
response = gemini_text_response(
prompt=prompt,
temperature=0.2,
top_p=0.9,
n=1,
max_tokens=1024,
system_prompt="You are ALwrity's Strategy Assistant. Validate field values and provide suggestions."
)
# Parse validation result
validation_result = self._parse_validation_response(response)
return validation_result
except Exception as e:
logger.error(f"Error validating field {field_id}: {str(e)}")
raise
async def analyze_strategy(self, form_data: Dict[str, Any]) -> Dict[str, Any]:
"""Analyze complete strategy for completeness and coherence."""
try:
# Get user data for context
user_id = 1 # TODO: Get from auth context
onboarding_data = self.onboarding_service.get_personalized_ai_inputs(user_id)
# Build analysis prompt
prompt = self._build_analysis_prompt(form_data, onboarding_data)
# Generate analysis using Gemini
response = gemini_text_response(
prompt=prompt,
temperature=0.3,
top_p=0.9,
n=1,
max_tokens=2048,
system_prompt="You are ALwrity's Strategy Assistant. Analyze strategies for completeness and coherence."
)
# Parse analysis result
analysis_result = self._parse_analysis_response(response)
return analysis_result
except Exception as e:
logger.error(f"Error analyzing strategy: {str(e)}")
raise
async def generate_field_suggestions(
self,
field_id: str,
current_form_data: Dict[str, Any]
) -> Dict[str, Any]:
"""Generate suggestions for a specific field."""
try:
# Get field definition
field_definition = self._get_field_definition(field_id)
# Get user data
user_id = 1 # TODO: Get from auth context
onboarding_data = self.onboarding_service.get_personalized_ai_inputs(user_id)
# Build suggestions prompt
prompt = self._build_suggestions_prompt(
field_definition, current_form_data, onboarding_data
)
# Generate suggestions using Gemini
response = gemini_text_response(
prompt=prompt,
temperature=0.4,
top_p=0.9,
n=1,
max_tokens=1024,
system_prompt="You are ALwrity's Strategy Assistant. Generate helpful suggestions for strategy fields."
)
# Parse suggestions
suggestions = self._parse_suggestions_response(response)
return suggestions
except Exception as e:
logger.error(f"Error generating suggestions for {field_id}: {str(e)}")
raise
def _build_category_generation_prompt(
self,
category: str,
user_description: str,
current_form_data: Dict[str, Any],
onboarding_data: Dict[str, Any]
) -> str:
"""Build prompt for category data generation."""
return f"""
You are ALwrity's Strategy Assistant. Generate data for the {category} category based on the user's description.
User Description: {user_description}
Current Form Data: {current_form_data}
Onboarding Data: {onboarding_data}
Category Fields: {self._get_category_fields(category)}
Generate appropriate values for all fields in the {category} category. Return only valid JSON with field IDs as keys.
Example response format:
{{
"field_id": "value",
"another_field": "value"
}}
"""
def _build_validation_prompt(self, field_definition: Dict[str, Any], value: Any) -> str:
"""Build prompt for field validation."""
return f"""
Validate the following field value:
Field: {field_definition['label']}
Description: {field_definition['description']}
Required: {field_definition['required']}
Type: {field_definition['type']}
Value: {value}
Return JSON with: {{"isValid": boolean, "suggestion": string, "confidence": number}}
Example response:
{{
"isValid": true,
"suggestion": "This looks good!",
"confidence": 0.95
}}
"""
def _build_analysis_prompt(
self,
form_data: Dict[str, Any],
onboarding_data: Dict[str, Any]
) -> str:
"""Build prompt for strategy analysis."""
return f"""
Analyze the following content strategy for completeness, coherence, and alignment:
Form Data: {form_data}
Onboarding Data: {onboarding_data}
Return JSON with: {{
"completeness": number,
"coherence": number,
"alignment": number,
"suggestions": [string],
"missingFields": [string],
"improvements": [string]
}}
Example response:
{{
"completeness": 85,
"coherence": 90,
"alignment": 88,
"suggestions": ["Consider adding more specific metrics"],
"missingFields": ["content_budget"],
"improvements": ["Add timeline details"]
}}
"""
def _build_suggestions_prompt(
self,
field_definition: Dict[str, Any],
current_form_data: Dict[str, Any],
onboarding_data: Dict[str, Any]
) -> str:
"""Build prompt for field suggestions."""
return f"""
Generate suggestions for the following field:
Field: {field_definition['label']}
Description: {field_definition['description']}
Required: {field_definition['required']}
Type: {field_definition['type']}
Current Form Data: {current_form_data}
Onboarding Data: {onboarding_data}
Return JSON with: {{
"suggestions": [string],
"reasoning": string,
"confidence": number
}}
Example response:
{{
"suggestions": ["Focus on measurable outcomes", "Align with business goals"],
"reasoning": "Based on your business context, measurable outcomes will be most effective",
"confidence": 0.92
}}
"""
def _get_field_definition(self, field_id: str) -> Dict[str, Any]:
"""Get field definition from STRATEGIC_INPUT_FIELDS."""
# This would be imported from the frontend field definitions
# For now, return a basic structure
return {
"id": field_id,
"label": field_id.replace("_", " ").title(),
"description": f"Description for {field_id}",
"required": True,
"type": "text"
}
def _get_category_fields(self, category: str) -> List[str]:
"""Get fields for a specific category."""
# This would be imported from the frontend field definitions
category_fields = {
"business_context": [
"business_objectives", "target_metrics", "content_budget", "team_size",
"implementation_timeline", "market_share", "competitive_position", "performance_metrics"
],
"audience_intelligence": [
"content_preferences", "consumption_patterns", "audience_pain_points",
"buying_journey", "seasonal_trends", "engagement_metrics"
],
"competitive_intelligence": [
"top_competitors", "competitor_content_strategies", "market_gaps",
"industry_trends", "emerging_trends"
],
"content_strategy": [
"preferred_formats", "content_mix", "content_frequency", "optimal_timing",
"quality_metrics", "editorial_guidelines", "brand_voice"
],
"performance_analytics": [
"traffic_sources", "conversion_rates", "content_roi_targets", "ab_testing_capabilities"
]
}
return category_fields.get(category, [])
def _parse_category_response(self, response: str, category: str) -> Dict[str, Any]:
"""Parse LLM response for category data."""
try:
import json
# Clean up the response to extract JSON
response = response.strip()
if response.startswith("```json"):
response = response[7:]
if response.endswith("```"):
response = response[:-3]
response = response.strip()
parsed_data = json.loads(response)
# Validate that we have actual data
if not isinstance(parsed_data, dict) or len(parsed_data) == 0:
raise Exception("Invalid or empty response data")
return parsed_data
except Exception as e:
logger.error(f"Error parsing category response: {str(e)}")
raise Exception(f"Failed to parse category response: {str(e)}")
def _parse_validation_response(self, response: str) -> Dict[str, Any]:
"""Parse LLM response for validation."""
try:
import json
# Clean up the response to extract JSON
response = response.strip()
if response.startswith("```json"):
response = response[7:]
if response.endswith("```"):
response = response[:-3]
response = response.strip()
parsed_data = json.loads(response)
# Validate required fields
if not isinstance(parsed_data, dict) or 'isValid' not in parsed_data:
raise Exception("Invalid validation response format")
return parsed_data
except Exception as e:
logger.error(f"Error parsing validation response: {str(e)}")
raise Exception(f"Failed to parse validation response: {str(e)}")
def _parse_analysis_response(self, response: str) -> Dict[str, Any]:
"""Parse LLM response for analysis."""
try:
import json
# Clean up the response to extract JSON
response = response.strip()
if response.startswith("```json"):
response = response[7:]
if response.endswith("```"):
response = response[:-3]
response = response.strip()
parsed_data = json.loads(response)
# Validate required fields
required_fields = ['completeness', 'coherence', 'alignment']
if not isinstance(parsed_data, dict) or not all(field in parsed_data for field in required_fields):
raise Exception("Invalid analysis response format")
return parsed_data
except Exception as e:
logger.error(f"Error parsing analysis response: {str(e)}")
raise Exception(f"Failed to parse analysis response: {str(e)}")
def _parse_suggestions_response(self, response: str) -> Dict[str, Any]:
"""Parse LLM response for suggestions."""
try:
import json
# Clean up the response to extract JSON
response = response.strip()
if response.startswith("```json"):
response = response[7:]
if response.endswith("```"):
response = response[:-3]
response = response.strip()
parsed_data = json.loads(response)
# Validate required fields
if not isinstance(parsed_data, dict) or 'suggestions' not in parsed_data:
raise Exception("Invalid suggestions response format")
return parsed_data
except Exception as e:
logger.error(f"Error parsing suggestions response: {str(e)}")
raise Exception(f"Failed to parse suggestions response: {str(e)}")