Merge branch 'main' into cursor/implement-usage-based-subscription-and-monitoring-0179

This commit is contained in:
ي
2025-09-10 13:56:54 +05:30
committed by GitHub
131 changed files with 24239 additions and 3483 deletions

View File

@@ -0,0 +1,702 @@
"""
Hallucination Detector Service
This service implements fact-checking functionality using Exa.ai API
to detect and verify claims in AI-generated content, similar to the
Exa.ai demo implementation.
"""
import json
import logging
from typing import List, Dict, Any, Optional
from dataclasses import dataclass
from datetime import datetime
import requests
import os
import asyncio
import concurrent.futures
try:
from google import genai
GOOGLE_GENAI_AVAILABLE = True
except Exception:
GOOGLE_GENAI_AVAILABLE = False
logger = logging.getLogger(__name__)
@dataclass
class Claim:
"""Represents a single verifiable claim extracted from text."""
text: str
confidence: float
assessment: str # "supported", "refuted", "insufficient_information"
supporting_sources: List[Dict[str, Any]]
refuting_sources: List[Dict[str, Any]]
reasoning: str = ""
@dataclass
class HallucinationResult:
"""Result of hallucination detection analysis."""
claims: List[Claim]
overall_confidence: float
total_claims: int
supported_claims: int
refuted_claims: int
insufficient_claims: int
timestamp: str
class HallucinationDetector:
"""
Hallucination detector using Exa.ai for fact-checking.
Implements the three-step process from Exa.ai demo:
1. Extract verifiable claims from text
2. Search for evidence using Exa.ai
3. Verify claims against sources
"""
def __init__(self):
self.exa_api_key = os.getenv('EXA_API_KEY')
self.gemini_api_key = os.getenv('GEMINI_API_KEY')
if not self.exa_api_key:
logger.warning("EXA_API_KEY not found. Hallucination detection will be limited.")
if not self.gemini_api_key:
logger.warning("GEMINI_API_KEY not found. Falling back to heuristic claim extraction.")
# Initialize Gemini client for claim extraction and assessment
self.gemini_client = genai.Client(api_key=self.gemini_api_key) if (GOOGLE_GENAI_AVAILABLE and self.gemini_api_key) else None
# Rate limiting to prevent API abuse
self.daily_api_calls = 0
self.daily_limit = 20 # Max 20 API calls per day for fact checking
self.last_reset_date = None
def _check_rate_limit(self) -> bool:
"""Check if we're within daily API usage limits."""
from datetime import date
today = date.today()
# Reset counter if it's a new day
if self.last_reset_date != today:
self.daily_api_calls = 0
self.last_reset_date = today
# Check if we've exceeded the limit
if self.daily_api_calls >= self.daily_limit:
logger.warning(f"Daily API limit reached ({self.daily_limit} calls). Fact checking disabled for today.")
return False
# Increment counter for this API call
self.daily_api_calls += 1
logger.info(f"Fact check API call #{self.daily_api_calls}/{self.daily_limit} today")
return True
async def detect_hallucinations(self, text: str) -> HallucinationResult:
"""
Main method to detect hallucinations in the given text.
Args:
text: The text to analyze for factual accuracy
Returns:
HallucinationResult with claims analysis and confidence scores
"""
try:
logger.info(f"Starting hallucination detection for text of length: {len(text)}")
logger.info(f"Text sample: {text[:200]}...")
# Check rate limits first
if not self._check_rate_limit():
return HallucinationResult(
claims=[],
overall_confidence=0.0,
total_claims=0,
supported_claims=0,
refuted_claims=0,
insufficient_claims=0,
timestamp=datetime.now().isoformat()
)
# Validate required API keys
if not self.gemini_api_key:
raise Exception("GEMINI_API_KEY not configured. Cannot perform hallucination detection.")
if not self.exa_api_key:
raise Exception("EXA_API_KEY not configured. Cannot search for evidence.")
# Step 1: Extract claims from text
claims_texts = await self._extract_claims(text)
logger.info(f"Extracted {len(claims_texts)} claims from text: {claims_texts}")
if not claims_texts:
logger.warning("No verifiable claims found in text")
return HallucinationResult(
claims=[],
overall_confidence=0.0,
total_claims=0,
supported_claims=0,
refuted_claims=0,
insufficient_claims=0,
timestamp=datetime.now().isoformat()
)
# Step 2 & 3: Verify claims in batch to reduce API calls
verified_claims = await self._verify_claims_batch(claims_texts)
# Calculate overall metrics
total_claims = len(verified_claims)
supported_claims = sum(1 for c in verified_claims if c.assessment == "supported")
refuted_claims = sum(1 for c in verified_claims if c.assessment == "refuted")
insufficient_claims = sum(1 for c in verified_claims if c.assessment == "insufficient_information")
# Calculate overall confidence (weighted average)
if total_claims > 0:
overall_confidence = sum(c.confidence for c in verified_claims) / total_claims
else:
overall_confidence = 0.0
result = HallucinationResult(
claims=verified_claims,
overall_confidence=overall_confidence,
total_claims=total_claims,
supported_claims=supported_claims,
refuted_claims=refuted_claims,
insufficient_claims=insufficient_claims,
timestamp=datetime.now().isoformat()
)
logger.info(f"Hallucination detection completed. Overall confidence: {overall_confidence:.2f}")
return result
except Exception as e:
logger.error(f"Error in hallucination detection: {str(e)}")
raise Exception(f"Hallucination detection failed: {str(e)}")
async def _extract_claims(self, text: str) -> List[str]:
"""
Extract verifiable claims from text using LLM.
Args:
text: Input text to extract claims from
Returns:
List of claim strings
"""
if not self.gemini_client:
raise Exception("Gemini client not available. Cannot extract claims without AI provider.")
try:
prompt = (
"Extract verifiable factual claims from the following text. "
"A verifiable claim is a statement that can be checked against external sources for accuracy.\n\n"
"Return ONLY a valid JSON array of strings, where each string is a single verifiable claim.\n\n"
"Examples of GOOD verifiable claims:\n"
"- \"The company was founded in 2020\"\n"
"- \"Sales increased by 25% last quarter\"\n"
"- \"The product has 10,000 users\"\n"
"- \"The market size is $50 billion\"\n"
"- \"The software supports 15 languages\"\n"
"- \"The company has offices in 5 countries\"\n\n"
"Examples of BAD claims (opinions, subjective statements):\n"
"- \"This is the best product\"\n"
"- \"Customers love our service\"\n"
"- \"We are innovative\"\n"
"- \"The future looks bright\"\n\n"
"IMPORTANT: Extract at least 2-3 verifiable claims if possible. "
"Look for specific facts, numbers, dates, locations, and measurable statements.\n\n"
f"Text to analyze: {text}\n\n"
"Return only the JSON array of verifiable claims:"
)
loop = asyncio.get_event_loop()
with concurrent.futures.ThreadPoolExecutor() as executor:
resp = await loop.run_in_executor(executor, lambda: self.gemini_client.models.generate_content(
model="gemini-1.5-flash",
contents=prompt
))
if not resp or not resp.text:
raise Exception("Empty response from Gemini API")
claims_text = resp.text.strip()
logger.info(f"Raw Gemini response for claims: {claims_text[:200]}...")
# Try to extract JSON from the response
try:
claims = json.loads(claims_text)
except json.JSONDecodeError:
# Try to find JSON array in the response (handle markdown code blocks)
import re
# First try to extract from markdown code blocks
code_block_match = re.search(r'```(?:json)?\s*(\[.*?\])\s*```', claims_text, re.DOTALL)
if code_block_match:
claims = json.loads(code_block_match.group(1))
else:
# Try to find JSON array directly
json_match = re.search(r'\[.*?\]', claims_text, re.DOTALL)
if json_match:
claims = json.loads(json_match.group())
else:
raise Exception(f"Could not parse JSON from Gemini response: {claims_text[:100]}")
if isinstance(claims, list):
valid_claims = [claim for claim in claims if isinstance(claim, str) and claim.strip()]
logger.info(f"Successfully extracted {len(valid_claims)} claims")
return valid_claims
else:
raise Exception(f"Expected JSON array, got: {type(claims)}")
except Exception as e:
logger.error(f"Error extracting claims: {str(e)}")
raise Exception(f"Failed to extract claims: {str(e)}")
async def _verify_claims_batch(self, claims: List[str]) -> List[Claim]:
"""
Verify multiple claims in batch to reduce API calls.
Args:
claims: List of claims to verify
Returns:
List of Claim objects with verification results
"""
try:
logger.info(f"Starting batch verification of {len(claims)} claims")
# Limit to maximum 3 claims to prevent excessive API usage
max_claims = min(len(claims), 3)
claims_to_verify = claims[:max_claims]
if len(claims) > max_claims:
logger.warning(f"Limited verification to {max_claims} claims to prevent API rate limits")
# Step 1: Search for evidence for all claims in one batch
all_sources = await self._search_evidence_batch(claims_to_verify)
# Step 2: Assess all claims against sources in one API call
verified_claims = await self._assess_claims_batch(claims_to_verify, all_sources)
# Add any remaining claims as insufficient information
for i in range(max_claims, len(claims)):
verified_claims.append(Claim(
text=claims[i],
confidence=0.0,
assessment="insufficient_information",
supporting_sources=[],
refuting_sources=[],
reasoning="Not verified due to API rate limit protection"
))
logger.info(f"Batch verification completed for {len(verified_claims)} claims")
return verified_claims
except Exception as e:
logger.error(f"Error in batch verification: {str(e)}")
# Return all claims as insufficient information
return [
Claim(
text=claim,
confidence=0.0,
assessment="insufficient_information",
supporting_sources=[],
refuting_sources=[],
reasoning=f"Batch verification failed: {str(e)}"
)
for claim in claims
]
async def _verify_claim(self, claim: str) -> Claim:
"""
Verify a single claim using Exa.ai search.
Args:
claim: The claim to verify
Returns:
Claim object with verification results
"""
try:
# Search for evidence using Exa.ai
sources = await self._search_evidence(claim)
if not sources:
return Claim(
text=claim,
confidence=0.5,
assessment="insufficient_information",
supporting_sources=[],
refuting_sources=[],
reasoning="No sources found for verification"
)
# Verify claim against sources using LLM
verification_result = await self._assess_claim_against_sources(claim, sources)
return Claim(
text=claim,
confidence=verification_result.get('confidence', 0.5),
assessment=verification_result.get('assessment', 'insufficient_information'),
supporting_sources=verification_result.get('supporting_sources', []),
refuting_sources=verification_result.get('refuting_sources', []),
reasoning=verification_result.get('reasoning', '')
)
except Exception as e:
logger.error(f"Error verifying claim '{claim}': {str(e)}")
return Claim(
text=claim,
confidence=0.5,
assessment="insufficient_information",
supporting_sources=[],
refuting_sources=[],
reasoning=f"Error during verification: {str(e)}"
)
async def _search_evidence_batch(self, claims: List[str]) -> List[Dict[str, Any]]:
"""
Search for evidence for multiple claims in one API call.
Args:
claims: List of claims to search for
Returns:
List of sources relevant to the claims
"""
try:
# Combine all claims into one search query
combined_query = " ".join(claims[:2]) # Use first 2 claims to avoid query length limits
logger.info(f"Searching for evidence for {len(claims)} claims with combined query")
# Use the existing search method with combined query
sources = await self._search_evidence(combined_query)
# Limit sources to prevent excessive processing
max_sources = 5
if len(sources) > max_sources:
sources = sources[:max_sources]
logger.info(f"Limited sources to {max_sources} to prevent API rate limits")
return sources
except Exception as e:
logger.error(f"Error in batch evidence search: {str(e)}")
return []
async def _assess_claims_batch(self, claims: List[str], sources: List[Dict[str, Any]]) -> List[Claim]:
"""
Assess multiple claims against sources in one API call.
Args:
claims: List of claims to assess
sources: List of sources to assess against
Returns:
List of Claim objects with assessment results
"""
if not self.gemini_client:
raise Exception("Gemini client not available. Cannot assess claims without AI provider.")
try:
# Limit to 3 claims to prevent excessive API usage
claims_to_assess = claims[:3]
# Prepare sources text
combined_sources = "\n\n".join([
f"Source {i+1}: {src.get('url','')}\nText: {src.get('text','')[:1000]}"
for i, src in enumerate(sources)
])
# Prepare claims text
claims_text = "\n".join([
f"Claim {i+1}: {claim}"
for i, claim in enumerate(claims_to_assess)
])
prompt = (
"You are a strict fact-checker. Analyze each claim against the provided sources.\n\n"
"Return ONLY a valid JSON object with this exact structure:\n"
"{\n"
' "assessments": [\n'
' {\n'
' "claim_index": 0,\n'
' "assessment": "supported" or "refuted" or "insufficient_information",\n'
' "confidence": number between 0.0 and 1.0,\n'
' "supporting_sources": [array of source indices that support the claim],\n'
' "refuting_sources": [array of source indices that refute the claim],\n'
' "reasoning": "brief explanation of your assessment"\n'
' }\n'
' ]\n'
"}\n\n"
f"Claims to verify:\n{claims_text}\n\n"
f"Sources:\n{combined_sources}\n\n"
"Return only the JSON object:"
)
loop = asyncio.get_event_loop()
with concurrent.futures.ThreadPoolExecutor() as executor:
resp = await loop.run_in_executor(executor, lambda: self.gemini_client.models.generate_content(
model="gemini-1.5-flash",
contents=prompt
))
if not resp or not resp.text:
raise Exception("Empty response from Gemini API for batch assessment")
result_text = resp.text.strip()
logger.info(f"Raw Gemini response for batch assessment: {result_text[:200]}...")
# Try to extract JSON from the response
try:
result = json.loads(result_text)
except json.JSONDecodeError:
# Try to find JSON object in the response (handle markdown code blocks)
import re
code_block_match = re.search(r'```(?:json)?\s*(\{.*?\})\s*```', result_text, re.DOTALL)
if code_block_match:
result = json.loads(code_block_match.group(1))
else:
json_match = re.search(r'\{.*?\}', result_text, re.DOTALL)
if json_match:
result = json.loads(json_match.group())
else:
raise Exception(f"Could not parse JSON from Gemini response: {result_text[:100]}")
# Process assessments
assessments = result.get('assessments', [])
verified_claims = []
for i, claim in enumerate(claims_to_assess):
# Find assessment for this claim
assessment = None
for a in assessments:
if a.get('claim_index') == i:
assessment = a
break
if assessment:
# Process supporting and refuting sources
supporting_sources = []
refuting_sources = []
if isinstance(assessment.get('supporting_sources'), list):
for idx in assessment['supporting_sources']:
if isinstance(idx, int) and 0 <= idx < len(sources):
supporting_sources.append(sources[idx])
if isinstance(assessment.get('refuting_sources'), list):
for idx in assessment['refuting_sources']:
if isinstance(idx, int) and 0 <= idx < len(sources):
refuting_sources.append(sources[idx])
verified_claims.append(Claim(
text=claim,
confidence=float(assessment.get('confidence', 0.5)),
assessment=assessment.get('assessment', 'insufficient_information'),
supporting_sources=supporting_sources,
refuting_sources=refuting_sources,
reasoning=assessment.get('reasoning', '')
))
else:
# No assessment found for this claim
verified_claims.append(Claim(
text=claim,
confidence=0.0,
assessment="insufficient_information",
supporting_sources=[],
refuting_sources=[],
reasoning="No assessment provided"
))
logger.info(f"Successfully assessed {len(verified_claims)} claims in batch")
return verified_claims
except Exception as e:
logger.error(f"Error in batch assessment: {str(e)}")
# Return all claims as insufficient information
return [
Claim(
text=claim,
confidence=0.0,
assessment="insufficient_information",
supporting_sources=[],
refuting_sources=[],
reasoning=f"Batch assessment failed: {str(e)}"
)
for claim in claims_to_assess
]
async def _search_evidence(self, claim: str) -> List[Dict[str, Any]]:
"""
Search for evidence using Exa.ai API.
Args:
claim: The claim to search evidence for
Returns:
List of source documents with evidence
"""
if not self.exa_api_key:
raise Exception("Exa API key not available. Cannot search for evidence without Exa.ai access.")
try:
headers = {
'x-api-key': self.exa_api_key,
'Content-Type': 'application/json'
}
payload = {
'query': claim,
'numResults': 5,
'text': True,
'useAutoprompt': True
}
response = requests.post(
'https://api.exa.ai/search',
headers=headers,
json=payload,
timeout=15
)
if response.status_code == 200:
data = response.json()
results = data.get('results', [])
if not results:
raise Exception(f"No search results found for claim: {claim}")
sources = []
for result in results:
source = {
'title': result.get('title', 'Untitled'),
'url': result.get('url', ''),
'text': result.get('text', ''),
'publishedDate': result.get('publishedDate', ''),
'author': result.get('author', ''),
'score': result.get('score', 0.5)
}
sources.append(source)
logger.info(f"Found {len(sources)} sources for claim: {claim[:50]}...")
return sources
else:
raise Exception(f"Exa API error: {response.status_code} - {response.text}")
except Exception as e:
logger.error(f"Error searching evidence with Exa: {str(e)}")
raise Exception(f"Failed to search evidence: {str(e)}")
async def _assess_claim_against_sources(self, claim: str, sources: List[Dict[str, Any]]) -> Dict[str, Any]:
"""
Assess whether sources support or refute the claim using LLM.
Args:
claim: The claim to assess
sources: List of source documents
Returns:
Dictionary with assessment results
"""
if not self.gemini_client:
raise Exception("Gemini client not available. Cannot assess claims without AI provider.")
try:
combined_sources = "\n\n".join([
f"Source {i+1}: {src.get('url','')}\nText: {src.get('text','')[:2000]}"
for i, src in enumerate(sources)
])
prompt = (
"You are a strict fact-checker. Analyze the claim against the provided sources.\n\n"
"Return ONLY a valid JSON object with this exact structure:\n"
"{\n"
' "assessment": "supported" or "refuted" or "insufficient_information",\n'
' "confidence": number between 0.0 and 1.0,\n'
' "supporting_sources": [array of source indices that support the claim],\n'
' "refuting_sources": [array of source indices that refute the claim],\n'
' "reasoning": "brief explanation of your assessment"\n'
"}\n\n"
f"Claim to verify: {claim}\n\n"
f"Sources:\n{combined_sources}\n\n"
"Return only the JSON object:"
)
loop = asyncio.get_event_loop()
with concurrent.futures.ThreadPoolExecutor() as executor:
resp = await loop.run_in_executor(executor, lambda: self.gemini_client.models.generate_content(
model="gemini-1.5-flash",
contents=prompt
))
if not resp or not resp.text:
raise Exception("Empty response from Gemini API for claim assessment")
result_text = resp.text.strip()
logger.info(f"Raw Gemini response for assessment: {result_text[:200]}...")
# Try to extract JSON from the response
try:
result = json.loads(result_text)
except json.JSONDecodeError:
# Try to find JSON object in the response (handle markdown code blocks)
import re
# First try to extract from markdown code blocks
code_block_match = re.search(r'```(?:json)?\s*(\{.*?\})\s*```', result_text, re.DOTALL)
if code_block_match:
result = json.loads(code_block_match.group(1))
else:
# Try to find JSON object directly
json_match = re.search(r'\{.*?\}', result_text, re.DOTALL)
if json_match:
result = json.loads(json_match.group())
else:
raise Exception(f"Could not parse JSON from Gemini response: {result_text[:100]}")
# Validate required fields
required_fields = ['assessment', 'confidence', 'supporting_sources', 'refuting_sources', 'reasoning']
for field in required_fields:
if field not in result:
raise Exception(f"Missing required field '{field}' in assessment response")
# Process supporting and refuting sources
supporting_sources = []
refuting_sources = []
if isinstance(result.get('supporting_sources'), list):
for idx in result['supporting_sources']:
if isinstance(idx, int) and 0 <= idx < len(sources):
supporting_sources.append(sources[idx])
if isinstance(result.get('refuting_sources'), list):
for idx in result['refuting_sources']:
if isinstance(idx, int) and 0 <= idx < len(sources):
refuting_sources.append(sources[idx])
# Validate assessment value
valid_assessments = ['supported', 'refuted', 'insufficient_information']
if result['assessment'] not in valid_assessments:
raise Exception(f"Invalid assessment value: {result['assessment']}")
# Validate confidence value
confidence = float(result['confidence'])
if not (0.0 <= confidence <= 1.0):
raise Exception(f"Invalid confidence value: {confidence}")
logger.info(f"Successfully assessed claim: {result['assessment']} (confidence: {confidence})")
return {
'assessment': result['assessment'],
'confidence': confidence,
'supporting_sources': supporting_sources,
'refuting_sources': refuting_sources,
'reasoning': result['reasoning']
}
except Exception as e:
logger.error(f"Error assessing claim against sources: {str(e)}")
raise Exception(f"Failed to assess claim: {str(e)}")

View File

@@ -355,7 +355,38 @@ class ContentGenerator:
except Exception as e:
logger.error(f"Error generating grounded post content: {str(e)}")
raise Exception(f"Failed to generate grounded post content: {str(e)}")
logger.info("Attempting fallback to standard content generation...")
# Fallback to standard content generation without grounding
try:
if not self.fallback_provider:
raise Exception("No fallback provider available")
# Build a simpler prompt for fallback generation
prompt = PostPromptBuilder.build_post_prompt(request)
# Generate content using fallback provider (it's a dict with functions)
if 'generate_text' in self.fallback_provider:
result = await self.fallback_provider['generate_text'](
prompt=prompt,
temperature=0.7,
max_tokens=request.max_length
)
else:
raise Exception("Fallback provider doesn't have generate_text method")
# Return result in the expected format
return {
'content': result.get('content', '') if isinstance(result, dict) else str(result),
'sources': [],
'citations': [],
'grounding_enabled': False,
'fallback_used': True
}
except Exception as fallback_error:
logger.error(f"Fallback generation also failed: {str(fallback_error)}")
raise Exception(f"Failed to generate content: {str(e)}. Fallback also failed: {str(fallback_error)}")
async def generate_grounded_article_content(self, request, research_sources: List) -> Dict[str, Any]:
"""Generate grounded article content using the enhanced Gemini provider with native grounding."""

View File

@@ -41,8 +41,9 @@ class GeminiGroundedProvider:
if not self.api_key:
raise ValueError("GEMINI_API_KEY environment variable is required")
# Initialize the Gemini client
# Initialize the Gemini client with timeout configuration
self.client = genai.Client(api_key=self.api_key)
self.timeout = 30 # 30 second timeout for API calls
logger.info("✅ Gemini Grounded Provider initialized with native Google Search grounding")
async def generate_grounded_content(
@@ -82,12 +83,27 @@ class GeminiGroundedProvider:
temperature=temperature
)
# Make the request with native grounding
response = self.client.models.generate_content(
model="gemini-2.5-flash",
contents=grounded_prompt,
config=config,
)
# Make the request with native grounding and timeout
import asyncio
import concurrent.futures
try:
# Run the synchronous generate_content in a thread pool to make it awaitable
loop = asyncio.get_event_loop()
with concurrent.futures.ThreadPoolExecutor() as executor:
response = await asyncio.wait_for(
loop.run_in_executor(
executor,
lambda: self.client.models.generate_content(
model="gemini-2.5-flash",
contents=grounded_prompt,
config=config,
)
),
timeout=self.timeout
)
except asyncio.TimeoutError:
raise Exception(f"Gemini API request timed out after {self.timeout} seconds")
# Process the grounded response
result = self._process_grounded_response(response, content_type)

View File

@@ -0,0 +1,106 @@
# Persona Services Package
This package contains platform-specific persona generation and analysis services, providing a modular and extensible architecture for creating platform-optimized writing personas.
## Structure
```
services/persona/
├── __init__.py # Package initialization
├── linkedin/ # LinkedIn-specific persona services
│ ├── __init__.py # LinkedIn package initialization
│ ├── linkedin_persona_service.py # Main LinkedIn persona service
│ ├── linkedin_persona_prompts.py # LinkedIn-specific prompts
│ └── linkedin_persona_schemas.py # LinkedIn-specific schemas
└── README.md # This documentation
```
## LinkedIn Persona Services
### LinkedInPersonaService
The main service class for generating LinkedIn-specific persona adaptations.
**Key Features:**
- Enhanced LinkedIn-specific prompt generation
- Professional networking optimization
- Industry-specific adaptations
- Algorithm optimization for LinkedIn
- Persona validation and quality scoring
**Methods:**
- `generate_linkedin_persona()` - Generate LinkedIn-optimized persona
- `validate_linkedin_persona()` - Validate persona data quality
- `optimize_for_linkedin_algorithm()` - Algorithm-specific optimizations
- `get_linkedin_constraints()` - Get LinkedIn platform constraints
### LinkedInPersonaPrompts
Handles LinkedIn-specific prompt generation with professional optimization.
**Key Features:**
- Industry-specific targeting (technology, business, etc.)
- Professional networking focus
- Thought leadership positioning
- B2B optimization
- LinkedIn algorithm awareness
### LinkedInPersonaSchemas
Defines LinkedIn-specific JSON schemas for persona generation.
**Key Features:**
- Enhanced LinkedIn schema with professional fields
- Algorithm optimization fields
- Professional networking elements
- LinkedIn feature-specific adaptations
## Usage
```python
from services.persona.linkedin.linkedin_persona_service import LinkedInPersonaService
# Initialize the service
linkedin_service = LinkedInPersonaService()
# Generate LinkedIn persona
linkedin_persona = linkedin_service.generate_linkedin_persona(
core_persona=core_persona_data,
onboarding_data=onboarding_data
)
# Validate persona quality
validation_results = linkedin_service.validate_linkedin_persona(linkedin_persona)
# Optimize for LinkedIn algorithm
optimized_persona = linkedin_service.optimize_for_linkedin_algorithm(linkedin_persona)
```
## Integration with Main Persona Service
The main `PersonaAnalysisService` automatically uses the LinkedIn service when generating LinkedIn personas:
```python
# In PersonaAnalysisService._generate_single_platform_persona()
if platform.lower() == "linkedin":
return self.linkedin_service.generate_linkedin_persona(core_persona, onboarding_data)
```
## Benefits of This Architecture
1. **Modularity**: Each platform has its own dedicated service
2. **Extensibility**: Easy to add new platforms (Facebook, Instagram, etc.)
3. **Maintainability**: Platform-specific logic is isolated
4. **Testability**: Each service can be tested independently
5. **Reusability**: Services can be used across different parts of the application
## Future Extensions
This architecture makes it easy to add new platform-specific services:
- `services/persona/facebook/` - Facebook-specific persona services
- `services/persona/instagram/` - Instagram-specific persona services
- `services/persona/twitter/` - Twitter-specific persona services
- `services/persona/blog/` - Blog-specific persona services
Each platform service would follow the same pattern:
- `{platform}_persona_service.py` - Main service class
- `{platform}_persona_prompts.py` - Platform-specific prompts
- `{platform}_persona_schemas.py` - Platform-specific schemas

View File

@@ -0,0 +1,8 @@
"""
Persona Services Package
Contains platform-specific persona generation and analysis services.
"""
from .linkedin.linkedin_persona_service import LinkedInPersonaService
__all__ = ['LinkedInPersonaService']

View File

@@ -0,0 +1,16 @@
"""
Core Persona Generation Module
This module contains the core persona generation logic extracted from persona_analysis_service.py
to improve maintainability and modularity.
"""
from .core_persona_service import CorePersonaService
from .data_collector import OnboardingDataCollector
from .prompt_builder import PersonaPromptBuilder
__all__ = [
'CorePersonaService',
'OnboardingDataCollector',
'PersonaPromptBuilder'
]

View File

@@ -0,0 +1,159 @@
"""
Core Persona Service
Handles the core persona generation logic using Gemini AI.
"""
from typing import Dict, Any, List
from loguru import logger
from datetime import datetime
from services.llm_providers.gemini_provider import gemini_structured_json_response
from .data_collector import OnboardingDataCollector
from .prompt_builder import PersonaPromptBuilder
from services.persona.linkedin.linkedin_persona_service import LinkedInPersonaService
class CorePersonaService:
"""Core service for generating writing personas using Gemini AI."""
def __init__(self):
"""Initialize the core persona service."""
self.data_collector = OnboardingDataCollector()
self.prompt_builder = PersonaPromptBuilder()
self.linkedin_service = LinkedInPersonaService()
logger.info("CorePersonaService initialized")
def generate_core_persona(self, onboarding_data: Dict[str, Any]) -> Dict[str, Any]:
"""Generate core writing persona using Gemini structured response."""
# Build analysis prompt
prompt = self.prompt_builder.build_persona_analysis_prompt(onboarding_data)
# Get schema for structured response
persona_schema = self.prompt_builder.get_persona_schema()
try:
# Generate structured response using Gemini
response = gemini_structured_json_response(
prompt=prompt,
schema=persona_schema,
temperature=0.2, # Low temperature for consistent analysis
max_tokens=8192,
system_prompt="You are an expert writing style analyst and persona developer. Analyze the provided data to create a precise, actionable writing persona."
)
if "error" in response:
logger.error(f"Gemini API error: {response['error']}")
return {"error": f"AI analysis failed: {response['error']}"}
logger.info("✅ Core persona generated successfully")
return response
except Exception as e:
logger.error(f"Error generating core persona: {str(e)}")
return {"error": f"Failed to generate core persona: {str(e)}"}
def generate_platform_adaptations(self, core_persona: Dict[str, Any], onboarding_data: Dict[str, Any]) -> Dict[str, Any]:
"""Generate platform-specific persona adaptations."""
platforms = ["twitter", "linkedin", "instagram", "facebook", "blog", "medium", "substack"]
platform_personas = {}
for platform in platforms:
try:
platform_persona = self._generate_single_platform_persona(core_persona, platform, onboarding_data)
if "error" not in platform_persona:
platform_personas[platform] = platform_persona
else:
logger.warning(f"Failed to generate {platform} persona: {platform_persona['error']}")
except Exception as e:
logger.error(f"Error generating {platform} persona: {str(e)}")
return platform_personas
def _generate_single_platform_persona(self, core_persona: Dict[str, Any], platform: str, onboarding_data: Dict[str, Any]) -> Dict[str, Any]:
"""Generate persona adaptation for a specific platform."""
# Use LinkedIn service for LinkedIn platform
if platform.lower() == "linkedin":
return self.linkedin_service.generate_linkedin_persona(core_persona, onboarding_data)
# Use generic platform adaptation for other platforms
platform_constraints = self._get_platform_constraints(platform)
prompt = self.prompt_builder.build_platform_adaptation_prompt(core_persona, platform, onboarding_data, platform_constraints)
# Get platform-specific schema
platform_schema = self.prompt_builder.get_platform_schema()
try:
response = gemini_structured_json_response(
prompt=prompt,
schema=platform_schema,
temperature=0.2,
max_tokens=4096,
system_prompt=f"You are an expert in {platform} content strategy and platform-specific writing optimization."
)
return response
except Exception as e:
logger.error(f"Error generating {platform} persona: {str(e)}")
return {"error": f"Failed to generate {platform} persona: {str(e)}"}
def _get_platform_constraints(self, platform: str) -> Dict[str, Any]:
"""Get platform-specific constraints and best practices."""
constraints = {
"twitter": {
"character_limit": 280,
"optimal_length": "120-150 characters",
"hashtag_limit": 3,
"image_support": True,
"thread_support": True,
"link_shortening": True
},
"linkedin": self.linkedin_service.get_linkedin_constraints(),
"instagram": {
"caption_limit": 2200,
"optimal_length": "125-150 words",
"hashtag_limit": 30,
"visual_first": True,
"story_support": True,
"emoji_friendly": True
},
"facebook": {
"character_limit": 63206,
"optimal_length": "40-80 words",
"algorithm_favors": "engagement",
"link_preview": True,
"event_support": True,
"group_sharing": True
},
"blog": {
"word_count": "800-2000 words",
"seo_important": True,
"header_structure": True,
"internal_linking": True,
"meta_descriptions": True,
"readability_score": True
},
"medium": {
"word_count": "1000-3000 words",
"storytelling_focus": True,
"subtitle_support": True,
"publication_support": True,
"clap_optimization": True,
"follower_building": True
},
"substack": {
"newsletter_format": True,
"email_optimization": True,
"subscription_focus": True,
"long_form": True,
"personal_connection": True,
"monetization_support": True
}
}
return constraints.get(platform, {})

View File

@@ -0,0 +1,306 @@
"""
Onboarding Data Collector
Handles comprehensive collection of onboarding data for persona generation.
"""
from typing import Dict, Any, List, Optional
from sqlalchemy.orm import Session
from loguru import logger
from services.database import get_db_session
from models.onboarding import OnboardingSession, WebsiteAnalysis, ResearchPreferences, APIKey
class OnboardingDataCollector:
"""Collects comprehensive onboarding data for persona analysis."""
def collect_onboarding_data(self, user_id: int, session_id: int = None) -> Optional[Dict[str, Any]]:
"""Collect comprehensive onboarding data for persona analysis."""
try:
session = get_db_session()
# Find onboarding session
if session_id:
onboarding_session = session.query(OnboardingSession).filter(
OnboardingSession.id == session_id,
OnboardingSession.user_id == user_id
).first()
else:
onboarding_session = session.query(OnboardingSession).filter(
OnboardingSession.user_id == user_id
).order_by(OnboardingSession.updated_at.desc()).first()
if not onboarding_session:
return None
# Get ALL website analyses (there might be multiple)
website_analyses = session.query(WebsiteAnalysis).filter(
WebsiteAnalysis.session_id == onboarding_session.id
).order_by(WebsiteAnalysis.updated_at.desc()).all()
# Get research preferences
research_prefs = session.query(ResearchPreferences).filter(
ResearchPreferences.session_id == onboarding_session.id
).first()
# Get API keys
api_keys = session.query(APIKey).filter(
APIKey.session_id == onboarding_session.id
).all()
# Compile comprehensive data with ALL available information
onboarding_data = {
"session_info": {
"session_id": onboarding_session.id,
"user_id": onboarding_session.user_id,
"current_step": onboarding_session.current_step,
"progress": onboarding_session.progress,
"started_at": onboarding_session.started_at.isoformat() if onboarding_session.started_at else None,
"updated_at": onboarding_session.updated_at.isoformat() if onboarding_session.updated_at else None
},
"api_keys": [key.to_dict() for key in api_keys] if api_keys else [],
"website_analyses": [analysis.to_dict() for analysis in website_analyses] if website_analyses else [],
"research_preferences": research_prefs.to_dict() if research_prefs else None,
# Legacy compatibility - use the latest website analysis
"website_analysis": website_analyses[0].to_dict() if website_analyses else None,
# Enhanced data extraction for persona generation
"enhanced_analysis": self._extract_enhanced_analysis_data(website_analyses, research_prefs)
}
session.close()
return onboarding_data
except Exception as e:
logger.error(f"Error collecting onboarding data: {str(e)}")
return None
def _extract_enhanced_analysis_data(self, website_analyses: List, research_prefs) -> Dict[str, Any]:
"""Extract and structure all the rich AI analysis data for persona generation."""
enhanced_data = {
"comprehensive_style_analysis": {},
"content_insights": {},
"audience_intelligence": {},
"brand_voice_analysis": {},
"technical_writing_metrics": {},
"competitive_analysis": {},
"content_strategy_insights": {}
}
if not website_analyses:
return enhanced_data
# Use the latest (most comprehensive) website analysis
latest_analysis = website_analyses[0]
# Extract comprehensive style analysis
if latest_analysis.writing_style:
enhanced_data["comprehensive_style_analysis"] = {
"tone_analysis": latest_analysis.writing_style.get("tone", ""),
"voice_characteristics": latest_analysis.writing_style.get("voice", ""),
"complexity_assessment": latest_analysis.writing_style.get("complexity", ""),
"engagement_level": latest_analysis.writing_style.get("engagement_level", ""),
"brand_personality": latest_analysis.writing_style.get("brand_personality", ""),
"formality_level": latest_analysis.writing_style.get("formality_level", ""),
"emotional_appeal": latest_analysis.writing_style.get("emotional_appeal", "")
}
# Extract content insights
if latest_analysis.content_characteristics:
enhanced_data["content_insights"] = {
"sentence_structure_analysis": latest_analysis.content_characteristics.get("sentence_structure", ""),
"vocabulary_level": latest_analysis.content_characteristics.get("vocabulary_level", ""),
"paragraph_organization": latest_analysis.content_characteristics.get("paragraph_organization", ""),
"content_flow": latest_analysis.content_characteristics.get("content_flow", ""),
"readability_score": latest_analysis.content_characteristics.get("readability_score", ""),
"content_density": latest_analysis.content_characteristics.get("content_density", ""),
"visual_elements_usage": latest_analysis.content_characteristics.get("visual_elements_usage", "")
}
# Extract audience intelligence
if latest_analysis.target_audience:
enhanced_data["audience_intelligence"] = {
"demographics": latest_analysis.target_audience.get("demographics", []),
"expertise_level": latest_analysis.target_audience.get("expertise_level", ""),
"industry_focus": latest_analysis.target_audience.get("industry_focus", ""),
"geographic_focus": latest_analysis.target_audience.get("geographic_focus", ""),
"psychographic_profile": latest_analysis.target_audience.get("psychographic_profile", ""),
"pain_points": latest_analysis.target_audience.get("pain_points", []),
"motivations": latest_analysis.target_audience.get("motivations", [])
}
# Extract brand voice analysis
if latest_analysis.content_type:
enhanced_data["brand_voice_analysis"] = {
"primary_content_type": latest_analysis.content_type.get("primary_type", ""),
"secondary_content_types": latest_analysis.content_type.get("secondary_types", []),
"content_purpose": latest_analysis.content_type.get("purpose", ""),
"call_to_action_style": latest_analysis.content_type.get("call_to_action", ""),
"conversion_focus": latest_analysis.content_type.get("conversion_focus", ""),
"educational_value": latest_analysis.content_type.get("educational_value", "")
}
# Extract technical writing metrics
if latest_analysis.style_patterns:
enhanced_data["technical_writing_metrics"] = {
"sentence_length_preference": latest_analysis.style_patterns.get("patterns", {}).get("sentence_length", ""),
"vocabulary_patterns": latest_analysis.style_patterns.get("patterns", {}).get("vocabulary_patterns", []),
"rhetorical_devices": latest_analysis.style_patterns.get("patterns", {}).get("rhetorical_devices", []),
"paragraph_structure": latest_analysis.style_patterns.get("patterns", {}).get("paragraph_structure", ""),
"transition_phrases": latest_analysis.style_patterns.get("patterns", {}).get("transition_phrases", []),
"style_consistency": latest_analysis.style_patterns.get("style_consistency", ""),
"unique_elements": latest_analysis.style_patterns.get("unique_elements", [])
}
# Extract competitive analysis from crawl results
if latest_analysis.crawl_result:
crawl_data = latest_analysis.crawl_result
enhanced_data["competitive_analysis"] = {
"domain_info": crawl_data.get("domain_info", {}),
"social_media_presence": crawl_data.get("social_media", {}),
"brand_info": crawl_data.get("brand_info", {}),
"content_structure": crawl_data.get("content_structure", {}),
"meta_optimization": crawl_data.get("meta_tags", {})
}
# Extract content strategy insights from style guidelines
if latest_analysis.style_guidelines:
guidelines = latest_analysis.style_guidelines
enhanced_data["content_strategy_insights"] = {
"tone_recommendations": guidelines.get("guidelines", {}).get("tone_recommendations", []),
"structure_guidelines": guidelines.get("guidelines", {}).get("structure_guidelines", []),
"vocabulary_suggestions": guidelines.get("guidelines", {}).get("vocabulary_suggestions", []),
"engagement_tips": guidelines.get("guidelines", {}).get("engagement_tips", []),
"audience_considerations": guidelines.get("guidelines", {}).get("audience_considerations", []),
"brand_alignment": guidelines.get("guidelines", {}).get("brand_alignment", []),
"seo_optimization": guidelines.get("guidelines", {}).get("seo_optimization", []),
"conversion_optimization": guidelines.get("guidelines", {}).get("conversion_optimization", []),
"best_practices": guidelines.get("best_practices", []),
"avoid_elements": guidelines.get("avoid_elements", []),
"content_strategy": guidelines.get("content_strategy", ""),
"ai_generation_tips": guidelines.get("ai_generation_tips", []),
"competitive_advantages": guidelines.get("competitive_advantages", []),
"content_calendar_suggestions": guidelines.get("content_calendar_suggestions", [])
}
# Add research preferences insights
if research_prefs:
enhanced_data["research_preferences"] = {
"research_depth": research_prefs.research_depth,
"content_types": research_prefs.content_types,
"auto_research": research_prefs.auto_research,
"factual_content": research_prefs.factual_content
}
return enhanced_data
def calculate_data_sufficiency(self, onboarding_data: Dict[str, Any]) -> float:
"""Calculate how sufficient the onboarding data is for persona generation."""
score = 0.0
# Get enhanced analysis data
enhanced_analysis = onboarding_data.get("enhanced_analysis", {})
website_analysis = onboarding_data.get("website_analysis", {}) or {}
research_prefs = onboarding_data.get("research_preferences", {}) or {}
# Enhanced scoring based on comprehensive data availability
# Comprehensive Style Analysis (25% of score)
style_analysis = enhanced_analysis.get("comprehensive_style_analysis", {})
if style_analysis.get("tone_analysis"):
score += 5
if style_analysis.get("voice_characteristics"):
score += 5
if style_analysis.get("brand_personality"):
score += 5
if style_analysis.get("formality_level"):
score += 5
if style_analysis.get("emotional_appeal"):
score += 5
# Content Insights (20% of score)
content_insights = enhanced_analysis.get("content_insights", {})
if content_insights.get("sentence_structure_analysis"):
score += 4
if content_insights.get("vocabulary_level"):
score += 4
if content_insights.get("readability_score"):
score += 4
if content_insights.get("content_flow"):
score += 4
if content_insights.get("visual_elements_usage"):
score += 4
# Audience Intelligence (15% of score)
audience_intel = enhanced_analysis.get("audience_intelligence", {})
if audience_intel.get("demographics"):
score += 3
if audience_intel.get("expertise_level"):
score += 3
if audience_intel.get("industry_focus"):
score += 3
if audience_intel.get("psychographic_profile"):
score += 3
if audience_intel.get("pain_points"):
score += 3
# Technical Writing Metrics (15% of score)
tech_metrics = enhanced_analysis.get("technical_writing_metrics", {})
if tech_metrics.get("vocabulary_patterns"):
score += 3
if tech_metrics.get("rhetorical_devices"):
score += 3
if tech_metrics.get("paragraph_structure"):
score += 3
if tech_metrics.get("style_consistency"):
score += 3
if tech_metrics.get("unique_elements"):
score += 3
# Content Strategy Insights (15% of score)
strategy_insights = enhanced_analysis.get("content_strategy_insights", {})
if strategy_insights.get("tone_recommendations"):
score += 3
if strategy_insights.get("best_practices"):
score += 3
if strategy_insights.get("competitive_advantages"):
score += 3
if strategy_insights.get("content_strategy"):
score += 3
if strategy_insights.get("ai_generation_tips"):
score += 3
# Research Preferences (10% of score)
if research_prefs.get("research_depth"):
score += 5
if research_prefs.get("content_types"):
score += 5
# Legacy compatibility - add points for basic data if enhanced data is missing
if score < 50: # If enhanced data is insufficient, fall back to legacy scoring
legacy_score = 0.0
# Website analysis components (70% of legacy score)
if website_analysis.get("writing_style"):
legacy_score += 25
if website_analysis.get("content_characteristics"):
legacy_score += 20
if website_analysis.get("target_audience"):
legacy_score += 15
if website_analysis.get("style_patterns"):
legacy_score += 10
# Research preferences components (30% of legacy score)
if research_prefs.get("research_depth"):
legacy_score += 10
if research_prefs.get("content_types"):
legacy_score += 10
if research_prefs.get("writing_style"):
legacy_score += 10
# Use the higher of enhanced or legacy score
score = max(score, legacy_score)
return min(score, 100.0)

View File

@@ -0,0 +1,313 @@
"""
Persona Prompt Builder
Handles building comprehensive prompts for persona generation.
"""
from typing import Dict, Any
import json
from loguru import logger
class PersonaPromptBuilder:
"""Builds comprehensive prompts for persona generation."""
def build_persona_analysis_prompt(self, onboarding_data: Dict[str, Any]) -> str:
"""Build the main persona analysis prompt with comprehensive data."""
# Get enhanced analysis data
enhanced_analysis = onboarding_data.get("enhanced_analysis", {})
website_analysis = onboarding_data.get("website_analysis", {}) or {}
research_prefs = onboarding_data.get("research_preferences", {}) or {}
prompt = f"""
COMPREHENSIVE PERSONA GENERATION TASK: Create a highly detailed, data-driven writing persona based on extensive AI analysis of user's website and content strategy.
=== COMPREHENSIVE ONBOARDING DATA ANALYSIS ===
WEBSITE ANALYSIS OVERVIEW:
- URL: {website_analysis.get('website_url', 'Not provided')}
- Analysis Date: {website_analysis.get('analysis_date', 'Not provided')}
- Status: {website_analysis.get('status', 'Not provided')}
=== DETAILED STYLE ANALYSIS ===
{json.dumps(enhanced_analysis.get('comprehensive_style_analysis', {}), indent=2)}
=== CONTENT INSIGHTS ===
{json.dumps(enhanced_analysis.get('content_insights', {}), indent=2)}
=== AUDIENCE INTELLIGENCE ===
{json.dumps(enhanced_analysis.get('audience_intelligence', {}), indent=2)}
=== BRAND VOICE ANALYSIS ===
{json.dumps(enhanced_analysis.get('brand_voice_analysis', {}), indent=2)}
=== TECHNICAL WRITING METRICS ===
{json.dumps(enhanced_analysis.get('technical_writing_metrics', {}), indent=2)}
=== COMPETITIVE ANALYSIS ===
{json.dumps(enhanced_analysis.get('competitive_analysis', {}), indent=2)}
=== CONTENT STRATEGY INSIGHTS ===
{json.dumps(enhanced_analysis.get('content_strategy_insights', {}), indent=2)}
=== RESEARCH PREFERENCES ===
{json.dumps(enhanced_analysis.get('research_preferences', {}), indent=2)}
=== LEGACY DATA (for compatibility) ===
Website Analysis: {json.dumps(website_analysis.get('writing_style', {}), indent=2)}
Content Characteristics: {json.dumps(website_analysis.get('content_characteristics', {}) or {}, indent=2)}
Target Audience: {json.dumps(website_analysis.get('target_audience', {}), indent=2)}
Style Patterns: {json.dumps(website_analysis.get('style_patterns', {}), indent=2)}
=== COMPREHENSIVE PERSONA GENERATION REQUIREMENTS ===
1. IDENTITY CREATION (Based on Brand Analysis):
- Create a memorable persona name that captures the essence of the brand personality and writing style
- Define a clear archetype that reflects the brand's positioning and audience appeal
- Articulate a core belief that drives the writing philosophy and brand values
- Write a comprehensive brand voice description incorporating all style elements
2. LINGUISTIC FINGERPRINT (Quantitative Analysis from Technical Metrics):
- Calculate precise average sentence length from sentence structure analysis
- Determine preferred sentence types based on paragraph organization patterns
- Analyze active vs passive voice ratio from voice characteristics
- Extract go-to words and phrases from vocabulary patterns and style analysis
- List words and phrases to avoid based on brand alignment guidelines
- Determine contraction usage patterns from formality level
- Assess vocabulary complexity level from readability scores
3. RHETORICAL ANALYSIS (From Style Patterns):
- Identify metaphor patterns and themes from rhetorical devices
- Analyze analogy usage from content strategy insights
- Assess rhetorical question frequency from engagement tips
- Determine storytelling approach from content flow analysis
4. TONAL RANGE (From Comprehensive Style Analysis):
- Define the default tone from tone analysis and brand personality
- List permissible tones based on emotional appeal and audience considerations
- Identify forbidden tones from avoid elements and brand alignment
- Describe emotional range from psychographic profile and engagement level
5. STYLISTIC CONSTRAINTS (From Technical Writing Metrics):
- Define punctuation preferences from paragraph structure analysis
- Set formatting guidelines from content structure insights
- Establish paragraph structure preferences from organization patterns
- Include transition phrase preferences from style patterns
6. PLATFORM-SPECIFIC ADAPTATIONS (From Content Strategy):
- Incorporate SEO optimization strategies
- Include conversion optimization techniques
- Apply engagement tips for different platforms
- Use competitive advantages for differentiation
7. CONTENT STRATEGY INTEGRATION:
- Incorporate best practices from content strategy insights
- Include AI generation tips for consistent output
- Apply content calendar suggestions for timing
- Use competitive advantages for positioning
=== ENHANCED ANALYSIS INSTRUCTIONS ===
- Base your analysis on ALL the comprehensive data provided above
- Use the detailed technical metrics for precise linguistic analysis
- Incorporate brand voice analysis for authentic personality
- Apply audience intelligence for targeted communication
- Include competitive analysis for market positioning
- Use content strategy insights for practical application
- Ensure the persona reflects the brand's unique elements and competitive advantages
- Provide a confidence score (0-100) based on data richness and quality
- Include detailed analysis notes explaining your reasoning and data sources
Generate a comprehensive, data-driven persona profile that can be used to replicate this writing style across different platforms while maintaining brand authenticity and competitive positioning.
"""
return prompt
def build_platform_adaptation_prompt(self, core_persona: Dict[str, Any], platform: str, onboarding_data: Dict[str, Any], platform_constraints: Dict[str, Any]) -> str:
"""Build prompt for platform-specific persona adaptation."""
prompt = f"""
PLATFORM ADAPTATION TASK: Adapt the core writing persona for {platform.upper()}.
CORE PERSONA:
{json.dumps(core_persona, indent=2)}
PLATFORM: {platform.upper()}
PLATFORM CONSTRAINTS:
{json.dumps(platform_constraints, indent=2)}
ADAPTATION REQUIREMENTS:
1. SENTENCE METRICS:
- Adjust sentence length for platform optimal performance
- Adapt sentence variety for platform engagement
- Consider platform reading patterns
2. LEXICAL ADAPTATIONS:
- Identify platform-specific vocabulary and slang
- Define hashtag strategy (if applicable)
- Set emoji usage guidelines
- Establish mention and tagging strategy
3. CONTENT FORMAT RULES:
- Respect character/word limits
- Optimize paragraph structure for platform
- Define call-to-action style
- Set link placement strategy
4. ENGAGEMENT PATTERNS:
- Determine optimal posting frequency
- Identify best posting times for audience
- Define engagement tactics
- Set community interaction guidelines
5. PLATFORM BEST PRACTICES:
- List platform-specific optimization techniques
- Consider algorithm preferences
- Include trending format adaptations
INSTRUCTIONS:
- Maintain the core persona identity while optimizing for platform performance
- Ensure all adaptations align with the original brand voice
- Consider platform-specific audience behavior
- Provide actionable, specific guidelines
Generate a platform-optimized persona adaptation that maintains brand consistency while maximizing platform performance.
"""
return prompt
def get_persona_schema(self) -> Dict[str, Any]:
"""Get the schema for core persona generation."""
return {
"type": "object",
"properties": {
"identity": {
"type": "object",
"properties": {
"persona_name": {"type": "string"},
"archetype": {"type": "string"},
"core_belief": {"type": "string"},
"brand_voice_description": {"type": "string"}
},
"required": ["persona_name", "archetype", "core_belief"]
},
"linguistic_fingerprint": {
"type": "object",
"properties": {
"sentence_metrics": {
"type": "object",
"properties": {
"average_sentence_length_words": {"type": "number"},
"preferred_sentence_type": {"type": "string"},
"active_to_passive_ratio": {"type": "string"},
"complexity_level": {"type": "string"}
}
},
"lexical_features": {
"type": "object",
"properties": {
"go_to_words": {"type": "array", "items": {"type": "string"}},
"go_to_phrases": {"type": "array", "items": {"type": "string"}},
"avoid_words": {"type": "array", "items": {"type": "string"}},
"contractions": {"type": "string"},
"filler_words": {"type": "string"},
"vocabulary_level": {"type": "string"}
}
},
"rhetorical_devices": {
"type": "object",
"properties": {
"metaphors": {"type": "string"},
"analogies": {"type": "string"},
"rhetorical_questions": {"type": "string"},
"storytelling_style": {"type": "string"}
}
}
}
},
"tonal_range": {
"type": "object",
"properties": {
"default_tone": {"type": "string"},
"permissible_tones": {"type": "array", "items": {"type": "string"}},
"forbidden_tones": {"type": "array", "items": {"type": "string"}},
"emotional_range": {"type": "string"}
}
},
"stylistic_constraints": {
"type": "object",
"properties": {
"punctuation": {
"type": "object",
"properties": {
"ellipses": {"type": "string"},
"em_dash": {"type": "string"},
"exclamation_points": {"type": "string"}
}
},
"formatting": {
"type": "object",
"properties": {
"paragraphs": {"type": "string"},
"lists": {"type": "string"},
"markdown": {"type": "string"}
}
}
}
},
"confidence_score": {"type": "number"},
"analysis_notes": {"type": "string"}
},
"required": ["identity", "linguistic_fingerprint", "tonal_range", "confidence_score"]
}
def get_platform_schema(self) -> Dict[str, Any]:
"""Get the schema for platform-specific persona adaptation."""
return {
"type": "object",
"properties": {
"platform_type": {"type": "string"},
"sentence_metrics": {
"type": "object",
"properties": {
"max_sentence_length": {"type": "number"},
"optimal_sentence_length": {"type": "number"},
"sentence_variety": {"type": "string"}
}
},
"lexical_adaptations": {
"type": "object",
"properties": {
"platform_specific_words": {"type": "array", "items": {"type": "string"}},
"hashtag_strategy": {"type": "string"},
"emoji_usage": {"type": "string"},
"mention_strategy": {"type": "string"}
}
},
"content_format_rules": {
"type": "object",
"properties": {
"character_limit": {"type": "number"},
"paragraph_structure": {"type": "string"},
"call_to_action_style": {"type": "string"},
"link_placement": {"type": "string"}
}
},
"engagement_patterns": {
"type": "object",
"properties": {
"posting_frequency": {"type": "string"},
"optimal_posting_times": {"type": "array", "items": {"type": "string"}},
"engagement_tactics": {"type": "array", "items": {"type": "string"}},
"community_interaction": {"type": "string"}
}
},
"platform_best_practices": {
"type": "array",
"items": {"type": "string"}
}
},
"required": ["platform_type", "sentence_metrics", "content_format_rules", "engagement_patterns"]
}

View File

@@ -0,0 +1,213 @@
"""
Facebook Persona Prompts
Contains Facebook-specific persona prompt generation logic.
"""
from typing import Dict, Any
from loguru import logger
class FacebookPersonaPrompts:
"""Facebook-specific persona prompt generation."""
@staticmethod
def build_facebook_system_prompt(core_persona: Dict[str, Any]) -> str:
"""
Build optimized system prompt with core persona for Facebook generation.
This moves the core persona to system prompt to free up context window.
"""
import json
return f"""You are an expert Facebook content strategist specializing in community engagement and social sharing optimization.
CORE PERSONA FOUNDATION:
{json.dumps(core_persona, indent=2)}
TASK: Create Facebook-optimized persona adaptations that maintain core identity while maximizing community engagement and Facebook algorithm performance.
FOCUS AREAS:
- Community-focused tone and engagement strategies
- Facebook algorithm optimization (engagement, reach, timing)
- Social sharing and viral content potential
- Facebook-specific features (Stories, Reels, Live, Groups, Events)
- Audience interaction and community building"""
@staticmethod
def build_focused_facebook_prompt(onboarding_data: Dict[str, Any]) -> str:
"""
Build focused Facebook prompt without core persona JSON to optimize context usage.
"""
# Extract audience context
audience_context = FacebookPersonaPrompts._extract_audience_context(onboarding_data)
target_audience = audience_context.get("target_audience", "general")
content_goals = audience_context.get("content_goals", "engagement")
business_type = audience_context.get("business_type", "general")
return f"""FACEBOOK OPTIMIZATION TASK: Create Facebook-specific adaptations for the core persona.
AUDIENCE CONTEXT:
- Target: {target_audience} | Goals: {content_goals} | Business: {business_type}
- Demographics: {audience_context.get('demographics', [])}
- Interests: {audience_context.get('interests', [])}
- Behaviors: {audience_context.get('behaviors', [])}
FACEBOOK SPECS:
- Character Limit: 63,206 | Optimal Length: 40-80 words
- Algorithm Priority: Engagement, meaningful interactions, community building
- Content Types: Posts, Stories, Reels, Live, Events, Groups, Carousels, Polls
- Hashtag Strategy: 1-2 recommended (max 30)
- Link Strategy: Native content performs better
OPTIMIZATION REQUIREMENTS:
1. COMMUNITY-FOCUSED TONE:
- Authentic, conversational, approachable language
- Balance professionalism with relatability
- Incorporate storytelling and personal anecdotes
- Community-building elements
2. CONTENT STRATEGY FOR {business_type.upper()}:
- Community engagement content for {target_audience}
- Social sharing optimization for {content_goals}
- Facebook-specific content formats
- Audience interaction strategies
- Viral content potential
3. FACEBOOK-SPECIFIC ADAPTATIONS:
- Algorithm optimization (engagement, reach, timing)
- Platform-specific vocabulary and terminology
- Engagement patterns for Facebook audience
- Community interaction strategies
- Facebook feature optimization (Stories, Reels, Live, Events, Groups)
4. AUDIENCE TARGETING:
- Demographic-specific positioning
- Interest-based content adaptation
- Behavioral targeting considerations
- Community building strategies
- Engagement optimization tactics
Generate comprehensive Facebook-optimized persona maintaining core identity while maximizing community engagement and social sharing potential."""
@staticmethod
def _extract_audience_context(onboarding_data: Dict[str, Any]) -> Dict[str, Any]:
"""Extract audience context from onboarding data."""
try:
# Get enhanced analysis data
enhanced_analysis = onboarding_data.get("enhanced_analysis", {})
website_analysis = onboarding_data.get("website_analysis", {}) or {}
research_prefs = onboarding_data.get("research_preferences", {}) or {}
# Extract audience intelligence
audience_intel = enhanced_analysis.get("audience_intelligence", {})
# Extract target audience from website analysis
target_audience_data = website_analysis.get("target_audience", {}) or {}
# Build audience context
audience_context = {
"target_audience": target_audience_data.get("primary_audience", "general"),
"content_goals": research_prefs.get("content_goals", "engagement"),
"business_type": website_analysis.get("business_type", "general"),
"demographics": audience_intel.get("demographics", []),
"interests": audience_intel.get("interests", []),
"behaviors": audience_intel.get("behaviors", []),
"psychographic_profile": audience_intel.get("psychographic_profile", "general"),
"pain_points": audience_intel.get("pain_points", []),
"engagement_level": audience_intel.get("engagement_level", "moderate")
}
return audience_context
except Exception as e:
logger.warning(f"Error extracting audience context: {str(e)}")
return {
"target_audience": "general",
"content_goals": "engagement",
"business_type": "general",
"demographics": [],
"interests": [],
"behaviors": [],
"psychographic_profile": "general",
"pain_points": [],
"engagement_level": "moderate"
}
@staticmethod
def build_facebook_validation_prompt(persona_data: Dict[str, Any]) -> str:
"""Build optimized prompt for validating Facebook persona data."""
return f"""FACEBOOK PERSONA VALIDATION TASK: Validate Facebook persona data for completeness and quality.
PERSONA DATA:
{persona_data}
VALIDATION REQUIREMENTS:
1. COMPLETENESS CHECK:
- Verify all required Facebook-specific fields are present
- Check for missing algorithm optimization strategies
- Validate engagement strategy completeness
- Ensure content format rules are defined
2. QUALITY ASSESSMENT:
- Evaluate Facebook algorithm optimization quality
- Assess engagement strategy effectiveness
- Check content format optimization
- Validate audience targeting strategies
3. FACEBOOK-SPECIFIC VALIDATION:
- Verify Facebook platform constraints are respected
- Check for Facebook-specific best practices
- Validate community building strategies
- Ensure Facebook feature optimization
4. RECOMMENDATIONS:
- Provide specific improvement suggestions
- Identify missing optimization opportunities
- Suggest Facebook-specific enhancements
- Recommend engagement strategy improvements
Generate comprehensive validation report with scores, recommendations, and specific improvement suggestions for Facebook optimization."""
@staticmethod
def build_facebook_optimization_prompt(persona_data: Dict[str, Any]) -> str:
"""Build optimized prompt for optimizing Facebook persona data."""
return f"""FACEBOOK PERSONA OPTIMIZATION TASK: Optimize Facebook persona data for maximum algorithm performance and community engagement.
CURRENT PERSONA DATA:
{persona_data}
OPTIMIZATION REQUIREMENTS:
1. ALGORITHM OPTIMIZATION:
- Enhance Facebook algorithm performance strategies
- Optimize for Facebook's engagement metrics
- Improve content timing and frequency
- Enhance audience targeting precision
2. ENGAGEMENT OPTIMIZATION:
- Strengthen community building strategies
- Enhance social sharing potential
- Improve audience interaction tactics
- Optimize content for viral potential
3. CONTENT FORMAT OPTIMIZATION:
- Optimize for Facebook's content formats
- Enhance visual content strategies
- Improve video content optimization
- Optimize for Facebook Stories and Reels
4. AUDIENCE TARGETING OPTIMIZATION:
- Refine demographic targeting
- Enhance interest-based targeting
- Improve behavioral targeting
- Optimize for Facebook's audience insights
5. COMMUNITY BUILDING OPTIMIZATION:
- Enhance group management strategies
- Improve event management tactics
- Optimize live streaming strategies
- Enhance community interaction methods
Generate optimized Facebook persona data with enhanced algorithm performance, engagement strategies, and community building tactics."""

View File

@@ -0,0 +1,364 @@
"""
Facebook Persona Schemas
Defines Facebook-specific persona data structures and validation schemas.
"""
from typing import Dict, Any, List, Optional
from pydantic import BaseModel, Field
class FacebookPersonaSchema(BaseModel):
"""Facebook-specific persona schema with platform optimizations."""
# Core persona fields (inherited from base persona)
persona_name: str = Field(..., description="Name of the persona")
archetype: str = Field(..., description="Persona archetype")
core_belief: str = Field(..., description="Core belief driving the persona")
# Facebook-specific optimizations
facebook_algorithm_optimization: Dict[str, Any] = Field(
default_factory=dict,
description="Facebook algorithm optimization strategies"
)
facebook_engagement_strategies: Dict[str, Any] = Field(
default_factory=dict,
description="Facebook-specific engagement strategies"
)
facebook_content_formats: Dict[str, Any] = Field(
default_factory=dict,
description="Facebook content format optimizations"
)
facebook_audience_targeting: Dict[str, Any] = Field(
default_factory=dict,
description="Facebook audience targeting strategies"
)
facebook_community_building: Dict[str, Any] = Field(
default_factory=dict,
description="Facebook community building strategies"
)
class FacebookPersonaConstraints:
"""Facebook platform constraints and best practices."""
@staticmethod
def get_facebook_constraints() -> Dict[str, Any]:
"""Get Facebook-specific platform constraints."""
return {
"character_limit": 63206,
"optimal_length": "40-80 words",
"hashtag_limit": 30,
"image_support": True,
"video_support": True,
"link_preview": True,
"event_support": True,
"group_sharing": True,
"story_support": True,
"reel_support": True,
"carousel_support": True,
"poll_support": True,
"live_support": True,
"algorithm_favors": [
"engagement",
"meaningful_interactions",
"video_content",
"community_posts",
"authentic_content"
],
"content_types": [
"text_posts",
"image_posts",
"video_posts",
"carousel_posts",
"story_posts",
"reel_posts",
"event_posts",
"poll_posts",
"live_posts"
],
"engagement_metrics": [
"likes",
"comments",
"shares",
"saves",
"clicks",
"reactions",
"video_views",
"story_views"
],
"posting_frequency": {
"optimal": "1-2 times per day",
"maximum": "3-4 times per day",
"minimum": "3-4 times per week"
},
"best_posting_times": [
"9:00 AM - 11:00 AM",
"1:00 PM - 3:00 PM",
"5:00 PM - 7:00 PM"
],
"content_guidelines": {
"authenticity": "High priority - Facebook favors authentic content",
"community_focus": "Build community and meaningful connections",
"visual_content": "Images and videos perform better than text-only",
"engagement_bait": "Avoid engagement bait - Facebook penalizes it",
"clickbait": "Avoid clickbait headlines and misleading content"
}
}
class FacebookPersonaValidation:
"""Facebook persona validation rules and scoring."""
@staticmethod
def validate_facebook_persona(persona_data: Dict[str, Any]) -> Dict[str, Any]:
"""Validate Facebook persona data for completeness and quality."""
validation_results = {
"is_valid": True,
"quality_score": 0.0,
"completeness_score": 0.0,
"facebook_optimization_score": 0.0,
"engagement_strategy_score": 0.0,
"missing_fields": [],
"incomplete_fields": [],
"recommendations": [],
"quality_issues": [],
"strengths": [],
"validation_details": {}
}
# Check required fields
required_fields = [
"persona_name", "archetype", "core_belief",
"facebook_algorithm_optimization", "facebook_engagement_strategies"
]
for field in required_fields:
if not persona_data.get(field):
validation_results["missing_fields"].append(field)
validation_results["is_valid"] = False
# Calculate completeness score
total_fields = len(required_fields)
present_fields = total_fields - len(validation_results["missing_fields"])
validation_results["completeness_score"] = (present_fields / total_fields) * 100
# Validate Facebook-specific optimizations
facebook_opt = persona_data.get("facebook_algorithm_optimization", {})
if facebook_opt:
validation_results["facebook_optimization_score"] = 85.0
validation_results["strengths"].append("Facebook algorithm optimization present")
else:
validation_results["quality_issues"].append("Missing Facebook algorithm optimization")
validation_results["recommendations"].append("Add Facebook-specific algorithm strategies")
# Validate engagement strategies
engagement_strategies = persona_data.get("facebook_engagement_strategies", {})
if engagement_strategies:
validation_results["engagement_strategy_score"] = 80.0
validation_results["strengths"].append("Facebook engagement strategies defined")
else:
validation_results["quality_issues"].append("Missing Facebook engagement strategies")
validation_results["recommendations"].append("Define Facebook-specific engagement tactics")
# Calculate overall quality score
validation_results["quality_score"] = (
validation_results["completeness_score"] * 0.4 +
validation_results["facebook_optimization_score"] * 0.3 +
validation_results["engagement_strategy_score"] * 0.3
)
# Add validation details
validation_results["validation_details"] = {
"total_fields_checked": total_fields,
"present_fields": present_fields,
"facebook_optimization_present": bool(facebook_opt),
"engagement_strategies_present": bool(engagement_strategies),
"validation_timestamp": "2024-01-01T00:00:00Z" # Will be updated with actual timestamp
}
return validation_results
class FacebookPersonaOptimization:
"""Facebook persona optimization strategies and techniques."""
@staticmethod
def get_facebook_optimization_strategies() -> Dict[str, Any]:
"""Get comprehensive Facebook optimization strategies."""
return {
"algorithm_optimization": {
"engagement_optimization": [
"Post when your audience is most active",
"Use Facebook's native video uploads instead of external links",
"Encourage meaningful comments and discussions",
"Respond to comments within 2 hours",
"Use Facebook Live for real-time engagement",
"Create shareable, valuable content",
"Use Facebook Stories for behind-the-scenes content",
"Leverage Facebook Groups for community building"
],
"content_quality_optimization": [
"Create authentic, original content",
"Use high-quality images and videos",
"Write compelling captions that encourage engagement",
"Use Facebook's built-in editing tools",
"Create content that sparks conversations",
"Share user-generated content",
"Use Facebook's trending topics and hashtags",
"Create content that provides value to your audience"
],
"timing_optimization": [
"Post during peak engagement hours (9-11 AM, 1-3 PM, 5-7 PM)",
"Use Facebook Insights to find your best posting times",
"Post consistently but not too frequently",
"Schedule posts for different time zones if global audience",
"Use Facebook's scheduling feature for optimal timing",
"Post when your competitors are less active",
"Consider your audience's daily routines and habits"
],
"audience_targeting_optimization": [
"Use Facebook's audience insights for targeting",
"Create content for specific audience segments",
"Use Facebook's lookalike audiences",
"Target based on interests and behaviors",
"Use Facebook's custom audiences",
"Create content that resonates with your core audience",
"Use Facebook's demographic targeting",
"Leverage Facebook's psychographic targeting"
]
},
"engagement_strategies": {
"community_building": [
"Create and moderate Facebook Groups",
"Host Facebook Live sessions regularly",
"Respond to all comments and messages",
"Share user-generated content",
"Create Facebook Events for community gatherings",
"Use Facebook's community features",
"Encourage user participation and feedback",
"Build relationships with your audience"
],
"content_engagement": [
"Ask questions in your posts",
"Use polls and surveys to engage audience",
"Create interactive content like quizzes",
"Use Facebook's reaction buttons strategically",
"Create content that encourages sharing",
"Use Facebook's tagging feature appropriately",
"Create content that sparks discussions",
"Use Facebook's story features for engagement"
],
"conversion_optimization": [
"Use clear call-to-actions in posts",
"Create Facebook-specific landing pages",
"Use Facebook's conversion tracking",
"Create content that drives traffic to your website",
"Use Facebook's lead generation features",
"Create content that builds trust and credibility",
"Use Facebook's retargeting capabilities",
"Create content that showcases your products/services"
]
},
"content_formats": {
"text_posts": {
"optimal_length": "40-80 words",
"best_practices": [
"Use compelling headlines",
"Include relevant hashtags (1-2)",
"Ask questions to encourage engagement",
"Use emojis sparingly but effectively",
"Include clear call-to-actions"
]
},
"image_posts": {
"optimal_specs": "1200x630 pixels",
"best_practices": [
"Use high-quality, original images",
"Include text overlay for key messages",
"Use consistent branding and colors",
"Create visually appealing graphics",
"Use Facebook's image editing tools"
]
},
"video_posts": {
"optimal_length": "15-60 seconds for feed, 2-3 minutes for longer content",
"best_practices": [
"Upload videos directly to Facebook",
"Create engaging thumbnails",
"Add captions for accessibility",
"Use Facebook's video editing tools",
"Create videos that work without sound"
]
},
"carousel_posts": {
"optimal_slides": "3-5 slides",
"best_practices": [
"Tell a story across slides",
"Use consistent design elements",
"Include clear navigation",
"Create slides that work individually",
"Use carousels for product showcases"
]
}
},
"audience_targeting": {
"demographic_targeting": [
"Age and gender targeting",
"Location-based targeting",
"Education and work targeting",
"Relationship status targeting",
"Language targeting"
],
"interest_targeting": [
"Hobbies and interests",
"Brand and product interests",
"Entertainment preferences",
"Lifestyle and behavior targeting",
"Purchase behavior targeting"
],
"behavioral_targeting": [
"Device usage patterns",
"Travel behavior",
"Purchase behavior",
"Digital activity patterns",
"Life events targeting"
]
},
"community_building": {
"group_management": [
"Create and moderate relevant Facebook Groups",
"Set clear group rules and guidelines",
"Encourage member participation",
"Share valuable content in groups",
"Use groups for customer support",
"Create group events and activities",
"Recognize and reward active members",
"Use groups for market research"
],
"event_management": [
"Create Facebook Events for promotions",
"Use events for product launches",
"Host virtual events and webinars",
"Create recurring events for consistency",
"Use events for community building",
"Promote events across all channels",
"Follow up with event attendees",
"Use events for lead generation"
],
"live_streaming": [
"Host regular Facebook Live sessions",
"Use live streaming for Q&A sessions",
"Create behind-the-scenes content",
"Use live streaming for product demos",
"Engage with viewers in real-time",
"Use live streaming for announcements",
"Create interactive live content",
"Use live streaming for customer support"
]
}
}

View File

@@ -0,0 +1,421 @@
"""
Facebook Persona Service
Encapsulates Facebook-specific persona generation logic.
"""
from typing import Dict, Any, Optional
from loguru import logger
from datetime import datetime
from .facebook_persona_schemas import (
FacebookPersonaSchema,
FacebookPersonaConstraints,
FacebookPersonaValidation,
FacebookPersonaOptimization
)
from .facebook_persona_prompts import FacebookPersonaPrompts
from services.llm_providers.gemini_provider import gemini_structured_json_response
class FacebookPersonaService:
"""Facebook-specific persona generation and optimization service."""
def __init__(self):
"""Initialize the Facebook persona service."""
self.schemas = FacebookPersonaSchema
self.constraints = FacebookPersonaConstraints()
self.validation = FacebookPersonaValidation()
self.optimization = FacebookPersonaOptimization()
self.prompts = FacebookPersonaPrompts()
logger.info("FacebookPersonaService initialized")
def generate_facebook_persona(self, core_persona: Dict[str, Any], onboarding_data: Dict[str, Any]) -> Dict[str, Any]:
"""
Generate Facebook-specific persona adaptation using optimized chained prompts.
Args:
core_persona: The core persona data
onboarding_data: User onboarding data
Returns:
Facebook-optimized persona data
"""
try:
logger.info("Generating Facebook-specific persona with optimized prompts")
# Build focused Facebook prompt (without core persona JSON)
prompt = self.prompts.build_focused_facebook_prompt(onboarding_data)
# Create system prompt with core persona
system_prompt = self.prompts.build_facebook_system_prompt(core_persona)
# Get Facebook-specific schema
schema = self._get_enhanced_facebook_schema()
# Generate structured response using Gemini with optimized prompts
response = gemini_structured_json_response(
prompt=prompt,
schema=schema,
temperature=0.2,
max_tokens=4096,
system_prompt=system_prompt
)
if not response or "error" in response:
logger.error(f"Failed to generate Facebook persona: {response}")
return {"error": f"Failed to generate Facebook persona: {response}"}
# Validate the generated persona
validation_results = self.validate_facebook_persona(response)
# Apply algorithm optimization
optimized_persona = self.optimize_for_facebook_algorithm(response)
# Add validation results to the persona
optimized_persona["validation_results"] = validation_results
logger.info(f"✅ Facebook persona generated successfully with {validation_results['quality_score']:.1f}% quality score")
return optimized_persona
except Exception as e:
logger.error(f"Error generating Facebook persona: {str(e)}")
return {"error": f"Failed to generate Facebook persona: {str(e)}"}
def get_facebook_constraints(self) -> Dict[str, Any]:
"""Get Facebook-specific platform constraints."""
return self.constraints.get_facebook_constraints()
def validate_facebook_persona(self, persona_data: Dict[str, Any]) -> Dict[str, Any]:
"""
Validate Facebook persona data for completeness and quality.
Args:
persona_data: Facebook persona data to validate
Returns:
Validation results with scores and recommendations
"""
try:
logger.info("Validating Facebook persona data")
# Use the validation class
validation_results = self.validation.validate_facebook_persona(persona_data)
# Initialize missing fields if they don't exist
if "content_format_score" not in validation_results:
validation_results["content_format_score"] = 0.0
if "audience_targeting_score" not in validation_results:
validation_results["audience_targeting_score"] = 0.0
if "community_building_score" not in validation_results:
validation_results["community_building_score"] = 0.0
# Add Facebook-specific validation
facebook_opt = persona_data.get("facebook_algorithm_optimization", {})
if facebook_opt:
validation_results["facebook_optimization_score"] = 90.0
validation_results["strengths"].append("Facebook algorithm optimization present")
else:
validation_results["quality_issues"].append("Missing Facebook algorithm optimization")
validation_results["recommendations"].append("Add Facebook-specific algorithm strategies")
# Validate engagement strategies
engagement_strategies = persona_data.get("facebook_engagement_strategies", {})
if engagement_strategies:
validation_results["engagement_strategy_score"] = 85.0
validation_results["strengths"].append("Facebook engagement strategies defined")
else:
validation_results["quality_issues"].append("Missing Facebook engagement strategies")
validation_results["recommendations"].append("Define Facebook-specific engagement tactics")
# Validate content formats
content_formats = persona_data.get("facebook_content_formats", {})
if content_formats:
validation_results["content_format_score"] = 80.0
validation_results["strengths"].append("Facebook content formats optimized")
else:
validation_results["quality_issues"].append("Missing Facebook content format optimization")
validation_results["recommendations"].append("Add Facebook-specific content format strategies")
# Validate audience targeting
audience_targeting = persona_data.get("facebook_audience_targeting", {})
if audience_targeting:
validation_results["audience_targeting_score"] = 75.0
validation_results["strengths"].append("Facebook audience targeting strategies present")
else:
validation_results["quality_issues"].append("Missing Facebook audience targeting")
validation_results["recommendations"].append("Add Facebook-specific audience targeting strategies")
# Validate community building
community_building = persona_data.get("facebook_community_building", {})
if community_building:
validation_results["community_building_score"] = 85.0
validation_results["strengths"].append("Facebook community building strategies defined")
else:
validation_results["quality_issues"].append("Missing Facebook community building strategies")
validation_results["recommendations"].append("Add Facebook-specific community building tactics")
# Recalculate overall quality score
validation_results["quality_score"] = (
validation_results["completeness_score"] * 0.2 +
validation_results["facebook_optimization_score"] * 0.25 +
validation_results["engagement_strategy_score"] * 0.2 +
validation_results["content_format_score"] * 0.15 +
validation_results["audience_targeting_score"] * 0.1 +
validation_results["community_building_score"] * 0.1
)
# Add validation timestamp
validation_results["validation_timestamp"] = datetime.utcnow().isoformat()
logger.info(f"Facebook persona validation completed: Quality Score: {validation_results['quality_score']:.1f}%")
return validation_results
except Exception as e:
logger.error(f"Error validating Facebook persona: {str(e)}")
return {
"is_valid": False,
"quality_score": 0.0,
"error": f"Validation failed: {str(e)}"
}
def optimize_for_facebook_algorithm(self, persona_data: Dict[str, Any]) -> Dict[str, Any]:
"""
Optimize Facebook persona data for maximum algorithm performance.
Args:
persona_data: Facebook persona data to optimize
Returns:
Optimized Facebook persona data
"""
try:
logger.info("Optimizing Facebook persona for algorithm performance")
# Get optimization strategies
optimization_strategies = self.optimization.get_facebook_optimization_strategies()
# Apply algorithm optimization
optimized_persona = persona_data.copy()
# Add comprehensive algorithm optimization
optimized_persona["algorithm_optimization"] = {
"engagement_optimization": optimization_strategies["algorithm_optimization"]["engagement_optimization"],
"content_quality_optimization": optimization_strategies["algorithm_optimization"]["content_quality_optimization"],
"timing_optimization": optimization_strategies["algorithm_optimization"]["timing_optimization"],
"audience_targeting_optimization": optimization_strategies["algorithm_optimization"]["audience_targeting_optimization"]
}
# Add engagement strategies
optimized_persona["engagement_strategies"] = {
"community_building": optimization_strategies["engagement_strategies"]["community_building"],
"content_engagement": optimization_strategies["engagement_strategies"]["content_engagement"],
"conversion_optimization": optimization_strategies["engagement_strategies"]["conversion_optimization"]
}
# Add content format optimization
optimized_persona["content_formats"] = optimization_strategies["content_formats"]
# Add audience targeting optimization
optimized_persona["audience_targeting"] = optimization_strategies["audience_targeting"]
# Add community building optimization
optimized_persona["community_building"] = optimization_strategies["community_building"]
# Add optimization metadata
total_strategies = 0
for category_name, category_data in optimization_strategies.items():
if isinstance(category_data, dict):
for strategy_name, strategies in category_data.items():
if isinstance(strategies, list):
total_strategies += len(strategies)
elif isinstance(strategies, dict):
# Handle nested dictionaries
for sub_strategy_name, sub_strategies in strategies.items():
if isinstance(sub_strategies, list):
total_strategies += len(sub_strategies)
else:
total_strategies += 1
else:
total_strategies += 1
elif isinstance(category_data, list):
total_strategies += len(category_data)
else:
total_strategies += 1
optimized_persona["optimization_metadata"] = {
"optimization_applied": True,
"optimization_timestamp": datetime.utcnow().isoformat(),
"optimization_categories": list(optimization_strategies.keys()),
"total_optimization_strategies": total_strategies
}
logger.info("✅ Facebook persona algorithm optimization completed successfully")
return optimized_persona
except Exception as e:
logger.error(f"Error optimizing Facebook persona: {str(e)}")
return persona_data # Return original data if optimization fails
def _get_enhanced_facebook_schema(self) -> Dict[str, Any]:
"""Get enhanced Facebook persona schema for Gemini structured response with improved JSON parsing reliability."""
return {
"type": "object",
"description": "Facebook-optimized persona data structure for community engagement and algorithm optimization",
"properties": {
"persona_name": {
"type": "string",
"description": "Name of the Facebook-optimized persona (e.g., 'Community Builder', 'Social Connector')",
"minLength": 3,
"maxLength": 50
},
"archetype": {
"type": "string",
"description": "Persona archetype for Facebook (e.g., 'The Community Catalyst', 'The Social Storyteller')",
"minLength": 5,
"maxLength": 50
},
"core_belief": {
"type": "string",
"description": "Core belief driving the Facebook persona (e.g., 'Building authentic connections through shared experiences')",
"minLength": 10,
"maxLength": 200
},
"facebook_algorithm_optimization": {
"type": "object",
"description": "Facebook algorithm optimization strategies",
"properties": {
"engagement_optimization": {
"type": "array",
"items": {"type": "string"},
"description": "Strategies for optimizing Facebook engagement (3-8 strategies)",
"minItems": 3,
"maxItems": 8
},
"content_quality_optimization": {
"type": "array",
"items": {"type": "string"},
"description": "Strategies for optimizing content quality on Facebook (3-8 strategies)",
"minItems": 3,
"maxItems": 8
},
"timing_optimization": {
"type": "array",
"items": {"type": "string"},
"description": "Strategies for optimizing posting timing on Facebook (3-8 strategies)",
"minItems": 3,
"maxItems": 8
},
"audience_targeting_optimization": {
"type": "array",
"items": {"type": "string"},
"description": "Strategies for optimizing audience targeting on Facebook (3-8 strategies)",
"minItems": 3,
"maxItems": 8
}
}
},
"facebook_engagement_strategies": {
"type": "object",
"description": "Facebook-specific engagement strategies",
"properties": {
"community_building": {
"type": "array",
"items": {"type": "string"},
"description": "Community building strategies for Facebook"
},
"content_engagement": {
"type": "array",
"items": {"type": "string"},
"description": "Content engagement strategies for Facebook"
},
"conversion_optimization": {
"type": "array",
"items": {"type": "string"},
"description": "Conversion optimization strategies for Facebook"
}
}
},
"facebook_content_formats": {
"type": "object",
"description": "Facebook content format optimizations",
"properties": {
"text_posts": {
"type": "object",
"description": "Text post optimization for Facebook"
},
"image_posts": {
"type": "object",
"description": "Image post optimization for Facebook"
},
"video_posts": {
"type": "object",
"description": "Video post optimization for Facebook"
},
"carousel_posts": {
"type": "object",
"description": "Carousel post optimization for Facebook"
}
}
},
"facebook_audience_targeting": {
"type": "object",
"description": "Facebook audience targeting strategies",
"properties": {
"demographic_targeting": {
"type": "array",
"items": {"type": "string"},
"description": "Demographic targeting strategies for Facebook"
},
"interest_targeting": {
"type": "array",
"items": {"type": "string"},
"description": "Interest targeting strategies for Facebook"
},
"behavioral_targeting": {
"type": "array",
"items": {"type": "string"},
"description": "Behavioral targeting strategies for Facebook"
}
}
},
"facebook_community_building": {
"type": "object",
"description": "Facebook community building strategies",
"properties": {
"group_management": {
"type": "array",
"items": {"type": "string"},
"description": "Facebook Group management strategies"
},
"event_management": {
"type": "array",
"items": {"type": "string"},
"description": "Facebook Event management strategies"
},
"live_streaming": {
"type": "array",
"items": {"type": "string"},
"description": "Facebook Live streaming strategies"
}
}
},
"confidence_score": {
"type": "number",
"description": "Confidence score for the Facebook persona (0-100)",
"minimum": 0,
"maximum": 100
}
},
"required": [
"persona_name",
"archetype",
"core_belief",
"facebook_algorithm_optimization",
"facebook_engagement_strategies",
"confidence_score"
],
"additionalProperties": False
}

View File

@@ -0,0 +1,10 @@
"""
LinkedIn Persona Services
Contains LinkedIn-specific persona generation and optimization services.
"""
from .linkedin_persona_service import LinkedInPersonaService
from .linkedin_persona_prompts import LinkedInPersonaPrompts
from .linkedin_persona_schemas import LinkedInPersonaSchemas
__all__ = ['LinkedInPersonaService', 'LinkedInPersonaPrompts', 'LinkedInPersonaSchemas']

View File

@@ -0,0 +1,319 @@
"""
LinkedIn Persona Prompts
Contains LinkedIn-specific prompt generation for persona analysis.
"""
from typing import Dict, Any
import json
from loguru import logger
class LinkedInPersonaPrompts:
"""Handles LinkedIn-specific persona prompt generation."""
@staticmethod
def build_enhanced_linkedin_prompt(core_persona: Dict[str, Any], onboarding_data: Dict[str, Any]) -> str:
"""Build enhanced LinkedIn-specific persona prompt with professional optimization."""
# Extract comprehensive professional context
professional_context = LinkedInPersonaPrompts._extract_professional_context(onboarding_data)
website_analysis = onboarding_data.get("website_analysis", {}) or {}
target_audience = website_analysis.get("target_audience", {})
industry_focus = professional_context.get("industry_focus", "general")
expertise_level = professional_context.get("expertise_level", "intermediate")
prompt = f"""
LINKEDIN PROFESSIONAL PERSONA OPTIMIZATION TASK: Create a comprehensive LinkedIn-optimized writing persona for professional networking and thought leadership.
CORE PERSONA FOUNDATION:
{json.dumps(core_persona, indent=2)}
PROFESSIONAL CONTEXT:
- Industry: {industry_focus}
- Expertise Level: {expertise_level}
- Company Size: {professional_context.get('company_size', 'Not specified')}
- Business Model: {professional_context.get('business_model', 'Not specified')}
- Professional Role: {professional_context.get('professional_role', 'Not specified')}
- Demographics: {professional_context.get('target_demographics', [])}
- Psychographic: {professional_context.get('psychographic_profile', 'Not specified')}
LINKEDIN PLATFORM SPECIFICATIONS:
- Character Limit: 3,000 characters
- Optimal Post Length: 150-300 words for maximum engagement
- Professional Network: B2B focused, career-oriented audience
- Algorithm Priority: Engagement, relevance, professional value
- Content Types: Posts, Articles, Polls, Videos, Carousels, Events
- Hashtag Limit: 3-5 hashtags for optimal reach
- Link Strategy: Place external links in first comment for algorithm optimization
LINKEDIN PROFESSIONAL OPTIMIZATION REQUIREMENTS:
1. PROFESSIONAL TONE & VOICE:
- Maintain authoritative yet approachable professional tone
- Use industry-specific terminology appropriately
- Balance expertise with accessibility for {expertise_level} audience
- Incorporate thought leadership elements
- Include professional storytelling and case studies
2. CONTENT STRATEGY FOR {industry_focus.upper()}:
- Industry insights for {expertise_level} professionals
- Professional development content for {professional_context.get('target_demographics', [])}
- Business strategy discussions for {professional_context.get('business_model', 'general business')}
- Networking focus for {professional_context.get('company_size', 'all company sizes')}
- Thought leadership positioning as {professional_context.get('professional_role', 'professional')}
3. ENGAGEMENT OPTIMIZATION:
- Professional question frameworks for discussion
- Industry-relevant polling strategies
- Professional networking call-to-actions
- Thought leadership positioning
- Community building through professional value
4. LINKEDIN-SPECIFIC FEATURES:
- Native video optimization for professional content
- LinkedIn Articles for long-form thought leadership
- LinkedIn Polls for industry insights and engagement
- LinkedIn Events for professional networking
- LinkedIn Carousels for educational content
- LinkedIn Live for professional discussions
5. PROFESSIONAL NETWORKING ELEMENTS:
- Industry-specific hashtag strategy
- Professional mention and tagging etiquette
- Thought leadership positioning
- Professional relationship building
- Career advancement focus
6. CONTENT FORMAT OPTIMIZATION:
- Hook strategies for professional feed
- "See More" optimization for longer posts
- Professional call-to-action frameworks
- Industry-specific content structures
- Professional storytelling techniques
7. LINKEDIN ALGORITHM OPTIMIZATION:
- Professional engagement patterns
- Industry-relevant content timing
- Professional network interaction strategies
- Thought leadership content performance
- Professional community building
8. INDUSTRY-SPECIFIC ADAPTATIONS FOR {industry_focus.upper()}:
- Terminology appropriate for {expertise_level} level
- Professional development for {professional_context.get('target_demographics', [])}
- Trend discussions for {professional_context.get('business_model', 'general business')}
- Networking strategies for {professional_context.get('company_size', 'all company sizes')}
- Thought leadership as {professional_context.get('professional_role', 'professional')}
- Content addressing {professional_context.get('psychographic_profile', 'professional needs')}
- Business insights for {professional_context.get('conversion_focus', 'business growth')}
PROFESSIONAL EXCELLENCE STANDARDS:
- Maintain high professional standards
- Focus on value-driven content
- Emphasize thought leadership and expertise
- Build professional credibility and authority
- Foster meaningful professional relationships
- Provide actionable business insights
- Support professional development and growth
Generate a comprehensive LinkedIn-optimized persona that positions the user as a thought leader in {industry_focus} while maintaining professional excellence and maximizing LinkedIn's professional networking potential.
"""
return prompt
@staticmethod
def get_linkedin_platform_constraints() -> Dict[str, Any]:
"""Get LinkedIn-specific platform constraints and best practices."""
return {
"character_limit": 3000,
"optimal_length": "150-300 words",
"professional_tone": True,
"hashtag_limit": 5,
"rich_media": True,
"long_form": True,
"thought_leadership": True,
"networking_focus": True,
"career_development": True,
"industry_insights": True,
"professional_storytelling": True,
"b2b_optimized": True,
"algorithm_engagement": True,
"native_video": True,
"linkedin_articles": True,
"linkedin_polls": True,
"linkedin_events": True,
"linkedin_carousels": True,
"linkedin_live": True
}
@staticmethod
def _extract_professional_context(onboarding_data: Dict[str, Any]) -> Dict[str, Any]:
"""Extract comprehensive professional context from onboarding data."""
professional_context = {
"industry_focus": "general",
"expertise_level": "intermediate",
"company_size": "Not specified",
"business_model": "Not specified",
"professional_role": "Not specified",
"geographic_focus": "global",
"target_demographics": [],
"psychographic_profile": "",
"content_purpose": "",
"conversion_focus": "",
"research_depth": "",
"content_types": []
}
# Extract from website analysis
website_analysis = onboarding_data.get("website_analysis", {}) or {}
# Target audience information
target_audience = website_analysis.get("target_audience", {})
if target_audience:
professional_context["industry_focus"] = target_audience.get("industry_focus", "general")
professional_context["expertise_level"] = target_audience.get("expertise_level", "intermediate")
professional_context["geographic_focus"] = target_audience.get("geographic_focus", "global")
professional_context["target_demographics"] = target_audience.get("demographics", [])
professional_context["psychographic_profile"] = target_audience.get("psychographic_profile", "")
# Content type and business context
content_type = website_analysis.get("content_type", {})
if content_type:
professional_context["content_purpose"] = content_type.get("purpose", "")
professional_context["conversion_focus"] = content_type.get("conversion_focus", "")
# Company and business information from crawl results
crawl_result = website_analysis.get("crawl_result", {})
if crawl_result:
domain_info = crawl_result.get("domain_info", {})
if domain_info:
professional_context["company_size"] = domain_info.get("company_size", "Not specified")
professional_context["business_model"] = domain_info.get("business_model", "Not specified")
brand_info = crawl_result.get("brand_info", {})
if brand_info:
professional_context["professional_role"] = brand_info.get("professional_role", "Not specified")
# Research preferences
research_prefs = onboarding_data.get("research_preferences", {})
if research_prefs:
professional_context["research_depth"] = research_prefs.get("research_depth", "")
professional_context["content_types"] = research_prefs.get("content_types", [])
# Enhanced analysis data
enhanced_analysis = onboarding_data.get("enhanced_analysis", {})
if enhanced_analysis:
audience_intel = enhanced_analysis.get("audience_intelligence", {})
if audience_intel:
# Override with more detailed information if available
if audience_intel.get("industry_focus"):
professional_context["industry_focus"] = audience_intel["industry_focus"]
if audience_intel.get("expertise_level"):
professional_context["expertise_level"] = audience_intel["expertise_level"]
if audience_intel.get("psychographic_profile"):
professional_context["psychographic_profile"] = audience_intel["psychographic_profile"]
brand_voice = enhanced_analysis.get("brand_voice_analysis", {})
if brand_voice:
if brand_voice.get("primary_content_type"):
professional_context["content_purpose"] = brand_voice["primary_content_type"]
if brand_voice.get("conversion_focus"):
professional_context["conversion_focus"] = brand_voice["conversion_focus"]
return professional_context
@staticmethod
def build_linkedin_system_prompt(core_persona: Dict[str, Any]) -> str:
"""
Build system prompt with core persona for LinkedIn generation.
This moves the core persona to system prompt to free up context window.
"""
import json
return f"""You are an expert LinkedIn content strategist and professional networking specialist.
CORE PERSONA FOUNDATION:
{json.dumps(core_persona, indent=2)}
Your task is to create LinkedIn-optimized persona adaptations that maintain the core persona's identity while optimizing for professional networking, thought leadership, and B2B engagement on LinkedIn.
Focus on:
- Professional tone and authority
- Industry-specific optimization
- LinkedIn algorithm best practices
- B2B engagement strategies
- Professional networking optimization"""
@staticmethod
def build_focused_linkedin_prompt(onboarding_data: Dict[str, Any]) -> str:
"""
Build focused LinkedIn prompt without core persona JSON to optimize context usage.
"""
# Extract professional context
professional_context = LinkedInPersonaPrompts._extract_professional_context(onboarding_data)
industry_focus = professional_context.get("industry_focus", "general")
expertise_level = professional_context.get("expertise_level", "intermediate")
return f"""LINKEDIN PROFESSIONAL OPTIMIZATION TASK: Create LinkedIn-specific adaptations for the core persona.
PROFESSIONAL CONTEXT:
- Industry: {industry_focus}
- Expertise Level: {expertise_level}
- Company Size: {professional_context.get('company_size', 'Not specified')}
- Business Model: {professional_context.get('business_model', 'Not specified')}
- Professional Role: {professional_context.get('professional_role', 'Not specified')}
- Demographics: {professional_context.get('target_demographics', [])}
- Psychographic: {professional_context.get('psychographic_profile', 'Not specified')}
LINKEDIN PLATFORM SPECIFICATIONS:
- Character Limit: 3,000 characters
- Optimal Post Length: 150-300 words for maximum engagement
- Professional Network: B2B focused, career-oriented audience
- Algorithm Priority: Engagement, relevance, professional value
- Content Types: Posts, Articles, Polls, Videos, Carousels, Events
- Hashtag Limit: 3-5 hashtags for optimal reach
- Link Strategy: Place external links in first comment for algorithm optimization
LINKEDIN OPTIMIZATION REQUIREMENTS:
1. PROFESSIONAL TONE & VOICE:
- Maintain authoritative yet approachable professional tone
- Use industry-specific terminology appropriately
- Balance expertise with accessibility for {expertise_level} audience
- Incorporate thought leadership elements
- Include professional storytelling and case studies
2. CONTENT STRATEGY FOR {industry_focus.upper()}:
- Industry insights for {expertise_level} professionals
- Professional development content for {professional_context.get('target_demographics', [])}
- Business strategy discussions for {professional_context.get('business_model', 'general business')}
- Networking focus for {professional_context.get('company_size', 'all company sizes')}
- Thought leadership positioning as {professional_context.get('professional_role', 'professional')}
3. LINKEDIN-SPECIFIC ADAPTATIONS:
- Optimize sentence structure for professional readability
- Create platform-specific vocabulary and terminology
- Define engagement patterns for B2B audience
- Establish professional networking strategies
- Include LinkedIn feature optimization (Articles, Polls, Events, etc.)
4. ALGORITHM OPTIMIZATION:
- Engagement patterns for professional audience
- Content timing for maximum reach
- Professional value metrics
- Network interaction strategies
5. PROFESSIONAL CONTEXT OPTIMIZATION:
- Industry-specific positioning
- Expertise level adaptation
- Company size considerations
- Business model alignment
- Professional role authority
- Demographic targeting
- Psychographic engagement
- Conversion optimization
Generate a comprehensive LinkedIn-optimized persona that maintains the core persona's identity while maximizing professional networking and thought leadership potential on LinkedIn."""

View File

@@ -0,0 +1,115 @@
"""
LinkedIn Persona Schemas
Contains LinkedIn-specific JSON schemas for persona generation.
"""
from typing import Dict, Any
class LinkedInPersonaSchemas:
"""Handles LinkedIn-specific persona schema definitions."""
@staticmethod
def get_linkedin_platform_schema() -> Dict[str, Any]:
"""Get LinkedIn-specific platform persona schema."""
return {
"type": "object",
"properties": {
"platform_type": {"type": "string"},
"sentence_metrics": {
"type": "object",
"properties": {
"max_sentence_length": {"type": "number"},
"optimal_sentence_length": {"type": "number"},
"sentence_variety": {"type": "string"}
}
},
"lexical_adaptations": {
"type": "object",
"properties": {
"platform_specific_words": {"type": "array", "items": {"type": "string"}},
"hashtag_strategy": {"type": "string"},
"emoji_usage": {"type": "string"},
"mention_strategy": {"type": "string"}
}
},
"content_format_rules": {
"type": "object",
"properties": {
"character_limit": {"type": "number"},
"paragraph_structure": {"type": "string"},
"call_to_action_style": {"type": "string"},
"link_placement": {"type": "string"}
}
},
"engagement_patterns": {
"type": "object",
"properties": {
"posting_frequency": {"type": "string"},
"optimal_posting_times": {"type": "array", "items": {"type": "string"}},
"engagement_tactics": {"type": "array", "items": {"type": "string"}},
"community_interaction": {"type": "string"}
}
},
"platform_best_practices": {
"type": "array",
"items": {"type": "string"}
}
},
"required": ["platform_type", "sentence_metrics", "content_format_rules", "engagement_patterns"]
}
@staticmethod
def get_enhanced_linkedin_schema() -> Dict[str, Any]:
"""Get enhanced LinkedIn schema with additional professional fields."""
base_schema = LinkedInPersonaSchemas.get_linkedin_platform_schema()
# Add LinkedIn-specific professional fields
base_schema["properties"]["professional_networking"] = {
"type": "object",
"properties": {
"thought_leadership_positioning": {"type": "string"},
"industry_authority_building": {"type": "string"},
"professional_relationship_strategies": {"type": "array", "items": {"type": "string"}},
"career_advancement_focus": {"type": "string"}
}
}
base_schema["properties"]["linkedin_features"] = {
"type": "object",
"properties": {
"articles_strategy": {"type": "string"},
"polls_optimization": {"type": "string"},
"events_networking": {"type": "string"},
"carousels_education": {"type": "string"},
"live_discussions": {"type": "string"},
"native_video": {"type": "string"}
}
}
base_schema["properties"]["algorithm_optimization"] = {
"type": "object",
"properties": {
"engagement_patterns": {"type": "array", "items": {"type": "string"}},
"content_timing": {"type": "array", "items": {"type": "string"}},
"professional_value_metrics": {"type": "array", "items": {"type": "string"}},
"network_interaction_strategies": {"type": "array", "items": {"type": "string"}}
}
}
# Add professional context optimization
base_schema["properties"]["professional_context_optimization"] = {
"type": "object",
"properties": {
"industry_specific_positioning": {"type": "string"},
"expertise_level_adaptation": {"type": "string"},
"company_size_considerations": {"type": "string"},
"business_model_alignment": {"type": "string"},
"professional_role_authority": {"type": "string"},
"demographic_targeting": {"type": "array", "items": {"type": "string"}},
"psychographic_engagement": {"type": "string"},
"conversion_optimization": {"type": "string"}
}
}
return base_schema

View File

@@ -0,0 +1,539 @@
"""
LinkedIn Persona Service
Handles LinkedIn-specific persona generation and optimization.
"""
from typing import Dict, Any, Optional
from loguru import logger
from services.llm_providers.gemini_provider import gemini_structured_json_response
from .linkedin_persona_prompts import LinkedInPersonaPrompts
from .linkedin_persona_schemas import LinkedInPersonaSchemas
class LinkedInPersonaService:
"""Service for generating LinkedIn-specific persona adaptations."""
def __init__(self):
"""Initialize the LinkedIn persona service."""
self.prompts = LinkedInPersonaPrompts()
self.schemas = LinkedInPersonaSchemas()
logger.info("LinkedInPersonaService initialized")
def generate_linkedin_persona(self, core_persona: Dict[str, Any], onboarding_data: Dict[str, Any]) -> Dict[str, Any]:
"""
Generate LinkedIn-specific persona adaptation using optimized chained prompts.
Args:
core_persona: The core writing persona
onboarding_data: User's onboarding data
Returns:
LinkedIn-optimized persona data
"""
try:
logger.info("Generating LinkedIn-specific persona with optimized prompts")
# Build focused LinkedIn prompt (without core persona JSON)
prompt = self.prompts.build_focused_linkedin_prompt(onboarding_data)
# Create system prompt with core persona
system_prompt = self.prompts.build_linkedin_system_prompt(core_persona)
# Get LinkedIn-specific schema
schema = self.schemas.get_enhanced_linkedin_schema()
# Generate structured response using Gemini with optimized prompts
response = gemini_structured_json_response(
prompt=prompt,
schema=schema,
temperature=0.2,
max_tokens=4096,
system_prompt=system_prompt
)
if "error" in response:
logger.error(f"LinkedIn persona generation failed: {response['error']}")
return {"error": f"LinkedIn persona generation failed: {response['error']}"}
# Validate the generated persona
validation_results = self.validate_linkedin_persona(response)
logger.info(f"LinkedIn persona validation: Quality Score: {validation_results['quality_score']:.1f}%, Valid: {validation_results['is_valid']}")
# Add validation results to persona data
response["validation_results"] = validation_results
# Apply comprehensive algorithm optimization
optimized_response = self.optimize_for_linkedin_algorithm(response)
logger.info("✅ LinkedIn persona algorithm optimization applied")
logger.info("✅ LinkedIn persona generated and optimized successfully")
return optimized_response
except Exception as e:
logger.error(f"Error generating LinkedIn persona: {str(e)}")
return {"error": f"Failed to generate LinkedIn persona: {str(e)}"}
def get_linkedin_constraints(self) -> Dict[str, Any]:
"""Get LinkedIn platform constraints."""
return self.prompts.get_linkedin_platform_constraints()
def validate_linkedin_persona(self, persona_data: Dict[str, Any]) -> Dict[str, Any]:
"""
Comprehensive validation of LinkedIn persona data for completeness and quality.
Args:
persona_data: LinkedIn persona data to validate
Returns:
Detailed validation results with quality metrics and recommendations
"""
try:
validation_results = {
"is_valid": True,
"quality_score": 0.0,
"completeness_score": 0.0,
"professional_context_score": 0.0,
"linkedin_optimization_score": 0.0,
"missing_fields": [],
"incomplete_fields": [],
"recommendations": [],
"quality_issues": [],
"strengths": [],
"validation_details": {}
}
# 1. CORE FIELDS VALIDATION (30% of score)
core_fields_score = self._validate_core_fields(persona_data, validation_results)
# 2. LINKEDIN-SPECIFIC FIELDS VALIDATION (40% of score)
linkedin_fields_score = self._validate_linkedin_specific_fields(persona_data, validation_results)
# 3. PROFESSIONAL CONTEXT VALIDATION (20% of score)
professional_context_score = self._validate_professional_context(persona_data, validation_results)
# 4. CONTENT QUALITY VALIDATION (10% of score)
content_quality_score = self._validate_content_quality(persona_data, validation_results)
# Calculate overall quality score
validation_results["quality_score"] = (
core_fields_score * 0.3 +
linkedin_fields_score * 0.4 +
professional_context_score * 0.2 +
content_quality_score * 0.1
)
# Set completeness score
validation_results["completeness_score"] = core_fields_score
validation_results["professional_context_score"] = professional_context_score
validation_results["linkedin_optimization_score"] = linkedin_fields_score
# Determine if persona is valid
validation_results["is_valid"] = (
validation_results["quality_score"] >= 70.0 and
len(validation_results["missing_fields"]) == 0
)
# Add quality assessment
self._assess_persona_quality(validation_results)
return validation_results
except Exception as e:
logger.error(f"Error validating LinkedIn persona: {str(e)}")
return {
"is_valid": False,
"quality_score": 0.0,
"error": str(e)
}
def _validate_core_fields(self, persona_data: Dict[str, Any], validation_results: Dict[str, Any]) -> float:
"""Validate core LinkedIn persona fields."""
core_fields = {
"platform_type": {"required": True, "type": str},
"sentence_metrics": {"required": True, "type": dict, "subfields": ["max_sentence_length", "optimal_sentence_length"]},
"lexical_adaptations": {"required": True, "type": dict, "subfields": ["platform_specific_words", "hashtag_strategy"]},
"content_format_rules": {"required": True, "type": dict, "subfields": ["character_limit", "paragraph_structure"]},
"engagement_patterns": {"required": True, "type": dict, "subfields": ["posting_frequency", "optimal_posting_times"]},
"platform_best_practices": {"required": True, "type": list}
}
score = 0.0
total_fields = len(core_fields)
for field, config in core_fields.items():
if field not in persona_data:
validation_results["missing_fields"].append(field)
continue
field_data = persona_data[field]
field_score = 0.0
# Check field type
if isinstance(field_data, config["type"]):
field_score += 0.5
else:
validation_results["quality_issues"].append(f"{field} has incorrect type: expected {config['type'].__name__}")
# Check subfields if specified
if "subfields" in config and isinstance(field_data, dict):
subfield_score = 0.0
for subfield in config["subfields"]:
if subfield in field_data and field_data[subfield]:
subfield_score += 1.0
else:
validation_results["incomplete_fields"].append(f"{field}.{subfield}")
if config["subfields"]:
field_score += (subfield_score / len(config["subfields"])) * 0.5
score += field_score
validation_results["validation_details"][field] = {
"present": True,
"type_correct": isinstance(field_data, config["type"]),
"completeness": field_score
}
return (score / total_fields) * 100
def _validate_linkedin_specific_fields(self, persona_data: Dict[str, Any], validation_results: Dict[str, Any]) -> float:
"""Validate LinkedIn-specific optimization fields."""
linkedin_fields = {
"professional_networking": {
"required": True,
"subfields": ["thought_leadership_positioning", "industry_authority_building", "professional_relationship_strategies"]
},
"linkedin_features": {
"required": True,
"subfields": ["articles_strategy", "polls_optimization", "events_networking", "carousels_education"]
},
"algorithm_optimization": {
"required": True,
"subfields": ["engagement_patterns", "content_timing", "professional_value_metrics"]
},
"professional_context_optimization": {
"required": True,
"subfields": ["industry_specific_positioning", "expertise_level_adaptation", "demographic_targeting"]
}
}
score = 0.0
total_fields = len(linkedin_fields)
for field, config in linkedin_fields.items():
if field not in persona_data:
validation_results["missing_fields"].append(field)
validation_results["recommendations"].append(f"Add {field} for enhanced LinkedIn optimization")
continue
field_data = persona_data[field]
if not isinstance(field_data, dict):
validation_results["quality_issues"].append(f"{field} should be a dictionary")
continue
field_score = 0.0
for subfield in config["subfields"]:
if subfield in field_data and field_data[subfield]:
field_score += 1.0
else:
validation_results["incomplete_fields"].append(f"{field}.{subfield}")
field_score = (field_score / len(config["subfields"])) * 100
score += field_score
validation_results["validation_details"][field] = {
"present": True,
"completeness": field_score,
"subfields_present": len([sf for sf in config["subfields"] if sf in field_data and field_data[sf]])
}
return score / total_fields
def _validate_professional_context(self, persona_data: Dict[str, Any], validation_results: Dict[str, Any]) -> float:
"""Validate professional context optimization."""
if "professional_context_optimization" not in persona_data:
validation_results["missing_fields"].append("professional_context_optimization")
return 0.0
context_data = persona_data["professional_context_optimization"]
if not isinstance(context_data, dict):
validation_results["quality_issues"].append("professional_context_optimization should be a dictionary")
return 0.0
professional_fields = [
"industry_specific_positioning",
"expertise_level_adaptation",
"company_size_considerations",
"business_model_alignment",
"professional_role_authority",
"demographic_targeting",
"psychographic_engagement",
"conversion_optimization"
]
score = 0.0
for field in professional_fields:
if field in context_data and context_data[field]:
score += 1.0
# Check for meaningful content (not just placeholder text)
if isinstance(context_data[field], str) and len(context_data[field]) > 50:
score += 0.5
else:
validation_results["incomplete_fields"].append(f"professional_context_optimization.{field}")
return (score / len(professional_fields)) * 100
def _validate_content_quality(self, persona_data: Dict[str, Any], validation_results: Dict[str, Any]) -> float:
"""Validate content quality and depth."""
score = 0.0
# Check for meaningful content in key fields
quality_checks = [
("sentence_metrics", "optimal_sentence_length"),
("lexical_adaptations", "platform_specific_words"),
("professional_networking", "thought_leadership_positioning"),
("linkedin_features", "articles_strategy")
]
for field, subfield in quality_checks:
if field in persona_data and subfield in persona_data[field]:
content = persona_data[field][subfield]
if isinstance(content, str) and len(content) > 30:
score += 1.0
elif isinstance(content, list) and len(content) > 3:
score += 1.0
else:
validation_results["quality_issues"].append(f"{field}.{subfield} content too brief")
else:
validation_results["quality_issues"].append(f"{field}.{subfield} missing or empty")
return (score / len(quality_checks)) * 100
def _assess_persona_quality(self, validation_results: Dict[str, Any]) -> None:
"""Assess overall persona quality and provide recommendations."""
quality_score = validation_results["quality_score"]
if quality_score >= 90:
validation_results["strengths"].append("Excellent LinkedIn persona with comprehensive optimization")
elif quality_score >= 80:
validation_results["strengths"].append("Strong LinkedIn persona with good optimization")
elif quality_score >= 70:
validation_results["strengths"].append("Good LinkedIn persona with basic optimization")
else:
validation_results["quality_issues"].append("LinkedIn persona needs significant improvement")
# Add specific recommendations based on missing fields
if "professional_context_optimization" in validation_results["missing_fields"]:
validation_results["recommendations"].append("Add professional context optimization for industry-specific positioning")
if "algorithm_optimization" in validation_results["missing_fields"]:
validation_results["recommendations"].append("Add algorithm optimization for better LinkedIn reach")
if validation_results["incomplete_fields"]:
validation_results["recommendations"].append(f"Complete {len(validation_results['incomplete_fields'])} incomplete fields for better optimization")
# Add enterprise-grade recommendations
if quality_score >= 80:
validation_results["recommendations"].append("Persona is enterprise-ready for professional LinkedIn content")
else:
validation_results["recommendations"].append("Consider regenerating persona with more comprehensive data")
def optimize_for_linkedin_algorithm(self, persona_data: Dict[str, Any]) -> Dict[str, Any]:
"""
Comprehensive LinkedIn algorithm optimization for maximum reach and engagement.
Args:
persona_data: LinkedIn persona data to optimize
Returns:
Algorithm-optimized persona data with advanced optimization features
"""
try:
optimized_persona = persona_data.copy()
# Initialize algorithm optimization if not present
if "algorithm_optimization" not in optimized_persona:
optimized_persona["algorithm_optimization"] = {}
# 1. CONTENT QUALITY OPTIMIZATION
optimized_persona["algorithm_optimization"]["content_quality_optimization"] = {
"original_insights_priority": [
"Share proprietary industry insights and case studies",
"Publish data-driven analyses and research findings",
"Create thought leadership content with unique perspectives",
"Avoid generic or recycled content that lacks value"
],
"professional_credibility_boost": [
"Include relevant credentials and expertise indicators",
"Reference industry experience and achievements",
"Use professional language and terminology appropriately",
"Maintain consistent brand voice and messaging"
],
"content_depth_requirements": [
"Provide actionable insights and practical advice",
"Include specific examples and real-world applications",
"Offer comprehensive analysis rather than surface-level content",
"Create content that solves professional problems"
]
}
# 2. MULTIMEDIA FORMAT OPTIMIZATION
optimized_persona["algorithm_optimization"]["multimedia_strategy"] = {
"native_video_optimization": [
"Upload videos directly to LinkedIn for maximum reach",
"Keep videos 1-3 minutes for optimal engagement",
"Include captions for accessibility and broader reach",
"Start with compelling hooks to retain viewers"
],
"carousel_document_strategy": [
"Create swipeable educational content and tutorials",
"Use 5-10 slides for optimal engagement",
"Include clear, scannable text and visuals",
"End with strong call-to-action"
],
"visual_content_optimization": [
"Use high-quality, professional images and graphics",
"Create infographics that convey complex information simply",
"Design visually appealing quote cards and statistics",
"Ensure all visuals align with professional brand"
]
}
# 3. ENGAGEMENT OPTIMIZATION
optimized_persona["algorithm_optimization"]["engagement_optimization"] = {
"comment_encouragement_strategies": [
"Ask thought-provoking questions that invite discussion",
"Pose industry-specific challenges or scenarios",
"Request personal experiences and insights",
"Create polls and surveys for interactive engagement"
],
"network_interaction_boost": [
"Respond to comments within 2-4 hours for maximum visibility",
"Engage meaningfully with others' content before posting",
"Share and comment on industry leaders' posts",
"Participate actively in relevant LinkedIn groups"
],
"professional_relationship_building": [
"Tag relevant connections when appropriate",
"Mention industry experts and thought leaders",
"Collaborate with peers on joint content",
"Build genuine professional relationships"
]
}
# 4. TIMING AND FREQUENCY OPTIMIZATION
optimized_persona["algorithm_optimization"]["timing_optimization"] = {
"optimal_posting_schedule": [
"Tuesday-Thursday: 8-11 AM EST for maximum professional engagement",
"Wednesday: Peak day for B2B content and thought leadership",
"Avoid posting on weekends unless targeting specific audiences",
"Maintain consistent posting schedule for algorithm recognition"
],
"frequency_optimization": [
"Post 3-5 times per week for consistent visibility",
"Balance original content with curated industry insights",
"Space posts 4-6 hours apart to avoid audience fatigue",
"Monitor engagement rates to adjust frequency"
],
"timezone_considerations": [
"Consider global audience time zones for international reach",
"Adjust posting times based on target audience location",
"Use LinkedIn Analytics to identify peak engagement times",
"Test different time slots to optimize reach"
]
}
# 5. HASHTAG AND DISCOVERABILITY OPTIMIZATION
optimized_persona["algorithm_optimization"]["discoverability_optimization"] = {
"strategic_hashtag_usage": [
"Use 3-5 relevant hashtags for optimal reach",
"Mix broad industry hashtags with niche-specific tags",
"Include trending hashtags when relevant to content",
"Create branded hashtags for consistent brand recognition"
],
"keyword_optimization": [
"Include industry-specific keywords naturally in content",
"Use professional terminology that resonates with target audience",
"Optimize for LinkedIn's search algorithm",
"Include location-based keywords for local reach"
],
"content_categorization": [
"Tag content appropriately for LinkedIn's content categorization",
"Use consistent themes and topics for algorithm recognition",
"Create content series for sustained engagement",
"Leverage LinkedIn's content suggestions and trending topics"
]
}
# 6. LINKEDIN FEATURES OPTIMIZATION
optimized_persona["algorithm_optimization"]["linkedin_features_optimization"] = {
"articles_strategy": [
"Publish long-form articles for thought leadership positioning",
"Use compelling headlines that encourage clicks",
"Include relevant images and formatting for readability",
"Cross-promote articles in regular posts"
],
"polls_and_surveys": [
"Create engaging polls to drive interaction",
"Ask industry-relevant questions that spark discussion",
"Use poll results to create follow-up content",
"Share poll insights to provide value to audience"
],
"events_and_networking": [
"Host or participate in LinkedIn events and webinars",
"Use LinkedIn's event features for promotion and networking",
"Create virtual networking opportunities",
"Leverage LinkedIn Live for real-time engagement"
]
}
# 7. PERFORMANCE MONITORING AND OPTIMIZATION
optimized_persona["algorithm_optimization"]["performance_monitoring"] = {
"key_metrics_tracking": [
"Monitor engagement rate (likes, comments, shares, saves)",
"Track reach and impression metrics",
"Analyze click-through rates on links and CTAs",
"Measure follower growth and network expansion"
],
"content_performance_analysis": [
"Identify top-performing content types and topics",
"Analyze posting times for optimal engagement",
"Track hashtag performance and reach",
"Monitor audience demographics and interests"
],
"optimization_recommendations": [
"A/B test different content formats and styles",
"Experiment with posting frequencies and timing",
"Test various hashtag combinations and strategies",
"Continuously refine content based on performance data"
]
}
# 8. PROFESSIONAL CONTEXT OPTIMIZATION
optimized_persona["algorithm_optimization"]["professional_context_optimization"] = {
"industry_specific_optimization": [
"Tailor content to industry-specific trends and challenges",
"Use industry terminology and references appropriately",
"Address current industry issues and developments",
"Position as thought leader within specific industry"
],
"career_stage_targeting": [
"Create content relevant to different career stages",
"Address professional development and growth topics",
"Share career insights and advancement strategies",
"Provide value to both junior and senior professionals"
],
"company_size_considerations": [
"Adapt content for different company sizes and structures",
"Address challenges specific to startups, SMBs, and enterprises",
"Provide relevant insights for different organizational contexts",
"Consider decision-making processes and hierarchies"
]
}
logger.info("✅ LinkedIn persona comprehensively optimized for 2024 algorithm performance")
return optimized_persona
except Exception as e:
logger.error(f"Error optimizing LinkedIn persona for algorithm: {str(e)}")
return persona_data

View File

@@ -12,13 +12,19 @@ import json
from services.database import get_db_session
from models.onboarding import OnboardingSession, WebsiteAnalysis, ResearchPreferences
from models.persona_models import WritingPersona, PlatformPersona, PersonaAnalysisResult
from services.llm_providers.gemini_provider import gemini_structured_json_response
from services.persona.core_persona import CorePersonaService, OnboardingDataCollector
from services.persona.linkedin.linkedin_persona_service import LinkedInPersonaService
from services.persona.facebook.facebook_persona_service import FacebookPersonaService
class PersonaAnalysisService:
"""Service for analyzing onboarding data and generating writing personas using Gemini AI."""
def __init__(self):
"""Initialize the persona analysis service."""
self.core_persona_service = CorePersonaService()
self.data_collector = OnboardingDataCollector()
self.linkedin_service = LinkedInPersonaService()
self.facebook_service = FacebookPersonaService()
logger.info("PersonaAnalysisService initialized")
def generate_persona_from_onboarding(self, user_id: int, onboarding_session_id: int = None) -> Dict[str, Any]:
@@ -36,20 +42,20 @@ class PersonaAnalysisService:
logger.info(f"Generating persona for user {user_id}")
# Get onboarding data
onboarding_data = self._collect_onboarding_data(user_id, onboarding_session_id)
onboarding_data = self.data_collector.collect_onboarding_data(user_id, onboarding_session_id)
if not onboarding_data:
logger.warning(f"No onboarding data found for user {user_id}")
return {"error": "No onboarding data available for persona generation"}
# Generate core persona using Gemini
core_persona = self._generate_core_persona(onboarding_data)
core_persona = self.core_persona_service.generate_core_persona(onboarding_data)
if "error" in core_persona:
return core_persona
# Generate platform-specific adaptations
platform_personas = self._generate_platform_adaptations(core_persona, onboarding_data)
platform_personas = self.core_persona_service.generate_platform_adaptations(core_persona, onboarding_data)
# Save to database
saved_persona = self._save_persona_to_db(user_id, core_persona, platform_personas, onboarding_data)
@@ -60,7 +66,7 @@ class PersonaAnalysisService:
"platform_personas": platform_personas,
"analysis_metadata": {
"confidence_score": core_persona.get("confidence_score", 0.0),
"data_sufficiency": self._calculate_data_sufficiency(onboarding_data),
"data_sufficiency": self.data_collector.calculate_data_sufficiency(onboarding_data),
"generated_at": datetime.utcnow().isoformat()
}
}
@@ -69,318 +75,114 @@ class PersonaAnalysisService:
logger.error(f"Error generating persona for user {user_id}: {str(e)}")
return {"error": f"Failed to generate persona: {str(e)}"}
def _collect_onboarding_data(self, user_id: int, session_id: int = None) -> Optional[Dict[str, Any]]:
"""Collect comprehensive onboarding data for persona analysis."""
try:
session = get_db_session()
# Find onboarding session
if session_id:
onboarding_session = session.query(OnboardingSession).filter(
OnboardingSession.id == session_id,
OnboardingSession.user_id == user_id
).first()
else:
onboarding_session = session.query(OnboardingSession).filter(
OnboardingSession.user_id == user_id
).order_by(OnboardingSession.updated_at.desc()).first()
if not onboarding_session:
return None
# Get website analysis
website_analysis = session.query(WebsiteAnalysis).filter(
WebsiteAnalysis.session_id == onboarding_session.id
).first()
# Get research preferences
research_prefs = session.query(ResearchPreferences).filter(
ResearchPreferences.session_id == onboarding_session.id
).first()
# Compile comprehensive data
onboarding_data = {
"session_info": {
"session_id": onboarding_session.id,
"current_step": onboarding_session.current_step,
"progress": onboarding_session.progress,
"started_at": onboarding_session.started_at.isoformat() if onboarding_session.started_at else None
},
"website_analysis": website_analysis.to_dict() if website_analysis else None,
"research_preferences": research_prefs.to_dict() if research_prefs else None
}
session.close()
return onboarding_data
except Exception as e:
logger.error(f"Error collecting onboarding data: {str(e)}")
return None
def _generate_core_persona(self, onboarding_data: Dict[str, Any]) -> Dict[str, Any]:
"""Generate core writing persona using Gemini structured response."""
# Build analysis prompt
prompt = self._build_persona_analysis_prompt(onboarding_data)
# Define schema for structured response
persona_schema = {
"type": "object",
"properties": {
"identity": {
"type": "object",
"properties": {
"persona_name": {"type": "string"},
"archetype": {"type": "string"},
"core_belief": {"type": "string"},
"brand_voice_description": {"type": "string"}
},
"required": ["persona_name", "archetype", "core_belief"]
},
"linguistic_fingerprint": {
"type": "object",
"properties": {
"sentence_metrics": {
"type": "object",
"properties": {
"average_sentence_length_words": {"type": "number"},
"preferred_sentence_type": {"type": "string"},
"active_to_passive_ratio": {"type": "string"},
"complexity_level": {"type": "string"}
}
},
"lexical_features": {
"type": "object",
"properties": {
"go_to_words": {"type": "array", "items": {"type": "string"}},
"go_to_phrases": {"type": "array", "items": {"type": "string"}},
"avoid_words": {"type": "array", "items": {"type": "string"}},
"contractions": {"type": "string"},
"filler_words": {"type": "string"},
"vocabulary_level": {"type": "string"}
}
},
"rhetorical_devices": {
"type": "object",
"properties": {
"metaphors": {"type": "string"},
"analogies": {"type": "string"},
"rhetorical_questions": {"type": "string"},
"storytelling_style": {"type": "string"}
}
}
}
},
"tonal_range": {
"type": "object",
"properties": {
"default_tone": {"type": "string"},
"permissible_tones": {"type": "array", "items": {"type": "string"}},
"forbidden_tones": {"type": "array", "items": {"type": "string"}},
"emotional_range": {"type": "string"}
}
},
"stylistic_constraints": {
"type": "object",
"properties": {
"punctuation": {
"type": "object",
"properties": {
"ellipses": {"type": "string"},
"em_dash": {"type": "string"},
"exclamation_points": {"type": "string"}
}
},
"formatting": {
"type": "object",
"properties": {
"paragraphs": {"type": "string"},
"lists": {"type": "string"},
"markdown": {"type": "string"}
}
}
}
},
"confidence_score": {"type": "number"},
"analysis_notes": {"type": "string"}
},
"required": ["identity", "linguistic_fingerprint", "tonal_range", "confidence_score"]
}
try:
# Generate structured response using Gemini
response = gemini_structured_json_response(
prompt=prompt,
schema=persona_schema,
temperature=0.2, # Low temperature for consistent analysis
max_tokens=8192,
system_prompt="You are an expert writing style analyst and persona developer. Analyze the provided data to create a precise, actionable writing persona."
)
if "error" in response:
logger.error(f"Gemini API error: {response['error']}")
return {"error": f"AI analysis failed: {response['error']}"}
logger.info("✅ Core persona generated successfully")
return response
except Exception as e:
logger.error(f"Error generating core persona: {str(e)}")
return {"error": f"Failed to generate core persona: {str(e)}"}
def _generate_platform_adaptations(self, core_persona: Dict[str, Any], onboarding_data: Dict[str, Any]) -> Dict[str, Any]:
"""Generate platform-specific persona adaptations."""
platforms = ["twitter", "linkedin", "instagram", "facebook", "blog", "medium", "substack"]
platform_personas = {}
for platform in platforms:
try:
platform_persona = self._generate_single_platform_persona(core_persona, platform, onboarding_data)
if "error" not in platform_persona:
platform_personas[platform] = platform_persona
else:
logger.warning(f"Failed to generate {platform} persona: {platform_persona['error']}")
except Exception as e:
logger.error(f"Error generating {platform} persona: {str(e)}")
return platform_personas
def _generate_single_platform_persona(self, core_persona: Dict[str, Any], platform: str, onboarding_data: Dict[str, Any]) -> Dict[str, Any]:
"""Generate persona adaptation for a specific platform."""
prompt = self._build_platform_adaptation_prompt(core_persona, platform, onboarding_data)
# Platform-specific schema
platform_schema = {
"type": "object",
"properties": {
"platform_type": {"type": "string"},
"sentence_metrics": {
"type": "object",
"properties": {
"max_sentence_length": {"type": "number"},
"optimal_sentence_length": {"type": "number"},
"sentence_variety": {"type": "string"}
}
},
"lexical_adaptations": {
"type": "object",
"properties": {
"platform_specific_words": {"type": "array", "items": {"type": "string"}},
"hashtag_strategy": {"type": "string"},
"emoji_usage": {"type": "string"},
"mention_strategy": {"type": "string"}
}
},
"content_format_rules": {
"type": "object",
"properties": {
"character_limit": {"type": "number"},
"paragraph_structure": {"type": "string"},
"call_to_action_style": {"type": "string"},
"link_placement": {"type": "string"}
}
},
"engagement_patterns": {
"type": "object",
"properties": {
"posting_frequency": {"type": "string"},
"optimal_posting_times": {"type": "array", "items": {"type": "string"}},
"engagement_tactics": {"type": "array", "items": {"type": "string"}},
"community_interaction": {"type": "string"}
}
},
"platform_best_practices": {
"type": "array",
"items": {"type": "string"}
}
},
"required": ["platform_type", "sentence_metrics", "content_format_rules", "engagement_patterns"]
}
try:
response = gemini_structured_json_response(
prompt=prompt,
schema=platform_schema,
temperature=0.2,
max_tokens=4096,
system_prompt=f"You are an expert in {platform} content strategy and platform-specific writing optimization."
)
return response
except Exception as e:
logger.error(f"Error generating {platform} persona: {str(e)}")
return {"error": f"Failed to generate {platform} persona: {str(e)}"}
def _build_persona_analysis_prompt(self, onboarding_data: Dict[str, Any]) -> str:
"""Build the main persona analysis prompt."""
"""Build the main persona analysis prompt with comprehensive data."""
website_analysis = onboarding_data.get("website_analysis", {})
research_prefs = onboarding_data.get("research_preferences", {})
# Get enhanced analysis data
enhanced_analysis = onboarding_data.get("enhanced_analysis", {})
website_analysis = onboarding_data.get("website_analysis", {}) or {}
research_prefs = onboarding_data.get("research_preferences", {}) or {}
prompt = f"""
PERSONA GENERATION TASK: Create a comprehensive writing persona based on user onboarding data.
COMPREHENSIVE PERSONA GENERATION TASK: Create a highly detailed, data-driven writing persona based on extensive AI analysis of user's website and content strategy.
ONBOARDING DATA ANALYSIS:
=== COMPREHENSIVE ONBOARDING DATA ANALYSIS ===
Website Analysis:
WEBSITE ANALYSIS OVERVIEW:
- URL: {website_analysis.get('website_url', 'Not provided')}
- Writing Style: {json.dumps(website_analysis.get('writing_style', {}), indent=2)}
- Content Characteristics: {json.dumps(website_analysis.get('content_characteristics', {}), indent=2)}
- Target Audience: {json.dumps(website_analysis.get('target_audience', {}), indent=2)}
- Content Type: {json.dumps(website_analysis.get('content_type', {}), indent=2)}
- Style Patterns: {json.dumps(website_analysis.get('style_patterns', {}), indent=2)}
- Analysis Date: {website_analysis.get('analysis_date', 'Not provided')}
- Status: {website_analysis.get('status', 'Not provided')}
Research Preferences:
- Research Depth: {research_prefs.get('research_depth', 'Not set')}
- Content Types: {research_prefs.get('content_types', [])}
- Auto Research: {research_prefs.get('auto_research', False)}
- Factual Content: {research_prefs.get('factual_content', False)}
=== DETAILED STYLE ANALYSIS ===
{json.dumps(enhanced_analysis.get('comprehensive_style_analysis', {}), indent=2)}
PERSONA GENERATION REQUIREMENTS:
=== CONTENT INSIGHTS ===
{json.dumps(enhanced_analysis.get('content_insights', {}), indent=2)}
1. IDENTITY CREATION:
- Create a memorable persona name that captures the essence of the writing style
- Define a clear archetype (e.g., "The Pragmatic Futurist", "The Thoughtful Educator")
- Articulate a core belief that drives the writing philosophy
- Write a comprehensive brand voice description
=== AUDIENCE INTELLIGENCE ===
{json.dumps(enhanced_analysis.get('audience_intelligence', {}), indent=2)}
2. LINGUISTIC FINGERPRINT (Quantitative Analysis):
- Calculate average sentence length based on website analysis
- Determine preferred sentence types (simple, compound, complex)
- Analyze active vs passive voice ratio
- Identify go-to words and phrases from the content analysis
- List words and phrases to avoid
- Determine contraction usage patterns
- Assess vocabulary complexity level
=== BRAND VOICE ANALYSIS ===
{json.dumps(enhanced_analysis.get('brand_voice_analysis', {}), indent=2)}
3. RHETORICAL ANALYSIS:
- Identify metaphor patterns and themes
- Analyze analogy usage
- Assess rhetorical question frequency and style
- Determine storytelling approach
=== TECHNICAL WRITING METRICS ===
{json.dumps(enhanced_analysis.get('technical_writing_metrics', {}), indent=2)}
4. TONAL RANGE:
- Define the default tone
- List permissible tones for different contexts
- Identify forbidden tones that don't match the brand
- Describe emotional range and expression
=== COMPETITIVE ANALYSIS ===
{json.dumps(enhanced_analysis.get('competitive_analysis', {}), indent=2)}
5. STYLISTIC CONSTRAINTS:
- Define punctuation preferences and rules
- Set formatting guidelines
- Establish paragraph structure preferences
=== CONTENT STRATEGY INSIGHTS ===
{json.dumps(enhanced_analysis.get('content_strategy_insights', {}), indent=2)}
ANALYSIS INSTRUCTIONS:
- Base your analysis on the actual data provided from the website analysis
- If data is limited, make reasonable inferences but note the confidence level
- Ensure the persona is actionable and specific enough for AI content generation
- Provide a confidence score (0-100) based on data availability and quality
- Include analysis notes explaining your reasoning
=== RESEARCH PREFERENCES ===
{json.dumps(enhanced_analysis.get('research_preferences', {}), indent=2)}
Generate a comprehensive persona profile that can be used to replicate this writing style across different platforms.
=== LEGACY DATA (for compatibility) ===
Website Analysis: {json.dumps(website_analysis.get('writing_style', {}), indent=2)}
Content Characteristics: {json.dumps(website_analysis.get('content_characteristics', {}) or {}, indent=2)}
Target Audience: {json.dumps(website_analysis.get('target_audience', {}), indent=2)}
Style Patterns: {json.dumps(website_analysis.get('style_patterns', {}), indent=2)}
=== COMPREHENSIVE PERSONA GENERATION REQUIREMENTS ===
1. IDENTITY CREATION (Based on Brand Analysis):
- Create a memorable persona name that captures the essence of the brand personality and writing style
- Define a clear archetype that reflects the brand's positioning and audience appeal
- Articulate a core belief that drives the writing philosophy and brand values
- Write a comprehensive brand voice description incorporating all style elements
2. LINGUISTIC FINGERPRINT (Quantitative Analysis from Technical Metrics):
- Calculate precise average sentence length from sentence structure analysis
- Determine preferred sentence types based on paragraph organization patterns
- Analyze active vs passive voice ratio from voice characteristics
- Extract go-to words and phrases from vocabulary patterns and style analysis
- List words and phrases to avoid based on brand alignment guidelines
- Determine contraction usage patterns from formality level
- Assess vocabulary complexity level from readability scores
3. RHETORICAL ANALYSIS (From Style Patterns):
- Identify metaphor patterns and themes from rhetorical devices
- Analyze analogy usage from content strategy insights
- Assess rhetorical question frequency from engagement tips
- Determine storytelling approach from content flow analysis
4. TONAL RANGE (From Comprehensive Style Analysis):
- Define the default tone from tone analysis and brand personality
- List permissible tones based on emotional appeal and audience considerations
- Identify forbidden tones from avoid elements and brand alignment
- Describe emotional range from psychographic profile and engagement level
5. STYLISTIC CONSTRAINTS (From Technical Writing Metrics):
- Define punctuation preferences from paragraph structure analysis
- Set formatting guidelines from content structure insights
- Establish paragraph structure preferences from organization patterns
- Include transition phrase preferences from style patterns
6. PLATFORM-SPECIFIC ADAPTATIONS (From Content Strategy):
- Incorporate SEO optimization strategies
- Include conversion optimization techniques
- Apply engagement tips for different platforms
- Use competitive advantages for differentiation
7. CONTENT STRATEGY INTEGRATION:
- Incorporate best practices from content strategy insights
- Include AI generation tips for consistent output
- Apply content calendar suggestions for timing
- Use competitive advantages for positioning
=== ENHANCED ANALYSIS INSTRUCTIONS ===
- Base your analysis on ALL the comprehensive data provided above
- Use the detailed technical metrics for precise linguistic analysis
- Incorporate brand voice analysis for authentic personality
- Apply audience intelligence for targeted communication
- Include competitive analysis for market positioning
- Use content strategy insights for practical application
- Ensure the persona reflects the brand's unique elements and competitive advantages
- Provide a confidence score (0-100) based on data richness and quality
- Include detailed analysis notes explaining your reasoning and data sources
Generate a comprehensive, data-driven persona profile that can be used to replicate this writing style across different platforms while maintaining brand authenticity and competitive positioning.
"""
return prompt
@@ -442,6 +244,7 @@ Generate a platform-optimized persona adaptation that maintains brand consistenc
return prompt
def _get_platform_constraints(self, platform: str) -> Dict[str, Any]:
"""Get platform-specific constraints and best practices."""
@@ -454,14 +257,8 @@ Generate a platform-optimized persona adaptation that maintains brand consistenc
"thread_support": True,
"link_shortening": True
},
"linkedin": {
"character_limit": 3000,
"optimal_length": "150-300 words",
"professional_tone": True,
"hashtag_limit": 5,
"rich_media": True,
"long_form": True
},
"linkedin": self.linkedin_service.get_linkedin_constraints(),
"facebook": self.facebook_service.get_facebook_constraints(),
"instagram": {
"caption_limit": 2200,
"optimal_length": "125-150 words",
@@ -521,8 +318,8 @@ Generate a platform-optimized persona adaptation that maintains brand consistenc
linguistic_fingerprint=core_persona.get("linguistic_fingerprint", {}),
platform_adaptations={"platforms": list(platform_personas.keys())},
onboarding_session_id=onboarding_data.get("session_info", {}).get("session_id"),
source_website_analysis=onboarding_data.get("website_analysis"),
source_research_preferences=onboarding_data.get("research_preferences"),
source_website_analysis=onboarding_data.get("website_analysis") or {},
source_research_preferences=onboarding_data.get("research_preferences") or {},
ai_analysis_version="gemini_v1.0",
confidence_score=core_persona.get("confidence_score", 0.0)
)
@@ -533,6 +330,24 @@ Generate a platform-optimized persona adaptation that maintains brand consistenc
# Create platform-specific persona records
for platform, platform_data in platform_personas.items():
# Prepare platform-specific data
platform_specific_data = {}
if platform.lower() == "linkedin":
platform_specific_data = {
"professional_networking": platform_data.get("professional_networking", {}),
"linkedin_features": platform_data.get("linkedin_features", {}),
"algorithm_optimization": platform_data.get("algorithm_optimization", {}),
"professional_context_optimization": platform_data.get("professional_context_optimization", {})
}
elif platform.lower() == "facebook":
platform_specific_data = {
"facebook_algorithm_optimization": platform_data.get("facebook_algorithm_optimization", {}),
"facebook_engagement_strategies": platform_data.get("facebook_engagement_strategies", {}),
"facebook_content_formats": platform_data.get("facebook_content_formats", {}),
"facebook_audience_targeting": platform_data.get("facebook_audience_targeting", {}),
"facebook_community_building": platform_data.get("facebook_community_building", {})
}
platform_persona = PlatformPersona(
writing_persona_id=writing_persona.id,
platform_type=platform,
@@ -543,7 +358,8 @@ Generate a platform-optimized persona adaptation that maintains brand consistenc
stylistic_constraints=core_persona.get("stylistic_constraints", {}),
content_format_rules=platform_data.get("content_format_rules", {}),
engagement_patterns=platform_data.get("engagement_patterns", {}),
platform_best_practices={"practices": platform_data.get("platform_best_practices", [])}
platform_best_practices={"practices": platform_data.get("platform_best_practices", [])},
algorithm_considerations=platform_specific_data if platform_specific_data else platform_data.get("algorithm_considerations", {})
)
session.add(platform_persona)
@@ -565,9 +381,10 @@ Generate a platform-optimized persona adaptation that maintains brand consistenc
session.add(analysis_result)
session.commit()
persona_id = writing_persona.id
session.close()
logger.info(f"✅ Persona saved to database with ID: {writing_persona.id}")
logger.info(f"✅ Persona saved to database with ID: {persona_id}")
return writing_persona
except Exception as e:
@@ -581,26 +398,108 @@ Generate a platform-optimized persona adaptation that maintains brand consistenc
"""Calculate how sufficient the onboarding data is for persona generation."""
score = 0.0
website_analysis = onboarding_data.get("website_analysis", {})
research_prefs = onboarding_data.get("research_preferences", {})
# Get enhanced analysis data
enhanced_analysis = onboarding_data.get("enhanced_analysis", {})
website_analysis = onboarding_data.get("website_analysis", {}) or {}
research_prefs = onboarding_data.get("research_preferences", {}) or {}
# Website analysis components (70% of score)
if website_analysis.get("writing_style"):
score += 25
if website_analysis.get("content_characteristics"):
score += 20
if website_analysis.get("target_audience"):
score += 15
if website_analysis.get("style_patterns"):
score += 10
# Enhanced scoring based on comprehensive data availability
# Research preferences components (30% of score)
# Comprehensive Style Analysis (25% of score)
style_analysis = enhanced_analysis.get("comprehensive_style_analysis", {})
if style_analysis.get("tone_analysis"):
score += 5
if style_analysis.get("voice_characteristics"):
score += 5
if style_analysis.get("brand_personality"):
score += 5
if style_analysis.get("formality_level"):
score += 5
if style_analysis.get("emotional_appeal"):
score += 5
# Content Insights (20% of score)
content_insights = enhanced_analysis.get("content_insights", {})
if content_insights.get("sentence_structure_analysis"):
score += 4
if content_insights.get("vocabulary_level"):
score += 4
if content_insights.get("readability_score"):
score += 4
if content_insights.get("content_flow"):
score += 4
if content_insights.get("visual_elements_usage"):
score += 4
# Audience Intelligence (15% of score)
audience_intel = enhanced_analysis.get("audience_intelligence", {})
if audience_intel.get("demographics"):
score += 3
if audience_intel.get("expertise_level"):
score += 3
if audience_intel.get("industry_focus"):
score += 3
if audience_intel.get("psychographic_profile"):
score += 3
if audience_intel.get("pain_points"):
score += 3
# Technical Writing Metrics (15% of score)
tech_metrics = enhanced_analysis.get("technical_writing_metrics", {})
if tech_metrics.get("vocabulary_patterns"):
score += 3
if tech_metrics.get("rhetorical_devices"):
score += 3
if tech_metrics.get("paragraph_structure"):
score += 3
if tech_metrics.get("style_consistency"):
score += 3
if tech_metrics.get("unique_elements"):
score += 3
# Content Strategy Insights (15% of score)
strategy_insights = enhanced_analysis.get("content_strategy_insights", {})
if strategy_insights.get("tone_recommendations"):
score += 3
if strategy_insights.get("best_practices"):
score += 3
if strategy_insights.get("competitive_advantages"):
score += 3
if strategy_insights.get("content_strategy"):
score += 3
if strategy_insights.get("ai_generation_tips"):
score += 3
# Research Preferences (10% of score)
if research_prefs.get("research_depth"):
score += 10
score += 5
if research_prefs.get("content_types"):
score += 10
if research_prefs.get("writing_style"):
score += 10
score += 5
# Legacy compatibility - add points for basic data if enhanced data is missing
if score < 50: # If enhanced data is insufficient, fall back to legacy scoring
legacy_score = 0.0
# Website analysis components (70% of legacy score)
if website_analysis.get("writing_style"):
legacy_score += 25
if website_analysis.get("content_characteristics"):
legacy_score += 20
if website_analysis.get("target_audience"):
legacy_score += 15
if website_analysis.get("style_patterns"):
legacy_score += 10
# Research preferences components (30% of legacy score)
if research_prefs.get("research_depth"):
legacy_score += 10
if research_prefs.get("content_types"):
legacy_score += 10
if research_prefs.get("writing_style"):
legacy_score += 10
# Use the higher of enhanced or legacy score
score = max(score, legacy_score)
return min(score, 100.0)

View File

@@ -0,0 +1,201 @@
import os
import asyncio
import concurrent.futures
from typing import Any, Dict, List
from dataclasses import dataclass
import requests
from loguru import logger
try:
from google import genai
GOOGLE_GENAI_AVAILABLE = True
except Exception:
GOOGLE_GENAI_AVAILABLE = False
@dataclass
class WritingSuggestion:
text: str
confidence: float
sources: List[Dict[str, Any]]
class WritingAssistantService:
"""
Minimal writing assistant that combines Exa search with Gemini continuation.
- Exa provides relevant sources with content snippets
- Gemini generates a short, cited continuation based on current text and sources
"""
def __init__(self) -> None:
self.exa_api_key = os.getenv("EXA_API_KEY")
self.gemini_api_key = os.getenv("GEMINI_API_KEY")
if not self.exa_api_key:
logger.warning("EXA_API_KEY not configured; writing assistant will fail")
if not (GOOGLE_GENAI_AVAILABLE and self.gemini_api_key):
logger.warning("Gemini not available; writing assistant will fail")
self.gemini_client = None
else:
self.gemini_client = genai.Client(api_key=self.gemini_api_key)
self.http_timeout_seconds = 15
# COST CONTROL: Daily usage limits
self.daily_api_calls = 0
self.daily_limit = 50 # Max 50 API calls per day (~$2.50 max cost)
self.last_reset_date = None
def _get_cached_suggestion(self, text: str) -> WritingSuggestion | None:
"""No cached suggestions - always use real API calls for authentic results."""
return None
def _check_daily_limit(self) -> bool:
"""Check if we're within daily API usage limits."""
import datetime
today = datetime.date.today()
# Reset counter if it's a new day
if self.last_reset_date != today:
self.daily_api_calls = 0
self.last_reset_date = today
# Check if we've exceeded the limit
if self.daily_api_calls >= self.daily_limit:
return False
# Increment counter for this API call
self.daily_api_calls += 1
logger.info(f"Writing assistant API call #{self.daily_api_calls}/{self.daily_limit} today")
return True
async def suggest(self, text: str, max_results: int = 1) -> List[WritingSuggestion]:
if not text or len(text.strip()) < 6:
return []
# COST OPTIMIZATION: Use cached/static suggestions for common patterns
# This reduces API calls by 90%+ while maintaining usefulness
cached_suggestion = self._get_cached_suggestion(text)
if cached_suggestion:
return [cached_suggestion]
# COST CONTROL: Check daily usage limits
if not self._check_daily_limit():
logger.warning("Daily API limit reached for writing assistant")
return []
# Only make expensive API calls for unique, substantial content
if len(text.strip()) < 50: # Skip API calls for very short text
return []
# 1) Find relevant sources via Exa (reduced results for cost)
sources = await self._search_sources(text)
# 2) Generate continuation suggestion via Gemini
suggestion_text, confidence = await self._generate_continuation(text, sources)
if not suggestion_text:
return []
return [WritingSuggestion(text=suggestion_text.strip(), confidence=confidence, sources=sources)]
async def _search_sources(self, text: str) -> List[Dict[str, Any]]:
if not self.exa_api_key:
raise Exception("EXA_API_KEY not configured")
# Follow Exa demo guidance: continuation-style prompt and 1000-char cap
exa_query = (
(text[-1000:] if len(text) > 1000 else text)
+ "\n\nIf you found the above interesting, here's another useful resource to read:"
)
payload = {
"query": exa_query,
"numResults": 3, # Reduced from 5 to 3 for cost savings
"text": True,
"type": "neural",
"highlights": {"numSentences": 1, "highlightsPerUrl": 1},
}
try:
resp = requests.post(
"https://api.exa.ai/search",
headers={"x-api-key": self.exa_api_key, "Content-Type": "application/json"},
json=payload,
timeout=self.http_timeout_seconds,
)
if resp.status_code != 200:
raise Exception(f"Exa error {resp.status_code}: {resp.text}")
data = resp.json()
results = data.get("results", [])
sources: List[Dict[str, Any]] = []
for r in results:
sources.append(
{
"title": r.get("title", "Untitled"),
"url": r.get("url", ""),
"text": r.get("text", ""),
"author": r.get("author", ""),
"published_date": r.get("publishedDate", ""),
"score": float(r.get("score", 0.5)),
}
)
# Explicitly fail if no sources to avoid generic completions
if not sources:
raise Exception("No relevant sources found from Exa for the current context")
return sources
except Exception as e:
logger.error(f"WritingAssistant _search_sources error: {e}")
raise
async def _generate_continuation(self, text: str, sources: List[Dict[str, Any]]) -> tuple[str, float]:
if not self.gemini_client:
raise Exception("Gemini client not available")
# Build compact sources context block
source_blocks: List[str] = []
for i, s in enumerate(sources[:5]):
excerpt = (s.get("text", "") or "")
excerpt = excerpt[:500]
source_blocks.append(
f"Source {i+1}: {s.get('title','') or 'Source'}\nURL: {s.get('url','')}\nExcerpt: {excerpt}"
)
sources_text = "\n\n".join(source_blocks) if source_blocks else "(No sources)"
# Based on Exa demo guidance for completion-only behavior and inline citations
system_prompt = (
"You are an essay-completion bot that completes a sentence or continues prose. "
"Only produce 1-2 SHORT sentences. Do not repeat or paraphrase the user's stub. "
"Continue in the same tone and topic as the stub. Prefer concrete, current facts from the provided sources. "
"Include exactly one brief, verifiable citation hint in parentheses with an author (or 'Source') and URL in square brackets, e.g., ((Doe, 2021)[https://example.com])."
)
user_prompt = (
f"User text to continue (do not repeat):\n{text}\n\n"
f"Relevant sources to inform your continuation:\n{sources_text}\n\n"
"Return only the continuation text, without quotes."
)
try:
loop = asyncio.get_event_loop()
with concurrent.futures.ThreadPoolExecutor() as executor:
resp = await loop.run_in_executor(
executor,
lambda: self.gemini_client.models.generate_content(
model="gemini-1.5-flash", contents=f"{system_prompt}\n\n{user_prompt}"
),
)
suggestion = (resp.text or "").strip()
if not suggestion:
raise Exception("Gemini returned empty suggestion")
# naive confidence from number of sources present
confidence = 0.7 if sources else 0.5
return suggestion, confidence
except Exception as e:
logger.error(f"WritingAssistant _generate_continuation error: {e}")
# Propagate to ensure frontend does not show stale/generic content
raise