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

Added citation and quality metrics to the content editor.

Browse Source
This commit is contained in:
ajaysi
2025-09-03 09:40:05 +05:30
parent 10b50f9732
commit 5efee4235d
35 changed files with 6987 additions and 1123 deletions
Download Patch File Download Diff File Expand all files Collapse all files

257
backend/models/linkedin_models.py
View File

@@ -2,6 +2,7 @@
LinkedIn Content Generation Models for ALwrity
This module defines the data models for LinkedIn content generation endpoints.
Enhanced to support grounding capabilities with source integration and quality metrics.
"""
from pydantic import BaseModel, Field, validator
@@ -37,6 +38,14 @@ class SearchEngine(str, Enum):
TAVILY = "tavily"
class GroundingLevel(str, Enum):
"""Levels of content grounding."""
NONE = "none"
BASIC = "basic"
ENHANCED = "enhanced"
ENTERPRISE = "enterprise"
class LinkedInPostRequest(BaseModel):
"""Request model for LinkedIn post generation."""
topic: str = Field(..., description="Main topic for the post", min_length=3, max_length=200)
@@ -48,8 +57,10 @@ class LinkedInPostRequest(BaseModel):
include_hashtags: bool = Field(default=True, description="Whether to include hashtags")
include_call_to_action: bool = Field(default=True, description="Whether to include call to action")
research_enabled: bool = Field(default=True, description="Whether to include research-backed content")
search_engine: SearchEngine = Field(default=SearchEngine.METAPHOR, description="Search engine for research")
search_engine: SearchEngine = Field(default=SearchEngine.GOOGLE, description="Search engine for research")
max_length: int = Field(default=3000, description="Maximum character count", ge=100, le=3000)
grounding_level: GroundingLevel = Field(default=GroundingLevel.ENHANCED, description="Level of content grounding")
include_citations: bool = Field(default=True, description="Whether to include inline citations")
class Config:
schema_extra = {
@@ -63,8 +74,10 @@ class LinkedInPostRequest(BaseModel):
"include_hashtags": True,
"include_call_to_action": True,
"research_enabled": True,
"search_engine": "metaphor",
"max_length": 2000
"search_engine": "google",
"max_length": 2000,
"grounding_level": "enhanced",
"include_citations": True
}
}
@@ -79,8 +92,10 @@ class LinkedInArticleRequest(BaseModel):
include_images: bool = Field(default=True, description="Whether to generate image suggestions")
seo_optimization: bool = Field(default=True, description="Whether to include SEO optimization")
research_enabled: bool = Field(default=True, description="Whether to include research-backed content")
search_engine: SearchEngine = Field(default=SearchEngine.METAPHOR, description="Search engine for research")
search_engine: SearchEngine = Field(default=SearchEngine.GOOGLE, description="Search engine for research")
word_count: int = Field(default=1500, description="Target word count", ge=500, le=5000)
grounding_level: GroundingLevel = Field(default=GroundingLevel.ENHANCED, description="Level of content grounding")
include_citations: bool = Field(default=True, description="Whether to include inline citations")
class Config:
schema_extra = {
@@ -93,124 +108,181 @@ class LinkedInArticleRequest(BaseModel):
"include_images": True,
"seo_optimization": True,
"research_enabled": True,
"search_engine": "metaphor",
"word_count": 2000
"search_engine": "google",
"word_count": 2000,
"grounding_level": "enhanced",
"include_citations": True
}
}
class LinkedInCarouselRequest(BaseModel):
"""Request model for LinkedIn carousel post generation."""
"""Request model for LinkedIn carousel generation."""
topic: str = Field(..., description="Main topic for the carousel", min_length=3, max_length=200)
industry: str = Field(..., description="Target industry context", min_length=2, max_length=100)
slide_count: int = Field(default=8, description="Number of slides", ge=3, le=15)
tone: LinkedInTone = Field(default=LinkedInTone.PROFESSIONAL, description="Tone of the carousel")
target_audience: Optional[str] = Field(None, description="Specific target audience", max_length=200)
key_takeaways: Optional[List[str]] = Field(None, description="Key takeaways to include", max_items=10)
number_of_slides: int = Field(default=5, description="Number of slides", ge=3, le=10)
include_cover_slide: bool = Field(default=True, description="Whether to include a cover slide")
include_cta_slide: bool = Field(default=True, description="Whether to include a call-to-action slide")
visual_style: Optional[str] = Field("modern", description="Visual style preference")
research_enabled: bool = Field(default=True, description="Whether to include research-backed content")
search_engine: SearchEngine = Field(default=SearchEngine.GOOGLE, description="Search engine for research")
grounding_level: GroundingLevel = Field(default=GroundingLevel.ENHANCED, description="Level of content grounding")
include_citations: bool = Field(default=True, description="Whether to include inline citations")
class Config:
schema_extra = {
"example": {
"topic": "5 Ways to Improve Team Productivity",
"industry": "Business Management",
"slide_count": 8,
"topic": "Future of remote work",
"industry": "Technology",
"tone": "professional",
"target_audience": "Team leaders and managers",
"key_takeaways": ["Clear communication", "Goal setting", "Tool optimization"],
"target_audience": "HR professionals and business leaders",
"number_of_slides": 6,
"include_cover_slide": True,
"include_cta_slide": True,
"visual_style": "modern"
"research_enabled": True,
"search_engine": "google",
"grounding_level": "enhanced",
"include_citations": True
}
}
class LinkedInVideoScriptRequest(BaseModel):
"""Request model for LinkedIn video script generation."""
topic: str = Field(..., description="Main topic for the video", min_length=3, max_length=200)
topic: str = Field(..., description="Main topic for the video script", min_length=3, max_length=200)
industry: str = Field(..., description="Target industry context", min_length=2, max_length=100)
video_length: int = Field(default=60, description="Target video length in seconds", ge=15, le=300)
tone: LinkedInTone = Field(default=LinkedInTone.PROFESSIONAL, description="Tone of the video")
tone: LinkedInTone = Field(default=LinkedInTone.PROFESSIONAL, description="Tone of the video script")
target_audience: Optional[str] = Field(None, description="Specific target audience", max_length=200)
key_messages: Optional[List[str]] = Field(None, description="Key messages to include", max_items=5)
include_hook: bool = Field(default=True, description="Whether to include an attention-grabbing hook")
include_captions: bool = Field(default=True, description="Whether to include caption suggestions")
video_duration: int = Field(default=60, description="Target video duration in seconds", ge=30, le=300)
include_captions: bool = Field(default=True, description="Whether to include captions")
include_thumbnail_suggestions: bool = Field(default=True, description="Whether to include thumbnail suggestions")
research_enabled: bool = Field(default=True, description="Whether to include research-backed content")
search_engine: SearchEngine = Field(default=SearchEngine.GOOGLE, description="Search engine for research")
grounding_level: GroundingLevel = Field(default=GroundingLevel.ENHANCED, description="Level of content grounding")
include_citations: bool = Field(default=True, description="Whether to include inline citations")
class Config:
schema_extra = {
"example": {
"topic": "Quick tips for remote team management",
"industry": "Human Resources",
"video_length": 90,
"tone": "conversational",
"target_audience": "Remote team managers",
"key_messages": ["Communication tools", "Regular check-ins", "Team building"],
"include_hook": True,
"include_captions": True
"topic": "Cybersecurity best practices",
"industry": "Technology",
"tone": "educational",
"target_audience": "IT professionals and business leaders",
"video_duration": 90,
"include_captions": True,
"include_thumbnail_suggestions": True,
"research_enabled": True,
"search_engine": "google",
"grounding_level": "enhanced",
"include_citations": True
}
}
class LinkedInCommentResponseRequest(BaseModel):
"""Request model for LinkedIn comment response generation."""
original_post: str = Field(..., description="Content of the original post", min_length=10, max_length=3000)
comment: str = Field(..., description="Comment to respond to", min_length=1, max_length=1000)
response_type: Literal["professional", "appreciative", "clarifying", "disagreement", "value_add"] = Field(
default="professional", description="Type of response"
)
tone: LinkedInTone = Field(default=LinkedInTone.PROFESSIONAL, description="Tone of the response")
include_question: bool = Field(default=False, description="Whether to include a follow-up question")
brand_voice: Optional[str] = Field(None, description="Specific brand voice guidelines", max_length=500)
original_comment: str = Field(..., description="Original comment to respond to", min_length=10, max_length=1000)
post_context: str = Field(..., description="Context of the post being commented on", min_length=10, max_length=500)
industry: str = Field(..., description="Industry context", min_length=2, max_length=100)
tone: LinkedInTone = Field(default=LinkedInTone.FRIENDLY, description="Tone of the response")
response_length: str = Field(default="medium", description="Length of response: short, medium, long")
include_questions: bool = Field(default=True, description="Whether to include engaging questions")
research_enabled: bool = Field(default=False, description="Whether to include research-backed content")
search_engine: SearchEngine = Field(default=SearchEngine.GOOGLE, description="Search engine for research")
grounding_level: GroundingLevel = Field(default=GroundingLevel.BASIC, description="Level of content grounding")
class Config:
schema_extra = {
"example": {
"original_post": "Just published an article about AI transformation in healthcare...",
"comment": "Great insights! How do you see this affecting smaller healthcare providers?",
"response_type": "value_add",
"tone": "professional",
"include_question": True,
"brand_voice": "Expert but approachable, data-driven"
"original_comment": "Great insights on AI implementation!",
"post_context": "Post about AI transformation in healthcare",
"industry": "Healthcare",
"tone": "friendly",
"response_length": "medium",
"include_questions": True,
"research_enabled": False,
"search_engine": "google",
"grounding_level": "basic"
}
}
# Enhanced Research Source Model
class ResearchSource(BaseModel):
"""Model for research source information."""
"""Enhanced model for research source information with grounding capabilities."""
title: str
url: str
content: str
relevance_score: Optional[float] = None
relevance_score: Optional[float] = Field(None, description="Relevance score (0.0-1.0)")
credibility_score: Optional[float] = Field(None, description="Credibility score (0.0-1.0)")
domain_authority: Optional[float] = Field(None, description="Domain authority score (0.0-1.0)")
source_type: Optional[str] = Field(None, description="Type of source (academic, business_news, etc.)")
publication_date: Optional[str] = Field(None, description="Publication date if available")
raw_result: Optional[Dict[str, Any]] = Field(None, description="Raw search result data")
# Enhanced Hashtag Suggestion Model
class HashtagSuggestion(BaseModel):
"""Model for hashtag suggestions."""
"""Enhanced model for hashtag suggestions."""
hashtag: str
category: str
popularity_score: Optional[float] = None
popularity_score: Optional[float] = Field(None, description="Popularity score (0.0-1.0)")
relevance_score: Optional[float] = Field(None, description="Relevance to topic (0.0-1.0)")
industry_alignment: Optional[float] = Field(None, description="Industry alignment score (0.0-1.0)")
# Enhanced Image Suggestion Model
class ImageSuggestion(BaseModel):
"""Model for image suggestions."""
"""Enhanced model for image suggestions."""
description: str
alt_text: str
style: Optional[str] = None
placement: Optional[str] = None
style: Optional[str] = Field(None, description="Visual style description")
placement: Optional[str] = Field(None, description="Suggested placement in content")
relevance_score: Optional[float] = Field(None, description="Relevance to content (0.0-1.0)")
# New Quality Metrics Model
class ContentQualityMetrics(BaseModel):
"""Model for content quality assessment metrics."""
overall_score: float = Field(..., description="Overall quality score (0.0-1.0)")
factual_accuracy: float = Field(..., description="Factual accuracy score (0.0-1.0)")
source_verification: float = Field(..., description="Source verification score (0.0-1.0)")
professional_tone: float = Field(..., description="Professional tone score (0.0-1.0)")
industry_relevance: float = Field(..., description="Industry relevance score (0.0-1.0)")
citation_coverage: float = Field(..., description="Citation coverage score (0.0-1.0)")
content_length: int = Field(..., description="Content length in characters")
word_count: int = Field(..., description="Word count")
analysis_timestamp: str = Field(..., description="Timestamp of quality analysis")
# New Citation Model
class Citation(BaseModel):
"""Model for inline citations in content."""
type: str = Field(..., description="Type of citation (inline, footnote, etc.)")
reference: str = Field(..., description="Citation reference (e.g., 'Source 1')")
position: Optional[int] = Field(None, description="Position in content")
source_index: Optional[int] = Field(None, description="Index of source in research_sources")
# Enhanced Post Content Model
class PostContent(BaseModel):
"""Model for generated post content."""
"""Enhanced model for generated post content with grounding capabilities."""
content: str
character_count: int
hashtags: List[HashtagSuggestion]
call_to_action: Optional[str] = None
engagement_prediction: Optional[Dict[str, Any]] = None
citations: List[Citation] = Field(default_factory=list, description="Inline citations")
source_list: Optional[str] = Field(None, description="Formatted source list")
quality_metrics: Optional[ContentQualityMetrics] = Field(None, description="Content quality metrics")
grounding_enabled: bool = Field(default=False, description="Whether grounding was used")
search_queries: Optional[List[str]] = Field(default_factory=list, description="Search queries used for research")
# Enhanced Article Content Model
class ArticleContent(BaseModel):
"""Model for generated article content."""
"""Enhanced model for generated article content with grounding capabilities."""
title: str
content: str
word_count: int
@@ -218,43 +290,62 @@ class ArticleContent(BaseModel):
seo_metadata: Optional[Dict[str, Any]] = None
image_suggestions: List[ImageSuggestion]
reading_time: Optional[int] = None
citations: List[Citation] = Field(default_factory=list, description="Inline citations")
source_list: Optional[str] = Field(None, description="Formatted source list")
quality_metrics: Optional[ContentQualityMetrics] = Field(None, description="Content quality metrics")
grounding_enabled: bool = Field(default=False, description="Whether grounding was used")
search_queries: Optional[List[str]] = Field(default_factory=list, description="Search queries used for research")
# Enhanced Carousel Slide Model
class CarouselSlide(BaseModel):
"""Model for carousel slide content."""
"""Enhanced model for carousel slide content."""
slide_number: int
title: str
content: str
visual_elements: List[str]
design_notes: Optional[str] = None
citations: List[Citation] = Field(default_factory=list, description="Inline citations for this slide")
# Enhanced Carousel Content Model
class CarouselContent(BaseModel):
"""Model for generated carousel content."""
"""Enhanced model for generated carousel content with grounding capabilities."""
title: str
slides: List[CarouselSlide]
cover_slide: Optional[CarouselSlide] = None
cta_slide: Optional[CarouselSlide] = None
design_guidelines: Dict[str, str]
citations: List[Citation] = Field(default_factory=list, description="Overall citations")
source_list: Optional[str] = Field(None, description="Formatted source list")
quality_metrics: Optional[ContentQualityMetrics] = Field(None, description="Content quality metrics")
grounding_enabled: bool = Field(default=False, description="Whether grounding was used")
# Enhanced Video Script Model
class VideoScript(BaseModel):
"""Model for video script content."""
"""Enhanced model for video script content with grounding capabilities."""
hook: str
main_content: List[Dict[str, str]] # scene_number, content, duration, visual_notes
conclusion: str
captions: Optional[List[str]] = None
thumbnail_suggestions: List[str]
video_description: str
citations: List[Citation] = Field(default_factory=list, description="Inline citations")
source_list: Optional[str] = Field(None, description="Formatted source list")
quality_metrics: Optional[ContentQualityMetrics] = Field(None, description="Content quality metrics")
grounding_enabled: bool = Field(default=False, description="Whether grounding was used")
# Enhanced LinkedIn Post Response Model
class LinkedInPostResponse(BaseModel):
"""Response model for LinkedIn post generation."""
"""Enhanced response model for LinkedIn post generation with grounding capabilities."""
success: bool = True
data: Optional[PostContent] = None
research_sources: List[ResearchSource] = []
generation_metadata: Dict[str, Any] = {}
error: Optional[str] = None
grounding_status: Optional[Dict[str, Any]] = Field(None, description="Grounding operation status")
class Config:
schema_extra = {
@@ -268,55 +359,91 @@ class LinkedInPostResponse(BaseModel):
{"hashtag": "#DigitalTransformation", "category": "general", "popularity_score": 0.8}
],
"call_to_action": "What's your experience with AI in healthcare? Share in the comments!",
"engagement_prediction": {"estimated_likes": 120, "estimated_comments": 15}
"engagement_prediction": {"estimated_likes": 120, "estimated_comments": 15},
"citations": [
{"type": "inline", "reference": "Source 1", "position": 45}
],
"source_list": "**Sources:**\n1. **AI in Healthcare: Current Trends**\n - URL: [https://example.com/ai-healthcare](https://example.com/ai-healthcare)",
"quality_metrics": {
"overall_score": 0.85,
"factual_accuracy": 0.9,
"source_verification": 0.8,
"professional_tone": 0.9,
"industry_relevance": 0.85,
"citation_coverage": 0.8,
"content_length": 1250,
"word_count": 180,
"analysis_timestamp": "2025-01-15T10:30:00Z"
},
"grounding_enabled": True
},
"research_sources": [
{
"title": "AI in Healthcare: Current Trends",
"url": "https://example.com/ai-healthcare",
"content": "Summary of AI healthcare trends...",
"relevance_score": 0.95
"relevance_score": 0.95,
"credibility_score": 0.85,
"domain_authority": 0.9,
"source_type": "business_news"
}
],
"generation_metadata": {
"model_used": "gemini-2.0-flash-001",
"generation_time": 3.2,
"research_time": 5.1
"research_time": 5.1,
"grounding_enabled": True
},
"grounding_status": {
"status": "success",
"sources_used": 3,
"citation_coverage": 0.8,
"quality_score": 0.85
}
}
}
# Enhanced LinkedIn Article Response Model
class LinkedInArticleResponse(BaseModel):
"""Response model for LinkedIn article generation."""
"""Enhanced response model for LinkedIn article generation with grounding capabilities."""
success: bool = True
data: Optional[ArticleContent] = None
research_sources: List[ResearchSource] = []
generation_metadata: Dict[str, Any] = {}
error: Optional[str] = None
grounding_status: Optional[Dict[str, Any]] = Field(None, description="Grounding operation status")
# Enhanced LinkedIn Carousel Response Model
class LinkedInCarouselResponse(BaseModel):
"""Response model for LinkedIn carousel generation."""
"""Enhanced response model for LinkedIn carousel generation with grounding capabilities."""
success: bool = True
data: Optional[CarouselContent] = None
research_sources: List[ResearchSource] = []
generation_metadata: Dict[str, Any] = {}
error: Optional[str] = None
grounding_status: Optional[Dict[str, Any]] = Field(None, description="Grounding operation status")
# Enhanced LinkedIn Video Script Response Model
class LinkedInVideoScriptResponse(BaseModel):
"""Response model for LinkedIn video script generation."""
"""Enhanced response model for LinkedIn video script generation with grounding capabilities."""
success: bool = True
data: Optional[VideoScript] = None
research_sources: List[ResearchSource] = []
generation_metadata: Dict[str, Any] = {}
error: Optional[str] = None
grounding_status: Optional[Dict[str, Any]] = Field(None, description="Grounding operation status")
# Enhanced LinkedIn Comment Response Result Model
class LinkedInCommentResponseResult(BaseModel):
"""Response model for LinkedIn comment response generation."""
"""Enhanced response model for LinkedIn comment response generation with grounding capabilities."""
success: bool = True
response: Optional[str] = None
alternative_responses: List[str] = []
tone_analysis: Optional[Dict[str, Any]] = None
generation_metadata: Dict[str, Any] = {}
error: Optional[str] = None
error: Optional[str] = None
grounding_status: Optional[Dict[str, Any]] = Field(None, description="Grounding operation status")

5
backend/requirements.txt
View File

@@ -15,7 +15,10 @@ copilotkit
openai>=1.3.0
anthropic>=0.7.0
mistralai>=0.0.12
google-genai>=1.9.0
google-genai>=0.3.0
google-api-python-client>=2.100.0
google-auth>=2.23.0
google-auth-oauthlib>=1.0.0
# Web scraping and content processing
beautifulsoup4>=4.12.0

15
backend/routers/linkedin.py
View File

@@ -18,7 +18,10 @@ from models.linkedin_models import (
LinkedInPostResponse, LinkedInArticleResponse, LinkedInCarouselResponse,
LinkedInVideoScriptResponse, LinkedInCommentResponseResult
)
from services.linkedin_service import linkedin_service
from services.linkedin_service import LinkedInService
# Initialize the LinkedIn service instance
linkedin_service = LinkedInService()
from middleware.monitoring_middleware import DatabaseAPIMonitor
from services.database import get_db_session
from sqlalchemy.orm import Session
@@ -117,7 +120,7 @@ async def generate_post(
raise HTTPException(status_code=422, detail="Industry cannot be empty")
# Generate post content
response = await linkedin_service.generate_post(request)
response = await linkedin_service.generate_linkedin_post(request)
# Log successful request
duration = time.time() - start_time
@@ -187,7 +190,7 @@ async def generate_article(
raise HTTPException(status_code=422, detail="Industry cannot be empty")
# Generate article content
response = await linkedin_service.generate_article(request)
response = await linkedin_service.generate_linkedin_article(request)
# Log successful request
duration = time.time() - start_time
@@ -259,7 +262,7 @@ async def generate_carousel(
raise HTTPException(status_code=422, detail="Slide count must be between 3 and 15")
# Generate carousel content
response = await linkedin_service.generate_carousel(request)
response = await linkedin_service.generate_linkedin_carousel(request)
# Log successful request
duration = time.time() - start_time
@@ -331,7 +334,7 @@ async def generate_video_script(
raise HTTPException(status_code=422, detail="Video length must be between 15 and 300 seconds")
# Generate video script content
response = await linkedin_service.generate_video_script(request)
response = await linkedin_service.generate_linkedin_video_script(request)
# Log successful request
duration = time.time() - start_time
@@ -400,7 +403,7 @@ async def generate_comment_response(
raise HTTPException(status_code=422, detail="Comment cannot be empty")
# Generate comment response
response = await linkedin_service.generate_comment_response(request)
response = await linkedin_service.generate_linkedin_comment_response(request)
# Log successful request
duration = time.time() - start_time

22
backend/services/citation/__init__.py Normal file
View File

@@ -0,0 +1,22 @@
"""
Citation Services Module for ALwrity
This module provides citation management capabilities for grounded content generation,
ensuring proper source attribution and citation validation.
Available Services:
- CitationManager: Handles inline citations, validation, and source attribution
- Citation pattern recognition and analysis
- Citation quality assessment and improvement suggestions
- Export formatting for different content types
Author: ALwrity Team
Version: 1.0
Last Updated: January 2025
"""
from services.citation.citation_manager import CitationManager
__all__ = [
"CitationManager"
]

532
backend/services/citation/citation_manager.py Normal file
View File

@@ -0,0 +1,532 @@
"""
Citation Manager Service for ALwrity
This service handles citation management for grounded content generation,
ensuring proper source attribution and citation validation.
Key Features:
- Inline citation formatting and management
- Citation validation and coverage analysis
- Source list generation
- Citation pattern recognition
- Quality assessment for citations
Dependencies:
- re (for pattern matching)
- typing (for type hints)
- logging (for debugging)
Author: ALwrity Team
Version: 1.0
Last Updated: January 2025
"""
import re
from typing import Dict, List, Optional, Any, Tuple
from loguru import logger
class CitationManager:
"""
Service for managing citations in grounded content.
This service handles the creation, validation, and management of citations
to ensure proper source attribution in generated content.
"""
def __init__(self):
"""Initialize the Citation Manager."""
# Citation patterns to recognize
self.citation_patterns = [
r'\[Source (\d+)\]', # [Source 1], [Source 2]
r'\[(\d+)\]', # [1], [2]
r'\(Source (\d+)\)', # (Source 1), (Source 2)
r'\((\d+)\)', # (1), (2)
r'Source (\d+)', # Source 1, Source 2
r'Ref\. (\d+)', # Ref. 1, Ref. 2
r'Reference (\d+)', # Reference 1, Reference 2
]
# Compile patterns for efficiency
self.compiled_patterns = [re.compile(pattern, re.IGNORECASE) for pattern in self.citation_patterns]
logger.info("Citation Manager initialized successfully")
def add_citations(
self,
content: str,
sources: List[Any],
citation_style: str = "brackets"
) -> str:
"""
Add citations to content based on source information.
Args:
content: The content to add citations to
sources: List of research sources (can be Dict or ResearchSource objects)
citation_style: Style of citations to use (brackets, parentheses, inline)
Returns:
Content with added citations
"""
if not sources:
return content
# Citation style templates
citation_templates = {
"brackets": "[Source {num}]",
"parentheses": "(Source {num})",
"inline": "Source {num}",
"numbered": "[{num}]"
}
template = citation_templates.get(citation_style, "[Source {num}]")
# Add source list at the end
source_list = self.generate_source_list(sources, citation_style)
# For now, we'll add a general citation at the end
# In a full implementation, you'd use NLP to identify claims and add specific citations
citation_text = f"\n\n{source_list}"
return content + citation_text
def validate_citations(
self,
content: str,
sources: List[Any]
) -> Dict[str, Any]:
"""
Validate citations in content for completeness and accuracy.
Args:
content: The content with citations
sources: List of research sources (can be Dict or ResearchSource objects)
Returns:
Citation validation results and metrics
"""
validation_result = {
"total_sources": len(sources),
"citations_found": 0,
"citation_coverage": 0.0,
"citation_quality": 0.0,
"missing_citations": [],
"invalid_citations": [],
"validation_score": 0.0
}
if not sources:
validation_result["validation_score"] = 0.0
return validation_result
# Find all citations in content
all_citations = []
for pattern in self.compiled_patterns:
matches = pattern.findall(content)
all_citations.extend(matches)
validation_result["citations_found"] = len(all_citations)
# Calculate citation coverage
validation_result["citation_coverage"] = min(
len(all_citations) / len(sources), 1.0
)
# Validate citation references
valid_citations = []
invalid_citations = []
for citation in all_citations:
try:
citation_num = int(citation)
if 1 <= citation_num <= len(sources):
valid_citations.append(citation_num)
else:
invalid_citations.append(citation_num)
except ValueError:
invalid_citations.append(citation)
validation_result["invalid_citations"] = invalid_citations
# Find missing citations
expected_citations = set(range(1, len(sources) + 1))
found_citations = set(valid_citations)
missing_citations = expected_citations - found_citations
validation_result["missing_citations"] = list(missing_citations)
# Calculate citation quality score
quality_factors = [
validation_result["citation_coverage"] * 0.4, # Coverage (40%)
(1.0 - len(invalid_citations) / max(len(all_citations), 1)) * 0.3, # Accuracy (30%)
(1.0 - len(missing_citations) / len(sources)) * 0.3 # Completeness (30%)
]
validation_result["citation_quality"] = sum(quality_factors)
validation_result["validation_score"] = (
validation_result["citation_coverage"] * 0.6 +
validation_result["citation_quality"] * 0.4
)
# Round scores
validation_result["citation_coverage"] = round(validation_result["citation_coverage"], 3)
validation_result["citation_quality"] = round(validation_result["citation_quality"], 3)
validation_result["validation_score"] = round(validation_result["validation_score"], 3)
return validation_result
def generate_source_list(
self,
sources: List[Any],
citation_style: str = "brackets"
) -> str:
"""
Generate a comprehensive list of sources with proper formatting.
Args:
sources: List of research sources (can be Dict or ResearchSource objects)
citation_style: Style of citations used in content
Returns:
Formatted source list
"""
if not sources:
return "**Sources:** No sources available."
# Header based on citation style
headers = {
"brackets": "**Sources:**",
"parentheses": "**Sources:**",
"inline": "**Sources:**",
"numbered": "**References:**"
}
header = headers.get(citation_style, "**Sources:**")
source_list = f"{header}\n\n"
for i, source in enumerate(sources, 1):
# Handle both Dict and ResearchSource objects
if hasattr(source, 'title'):
# ResearchSource Pydantic model
title = source.title
url = source.url
relevance = source.relevance_score or 0
credibility = source.credibility_score or 0
source_type = source.source_type or "general"
publication_date = source.publication_date or ""
else:
# Dictionary object
title = source.get("title", "Untitled")
url = source.get("url", "")
relevance = source.get("relevance_score", 0)
credibility = source.get("credibility_score", 0)
source_type = source.get("source_type", "general")
publication_date = source.get("publication_date", "")
# Format the source entry
source_entry = f"{i}. **{title}**\n"
if url:
source_entry += f" - URL: [{url}]({url})\n"
if relevance and relevance > 0:
source_entry += f" - Relevance: {relevance:.2f}\n"
if credibility and credibility > 0:
source_entry += f" - Credibility: {credibility:.2f}\n"
if source_type and source_type != "general":
source_entry += f" - Type: {source_type.replace('_', ' ').title()}\n"
if publication_date:
source_entry += f" - Published: {publication_date}\n"
source_list += source_entry + "\n"
return source_list
def extract_citations(self, content: str) -> List[Dict[str, Any]]:
"""
Extract all citations from content with their positions and references.
Args:
content: The content to extract citations from
Returns:
List of citation objects with metadata
"""
citations = []
for pattern in self.compiled_patterns:
matches = pattern.finditer(content)
for match in matches:
citation_text = match.group(0)
citation_num = match.group(1) if len(match.groups()) > 0 else None
position = match.start()
citation_obj = {
"text": citation_text,
"number": citation_num,
"position": position,
"pattern": pattern.pattern,
"line_number": content[:position].count('\n') + 1
}
citations.append(citation_obj)
# Sort by position
citations.sort(key=lambda x: x["position"])
return citations
def analyze_citation_patterns(self, content: str) -> Dict[str, Any]:
"""
Analyze citation patterns in content for insights.
Args:
content: The content to analyze
Returns:
Analysis results and pattern insights
"""
citations = self.extract_citations(content)
analysis = {
"total_citations": len(citations),
"citation_patterns": {},
"distribution": {},
"quality_indicators": {}
}
# Analyze citation patterns
for citation in citations:
pattern = citation["pattern"]
if pattern not in analysis["citation_patterns"]:
analysis["citation_patterns"][pattern] = 0
analysis["citation_patterns"][pattern] += 1
# Analyze citation distribution
if citations:
positions = [c["position"] for c in citations]
content_length = len(content)
# Distribution by content thirds
third_length = content_length // 3
first_third = sum(1 for pos in positions if pos < third_length)
second_third = sum(1 for pos in positions if third_length <= pos < 2 * third_length)
third_third = sum(1 for pos in positions if pos >= 2 * third_length)
analysis["distribution"] = {
"first_third": first_third,
"second_third": second_third,
"third_third": third_third,
"evenly_distributed": abs(first_third - second_third) <= 1 and abs(second_third - third_third) <= 1
}
# Quality indicators
analysis["quality_indicators"] = {
"has_citations": len(citations) > 0,
"multiple_citations": len(citations) > 1,
"even_distribution": analysis["distribution"].get("evenly_distributed", False),
"consistent_pattern": len(analysis["citation_patterns"]) <= 2
}
return analysis
def suggest_citation_improvements(
self,
content: str,
sources: List[Dict[str, Any]]
) -> List[str]:
"""
Suggest improvements for citation usage in content.
Args:
content: The content to analyze
sources: List of research sources
Returns:
List of improvement suggestions
"""
suggestions = []
if not sources:
suggestions.append("No sources available for citation.")
return suggestions
# Analyze current citations
citations = self.extract_citations(content)
validation = self.validate_citations(content, sources)
# Coverage suggestions
if validation["citation_coverage"] < 0.5:
suggestions.append(f"Low citation coverage ({validation['citation_coverage']:.1%}). Consider adding more citations to support factual claims.")
if validation["citation_coverage"] < 0.8:
suggestions.append("Moderate citation coverage. Aim for at least 80% of sources to be cited.")
# Distribution suggestions
analysis = self.analyze_citation_patterns(content)
if not analysis["distribution"].get("evenly_distributed", False):
suggestions.append("Citations appear clustered. Consider distributing citations more evenly throughout the content.")
# Pattern suggestions
if len(analysis["citation_patterns"]) > 2:
suggestions.append("Multiple citation patterns detected. Consider using consistent citation formatting for better readability.")
# Source quality suggestions
if sources:
avg_credibility = sum(s.get("credibility_score", 0) for s in sources) / len(sources)
if avg_credibility < 0.6:
suggestions.append("Low average source credibility. Consider using more authoritative sources when available.")
# Content length suggestions
if len(content) > 1000 and len(citations) < 3:
suggestions.append("Long content with few citations. Consider adding more citations to support key claims.")
if not suggestions:
suggestions.append("Citation usage looks good! Consider adding more specific citations if you have additional factual claims.")
return suggestions
def format_citation_for_export(
self,
content: str,
sources: List[Dict[str, Any]],
format_type: str = "markdown"
) -> str:
"""
Format content with citations for export in different formats.
Args:
content: The content with citations
sources: List of research sources
format_type: Export format (markdown, html, plain_text)
Returns:
Formatted content for export
"""
if format_type == "markdown":
return self._format_markdown_export(content, sources)
elif format_type == "html":
return self._format_html_export(content, sources)
elif format_type == "plain_text":
return self._format_plain_text_export(content, sources)
else:
logger.warning(f"Unknown format type: {format_type}, using markdown")
return self._format_markdown_export(content, sources)
def _format_markdown_export(self, content: str, sources: List[Dict[str, Any]]) -> str:
"""Format content for markdown export."""
# Add source list at the end
source_list = self.generate_source_list(sources, "brackets")
# Ensure proper markdown formatting
formatted_content = content
# Add source list
if sources:
formatted_content += f"\n\n{source_list}"
return formatted_content
def _format_html_export(self, content: str, sources: List[Dict[str, Any]]) -> str:
"""Format content for HTML export."""
# Convert markdown to basic HTML
html_content = content
# Convert markdown links to HTML
html_content = re.sub(r'\[([^\]]+)\]\(([^)]+)\)', r'<a href="\2">\1</a>', html_content)
# Convert markdown bold to HTML
html_content = re.sub(r'\*\*([^*]+)\*\*', r'<strong>\1</strong>', html_content)
# Convert line breaks to HTML
html_content = html_content.replace('\n', '<br>\n')
# Add source list
if sources:
source_list = self.generate_source_list(sources, "brackets")
# Convert markdown source list to HTML
html_source_list = re.sub(r'\*\*([^*]+)\*\*', r'<strong>\1</strong>', source_list)
html_source_list = re.sub(r'\[([^\]]+)\]\(([^)]+)\)', r'<a href="\2">\1</a>', html_source_list)
html_source_list = html_source_list.replace('\n', '<br>\n')
html_content += f"<br><br>{html_source_list}"
return html_content
def _format_plain_text_export(self, content: str, sources: List[Dict[str, Any]]) -> str:
"""Format content for plain text export."""
# Remove markdown formatting
plain_content = content
# Remove markdown links, keeping just the text
plain_content = re.sub(r'\[([^\]]+)\]\([^)]+\)', r'\1', plain_content)
# Remove markdown bold
plain_content = re.sub(r'\*\*([^*]+)\*\*', r'\1', plain_content)
# Add source list
if sources:
source_list = self.generate_source_list(sources, "brackets")
# Remove markdown formatting from source list
plain_source_list = re.sub(r'\[([^\]]+)\]\([^)]+\)', r'\1', plain_source_list)
plain_source_list = re.sub(r'\*\*([^*]+)\*\*', r'\1', plain_source_list)
plain_content += f"\n\n{plain_source_list}"
return plain_content
def get_citation_statistics(self, content: str, sources: List[Dict[str, Any]]) -> Dict[str, Any]:
"""
Get comprehensive statistics about citations in content.
Args:
content: The content to analyze
sources: List of research sources
Returns:
Citation statistics and metrics
"""
citations = self.extract_citations(content)
validation = self.validate_citations(content, sources)
analysis = self.analyze_citation_patterns(content)
stats = {
"content_metrics": {
"total_length": len(content),
"word_count": len(content.split()),
"paragraph_count": content.count('\n\n') + 1
},
"citation_metrics": {
"total_citations": len(citations),
"unique_citations": len(set(c.get("number") for c in citations if c.get("number"))),
"citation_density": len(citations) / max(len(content.split()), 1) * 1000, # citations per 1000 words
"citation_coverage": validation["citation_coverage"],
"citation_quality": validation["citation_quality"]
},
"source_metrics": {
"total_sources": len(sources),
"sources_cited": len(set(c.get("number") for c in citations if c.get("number"))),
"citation_efficiency": len(set(c.get("number") for c in citations if c.get("number"))) / max(len(sources), 1)
},
"quality_metrics": {
"validation_score": validation["validation_score"],
"distribution_score": 1.0 if analysis["distribution"].get("evenly_distributed", False) else 0.5,
"pattern_consistency": 1.0 if len(analysis["citation_patterns"]) <= 2 else 0.5
}
}
# Calculate overall citation score
overall_score = (
stats["citation_metrics"]["citation_coverage"] * 0.3 +
stats["citation_metrics"]["citation_quality"] * 0.3 +
stats["quality_metrics"]["validation_score"] * 0.2 +
stats["quality_metrics"]["distribution_score"] * 0.1 +
stats["quality_metrics"]["pattern_consistency"] * 0.1
)
stats["overall_citation_score"] = round(overall_score, 3)
return stats

11
backend/services/linkedin/__init__.py Normal file
View File

@@ -0,0 +1,11 @@
"""
LinkedIn Services Package
Contains specialized services for LinkedIn content generation.
"""
from .quality_handler import QualityHandler
from .content_generator import ContentGenerator
from .research_handler import ResearchHandler
__all__ = ["QualityHandler", "ContentGenerator", "ResearchHandler"]

748
backend/services/linkedin/content_generator.py Normal file
View File

@@ -0,0 +1,748 @@
"""
Content Generator for LinkedIn Content Generation
Handles the main content generation logic for posts and articles.
"""
from typing import Dict, Any, List, Optional
from datetime import datetime
from loguru import logger
from models.linkedin_models import (
LinkedInPostRequest, LinkedInArticleRequest, LinkedInPostResponse, LinkedInArticleResponse,
PostContent, ArticleContent, GroundingLevel, ResearchSource
)
from services.linkedin.quality_handler import QualityHandler
class ContentGenerator:
"""Handles content generation for all LinkedIn content types."""
def __init__(self, citation_manager=None, quality_analyzer=None, gemini_grounded=None, fallback_provider=None):
self.citation_manager = citation_manager
self.quality_analyzer = quality_analyzer
self.gemini_grounded = gemini_grounded
self.fallback_provider = fallback_provider
def _transform_gemini_sources(self, gemini_sources):
"""Transform Gemini sources to ResearchSource format."""
transformed_sources = []
for source in gemini_sources:
transformed_source = ResearchSource(
title=source.get('title', 'Unknown Source'),
url=source.get('url', ''),
content=f"Source from {source.get('title', 'Unknown')}",
relevance_score=0.8, # Default relevance score
credibility_score=0.7, # Default credibility score
domain_authority=0.6, # Default domain authority
source_type=source.get('type', 'web'),
publication_date=datetime.now().strftime('%Y-%m-%d')
)
transformed_sources.append(transformed_source)
return transformed_sources
async def generate_post(
self,
request: LinkedInPostRequest,
research_sources: List,
research_time: float,
content_result: Dict[str, Any],
grounding_enabled: bool
) -> LinkedInPostResponse:
"""Generate LinkedIn post with all processing steps."""
try:
start_time = datetime.now()
# Debug: Log what we received
logger.info(f"ContentGenerator.generate_post called with:")
logger.info(f" - research_sources count: {len(research_sources) if research_sources else 0}")
logger.info(f" - research_sources type: {type(research_sources)}")
logger.info(f" - content_result keys: {list(content_result.keys()) if content_result else 'None'}")
logger.info(f" - grounding_enabled: {grounding_enabled}")
logger.info(f" - include_citations: {request.include_citations}")
# Debug: Log content_result details
if content_result:
logger.info(f" - content_result has citations: {'citations' in content_result}")
logger.info(f" - content_result has sources: {'sources' in content_result}")
if 'citations' in content_result:
logger.info(f" - citations count: {len(content_result['citations']) if content_result['citations'] else 0}")
if 'sources' in content_result:
logger.info(f" - sources count: {len(content_result['sources']) if content_result['sources'] else 0}")
if research_sources:
logger.info(f" - First research source: {research_sources[0] if research_sources else 'None'}")
logger.info(f" - Research sources types: {[type(s) for s in research_sources[:3]]}")
# Step 3: Add citations if requested - POST METHOD
citations = []
source_list = None
final_research_sources = research_sources # Default to passed research_sources
# Use sources and citations from content_result if available (from Gemini grounding)
if content_result.get('citations') and content_result.get('sources'):
logger.info(f"Using citations and sources from Gemini grounding: {len(content_result['citations'])} citations, {len(content_result['sources'])} sources")
citations = content_result['citations']
# Transform Gemini sources to ResearchSource format
gemini_sources = self._transform_gemini_sources(content_result['sources'])
source_list = self.citation_manager.generate_source_list(gemini_sources) if self.citation_manager else None
# Use transformed sources for the response
final_research_sources = gemini_sources
elif request.include_citations and research_sources and self.citation_manager:
try:
logger.info(f"Processing citations for content length: {len(content_result['content'])}")
citations = self.citation_manager.extract_citations(content_result['content'])
logger.info(f"Extracted {len(citations)} citations from content")
source_list = self.citation_manager.generate_source_list(research_sources)
logger.info(f"Generated source list: {source_list[:200] if source_list else 'None'}")
except Exception as e:
logger.warning(f"Citation processing failed: {e}")
else:
logger.info(f"Citation processing skipped: include_citations={request.include_citations}, research_sources={len(research_sources) if research_sources else 0}, citation_manager={self.citation_manager is not None}")
# Step 4: Analyze content quality
quality_metrics = None
if grounding_enabled and self.quality_analyzer:
try:
quality_handler = QualityHandler(self.quality_analyzer)
quality_metrics = quality_handler.create_quality_metrics(
content=content_result['content'],
sources=final_research_sources, # Use final_research_sources
industry=request.industry,
grounding_enabled=grounding_enabled
)
except Exception as e:
logger.warning(f"Quality analysis failed: {e}")
# Step 5: Build response
post_content = PostContent(
content=content_result['content'],
character_count=len(content_result['content']),
hashtags=content_result.get('hashtags', []),
call_to_action=content_result.get('call_to_action'),
engagement_prediction=content_result.get('engagement_prediction'),
citations=citations,
source_list=source_list,
quality_metrics=quality_metrics,
grounding_enabled=grounding_enabled,
search_queries=content_result.get('search_queries', [])
)
generation_time = (datetime.now() - start_time).total_seconds()
# Build grounding status
grounding_status = {
'status': 'success' if grounding_enabled else 'disabled',
'sources_used': len(final_research_sources), # Use final_research_sources
'citation_coverage': len(citations) / max(len(final_research_sources), 1) if final_research_sources else 0,
'quality_score': quality_metrics.overall_score if quality_metrics else 0.0
} if grounding_enabled else None
return LinkedInPostResponse(
success=True,
data=post_content,
research_sources=final_research_sources, # Use final_research_sources
generation_metadata={
'model_used': 'gemini-2.0-flash-001',
'generation_time': generation_time,
'research_time': research_time,
'grounding_enabled': grounding_enabled
},
grounding_status=grounding_status
)
except Exception as e:
logger.error(f"Error generating LinkedIn post: {str(e)}")
return LinkedInPostResponse(
success=False,
error=f"Failed to generate LinkedIn post: {str(e)}"
)
async def generate_article(
self,
request: LinkedInArticleRequest,
research_sources: List,
research_time: float,
content_result: Dict[str, Any],
grounding_enabled: bool
) -> LinkedInArticleResponse:
"""Generate LinkedIn article with all processing steps."""
try:
start_time = datetime.now()
# Step 3: Add citations if requested - ARTICLE METHOD
citations = []
source_list = None
final_research_sources = research_sources # Default to passed research_sources
# Use sources and citations from content_result if available (from Gemini grounding)
if content_result.get('citations') and content_result.get('sources'):
logger.info(f"Using citations and sources from Gemini grounding: {len(content_result['citations'])} citations, {len(content_result['sources'])} sources")
citations = content_result['citations']
# Transform Gemini sources to ResearchSource format
gemini_sources = self._transform_gemini_sources(content_result['sources'])
source_list = self.citation_manager.generate_source_list(gemini_sources) if self.citation_manager else None
# Use transformed sources for the response
final_research_sources = gemini_sources
elif request.include_citations and research_sources and self.citation_manager:
try:
citations = self.citation_manager.extract_citations(content_result['content'])
source_list = self.citation_manager.generate_source_list(research_sources)
except Exception as e:
logger.warning(f"Citation processing failed: {e}")
# Step 4: Analyze content quality
quality_metrics = None
if grounding_enabled and self.quality_analyzer:
try:
quality_handler = QualityHandler(self.quality_analyzer)
quality_metrics = quality_handler.create_quality_metrics(
content=content_result['content'],
sources=final_research_sources, # Use final_research_sources
industry=request.industry,
grounding_enabled=grounding_enabled
)
except Exception as e:
logger.warning(f"Quality analysis failed: {e}")
# Step 5: Build response
article_content = ArticleContent(
title=content_result['title'],
content=content_result['content'],
word_count=len(content_result['content'].split()),
sections=content_result.get('sections', []),
seo_metadata=content_result.get('seo_metadata'),
image_suggestions=content_result.get('image_suggestions', []),
reading_time=content_result.get('reading_time'),
citations=citations,
source_list=source_list,
quality_metrics=quality_metrics,
grounding_enabled=grounding_enabled,
search_queries=content_result.get('search_queries', [])
)
generation_time = (datetime.now() - start_time).total_seconds()
# Build grounding status
grounding_status = {
'status': 'success' if grounding_enabled else 'disabled',
'sources_used': len(final_research_sources), # Use final_research_sources
'citation_coverage': len(citations) / max(len(final_research_sources), 1) if final_research_sources else 0,
'quality_score': quality_metrics.overall_score if quality_metrics else 0.0
} if grounding_enabled else None
return LinkedInArticleResponse(
success=True,
data=article_content,
research_sources=final_research_sources, # Use final_research_sources
generation_metadata={
'model_used': 'gemini-2.0-flash-001',
'generation_time': generation_time,
'research_time': research_time,
'grounding_enabled': grounding_enabled
},
grounding_status=grounding_status
)
except Exception as e:
logger.error(f"Error generating LinkedIn article: {str(e)}")
return LinkedInArticleResponse(
success=False,
error=f"Failed to generate LinkedIn article: {str(e)}"
)
async def generate_carousel(
self,
request,
research_sources: List,
research_time: float,
content_result: Dict[str, Any],
grounding_enabled: bool
):
"""Generate LinkedIn carousel with all processing steps."""
try:
start_time = datetime.now()
# Step 3: Add citations if requested
citations = []
source_list = None
if request.include_citations and research_sources:
# Extract citations from all slides
all_content = " ".join([slide['content'] for slide in content_result['slides']])
citations = self.citation_manager.extract_citations(all_content) if self.citation_manager else []
source_list = self.citation_manager.generate_source_list(research_sources) if self.citation_manager else None
# Step 4: Analyze content quality
quality_metrics = None
if grounding_enabled and self.quality_analyzer:
try:
all_content = " ".join([slide['content'] for slide in content_result['slides']])
quality_handler = QualityHandler(self.quality_analyzer)
quality_metrics = quality_handler.create_quality_metrics(
content=all_content,
sources=research_sources,
industry=request.industry,
grounding_enabled=grounding_enabled
)
except Exception as e:
logger.warning(f"Quality analysis failed: {e}")
# Step 5: Build response
slides = []
for i, slide_data in enumerate(content_result['slides']):
slide_citations = []
if request.include_citations and research_sources and self.citation_manager:
slide_citations = self.citation_manager.extract_citations(slide_data['content'])
slides.append({
'slide_number': i + 1,
'title': slide_data['title'],
'content': slide_data['content'],
'visual_elements': slide_data.get('visual_elements', []),
'design_notes': slide_data.get('design_notes'),
'citations': slide_citations
})
carousel_content = {
'title': content_result['title'],
'slides': slides,
'cover_slide': content_result.get('cover_slide'),
'cta_slide': content_result.get('cta_slide'),
'design_guidelines': content_result.get('design_guidelines', {}),
'citations': citations,
'source_list': source_list,
'quality_metrics': quality_metrics,
'grounding_enabled': grounding_enabled
}
generation_time = (datetime.now() - start_time).total_seconds()
# Build grounding status
grounding_status = {
'status': 'success' if grounding_enabled else 'disabled',
'sources_used': len(research_sources),
'citation_coverage': len(citations) / max(len(research_sources), 1) if research_sources else 0,
'quality_score': quality_metrics.overall_score if quality_metrics else 0.0
} if grounding_enabled else None
return {
'success': True,
'data': carousel_content,
'research_sources': research_sources,
'generation_metadata': {
'model_used': 'gemini-2.0-flash-001',
'generation_time': generation_time,
'research_time': research_time,
'grounding_enabled': grounding_enabled
},
'grounding_status': grounding_status
}
except Exception as e:
logger.error(f"Error generating LinkedIn carousel: {str(e)}")
return {
'success': False,
'error': f"Failed to generate LinkedIn carousel: {str(e)}"
}
async def generate_video_script(
self,
request,
research_sources: List,
research_time: float,
content_result: Dict[str, Any],
grounding_enabled: bool
):
"""Generate LinkedIn video script with all processing steps."""
try:
start_time = datetime.now()
# Step 3: Add citations if requested
citations = []
source_list = None
if request.include_citations and research_sources and self.citation_manager:
all_content = f"{content_result['hook']} {' '.join([scene['content'] for scene in content_result['main_content']])} {content_result['conclusion']}"
citations = self.citation_manager.extract_citations(all_content)
source_list = self.citation_manager.generate_source_list(research_sources)
# Step 4: Analyze content quality
quality_metrics = None
if grounding_enabled and self.quality_analyzer:
try:
all_content = f"{content_result['hook']} {' '.join([scene['content'] for scene in content_result['main_content']])} {content_result['conclusion']}"
quality_handler = QualityHandler(self.quality_analyzer)
quality_metrics = quality_handler.create_quality_metrics(
content=all_content,
sources=research_sources,
industry=request.industry,
grounding_enabled=grounding_enabled
)
except Exception as e:
logger.warning(f"Quality analysis failed: {e}")
# Step 5: Build response
video_script = {
'hook': content_result['hook'],
'main_content': content_result['main_content'],
'conclusion': content_result['conclusion'],
'captions': content_result.get('captions'),
'thumbnail_suggestions': content_result.get('thumbnail_suggestions', []),
'video_description': content_result.get('video_description', ''),
'citations': citations,
'source_list': source_list,
'quality_metrics': quality_metrics,
'grounding_enabled': grounding_enabled
}
generation_time = (datetime.now() - start_time).total_seconds()
# Build grounding status
grounding_status = {
'status': 'success' if grounding_enabled else 'disabled',
'sources_used': len(research_sources),
'citation_coverage': len(citations) / max(len(research_sources), 1) if research_sources else 0,
'quality_score': quality_metrics.overall_score if quality_metrics else 0.0
} if grounding_enabled else None
return {
'success': True,
'data': video_script,
'research_sources': research_sources,
'generation_metadata': {
'model_used': 'gemini-2.0-flash-001',
'generation_time': generation_time,
'research_time': research_time,
'grounding_enabled': grounding_enabled
},
'grounding_status': grounding_status
}
except Exception as e:
logger.error(f"Error generating LinkedIn video script: {str(e)}")
return {
'success': False,
'error': f"Failed to generate LinkedIn video script: {str(e)}"
}
async def generate_comment_response(
self,
request,
research_sources: List,
research_time: float,
content_result: Dict[str, Any],
grounding_enabled: bool
):
"""Generate LinkedIn comment response with all processing steps."""
try:
start_time = datetime.now()
generation_time = (datetime.now() - start_time).total_seconds()
# Build grounding status
grounding_status = {
'status': 'success' if grounding_enabled else 'disabled',
'sources_used': len(research_sources),
'citation_coverage': 0, # Comments typically don't have citations
'quality_score': 0.8 # Default quality for comments
} if grounding_enabled else None
return {
'success': True,
'response': content_result['response'],
'alternative_responses': content_result.get('alternative_responses', []),
'tone_analysis': content_result.get('tone_analysis'),
'generation_metadata': {
'model_used': 'gemini-2.0-flash-001',
'generation_time': generation_time,
'research_time': research_time,
'grounding_enabled': grounding_enabled
},
'grounding_status': grounding_status
}
except Exception as e:
logger.error(f"Error generating LinkedIn comment response: {str(e)}")
return {
'success': False,
'error': f"Failed to generate LinkedIn comment response: {str(e)}"
}
# Grounded content generation methods
async def generate_grounded_post_content(self, request, research_sources: List) -> Dict[str, Any]:
"""Generate grounded post content using the enhanced Gemini provider with native grounding."""
try:
if not self.gemini_grounded:
logger.warning("Gemini Grounded Provider not available, using fallback")
return await self.generate_fallback_post_content(request)
# Build the prompt for grounded generation
prompt = self._build_post_prompt(request)
# Generate grounded content using native Google Search grounding
result = await self.gemini_grounded.generate_grounded_content(
prompt=prompt,
content_type="linkedin_post",
temperature=0.7,
max_tokens=request.max_length
)
return result
except Exception as e:
logger.error(f"Error generating grounded post content: {str(e)}")
# Fallback to basic generation
return await self.generate_fallback_post_content(request)
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."""
try:
if not self.gemini_grounded:
logger.warning("Gemini Grounded Provider not available, using fallback")
return await self.generate_fallback_article_content(request)
# Build the prompt for grounded generation
prompt = self._build_article_prompt(request)
# Generate grounded content using native Google Search grounding
result = await self.gemini_grounded.generate_grounded_content(
prompt=prompt,
content_type="linkedin_article",
temperature=0.7,
max_tokens=request.word_count * 10 # Approximate character count
)
return result
except Exception as e:
logger.error(f"Error generating grounded article content: {str(e)}")
# Fallback to basic generation
return await self.generate_fallback_article_content(request)
async def generate_grounded_carousel_content(self, request, research_sources: List) -> Dict[str, Any]:
"""Generate grounded carousel content using the enhanced Gemini provider with native grounding."""
try:
if not self.gemini_grounded:
logger.warning("Gemini Grounded Provider not available, using fallback")
return await self.generate_fallback_carousel_content(request)
# Build the prompt for grounded generation
prompt = self._build_carousel_prompt(request)
# Generate grounded content using native Google Search grounding
result = await self.gemini_grounded.generate_grounded_content(
prompt=prompt,
content_type="linkedin_carousel",
temperature=0.7,
max_tokens=2000
)
return result
except Exception as e:
logger.error(f"Error generating grounded carousel content: {str(e)}")
# Fallback to basic generation
return await self.generate_fallback_carousel_content(request)
async def generate_grounded_video_script_content(self, request, research_sources: List) -> Dict[str, Any]:
"""Generate grounded video script content using the enhanced Gemini provider with native grounding."""
try:
if not self.gemini_grounded:
logger.warning("Gemini Grounded Provider not available, using fallback")
return await self.generate_fallback_video_script_content(request)
# Build the prompt for grounded generation
prompt = self._build_video_script_prompt(request)
# Generate grounded content using native Google Search grounding
result = await self.gemini_grounded.generate_grounded_content(
prompt=prompt,
content_type="linkedin_video_script",
temperature=0.7,
max_tokens=1500
)
return result
except Exception as e:
logger.error(f"Error generating grounded video script content: {str(e)}")
# Fallback to basic generation
return await self.generate_fallback_video_script_content(request)
async def generate_grounded_comment_response(self, request, research_sources: List) -> Dict[str, Any]:
"""Generate grounded comment response using the enhanced Gemini provider with native grounding."""
try:
if not self.gemini_grounded:
logger.warning("Gemini Grounded Provider not available, using fallback")
return await self.generate_fallback_comment_response(request)
# Build the prompt for grounded generation
prompt = self._build_comment_response_prompt(request)
# Generate grounded content using native Google Search grounding
result = await self.gemini_grounded.generate_grounded_content(
prompt=prompt,
content_type="linkedin_comment_response",
temperature=0.7,
max_tokens=500
)
return result
except Exception as e:
logger.error(f"Error generating grounded comment response: {str(e)}")
# Fallback to basic generation
return await self.generate_fallback_comment_response(request)
# Fallback content generation methods
async def generate_fallback_post_content(self, request) -> Dict[str, Any]:
"""Generate post content using fallback provider."""
if not self.fallback_provider:
raise Exception("No fallback provider available")
return {
'content': f"Professional LinkedIn post about {request.topic} in the {request.industry} industry.",
'hashtags': [{'hashtag': f'#{request.industry.lower().replace(" ", "")}', 'category': 'industry', 'popularity_score': 0.8}],
'call_to_action': "What are your thoughts on this? Share in the comments!",
'engagement_prediction': {'estimated_likes': 50, 'estimated_comments': 5}
}
async def generate_fallback_article_content(self, request) -> Dict[str, Any]:
"""Generate article content using fallback provider."""
if not self.fallback_provider:
raise Exception("No fallback provider available")
return {
'title': f"Comprehensive Guide to {request.topic} in {request.industry}",
'content': f"Detailed article about {request.topic} in the {request.industry} industry.",
'sections': [{'title': 'Introduction', 'content': 'Industry overview and context'}],
'seo_metadata': {'keywords': [request.topic, request.industry]},
'image_suggestions': ['Industry-related visual content'],
'reading_time': '5 minutes'
}
async def generate_fallback_carousel_content(self, request) -> Dict[str, Any]:
"""Generate carousel content using fallback provider."""
if not self.fallback_provider:
raise Exception("No fallback provider available")
return {
'title': f"Key Insights: {request.topic} in {request.industry}",
'slides': [
{'title': 'Overview', 'content': f'Introduction to {request.topic}', 'visual_elements': [], 'design_notes': 'Clean, professional design'},
{'title': 'Key Points', 'content': f'Main insights about {request.topic}', 'visual_elements': [], 'design_notes': 'Bullet points with icons'}
],
'cover_slide': {'title': 'Cover', 'content': 'Professional cover slide', 'visual_elements': [], 'design_notes': 'Eye-catching design'},
'cta_slide': {'title': 'Call to Action', 'content': 'Engage with this content', 'visual_elements': [], 'design_notes': 'Clear CTA design'},
'design_guidelines': {'style': 'professional', 'colors': 'brand colors'}
}
async def generate_fallback_video_script_content(self, request) -> Dict[str, Any]:
"""Generate video script content using fallback provider."""
if not self.fallback_provider:
raise Exception("No fallback provider available")
return {
'hook': f"Discover how {request.topic} is transforming the {request.industry} industry!",
'main_content': [
{'content': f'Introduction to {request.topic}', 'duration': '30s'},
{'content': f'Key insights about {request.topic}', 'duration': '45s'}
],
'conclusion': f"Ready to explore {request.topic}? Let's dive in!",
'captions': [f'Key point about {request.topic}'],
'thumbnail_suggestions': ['Professional thumbnail with industry imagery'],
'video_description': f"Video description about {request.topic}"
}
async def generate_fallback_comment_response(self, request) -> Dict[str, Any]:
"""Generate comment response using fallback provider."""
if not self.fallback_provider:
raise Exception("No fallback provider available")
return {
'response': f"Thank you for your comment about {request.original_comment}",
'alternative_responses': [],
'tone_analysis': None
}
# Prompt building methods
def _build_post_prompt(self, request) -> str:
"""Build prompt for post generation."""
prompt = f"""
Generate a professional LinkedIn post about {request.topic} in the {request.industry} industry.
Requirements:
- Tone: {request.tone}
- Target audience: {request.target_audience or 'Industry professionals'}
- Maximum length: {request.max_length} characters
- Include engaging hashtags
- Include a call to action
- Make it informative and shareable
Key points to include: {', '.join(request.key_points) if request.key_points else 'Industry insights and trends'}
"""
return prompt.strip()
def _build_article_prompt(self, request) -> str:
"""Build prompt for article generation."""
prompt = f"""
Generate a comprehensive LinkedIn article about {request.topic} in the {request.industry} industry.
Requirements:
- Tone: {request.tone}
- Target audience: {request.target_audience or 'Industry professionals'}
- Word count: {request.word_count} words
- Include SEO optimization
- Include image suggestions
- Make it informative and engaging
Key sections to include: {', '.join(request.key_sections) if request.key_sections else 'Introduction, main content, conclusion'}
"""
return prompt.strip()
def _build_carousel_prompt(self, request) -> str:
"""Build prompt for carousel generation."""
prompt = f"""
Generate a LinkedIn carousel about {request.topic} in the {request.industry} industry.
Requirements:
- Tone: {request.tone}
- Target audience: {request.target_audience or 'Industry professionals'}
- Number of slides: {request.number_of_slides}
- Include cover slide: {request.include_cover_slide}
- Include CTA slide: {request.include_cta_slide}
- Make each slide informative and visually appealing
Each slide should contain valuable insights and be designed for social media engagement.
"""
return prompt.strip()
def _build_video_script_prompt(self, request) -> str:
"""Build prompt for video script generation."""
prompt = f"""
Generate a LinkedIn video script about {request.topic} in the {request.industry} industry.
Requirements:
- Tone: {request.tone}
- Target audience: {request.target_audience or 'Industry professionals'}
- Duration: {request.video_duration} seconds
- Include captions: {request.include_captions}
- Include thumbnail suggestions: {request.include_thumbnail_suggestions}
- Make it engaging and informative
Structure: Hook, main content (divided into scenes), conclusion
"""
return prompt.strip()
def _build_comment_response_prompt(self, request) -> str:
"""Build prompt for comment response generation."""
prompt = f"""
Generate a LinkedIn comment response to: "{request.original_comment}"
Context: {request.post_context}
Industry: {request.industry}
Tone: {request.tone}
Response length: {request.response_length}
Include questions: {request.include_questions}
Make the response engaging, professional, and add value to the conversation.
"""
return prompt.strip()

61
backend/services/linkedin/quality_handler.py Normal file
View File

@@ -0,0 +1,61 @@
"""
Quality Handler for LinkedIn Content Generation
Handles content quality analysis and metrics conversion.
"""
from typing import Dict, Any, Optional
from models.linkedin_models import ContentQualityMetrics
from loguru import logger
class QualityHandler:
"""Handles content quality analysis and metrics conversion."""
def __init__(self, quality_analyzer=None):
self.quality_analyzer = quality_analyzer
def create_quality_metrics(
self,
content: str,
sources: list,
industry: str,
grounding_enabled: bool = False
) -> Optional[ContentQualityMetrics]:
"""
Create ContentQualityMetrics object from quality analysis.
Args:
content: Content to analyze
sources: Research sources used
industry: Target industry
grounding_enabled: Whether grounding was used
Returns:
ContentQualityMetrics object or None if analysis fails
"""
if not grounding_enabled or not self.quality_analyzer:
return None
try:
quality_analysis = self.quality_analyzer.analyze_content_quality(
content=content,
sources=sources,
industry=industry
)
# Convert the analysis result to ContentQualityMetrics format
return ContentQualityMetrics(
overall_score=quality_analysis.get('overall_score', 0.0),
factual_accuracy=quality_analysis.get('metrics', {}).get('factual_accuracy', 0.0),
source_verification=quality_analysis.get('metrics', {}).get('source_verification', 0.0),
professional_tone=quality_analysis.get('metrics', {}).get('professional_tone', 0.0),
industry_relevance=quality_analysis.get('metrics', {}).get('industry_relevance', 0.0),
citation_coverage=quality_analysis.get('metrics', {}).get('citation_coverage', 0.0),
content_length=quality_analysis.get('content_length', 0),
word_count=quality_analysis.get('word_count', 0),
analysis_timestamp=quality_analysis.get('analysis_timestamp', '')
)
except Exception as e:
logger.warning(f"Quality metrics creation failed: {e}")
return None

76
backend/services/linkedin/research_handler.py Normal file
View File

@@ -0,0 +1,76 @@
"""
Research Handler for LinkedIn Content Generation
Handles research operations and timing for content generation.
"""
from typing import List
from datetime import datetime
from loguru import logger
from models.linkedin_models import ResearchSource
class ResearchHandler:
"""Handles research operations and timing for LinkedIn content."""
def __init__(self, linkedin_service):
self.linkedin_service = linkedin_service
async def conduct_research(
self,
request,
research_enabled: bool,
search_engine: str,
max_results: int = 10
) -> tuple[List[ResearchSource], float]:
"""
Conduct research if enabled and return sources with timing.
Returns:
Tuple of (research_sources, research_time)
"""
research_sources = []
research_time = 0
if research_enabled:
# Debug: Log the search engine value being passed
logger.info(f"ResearchHandler: search_engine='{search_engine}' (type: {type(search_engine)})")
research_start = datetime.now()
research_sources = await self.linkedin_service._conduct_research(
topic=request.topic,
industry=request.industry,
search_engine=search_engine,
max_results=max_results
)
research_time = (datetime.now() - research_start).total_seconds()
logger.info(f"Research completed in {research_time:.2f}s, found {len(research_sources)} sources")
return research_sources, research_time
def determine_grounding_enabled(self, request, research_sources: List[ResearchSource]) -> bool:
"""Determine if grounding should be enabled based on request and research results."""
# Normalize values from possible Enum or string
try:
level_raw = getattr(request, 'grounding_level', 'enhanced')
level = (getattr(level_raw, 'value', level_raw) or '').strip().lower()
except Exception:
level = 'enhanced'
try:
engine_raw = getattr(request, 'search_engine', 'google')
engine_val = getattr(engine_raw, 'value', engine_raw)
engine_str = str(engine_val).split('.')[-1].strip().lower()
except Exception:
engine_str = 'google'
research_enabled = bool(getattr(request, 'research_enabled', True))
if not research_enabled or level == 'none':
return False
# For Google native grounding, Gemini returns sources in the generation metadata,
# so we should not require pre-fetched research_sources.
if engine_str == 'google':
return True
# For other engines, require that research actually returned sources
return bool(research_sources)

1385
backend/services/linkedin_service.py
View File

File diff suppressed because it is too large Load Diff

11
backend/services/llm_providers/__init__.py
View File

@@ -4,11 +4,11 @@ This service handles all LLM (Language Model) provider integrations,
migrated from the legacy lib/gpt_providers functionality.
"""
from .main_text_generation import llm_text_gen
from .openai_provider import openai_chatgpt, test_openai_api_key
from .gemini_provider import gemini_text_response, gemini_structured_json_response, test_gemini_api_key
from .anthropic_provider import anthropic_text_response
from .deepseek_provider import deepseek_text_response
from services.llm_providers.main_text_generation import llm_text_gen
from services.llm_providers.openai_provider import openai_chatgpt, test_openai_api_key
from services.llm_providers.gemini_provider import gemini_text_response, gemini_structured_json_response
from services.llm_providers.anthropic_provider import anthropic_text_response
from services.llm_providers.deepseek_provider import deepseek_text_response
__all__ = [
"llm_text_gen",
@@ -16,7 +16,6 @@ __all__ = [
"test_openai_api_key",
"gemini_text_response",
"gemini_structured_json_response",
"test_gemini_api_key",
"anthropic_text_response",
"deepseek_text_response"
]

577
backend/services/llm_providers/gemini_grounded_provider.py Normal file
View File

@@ -0,0 +1,577 @@
"""
Enhanced Gemini Provider for Grounded Content Generation
This provider uses native Google Search grounding to generate content that is
factually grounded in current web sources, with automatic citation generation.
Based on Google AI's official grounding documentation.
"""
import os
import json
import re
from typing import List, Dict, Any, Optional
from datetime import datetime
from loguru import logger
try:
from google import genai
from google.genai import types
GOOGLE_GENAI_AVAILABLE = True
except ImportError:
GOOGLE_GENAI_AVAILABLE = False
logger.warn("Google GenAI not available. Install with: pip install google-genai")
class GeminiGroundedProvider:
"""
Enhanced Gemini provider for grounded content generation with native Google Search.
This provider uses the official Google Search grounding tool to generate content
that is factually grounded in current web sources, with automatic citation generation.
Based on: https://ai.google.dev/gemini-api/docs/google-search
"""
def __init__(self):
"""Initialize the Gemini Grounded Provider."""
if not GOOGLE_GENAI_AVAILABLE:
raise ImportError("Google GenAI library not available. Install with: pip install google-genai")
self.api_key = os.getenv('GEMINI_API_KEY')
if not self.api_key:
raise ValueError("GEMINI_API_KEY environment variable is required")
# Initialize the Gemini client
self.client = genai.Client(api_key=self.api_key)
logger.info("✅ Gemini Grounded Provider initialized with native Google Search grounding")
async def generate_grounded_content(
self,
prompt: str,
content_type: str = "linkedin_post",
temperature: float = 0.7,
max_tokens: int = 2048
) -> Dict[str, Any]:
"""
Generate grounded content using native Google Search grounding.
Args:
prompt: The content generation prompt
content_type: Type of content to generate
temperature: Creativity level (0.0-1.0)
max_tokens: Maximum tokens in response
Returns:
Dictionary containing generated content and grounding metadata
"""
try:
logger.info(f"Generating grounded content for {content_type} using native Google Search")
# Build the grounded prompt
grounded_prompt = self._build_grounded_prompt(prompt, content_type)
# Configure the grounding tool
grounding_tool = types.Tool(
google_search=types.GoogleSearch()
)
# Configure generation settings
config = types.GenerateContentConfig(
tools=[grounding_tool],
max_output_tokens=max_tokens,
temperature=temperature
)
# Make the request with native grounding
response = self.client.models.generate_content(
model="gemini-2.5-flash",
contents=grounded_prompt,
config=config,
)
# Process the grounded response
result = self._process_grounded_response(response, content_type)
logger.info(f"✅ Grounded content generated successfully with {len(result.get('sources', []))} sources")
return result
except Exception as e:
logger.error(f"❌ Error generating grounded content: {str(e)}")
raise
def _build_grounded_prompt(self, prompt: str, content_type: str) -> str:
"""
Build a prompt optimized for grounded content generation.
Args:
prompt: Base prompt
content_type: Type of content being generated
Returns:
Enhanced prompt for grounded generation
"""
content_type_instructions = {
"linkedin_post": "Generate a professional LinkedIn post that is factually accurate and cites current sources. Include engaging hashtags and a call-to-action.",
"linkedin_article": "Generate a comprehensive LinkedIn article with proper structure, factual accuracy, and source citations. Include an engaging title and conclusion.",
"linkedin_carousel": "Generate LinkedIn carousel content with multiple slides, each containing factual information with proper source attribution.",
"linkedin_video_script": "Generate a video script with hook, main content, and conclusion. Ensure all claims are factually grounded.",
"linkedin_comment_response": "Generate a professional comment response that adds value to the conversation."
}
instruction = content_type_instructions.get(content_type, "Generate professional content with factual accuracy.")
grounded_prompt = f"""
{instruction}
IMPORTANT: Use current, factual information from reliable sources. Cite specific sources for any claims, statistics, or recent developments.
User Request: {prompt}
Requirements:
- Ensure all factual claims are backed by current sources
- Use professional, engaging language appropriate for LinkedIn
- Include relevant industry insights and trends
- Make content shareable and valuable for the target audience
"""
return grounded_prompt.strip()
def _process_grounded_response(self, response, content_type: str) -> Dict[str, Any]:
"""
Process the Gemini response with grounding metadata.
Args:
response: Gemini API response
content_type: Type of content generated
Returns:
Processed content with sources and citations
"""
try:
# Extract the main content
content = ""
if hasattr(response, 'text'):
content = response.text
elif hasattr(response, 'candidates') and response.candidates:
candidate = response.candidates[0]
if hasattr(candidate, 'content') and candidate.content:
# Extract text from content parts
text_parts = []
for part in candidate.content:
if hasattr(part, 'text'):
text_parts.append(part.text)
content = " ".join(text_parts)
logger.info(f"Extracted content length: {len(content) if content else 0}")
if not content:
logger.warning("No content extracted from response")
content = "Generated content about the requested topic."
# Initialize result structure
result = {
'content': content,
'sources': [],
'citations': [],
'search_queries': [],
'grounding_metadata': {},
'content_type': content_type,
'generation_timestamp': datetime.now().isoformat()
}
# Debug: Log response structure
logger.info(f"Response type: {type(response)}")
logger.info(f"Response attributes: {dir(response)}")
# Extract grounding metadata if available
if hasattr(response, 'candidates') and response.candidates:
candidate = response.candidates[0]
logger.info(f"Candidate attributes: {dir(candidate)}")
if hasattr(candidate, 'grounding_metadata') and candidate.grounding_metadata:
grounding_metadata = candidate.grounding_metadata
result['grounding_metadata'] = grounding_metadata
logger.info(f"Grounding metadata attributes: {dir(grounding_metadata)}")
logger.info(f"Grounding metadata type: {type(grounding_metadata)}")
logger.info(f"Grounding metadata value: {grounding_metadata}")
# Log all available attributes and their values
for attr in dir(grounding_metadata):
if not attr.startswith('_'):
try:
value = getattr(grounding_metadata, attr)
logger.info(f" {attr}: {type(value)} = {value}")
except Exception as e:
logger.warning(f" {attr}: Error accessing - {e}")
# Extract search queries
if hasattr(grounding_metadata, 'web_search_queries'):
result['search_queries'] = grounding_metadata.web_search_queries
logger.info(f"Search queries: {grounding_metadata.web_search_queries}")
# Extract sources from grounding chunks
if hasattr(grounding_metadata, 'grounding_chunks') and grounding_metadata.grounding_chunks:
sources = []
for i, chunk in enumerate(grounding_metadata.grounding_chunks):
logger.info(f"Chunk {i} attributes: {dir(chunk)}")
if hasattr(chunk, 'web'):
source = {
'index': i,
'title': getattr(chunk.web, 'title', f'Source {i+1}'),
'url': getattr(chunk.web, 'uri', ''),
'type': 'web'
}
sources.append(source)
result['sources'] = sources
logger.info(f"Extracted {len(sources)} sources")
else:
logger.error("❌ CRITICAL: No grounding chunks found in response")
logger.error(f"Grounding metadata structure: {dir(grounding_metadata)}")
if hasattr(grounding_metadata, 'grounding_chunks'):
logger.error(f"Grounding chunks type: {type(grounding_metadata.grounding_chunks)}")
logger.error(f"Grounding chunks value: {grounding_metadata.grounding_chunks}")
raise ValueError("No grounding chunks found - grounding is not working properly")
# Extract citations from grounding supports
if hasattr(grounding_metadata, 'grounding_supports') and grounding_metadata.grounding_supports:
citations = []
for support in grounding_metadata.grounding_supports:
if hasattr(support, 'segment') and hasattr(support, 'grounding_chunk_indices'):
citation = {
'type': 'inline',
'start_index': getattr(support.segment, 'start_index', 0),
'end_index': getattr(support.segment, 'end_index', 0),
'text': getattr(support.segment, 'text', ''),
'source_indices': support.grounding_chunk_indices,
'reference': f"Source {support.grounding_chunk_indices[0] + 1}" if support.grounding_chunk_indices else "Unknown"
}
citations.append(citation)
result['citations'] = citations
logger.info(f"Extracted {len(citations)} citations")
else:
logger.error("❌ CRITICAL: No grounding supports found in response")
logger.error(f"Grounding metadata structure: {dir(grounding_metadata)}")
if hasattr(grounding_metadata, 'grounding_supports'):
logger.error(f"Grounding supports type: {type(grounding_metadata.grounding_supports)}")
logger.error(f"Grounding supports value: {grounding_metadata.grounding_supports}")
raise ValueError("No grounding supports found - grounding is not working properly")
logger.info(f"✅ Successfully extracted {len(result['sources'])} sources and {len(result['citations'])} citations from grounding metadata")
logger.info(f"Sources: {result['sources']}")
logger.info(f"Citations: {result['citations']}")
else:
logger.error("❌ CRITICAL: No grounding metadata found in response")
logger.error(f"Response structure: {dir(response)}")
logger.error(f"First candidate structure: {dir(candidates[0]) if candidates else 'No candidates'}")
raise ValueError("No grounding metadata found - grounding is not working properly")
else:
logger.error("❌ CRITICAL: No candidates found in response")
logger.error(f"Response structure: {dir(response)}")
raise ValueError("No candidates found in response - grounding is not working properly")
# Add content-specific processing
if content_type == "linkedin_post":
result.update(self._process_post_content(content))
elif content_type == "linkedin_article":
result.update(self._process_article_content(content))
elif content_type == "linkedin_carousel":
result.update(self._process_carousel_content(content))
elif content_type == "linkedin_video_script":
result.update(self._process_video_script_content(content))
return result
except Exception as e:
logger.error(f"❌ CRITICAL: Error processing grounded response: {str(e)}")
logger.error(f"Exception type: {type(e)}")
logger.error(f"Exception details: {e}")
raise ValueError(f"Failed to process grounded response: {str(e)}")
def _process_post_content(self, content: str) -> Dict[str, Any]:
"""Process LinkedIn post content for hashtags and engagement elements."""
try:
# Handle None content
if content is None:
content = ""
logger.warning("Content is None, using empty string")
# Extract hashtags
hashtags = re.findall(r'#\w+', content)
# Generate call-to-action if not present
cta_patterns = [
r'What do you think\?',
r'Share your thoughts',
r'Comment below',
r'What\'s your experience\?',
r'Let me know in the comments'
]
has_cta = any(re.search(pattern, content, re.IGNORECASE) for pattern in cta_patterns)
call_to_action = None
if not has_cta:
call_to_action = "What are your thoughts on this? Share in the comments!"
return {
'hashtags': [{'hashtag': tag, 'category': 'general', 'popularity_score': 0.8} for tag in hashtags],
'call_to_action': call_to_action,
'engagement_prediction': {
'estimated_likes': max(50, len(content) // 10),
'estimated_comments': max(5, len(content) // 100)
}
}
except Exception as e:
logger.error(f"Error processing post content: {str(e)}")
return {}
def _process_article_content(self, content: str) -> Dict[str, Any]:
"""Process LinkedIn article content for structure and SEO."""
try:
# Extract title (first line or first sentence)
lines = content.split('\n')
title = lines[0].strip() if lines else "Article Title"
# Estimate word count
word_count = len(content.split())
# Generate sections based on content structure
sections = []
current_section = ""
for line in lines:
if line.strip().startswith('#') or line.strip().startswith('##'):
if current_section:
sections.append({'title': 'Section', 'content': current_section.strip()})
current_section = ""
else:
current_section += line + "\n"
if current_section:
sections.append({'title': 'Content', 'content': current_section.strip()})
return {
'title': title,
'word_count': word_count,
'sections': sections,
'reading_time': max(1, word_count // 200), # 200 words per minute
'seo_metadata': {
'meta_description': content[:160] + "..." if len(content) > 160 else content,
'keywords': self._extract_keywords(content)
}
}
except Exception as e:
logger.error(f"Error processing article content: {str(e)}")
return {}
def _process_carousel_content(self, content: str) -> Dict[str, Any]:
"""Process LinkedIn carousel content for slide structure."""
try:
# Split content into slides (basic implementation)
slides = []
content_parts = content.split('\n\n')
for i, part in enumerate(content_parts[:10]): # Max 10 slides
if part.strip():
slides.append({
'slide_number': i + 1,
'title': f"Slide {i + 1}",
'content': part.strip(),
'visual_elements': [],
'design_notes': None
})
return {
'title': f"Carousel on {content[:50]}...",
'slides': slides,
'design_guidelines': {
'color_scheme': 'professional',
'typography': 'clean',
'layout': 'minimal'
}
}
except Exception as e:
logger.error(f"Error processing carousel content: {str(e)}")
return {}
def _process_video_script_content(self, content: str) -> Dict[str, Any]:
"""Process LinkedIn video script content for structure."""
try:
# Basic video script processing
lines = content.split('\n')
hook = ""
main_content = []
conclusion = ""
# Extract hook (first few lines)
hook_lines = []
for line in lines[:3]:
if line.strip() and not line.strip().startswith('#'):
hook_lines.append(line.strip())
if len(' '.join(hook_lines)) > 100:
break
hook = ' '.join(hook_lines)
# Extract conclusion (last few lines)
conclusion_lines = []
for line in lines[-3:]:
if line.strip() and not line.strip().startswith('#'):
conclusion_lines.insert(0, line.strip())
if len(' '.join(conclusion_lines)) > 100:
break
conclusion = ' '.join(conclusion_lines)
# Main content (everything in between)
main_content_text = content[len(hook):len(content)-len(conclusion)].strip()
return {
'hook': hook,
'main_content': [{
'scene_number': 1,
'content': main_content_text,
'duration': 60,
'visual_notes': 'Professional presentation style'
}],
'conclusion': conclusion,
'thumbnail_suggestions': ['Professional thumbnail', 'Industry-focused image'],
'video_description': f"Professional insights on {content[:100]}..."
}
except Exception as e:
logger.error(f"Error processing video script content: {str(e)}")
return {}
def _extract_keywords(self, content: str) -> List[str]:
"""Extract relevant keywords from content."""
try:
# Simple keyword extraction (can be enhanced with NLP)
words = re.findall(r'\b\w+\b', content.lower())
word_freq = {}
# Filter out common words
stop_words = {'the', 'and', 'or', 'but', 'in', 'on', 'at', 'to', 'for', 'of', 'with', 'by', 'is', 'are', 'was', 'were', 'be', 'been', 'have', 'has', 'had', 'do', 'does', 'did', 'will', 'would', 'could', 'should', 'may', 'might', 'can', 'this', 'that', 'these', 'those', 'a', 'an'}
for word in words:
if word not in stop_words and len(word) > 3:
word_freq[word] = word_freq.get(word, 0) + 1
# Return top keywords
sorted_words = sorted(word_freq.items(), key=lambda x: x[1], reverse=True)
return [word for word, freq in sorted_words[:10]]
except Exception as e:
logger.error(f"Error extracting keywords: {str(e)}")
return []
def add_citations(self, content: str, sources: List[Dict[str, Any]]) -> str:
"""
Add inline citations to content based on grounding metadata.
Args:
content: The content to add citations to
sources: List of sources from grounding metadata
Returns:
Content with inline citations
"""
try:
if not sources:
return content
# Create citation mapping
citation_map = {}
for source in sources:
index = source.get('index', 0)
citation_map[index] = f"[Source {index + 1}]({source.get('url', '')})"
# Add citations at the end of sentences or paragraphs
# This is a simplified approach - in practice, you'd use the groundingSupports data
citation_text = "\n\n**Sources:**\n"
for i, source in enumerate(sources):
citation_text += f"{i+1}. **{source.get('title', f'Source {i+1}')}**\n - URL: [{source.get('url', '')}]({source.get('url', '')})\n\n"
return content + citation_text
except Exception as e:
logger.error(f"Error adding citations: {str(e)}")
return content
def extract_citations(self, content: str) -> List[Dict[str, Any]]:
"""
Extract citations from content.
Args:
content: Content to extract citations from
Returns:
List of citation objects
"""
try:
citations = []
# Look for citation patterns
citation_patterns = [
r'\[Source (\d+)\]',
r'\[(\d+)\]',
r'\(Source (\d+)\)'
]
for pattern in citation_patterns:
matches = re.finditer(pattern, content)
for match in matches:
citations.append({
'type': 'inline',
'reference': match.group(0),
'position': match.start(),
'source_index': int(match.group(1)) - 1
})
return citations
except Exception as e:
logger.error(f"Error extracting citations: {str(e)}")
return []
def assess_content_quality(self, content: str, sources: List[Dict[str, Any]]) -> Dict[str, Any]:
"""
Assess the quality of generated content.
Args:
content: The generated content
sources: List of sources used
Returns:
Quality metrics dictionary
"""
try:
# Basic quality metrics
word_count = len(content.split())
char_count = len(content)
# Source coverage
source_coverage = min(1.0, len(sources) / max(1, word_count / 100))
# Professional tone indicators
professional_indicators = ['research', 'analysis', 'insights', 'trends', 'industry', 'professional']
unprofessional_indicators = ['awesome', 'amazing', 'incredible', 'mind-blowing']
professional_score = sum(1 for indicator in professional_indicators if indicator.lower() in content.lower()) / len(professional_indicators)
unprofessional_score = sum(1 for indicator in unprofessional_indicators if indicator.lower() in content.lower()) / len(unprofessional_indicators)
tone_score = max(0, professional_score - unprofessional_score)
# Overall quality score
overall_score = (source_coverage * 0.4 + tone_score * 0.3 + min(1.0, word_count / 500) * 0.3)
return {
'overall_score': round(overall_score, 2),
'source_coverage': round(source_coverage, 2),
'tone_score': round(tone_score, 2),
'word_count': word_count,
'char_count': char_count,
'sources_count': len(sources),
'quality_level': 'high' if overall_score > 0.8 else 'medium' if overall_score > 0.6 else 'low'
}
except Exception as e:
logger.error(f"Error assessing content quality: {str(e)}")
return {
'overall_score': 0.0,
'error': str(e)
}

22
backend/services/quality/__init__.py Normal file
View File

@@ -0,0 +1,22 @@
"""
Quality Services Module for ALwrity
This module provides content quality assessment and analysis capabilities,
ensuring generated content meets enterprise standards and quality requirements.
Available Services:
- ContentQualityAnalyzer: Comprehensive content quality assessment
- Quality metrics and scoring systems
- Improvement recommendations and tracking
- Content comparison and analysis
Author: ALwrity Team
Version: 1.0
Last Updated: January 2025
"""
from services.quality.content_analyzer import ContentQualityAnalyzer
__all__ = [
"ContentQualityAnalyzer"
]

755
backend/services/quality/content_analyzer.py Normal file
View File

@@ -0,0 +1,755 @@
"""
Content Quality Analyzer Service for ALwrity
This service provides comprehensive quality assessment for generated content,
evaluating factual accuracy, source verification, professional tone, and industry relevance.
Key Features:
- Factual accuracy scoring against source verification
- Professional tone analysis for enterprise content
- Industry relevance metrics and assessment
- Overall quality scoring and recommendations
- Content quality tracking over time
Dependencies:
- re (for pattern matching)
- typing (for type hints)
- logging (for debugging)
Author: ALwrity Team
Version: 1.0
Last Updated: January 2025
"""
import re
from typing import Dict, List, Optional, Any, Tuple
from loguru import logger
class ContentQualityAnalyzer:
"""
Service for analyzing and scoring content quality.
This service evaluates content across multiple dimensions including
factual accuracy, professional tone, industry relevance, and overall quality.
"""
def __init__(self):
"""Initialize the Content Quality Analyzer."""
# Professional tone indicators
self.professional_indicators = [
"research", "analysis", "insights", "trends", "strategies",
"implementation", "optimization", "innovation", "development",
"leadership", "expertise", "professional", "industry", "enterprise"
]
# Unprofessional tone indicators
self.unprofessional_indicators = [
"awesome", "amazing", "incredible", "mind-blowing", "crazy",
"totally", "absolutely", "literally", "basically", "actually",
"you know", "like", "um", "uh", "lol", "omg"
]
# Industry-specific terminology patterns
self.industry_terminology = {
"Technology": ["ai", "machine learning", "automation", "digital transformation", "cloud computing"],
"Healthcare": ["patient care", "medical", "treatment", "diagnosis", "healthcare"],
"Finance": ["investment", "market", "financial", "portfolio", "risk management"],
"Marketing": ["brand", "campaign", "audience", "conversion", "engagement"],
"Education": ["learning", "curriculum", "pedagogy", "student", "academic"]
}
logger.info("Content Quality Analyzer initialized successfully")
def analyze_content_quality(
self,
content: str,
sources: List[Dict[str, Any]],
industry: str = "general"
) -> Dict[str, Any]:
"""
Analyze content quality across multiple dimensions.
Args:
content: The content to analyze
sources: List of research sources used
industry: The target industry for relevance assessment
Returns:
Comprehensive quality analysis results
"""
try:
# Analyze different quality aspects
logger.info("🔍 [Quality Analysis] Starting content quality analysis")
logger.info(f"🔍 [Quality Analysis] Content length: {len(content)} characters")
logger.info(f"🔍 [Quality Analysis] Sources count: {len(sources)}")
factual_accuracy = self._assess_factual_accuracy(content, sources)
logger.info(f"🔍 [Quality Analysis] Factual accuracy score: {factual_accuracy}")
source_verification = self._assess_source_verification(content, sources)
logger.info(f"🔍 [Quality Analysis] Source verification score: {source_verification}")
professional_tone = self._assess_professional_tone(content)
logger.info(f"🔍 [Quality Analysis] Professional tone score: {professional_tone}")
industry_relevance = self._assess_industry_relevance(content, industry)
logger.info(f"🔍 [Quality Analysis] Industry relevance score: {industry_relevance}")
citation_coverage = self._assess_citation_coverage(content, sources)
logger.info(f"🔍 [Quality Analysis] Citation coverage score: {citation_coverage}")
# Calculate overall quality score
overall_score = self._calculate_overall_score({
"factual_accuracy": factual_accuracy,
"source_verification": source_verification,
"professional_tone": professional_tone,
"industry_relevance": industry_relevance,
"citation_coverage": citation_coverage
})
logger.info(f"🔍 [Quality Analysis] Overall score calculated: {overall_score}")
# Generate recommendations
recommendations = self._generate_recommendations({
"factual_accuracy": factual_accuracy,
"source_verification": source_verification,
"professional_tone": professional_tone,
"industry_relevance": industry_relevance,
"citation_coverage": citation_coverage
})
logger.info(f"🔍 [Quality Analysis] Generated {len(recommendations)} recommendations")
result = {
"overall_score": overall_score,
"metrics": {
"factual_accuracy": factual_accuracy,
"source_verification": source_verification,
"professional_tone": professional_tone,
"industry_relevance": industry_relevance,
"citation_coverage": citation_coverage
},
"recommendations": recommendations,
"content_length": len(content),
"word_count": len(content.split()),
"analysis_timestamp": self._get_timestamp()
}
logger.info(f"🔍 [Quality Analysis] Final result: {result}")
return result
except Exception as e:
logger.error(f"Content quality analysis failed: {str(e)}")
return {
"overall_score": 0.0,
"error": str(e),
"metrics": {},
"recommendations": ["Content quality analysis failed. Please try again."]
}
def _assess_factual_accuracy(self, content: str, sources: List[Dict[str, Any]]) -> float:
"""
Assess factual accuracy based on source verification.
Args:
content: The content to analyze
sources: Research sources used
Returns:
Factual accuracy score between 0.0 and 1.0
"""
logger.info(f"🔍 [Factual Accuracy] Starting analysis with {len(sources)} sources")
logger.info(f"🔍 [Factual Accuracy] Content length: {len(content)} characters")
if not sources:
logger.warning("🔍 [Factual Accuracy] No sources provided, returning 0.0")
return 0.0
# Look for factual indicators in the content
factual_indicators = [
r'\d+%', r'\d+ percent', # Percentages
r'\$\d+', r'\d+ dollars', # Dollar amounts
r'\d+ million', r'\d+ billion', # Billions
r'research shows', r'studies indicate', r'data reveals',
r'experts say', r'according to', r'statistics show',
r'\d{4}', # Years
r'\d+ organizations', r'\d+ companies', r'\d+ enterprises',
r'AI', r'artificial intelligence', r'machine learning', # Technology terms
r'content creation', r'digital marketing', r'technology industry', # Industry terms
r'efficiency', r'innovation', r'development', r'growth', # Business terms
r'businesses', r'companies', r'organizations', # Entity terms
r'tools', r'platforms', r'systems', r'solutions' # Product terms
]
factual_claims = 0
supported_claims = 0
for pattern in factual_indicators:
matches = re.findall(pattern, content, re.IGNORECASE)
if matches:
logger.info(f"🔍 [Factual Accuracy] Pattern {pattern} found {len(matches)} matches: {matches}")
factual_claims += len(matches)
# Check if claims are near citations
for match in matches:
if self._is_claim_supported(match, content, sources):
supported_claims += 1
logger.info(f"🔍 [Factual Accuracy] Total factual claims: {factual_claims}")
logger.info(f"🔍 [Factual Accuracy] Supported claims: {supported_claims}")
# Calculate accuracy score - be more lenient
if factual_claims == 0:
logger.info("🔍 [Factual Accuracy] No factual claims to verify, returning 0.8")
return 0.8 # No factual claims to verify
# Base accuracy score
accuracy_score = supported_claims / factual_claims
logger.info(f"🔍 [Factual Accuracy] Base accuracy score: {accuracy_score}")
# Boost score if we have good source quality
if sources:
avg_credibility = sum(
(s.credibility_score or 0) if hasattr(s, 'credibility_score') else (s.get("credibility_score", 0) or 0)
for s in sources
) / len(sources)
logger.info(f"🔍 [Factual Accuracy] Average credibility: {avg_credibility}")
# Boost accuracy if sources are credible
if avg_credibility > 0.7:
accuracy_score = min(accuracy_score * 1.3, 1.0)
logger.info(f"🔍 [Factual Accuracy] Applied high credibility boost: {accuracy_score}")
elif avg_credibility > 0.5:
accuracy_score = min(accuracy_score * 1.1, 1.0)
logger.info(f"🔍 [Factual Accuracy] Applied medium credibility boost: {accuracy_score}")
# Boost score if we have multiple sources (diversity)
if len(sources) >= 3:
accuracy_score = min(accuracy_score * 1.2, 1.0)
logger.info(f"🔍 [Factual Accuracy] Applied diversity boost: {accuracy_score}")
final_score = round(min(accuracy_score, 1.0), 3)
logger.info(f"🔍 [Factual Accuracy] Final accuracy score: {final_score}")
return final_score
def _assess_source_verification(self, content: str, sources: List[Dict[str, Any]]) -> float:
"""
Assess source verification quality.
Args:
content: The content to analyze
sources: Research sources used
Returns:
Source verification score between 0.0 and 1.0
"""
if not sources:
return 0.0
# Calculate source quality metrics
total_sources = len(sources)
# Source credibility scores - handle both Dict and ResearchSource objects
credibility_scores = []
relevance_scores = []
domain_scores = []
source_types = set()
for s in sources:
if hasattr(s, 'credibility_score'):
# ResearchSource Pydantic model
credibility_scores.append(s.credibility_score or 0)
relevance_scores.append(s.relevance_score or 0)
domain_scores.append(s.domain_authority or 0)
source_types.add(s.source_type or "general")
else:
# Dictionary object
credibility_scores.append(s.get("credibility_score", 0))
relevance_scores.append(s.get("relevance_score", 0))
domain_scores.append(s.get("domain_authority", 0))
source_types.add(s.get("source_type", "general"))
avg_credibility = sum(credibility_scores) / len(credibility_scores) if credibility_scores else 0
avg_relevance = sum(relevance_scores) / len(relevance_scores) if relevance_scores else 0
avg_domain_authority = sum(domain_scores) / len(domain_scores) if domain_scores else 0
diversity_score = min(len(source_types) / 3, 1.0) # Normalize to 3+ types
# Calculate verification score
verification_score = (
avg_credibility * 0.3 +
avg_relevance * 0.3 +
avg_domain_authority * 0.2 +
diversity_score * 0.2
)
return round(verification_score, 3)
def _assess_professional_tone(self, content: str) -> float:
"""
Assess professional tone appropriateness.
Args:
content: The content to analyze
Returns:
Professional tone score between 0.0 and 1.0
"""
content_lower = content.lower()
# Count professional indicators
professional_count = sum(1 for indicator in self.professional_indicators if indicator in content_lower)
# Count unprofessional indicators
unprofessional_count = sum(1 for indicator in self.unprofessional_indicators if indicator in content_lower)
# Calculate tone score
total_indicators = len(self.professional_indicators) + len(self.unprofessional_indicators)
if total_indicators == 0:
return 0.7 # Neutral score
professional_score = professional_count / len(self.professional_indicators)
unprofessional_penalty = unprofessional_count / len(self.unprofessional_indicators)
tone_score = professional_score - unprofessional_penalty
tone_score = max(0.0, min(1.0, tone_score)) # Clamp between 0 and 1
return round(tone_score, 3)
def _assess_industry_relevance(self, content: str, industry: str) -> float:
"""
Assess industry relevance of the content.
Args:
content: The content to analyze
industry: The target industry
Returns:
Industry relevance score between 0.0 and 1.0
"""
if industry.lower() == "general":
return 0.7 # Neutral score for general industry
content_lower = content.lower()
industry_lower = industry.lower()
# Get industry-specific terminology
industry_terms = self.industry_terminology.get(industry, [])
# Count industry-specific terms
industry_term_count = sum(1 for term in industry_terms if term in content_lower)
# Count industry mentions
industry_mentions = content_lower.count(industry_lower)
# Calculate relevance score
if not industry_terms:
return 0.6 # Fallback score
term_relevance = min(industry_term_count / len(industry_terms), 1.0)
mention_relevance = min(industry_mentions / 3, 1.0) # Normalize to 3+ mentions
relevance_score = (term_relevance * 0.7) + (mention_relevance * 0.3)
return round(relevance_score, 3)
def _assess_citation_coverage(self, content: str, sources: List[Dict[str, Any]]) -> float:
"""
Assess citation coverage in the content.
Args:
content: The content to analyze
sources: Research sources used
Returns:
Citation coverage score between 0.0 and 1.0
"""
logger.info(f"🔍 [Citation Coverage] Starting analysis with {len(sources)} sources")
logger.info(f"🔍 [Citation Coverage] Content length: {len(content)} characters")
# Debug: Show sample of content to see what we're analyzing
content_sample = content[:500] + "..." if len(content) > 500 else content
logger.info(f"🔍 [Citation Coverage] Content sample: {content_sample}")
if not sources:
logger.warning("🔍 [Citation Coverage] No sources provided, returning 0.0")
return 0.0
# Look for citation patterns - updated to match our actual citation format
citation_patterns = [
r'<sup class="liw-cite"[^>]*>\[(\d+)\]</sup>', # HTML format - PRIORITY 1
r'\[(\d+)\]', # Our primary format: [1], [2], etc.
r'\[Source (\d+)\]', r'\(Source (\d+)\)',
r'\((\d+)\)', r'Source (\d+)', r'Ref\. (\d+)', r'Reference (\d+)'
]
total_citations = 0
for pattern in citation_patterns:
matches = re.findall(pattern, content, re.IGNORECASE)
if matches:
logger.info(f"🔍 [Citation Coverage] Pattern {pattern} found {len(matches)} matches: {matches}")
total_citations += len(matches)
logger.info(f"🔍 [Citation Coverage] Total citations found: {total_citations}")
# Calculate coverage score - be more lenient since we strategically place citations
expected_citations = min(len(sources), len(sources) * 0.8) # Allow 80% coverage
if expected_citations == 0:
logger.warning("🔍 [Citation Coverage] Expected citations is 0, returning 0.0")
return 0.0
coverage_score = min(total_citations / expected_citations, 1.0)
logger.info(f"🔍 [Citation Coverage] Coverage score before boost: {coverage_score}")
# Boost score if we have good source diversity
if len(sources) >= 3:
coverage_score = min(coverage_score * 1.2, 1.0)
logger.info(f"🔍 [Citation Coverage] Applied diversity boost, final score: {coverage_score}")
final_score = round(coverage_score, 3)
logger.info(f"🔍 [Citation Coverage] Final coverage score: {final_score}")
return final_score
def _is_claim_supported(self, claim: str, content: str, sources: List[Dict[str, Any]]) -> bool:
"""
Check if a factual claim is supported by nearby citations.
Args:
claim: The factual claim to check
content: The content containing the claim
sources: Research sources used
Returns:
True if the claim appears to be supported
"""
# Find the position of the claim
claim_pos = content.lower().find(claim.lower())
if claim_pos == -1:
return False
# Look for citations within 300 characters of the claim (increased range)
start_pos = max(0, claim_pos - 150)
end_pos = min(len(content), claim_pos + len(claim) + 150)
nearby_text = content[start_pos:end_pos]
# Check for citation patterns - updated to match our actual format
citation_patterns = [
r'<sup class="liw-cite"[^>]*>\[(\d+)\]</sup>', # HTML format - PRIORITY 1
r'\[(\d+)\]', # Our primary format: [1], [2], etc.
r'\[Source (\d+)\]', r'\[(\d+)\]', r'\(Source (\d+)\)',
r'\((\d+)\)', r'Source (\d+)', r'Ref\. (\d+)', r'Reference (\d+)'
]
for pattern in citation_patterns:
if re.search(pattern, nearby_text, re.IGNORECASE):
return True
return False
def _calculate_overall_score(self, metrics: Dict[str, float]) -> float:
"""
Calculate overall quality score from individual metrics.
Args:
metrics: Dictionary of quality metrics
Returns:
Overall quality score between 0.0 and 1.0
"""
# Weighted scoring system
weights = {
"factual_accuracy": 0.25,
"source_verification": 0.25,
"professional_tone": 0.20,
"industry_relevance": 0.15,
"citation_coverage": 0.15
}
overall_score = 0.0
total_weight = 0.0
for metric_name, weight in weights.items():
if metric_name in metrics:
overall_score += metrics[metric_name] * weight
total_weight += weight
if total_weight == 0:
return 0.0
final_score = overall_score / total_weight
return round(final_score, 3)
def _generate_recommendations(self, metrics: Dict[str, float]) -> List[str]:
"""
Generate improvement recommendations based on quality metrics.
Args:
metrics: Dictionary of quality metrics
Returns:
List of improvement recommendations
"""
recommendations = []
# Factual accuracy recommendations
if metrics.get("factual_accuracy", 0) < 0.7:
recommendations.append("Improve factual accuracy by ensuring all claims are properly supported by sources.")
if metrics.get("factual_accuracy", 0) < 0.5:
recommendations.append("Significant factual accuracy issues detected. Review and verify all claims against sources.")
# Source verification recommendations
if metrics.get("source_verification", 0) < 0.6:
recommendations.append("Enhance source quality by using more credible and relevant sources.")
if metrics.get("source_verification", 0) < 0.4:
recommendations.append("Low source verification quality. Consider using more authoritative and recent sources.")
# Professional tone recommendations
if metrics.get("professional_tone", 0) < 0.7:
recommendations.append("Improve professional tone by using more industry-appropriate language.")
if metrics.get("professional_tone", 0) < 0.5:
recommendations.append("Content tone needs significant improvement for professional audiences.")
# Industry relevance recommendations
if metrics.get("industry_relevance", 0) < 0.6:
recommendations.append("Increase industry relevance by using more industry-specific terminology and examples.")
if metrics.get("industry_relevance", 0) < 0.4:
recommendations.append("Content lacks industry focus. Add more industry-specific content and context.")
# Citation coverage recommendations
if metrics.get("citation_coverage", 0) < 0.8:
recommendations.append("Improve citation coverage by adding more inline citations throughout the content.")
if metrics.get("citation_coverage", 0) < 0.5:
recommendations.append("Low citation coverage. Add citations for all factual claims and data points.")
# General recommendations
if not recommendations:
recommendations.append("Content quality is good. Consider adding more specific examples or expanding on key points.")
return recommendations
def _get_timestamp(self) -> str:
"""Get current timestamp for analysis tracking."""
from datetime import datetime
return datetime.utcnow().isoformat()
def track_quality_over_time(
self,
content_id: str,
quality_metrics: Dict[str, Any]
) -> Dict[str, Any]:
"""
Track content quality metrics over time for analysis.
Args:
content_id: Unique identifier for the content
quality_metrics: Quality analysis results
Returns:
Tracking information and trends
"""
# This would typically integrate with a database or analytics system
# For now, we'll return the tracking structure
tracking_data = {
"content_id": content_id,
"timestamp": quality_metrics.get("analysis_timestamp"),
"overall_score": quality_metrics.get("overall_score", 0.0),
"metrics": quality_metrics.get("metrics", {}),
"content_length": quality_metrics.get("content_length", 0),
"word_count": quality_metrics.get("word_count", 0)
}
logger.info(f"Quality metrics tracked for content {content_id}: {tracking_data['overall_score']}")
return {
"tracked": True,
"tracking_data": tracking_data,
"message": f"Quality metrics tracked for content {content_id}"
}
def compare_content_quality(
self,
content_a: Dict[str, Any],
content_b: Dict[str, Any]
) -> Dict[str, Any]:
"""
Compare quality between two pieces of content.
Args:
content_a: Quality metrics for first content piece
content_b: Quality metrics for second content piece
Returns:
Comparison analysis and recommendations
"""
comparison = {
"content_a_score": content_a.get("overall_score", 0.0),
"content_b_score": content_b.get("overall_score", 0.0),
"score_difference": 0.0,
"better_content": "content_a",
"improvement_areas": [],
"strength_areas": []
}
# Calculate score difference
score_a = content_a.get("overall_score", 0.0)
score_b = content_b.get("overall_score", 0.0)
comparison["score_difference"] = round(abs(score_a - score_b), 3)
# Determine better content
if score_a > score_b:
comparison["better_content"] = "content_a"
better_metrics = content_a.get("metrics", {})
worse_metrics = content_b.get("metrics", {})
else:
comparison["better_content"] = "content_b"
better_metrics = content_b.get("metrics", {})
worse_metrics = content_a.get("metrics", {})
# Identify improvement areas
for metric_name in better_metrics:
if metric_name in worse_metrics:
if worse_metrics[metric_name] < better_metrics[metric_name] - 0.2:
comparison["improvement_areas"].append(f"Improve {metric_name.replace('_', ' ')}")
# Identify strength areas
for metric_name in better_metrics:
if better_metrics[metric_name] > 0.8:
comparison["strength_areas"].append(f"Strong {metric_name.replace('_', ' ')}")
return comparison
def generate_quality_report(
self,
content: str,
sources: List[Any],
industry: str = "general"
) -> Dict[str, Any]:
"""
Generate a comprehensive quality report for content.
Args:
content: The content to analyze
sources: Research sources used (can be Dict or ResearchSource objects)
industry: Target industry
Returns:
Comprehensive quality report
"""
# Perform full quality analysis
quality_analysis = self.analyze_content_quality(content, sources, industry)
# Generate detailed report
report = {
"summary": {
"overall_score": quality_analysis["overall_score"],
"quality_level": self._get_quality_level(quality_analysis["overall_score"]),
"content_length": quality_analysis["content_length"],
"word_count": quality_analysis["word_count"]
},
"detailed_metrics": quality_analysis["metrics"],
"recommendations": quality_analysis["recommendations"],
"source_analysis": {
"total_sources": len(sources),
"source_types": self._extract_source_types(sources),
"avg_credibility": self._calculate_avg_score(sources, "credibility_score"),
"avg_relevance": self._calculate_avg_score(sources, "relevance_score")
},
"improvement_plan": self._generate_improvement_plan(quality_analysis["metrics"]),
"analysis_timestamp": quality_analysis["analysis_timestamp"]
}
return report
def _get_quality_level(self, score: float) -> str:
"""Convert numerical score to quality level description."""
if score >= 0.9:
return "Excellent"
elif score >= 0.8:
return "Very Good"
elif score >= 0.7:
return "Good"
elif score >= 0.6:
return "Fair"
elif score >= 0.5:
return "Below Average"
else:
return "Poor"
def _generate_improvement_plan(self, metrics: Dict[str, float]) -> Dict[str, Any]:
"""
Generate a structured improvement plan based on quality metrics.
Args:
metrics: Quality metrics dictionary
Returns:
Structured improvement plan
"""
improvement_plan = {
"priority_high": [],
"priority_medium": [],
"priority_low": [],
"estimated_effort": "medium"
}
# Categorize improvements by priority
for metric_name, score in metrics.items():
if score < 0.4:
improvement_plan["priority_high"].append(f"Significantly improve {metric_name.replace('_', ' ')}")
elif score < 0.6:
improvement_plan["priority_medium"].append(f"Improve {metric_name.replace('_', ' ')}")
elif score < 0.8:
improvement_plan["priority_low"].append(f"Enhance {metric_name.replace('_', ' ')}")
# Estimate effort based on number of high-priority items
high_priority_count = len(improvement_plan["priority_high"])
if high_priority_count >= 3:
improvement_plan["estimated_effort"] = "high"
elif high_priority_count >= 1:
improvement_plan["estimated_effort"] = "medium"
else:
improvement_plan["estimated_effort"] = "low"
return improvement_plan
def _extract_source_types(self, sources: List[Any]) -> List[str]:
"""Extract source types from sources, handling both Dict and ResearchSource objects."""
source_types = set()
for s in sources:
if hasattr(s, 'source_type'):
# ResearchSource Pydantic model
source_types.add(s.source_type or "general")
else:
# Dictionary object
source_types.add(s.get("source_type", "general"))
return list(source_types)
def _calculate_avg_score(self, sources: List[Any], score_field: str) -> float:
"""Calculate average score from sources, handling both Dict and ResearchSource objects."""
if not sources:
return 0.0
total_score = 0.0
valid_sources = 0
for s in sources:
if hasattr(s, score_field):
# ResearchSource Pydantic model
score = getattr(s, score_field)
if score is not None:
total_score += score
valid_sources += 1
else:
# Dictionary object
score = s.get(score_field, 0)
if score:
total_score += score
valid_sources += 1
return total_score / valid_sources if valid_sources > 0 else 0.0

21
backend/services/research/__init__.py Normal file
View File

@@ -0,0 +1,21 @@
"""
Research Services Module for ALwrity
This module provides research and grounding capabilities for content generation,
replacing mock research with real-time industry information.
Available Services:
- GoogleSearchService: Real-time industry research using Google Custom Search API
- Source ranking and credibility assessment
- Content extraction and insight generation
Author: ALwrity Team
Version: 1.0
Last Updated: January 2025
"""
from services.research.google_search_service import GoogleSearchService
__all__ = [
"GoogleSearchService"
]

542
backend/services/research/google_search_service.py Normal file
View File

@@ -0,0 +1,542 @@
"""
Google Search Service for ALwrity
This service provides real-time industry research using Google Custom Search API,
replacing the mock research system with actual web search capabilities.
Key Features:
- Industry-specific search queries
- Source credibility scoring and ranking
- Content extraction and insight generation
- Real-time information from the last month
- Fallback mechanisms for API failures
Dependencies:
- google-api-python-client
- aiohttp (for async HTTP requests)
- os (for environment variables)
- logging (for debugging)
Author: ALwrity Team
Version: 1.0
Last Updated: January 2025
"""
import os
import json
import asyncio
import aiohttp
from typing import Dict, List, Optional, Any
from datetime import datetime, timedelta
from loguru import logger
class GoogleSearchService:
"""
Service for conducting real industry research using Google Custom Search API.
This service replaces the mock research system with actual web search capabilities,
providing current, relevant industry information for content grounding.
"""
def __init__(self):
"""Initialize the Google Search Service with API credentials."""
self.api_key = os.getenv("GOOGLE_SEARCH_API_KEY")
self.search_engine_id = os.getenv("GOOGLE_SEARCH_ENGINE_ID")
self.base_url = "https://www.googleapis.com/customsearch/v1"
if not self.api_key or not self.search_engine_id:
logger.warning("Google Search API credentials not configured. Service will use fallback methods.")
self.enabled = False
else:
self.enabled = True
logger.info("Google Search Service initialized successfully")
async def search_industry_trends(
self,
topic: str,
industry: str,
max_results: int = 10
) -> List[Dict[str, Any]]:
"""
Search for current industry trends and insights.
Args:
topic: The specific topic to research
industry: The industry context for the search
max_results: Maximum number of search results to return
Returns:
List of search results with credibility scoring
"""
if not self.enabled:
logger.warning("Google Search Service not enabled, using fallback research")
return await self._fallback_research(topic, industry)
try:
# Construct industry-specific search query
search_query = self._build_search_query(topic, industry)
logger.info(f"Searching for: {search_query}")
# Perform the search
search_results = await self._perform_search(search_query, max_results)
# Process and rank results
processed_results = await self._process_search_results(search_results, topic, industry)
# Extract insights and statistics
insights = await self._extract_insights(processed_results, topic, industry)
logger.info(f"Search completed successfully. Found {len(processed_results)} relevant sources.")
return {
"sources": processed_results,
"key_insights": insights["insights"],
"statistics": insights["statistics"],
"grounding_enabled": True,
"search_query": search_query,
"timestamp": datetime.utcnow().isoformat()
}
except Exception as e:
logger.error(f"Google search failed: {str(e)}")
return await self._fallback_research(topic, industry)
def _build_search_query(self, topic: str, industry: str) -> str:
"""
Build an optimized search query for industry research.
Args:
topic: The specific topic to research
industry: The industry context
Returns:
Optimized search query string
"""
# Add industry-specific terms and current year for relevance
current_year = datetime.now().year
# Industry-specific search patterns
industry_patterns = {
"Technology": ["trends", "innovations", "developments", "insights"],
"Healthcare": ["advances", "research", "treatments", "studies"],
"Finance": ["market analysis", "trends", "reports", "insights"],
"Marketing": ["strategies", "trends", "best practices", "case studies"],
"Education": ["innovations", "trends", "research", "best practices"]
}
# Get industry-specific terms
industry_terms = industry_patterns.get(industry, ["trends", "insights", "developments"])
# Build the query
query_components = [
topic,
industry,
f"{current_year}",
"latest",
"trends",
"insights"
]
# Add industry-specific terms
query_components.extend(industry_terms[:2])
return " ".join(query_components)
async def _perform_search(self, query: str, max_results: int) -> List[Dict[str, Any]]:
"""
Perform the actual Google Custom Search API call.
Args:
query: The search query to execute
max_results: Maximum number of results to return
Returns:
Raw search results from Google API
"""
params = {
"key": self.api_key,
"cx": self.search_engine_id,
"q": query,
"num": min(max_results, 10), # Google CSE max is 10 per request
"dateRestrict": "m1", # Last month
"sort": "date", # Sort by date for current information
"safe": "active" # Safe search for professional content
}
async with aiohttp.ClientSession() as session:
async with session.get(self.base_url, params=params) as response:
if response.status == 200:
data = await response.json()
return data.get("items", [])
else:
error_text = await response.text()
logger.error(f"Google Search API error: {response.status} - {error_text}")
raise Exception(f"Search API returned status {response.status}")
async def _process_search_results(
self,
raw_results: List[Dict[str, Any]],
topic: str,
industry: str
) -> List[Dict[str, Any]]:
"""
Process and rank search results by relevance and credibility.
Args:
raw_results: Raw search results from Google API
topic: The research topic for relevance scoring
industry: The industry context for relevance scoring
Returns:
Processed and ranked search results
"""
processed_results = []
for result in raw_results:
try:
# Extract basic information
title = result.get("title", "")
url = result.get("link", "")
snippet = result.get("snippet", "")
# Calculate relevance score
relevance_score = self._calculate_relevance_score(title, snippet, topic, industry)
# Calculate credibility score
credibility_score = self._calculate_credibility_score(url, title)
# Extract publication date if available
publication_date = self._extract_publication_date(result)
# Calculate domain authority
domain_authority = self._calculate_domain_authority(url)
processed_result = {
"title": title,
"url": url,
"content": snippet,
"relevance_score": relevance_score,
"credibility_score": credibility_score,
"domain_authority": domain_authority,
"publication_date": publication_date,
"source_type": self._categorize_source(url, title),
"raw_result": result
}
processed_results.append(processed_result)
except Exception as e:
logger.warning(f"Failed to process search result: {str(e)}")
continue
# Sort by combined score (relevance + credibility)
processed_results.sort(
key=lambda x: (x["relevance_score"] + x["credibility_score"]) / 2,
reverse=True
)
return processed_results
def _calculate_relevance_score(self, title: str, snippet: str, topic: str, industry: str) -> float:
"""
Calculate relevance score based on topic and industry alignment.
Args:
title: The title of the search result
snippet: The snippet/description of the result
topic: The research topic
industry: The industry context
Returns:
Relevance score between 0.0 and 1.0
"""
score = 0.0
text = f"{title} {snippet}".lower()
# Topic relevance (40% of score)
topic_words = topic.lower().split()
topic_matches = sum(1 for word in topic_words if word in text)
topic_score = min(topic_matches / len(topic_words), 1.0) * 0.4
# Industry relevance (30% of score)
industry_words = industry.lower().split()
industry_matches = sum(1 for word in industry_words if word in text)
industry_score = min(industry_matches / len(industry_words), 1.0) * 0.3
# Content quality indicators (30% of score)
quality_indicators = [
"research", "study", "analysis", "report", "insights",
"trends", "data", "statistics", "findings", "expert"
]
quality_matches = sum(1 for indicator in quality_indicators if indicator in text)
quality_score = min(quality_matches / len(quality_indicators), 1.0) * 0.3
score = topic_score + industry_score + quality_score
return round(score, 3)
def _calculate_credibility_score(self, url: str, title: str) -> float:
"""
Calculate credibility score based on URL and title analysis.
Args:
url: The URL of the source
title: The title of the content
Returns:
Credibility score between 0.0 and 1.0
"""
score = 0.5 # Base score
# Domain credibility indicators
credible_domains = [
"harvard.edu", "stanford.edu", "mit.edu", "berkeley.edu", # Academic
"forbes.com", "bloomberg.com", "reuters.com", "wsj.com", # Business
"nature.com", "science.org", "ieee.org", "acm.org", # Scientific
"linkedin.com", "medium.com", "substack.com" # Professional
]
# Check if domain is in credible list
domain = self._extract_domain(url)
if any(credible_domain in domain for credible_domain in credible_domains):
score += 0.3
# Title credibility indicators
credible_indicators = [
"research", "study", "analysis", "report", "insights",
"expert", "professional", "industry", "trends"
]
title_lower = title.lower()
credible_matches = sum(1 for indicator in credible_indicators if indicator in title_lower)
score += min(credible_matches * 0.1, 0.2)
return round(min(score, 1.0), 3)
def _calculate_domain_authority(self, url: str) -> float:
"""
Calculate domain authority based on URL analysis.
Args:
url: The URL to analyze
Returns:
Domain authority score between 0.0 and 1.0
"""
domain = self._extract_domain(url)
# High authority domains
high_authority = [
"harvard.edu", "stanford.edu", "mit.edu", "berkeley.edu",
"forbes.com", "bloomberg.com", "reuters.com", "wsj.com",
"nature.com", "science.org", "ieee.org", "acm.org"
]
# Medium authority domains
medium_authority = [
"linkedin.com", "medium.com", "substack.com", "techcrunch.com",
"venturebeat.com", "wired.com", "theverge.com"
]
if any(auth_domain in domain for auth_domain in high_authority):
return 0.9
elif any(auth_domain in domain for auth_domain in medium_authority):
return 0.7
else:
# Basic scoring for other domains
return 0.5
def _extract_domain(self, url: str) -> str:
"""Extract domain from URL."""
try:
from urllib.parse import urlparse
parsed = urlparse(url)
return parsed.netloc.lower()
except:
return url.lower()
def _extract_publication_date(self, result: Dict[str, Any]) -> Optional[str]:
"""Extract publication date from search result if available."""
# Check for various date fields
date_fields = ["pagemap", "metatags", "date"]
for field in date_fields:
if field in result:
date_value = result[field]
if isinstance(date_value, dict):
# Look for common date keys
for date_key in ["date", "pubdate", "article:published_time"]:
if date_key in date_value:
return date_value[date_key]
elif isinstance(date_value, str):
return date_value
return None
def _categorize_source(self, url: str, title: str) -> str:
"""Categorize the source type based on URL and title."""
domain = self._extract_domain(url)
title_lower = title.lower()
# Academic sources
if any(edu in domain for edu in [".edu", "harvard", "stanford", "mit"]):
return "academic"
# Business/News sources
if any(biz in domain for biz in ["forbes", "bloomberg", "reuters", "wsj"]):
return "business_news"
# Professional platforms
if any(prof in domain for prof in ["linkedin", "medium", "substack"]):
return "professional_platform"
# Research/Scientific
if any(research in domain for research in ["nature", "science", "ieee", "acm"]):
return "research_scientific"
# Industry reports
if any(report in title_lower for report in ["report", "study", "analysis", "research"]):
return "industry_report"
return "general"
async def _extract_insights(
self,
sources: List[Dict[str, Any]],
topic: str,
industry: str
) -> Dict[str, List[str]]:
"""
Extract key insights and statistics from search results.
Args:
sources: Processed search results
topic: The research topic
industry: The industry context
Returns:
Dictionary containing insights and statistics
"""
insights = []
statistics = []
# Extract insights from top sources
top_sources = sources[:5] # Top 5 most relevant sources
for source in top_sources:
content = source.get("content", "")
# Look for insight patterns
insight_patterns = [
"shows", "indicates", "suggests", "reveals", "demonstrates",
"highlights", "emphasizes", "points to", "suggests that"
]
for pattern in insight_patterns:
if pattern in content.lower():
# Extract the sentence containing the insight
sentences = content.split(". ")
for sentence in sentences:
if pattern in sentence.lower():
insights.append(sentence.strip())
break
# Look for statistical patterns
stat_patterns = [
r'\d+%', # Percentages
r'\d+ percent', # Written percentages
r'\$\d+', # Dollar amounts
r'\d+ million', # Millions
r'\d+ billion', # Billions
r'\d+ out of \d+', # Ratios
]
import re
for pattern in stat_patterns:
matches = re.findall(pattern, content, re.IGNORECASE)
for match in matches:
statistics.append(f"{match}")
# Limit the number of insights and statistics
insights = insights[:10] # Top 10 insights
statistics = statistics[:10] # Top 10 statistics
return {
"insights": insights,
"statistics": statistics
}
async def _fallback_research(self, topic: str, industry: str) -> Dict[str, Any]:
"""
Fallback research method when Google Search is not available.
Args:
topic: The research topic
industry: The industry context
Returns:
Fallback research data
"""
logger.info(f"Using fallback research for {topic} in {industry}")
return {
"sources": [
{
"title": f"Industry insights on {topic} in {industry}",
"url": f"https://example.com/{topic.lower().replace(' ', '-')}",
"content": f"Professional insights and trends related to {topic} in the {industry} sector...",
"relevance_score": 0.8,
"credibility_score": 0.6,
"domain_authority": 0.5,
"source_type": "general",
"grounding_enabled": False
}
],
"key_insights": [
f"{topic} is transforming {industry} operations",
f"Industry leaders are investing in {topic}",
f"Expected growth in {topic} adoption within {industry}"
],
"statistics": [
f"85% of {industry} companies are exploring {topic}",
f"Investment in {topic} increased by 40% this year"
],
"grounding_enabled": False,
"search_query": f"{topic} {industry} trends",
"timestamp": datetime.utcnow().isoformat()
}
async def test_api_connection(self) -> Dict[str, Any]:
"""
Test the Google Search API connection.
Returns:
Test results and status information
"""
if not self.enabled:
return {
"status": "disabled",
"message": "Google Search API credentials not configured",
"enabled": False
}
try:
# Perform a simple test search
test_query = "AI technology trends 2024"
test_results = await self._perform_search(test_query, 1)
return {
"status": "success",
"message": "Google Search API connection successful",
"enabled": True,
"test_results_count": len(test_results),
"api_key_configured": bool(self.api_key),
"search_engine_configured": bool(self.search_engine_id)
}
except Exception as e:
return {
"status": "error",
"message": f"Google Search API connection failed: {str(e)}",
"enabled": False,
"error": str(e)
}

62
backend/test_grounding_flow.py Normal file
View File

@@ -0,0 +1,62 @@
#!/usr/bin/env python3
"""
Test script to debug the grounding data flow
"""
import asyncio
import sys
import os
# Add the backend directory to the path
sys.path.append(os.path.dirname(os.path.abspath(__file__)))
from services.linkedin_service import LinkedInService
from models.linkedin_models import LinkedInPostRequest, GroundingLevel
async def test_grounding_flow():
"""Test the grounding data flow"""
try:
print("🔍 Testing grounding data flow...")
# Initialize the service
service = LinkedInService()
print("✅ LinkedInService initialized")
# Create a test request
request = LinkedInPostRequest(
topic="AI in healthcare transformation",
industry="Healthcare",
grounding_level=GroundingLevel.ENHANCED,
include_citations=True,
research_enabled=True,
search_engine="google",
max_length=2000
)
print("✅ Test request created")
# Generate post
print("🚀 Generating LinkedIn post...")
response = await service.generate_linkedin_post(request)
if response.success:
print("✅ Post generated successfully!")
print(f"📊 Research sources count: {len(response.research_sources) if response.research_sources else 0}")
print(f"📝 Citations count: {len(response.data.citations) if response.data.citations else 0}")
print(f"🔗 Source list: {response.data.source_list[:200] if response.data.source_list else 'None'}")
if response.research_sources:
print(f"📚 First research source: {response.research_sources[0]}")
print(f"📚 Research source types: {[type(s) for s in response.research_sources[:3]]}")
if response.data.citations:
print(f"📝 First citation: {response.data.citations[0]}")
else:
print(f"❌ Post generation failed: {response.error}")
except Exception as e:
print(f"❌ Error during test: {str(e)}")
import traceback
traceback.print_exc()
if __name__ == "__main__":
asyncio.run(test_grounding_flow())

228
backend/test_grounding_integration.py Normal file
View File

@@ -0,0 +1,228 @@
"""
Test script for LinkedIn grounding integration.
This script tests the integration of the new grounding services:
- GoogleSearchService
- GeminiGroundedProvider
- CitationManager
- ContentQualityAnalyzer
- Enhanced LinkedInService
"""
import asyncio
import os
from datetime import datetime
from loguru import logger
# Set up environment variables for testing
os.environ.setdefault('GOOGLE_SEARCH_API_KEY', 'test_key')
os.environ.setdefault('GOOGLE_SEARCH_ENGINE_ID', 'test_engine_id')
os.environ.setdefault('GEMINI_API_KEY', 'test_gemini_key')
from services.linkedin_service import LinkedInService
from models.linkedin_models import (
LinkedInPostRequest, LinkedInArticleRequest, LinkedInCarouselRequest,
LinkedInVideoScriptRequest, LinkedInCommentResponseRequest,
GroundingLevel, SearchEngine, LinkedInTone, LinkedInPostType
)
async def test_grounding_integration():
"""Test the complete grounding integration."""
logger.info("Starting LinkedIn grounding integration test")
try:
# Initialize the enhanced LinkedIn service
linkedin_service = LinkedInService()
logger.info("LinkedIn service initialized successfully")
# Test 1: Basic post generation with grounding disabled
logger.info("\n=== Test 1: Basic Post Generation (No Grounding) ===")
basic_request = LinkedInPostRequest(
topic="AI in Marketing",
industry="Marketing",
post_type=LinkedInPostType.PROFESSIONAL,
tone=LinkedInTone.PROFESSIONAL,
research_enabled=False,
grounding_level=GroundingLevel.NONE,
include_citations=False
)
basic_response = await linkedin_service.generate_linkedin_post(basic_request)
logger.info(f"Basic post generation: {'SUCCESS' if basic_response.success else 'FAILED'}")
if basic_response.success:
logger.info(f"Content length: {basic_response.data.character_count}")
logger.info(f"Grounding enabled: {basic_response.data.grounding_enabled}")
# Test 2: Enhanced post generation with grounding enabled
logger.info("\n=== Test 2: Enhanced Post Generation (With Grounding) ===")
enhanced_request = LinkedInPostRequest(
topic="Digital Transformation in Healthcare",
industry="Healthcare",
post_type=LinkedInPostType.THOUGHT_LEADERSHIP,
tone=LinkedInTone.AUTHORITATIVE,
research_enabled=True,
search_engine=SearchEngine.GOOGLE,
grounding_level=GroundingLevel.ENHANCED,
include_citations=True,
max_length=2000
)
enhanced_response = await linkedin_service.generate_linkedin_post(enhanced_request)
logger.info(f"Enhanced post generation: {'SUCCESS' if enhanced_response.success else 'FAILED'}")
if enhanced_response.success:
logger.info(f"Content length: {enhanced_response.data.character_count}")
logger.info(f"Grounding enabled: {enhanced_response.data.grounding_enabled}")
logger.info(f"Research sources: {len(enhanced_response.research_sources)}")
logger.info(f"Citations: {len(enhanced_response.data.citations)}")
if enhanced_response.data.quality_metrics:
logger.info(f"Quality score: {enhanced_response.data.quality_metrics.overall_score:.2f}")
if enhanced_response.grounding_status:
logger.info(f"Grounding status: {enhanced_response.grounding_status['status']}")
# Test 3: Article generation with grounding
logger.info("\n=== Test 3: Article Generation (With Grounding) ===")
article_request = LinkedInArticleRequest(
topic="Future of Remote Work",
industry="Technology",
tone=LinkedInTone.EDUCATIONAL,
research_enabled=True,
search_engine=SearchEngine.GOOGLE,
grounding_level=GroundingLevel.ENHANCED,
include_citations=True,
word_count=1500
)
article_response = await linkedin_service.generate_linkedin_article(article_request)
logger.info(f"Article generation: {'SUCCESS' if article_response.success else 'FAILED'}")
if article_response.success:
logger.info(f"Word count: {article_response.data.word_count}")
logger.info(f"Grounding enabled: {article_response.data.grounding_enabled}")
logger.info(f"Research sources: {len(article_response.research_sources)}")
logger.info(f"Citations: {len(article_response.data.citations)}")
# Test 4: Carousel generation with grounding
logger.info("\n=== Test 4: Carousel Generation (With Grounding) ===")
carousel_request = LinkedInCarouselRequest(
topic="Cybersecurity Best Practices",
industry="Technology",
tone=LinkedInTone.EDUCATIONAL,
research_enabled=True,
search_engine=SearchEngine.GOOGLE,
grounding_level=GroundingLevel.ENHANCED,
include_citations=True,
number_of_slides=5
)
carousel_response = await linkedin_service.generate_linkedin_carousel(carousel_request)
logger.info(f"Carousel generation: {'SUCCESS' if carousel_response.success else 'FAILED'}")
if carousel_response.success:
logger.info(f"Number of slides: {len(carousel_response.data.slides)}")
logger.info(f"Grounding enabled: {carousel_response.data.grounding_enabled}")
logger.info(f"Research sources: {len(carousel_response.research_sources)}")
# Test 5: Video script generation with grounding
logger.info("\n=== Test 5: Video Script Generation (With Grounding) ===")
video_request = LinkedInVideoScriptRequest(
topic="AI Ethics in Business",
industry="Technology",
tone=LinkedInTone.EDUCATIONAL,
research_enabled=True,
search_engine=SearchEngine.GOOGLE,
grounding_level=GroundingLevel.ENHANCED,
include_citations=True,
video_duration=90
)
video_response = await linkedin_service.generate_linkedin_video_script(video_request)
logger.info(f"Video script generation: {'SUCCESS' if video_response.success else 'FAILED'}")
if video_response.success:
logger.info(f"Grounding enabled: {video_response.data.grounding_enabled}")
logger.info(f"Research sources: {len(video_response.research_sources)}")
logger.info(f"Citations: {len(video_response.data.citations)}")
# Test 6: Comment response generation
logger.info("\n=== Test 6: Comment Response Generation ===")
comment_request = LinkedInCommentResponseRequest(
original_comment="Great insights on AI implementation!",
post_context="Post about AI transformation in healthcare",
industry="Healthcare",
tone=LinkedInTone.FRIENDLY,
response_length="medium",
include_questions=True,
research_enabled=False,
grounding_level=GroundingLevel.BASIC
)
comment_response = await linkedin_service.generate_linkedin_comment_response(comment_request)
logger.info(f"Comment response generation: {'SUCCESS' if comment_response.success else 'FAILED'}")
if comment_response.success:
logger.info(f"Response length: {len(comment_response.response) if comment_response.response else 0}")
logger.info(f"Grounding enabled: {comment_response.grounding_status['status'] if comment_response.grounding_status else 'N/A'}")
logger.info("\n=== Integration Test Summary ===")
logger.info("All tests completed successfully!")
except Exception as e:
logger.error(f"Integration test failed: {str(e)}")
raise
async def test_individual_services():
"""Test individual service components."""
logger.info("\n=== Testing Individual Service Components ===")
try:
# Test Google Search Service
from services.research import GoogleSearchService
google_search = GoogleSearchService()
logger.info("GoogleSearchService initialized successfully")
# Test Citation Manager
from services.citation import CitationManager
citation_manager = CitationManager()
logger.info("CitationManager initialized successfully")
# Test Content Quality Analyzer
from services.quality import ContentQualityAnalyzer
quality_analyzer = ContentQualityAnalyzer()
logger.info("ContentQualityAnalyzer initialized successfully")
# Test Gemini Grounded Provider
from services.llm_providers.gemini_grounded_provider import GeminiGroundedProvider
gemini_grounded = GeminiGroundedProvider()
logger.info("GeminiGroundedProvider initialized successfully")
logger.info("All individual services initialized successfully!")
except Exception as e:
logger.error(f"Service component test failed: {str(e)}")
raise
async def main():
"""Main test function."""
logger.info("Starting LinkedIn Grounding Integration Tests")
logger.info(f"Test timestamp: {datetime.now().isoformat()}")
try:
# Test individual services first
await test_individual_services()
# Test complete integration
await test_grounding_integration()
logger.info("\n🎉 All tests completed successfully!")
except Exception as e:
logger.error(f"Test suite failed: {str(e)}")
logger.error("Please check the error details above and ensure all services are properly configured.")
return 1
return 0
if __name__ == "__main__":
# Run the tests
exit_code = asyncio.run(main())
exit(exit_code)

139
backend/test_imports.py Normal file
View File

@@ -0,0 +1,139 @@
#!/usr/bin/env python3
"""
Simple test script to verify import issues are fixed.
This script tests that all the required services can be imported and initialized
without import errors.
Usage:
python test_imports.py
"""
import sys
import os
from pathlib import Path
# Add the backend directory to the Python path
backend_dir = Path(__file__).parent
sys.path.insert(0, str(backend_dir))
def test_imports():
"""Test that all required modules can be imported."""
print("🧪 Testing Imports...")
try:
print("📦 Testing LinkedIn Models...")
from models.linkedin_models import (
LinkedInPostRequest, LinkedInPostResponse, PostContent, ResearchSource,
LinkedInArticleRequest, LinkedInArticleResponse, ArticleContent,
LinkedInCarouselRequest, LinkedInCarouselResponse, CarouselContent, CarouselSlide,
LinkedInVideoScriptRequest, LinkedInVideoScriptResponse, VideoScript,
LinkedInCommentResponseRequest, LinkedInCommentResponseResult,
HashtagSuggestion, ImageSuggestion, Citation, ContentQualityMetrics,
GroundingLevel
)
print("✅ LinkedIn Models imported successfully")
except Exception as e:
print(f"❌ LinkedIn Models import failed: {e}")
return False
try:
print("📦 Testing Research Service...")
from services.research import GoogleSearchService
print("✅ Research Service imported successfully")
except Exception as e:
print(f"❌ Research Service import failed: {e}")
return False
try:
print("📦 Testing Citation Service...")
from services.citation import CitationManager
print("✅ Citation Service imported successfully")
except Exception as e:
print(f"❌ Citation Service import failed: {e}")
return False
try:
print("📦 Testing Quality Service...")
from services.quality import ContentQualityAnalyzer
print("✅ Quality Service imported successfully")
except Exception as e:
print(f"❌ Quality Service import failed: {e}")
return False
try:
print("📦 Testing LLM Providers...")
from services.llm_providers.gemini_provider import gemini_structured_json_response, gemini_text_response
print("✅ LLM Providers imported successfully")
except Exception as e:
print(f"❌ LLM Providers import failed: {e}")
return False
try:
print("📦 Testing Gemini Grounded Provider...")
from services.llm_providers.gemini_grounded_provider import GeminiGroundedProvider
print("✅ Gemini Grounded Provider imported successfully")
except Exception as e:
print(f"❌ Gemini Grounded Provider import failed: {e}")
return False
try:
print("📦 Testing LinkedIn Service...")
from services.linkedin_service import LinkedInService
print("✅ LinkedIn Service imported successfully")
except Exception as e:
print(f"❌ LinkedIn Service import failed: {e}")
return False
print("\n🎉 All imports successful!")
return True
def test_service_initialization():
"""Test that services can be initialized without errors."""
print("\n🔧 Testing Service Initialization...")
try:
print("📦 Initializing LinkedIn Service...")
from services.linkedin_service import LinkedInService
service = LinkedInService()
print("✅ LinkedIn Service initialized successfully")
# Check which services are available
print(f" - Google Search: {'✅' if service.google_search else '❌'}")
print(f" - Gemini Grounded: {'✅' if service.gemini_grounded else '❌'}")
print(f" - Citation Manager: {'✅' if service.citation_manager else '❌'}")
print(f" - Quality Analyzer: {'✅' if service.quality_analyzer else '❌'}")
print(f" - Fallback Provider: {'✅' if service.fallback_provider else '❌'}")
return True
except Exception as e:
print(f"❌ LinkedIn Service initialization failed: {e}")
return False
def main():
"""Main test function."""
print("🚀 Starting Import Tests")
print("=" * 50)
# Test imports
import_success = test_imports()
if import_success:
# Test service initialization
init_success = test_service_initialization()
if init_success:
print("\n🎉 SUCCESS: All tests passed!")
print("✅ Import issues have been resolved")
print("✅ Services can be initialized")
print("✅ Ready for testing native grounding")
else:
print("\n⚠️ PARTIAL SUCCESS: Imports work but initialization failed")
print("💡 This may be due to missing dependencies or configuration")
else:
print("\n❌ FAILURE: Import tests failed")
print("💡 There are still import issues to resolve")
sys.exit(1)
if __name__ == "__main__":
main()

105
backend/test_linkedin_service.py Normal file
View File

@@ -0,0 +1,105 @@
#!/usr/bin/env python3
"""
Test script for LinkedIn service functionality.
This script tests that the LinkedIn service can be initialized and
basic functionality works without errors.
Usage:
python test_linkedin_service.py
"""
import asyncio
import sys
import os
from pathlib import Path
# Add the backend directory to the Python path
backend_dir = Path(__file__).parent
sys.path.insert(0, str(backend_dir))
from loguru import logger
from models.linkedin_models import LinkedInPostRequest, GroundingLevel
from services.linkedin_service import LinkedInService
async def test_linkedin_service():
"""Test the LinkedIn service functionality."""
try:
logger.info("🧪 Testing LinkedIn Service Functionality")
# Initialize the service
logger.info("📦 Initializing LinkedIn Service...")
service = LinkedInService()
logger.info("✅ LinkedIn Service initialized successfully")
# Create a test request
test_request = LinkedInPostRequest(
topic="AI in Marketing",
industry="Technology",
tone="professional",
max_length=500,
target_audience="Marketing professionals",
key_points=["AI automation", "Personalization", "ROI improvement"],
research_enabled=True,
search_engine="google",
grounding_level=GroundingLevel.BASIC,
include_citations=True
)
logger.info("📝 Testing LinkedIn Post Generation...")
# Test post generation
response = await service.generate_linkedin_post(test_request)
if response.success:
logger.info("✅ LinkedIn post generation successful")
logger.info(f"📊 Content length: {len(response.data.content)} characters")
logger.info(f"🔗 Sources: {len(response.research_sources)}")
logger.info(f"📚 Citations: {len(response.data.citations)}")
logger.info(f"🏆 Quality score: {response.data.quality_metrics.overall_score if response.data.quality_metrics else 'N/A'}")
# Display a snippet of the generated content
content_preview = response.data.content[:200] + "..." if len(response.data.content) > 200 else response.data.content
logger.info(f"📄 Content preview: {content_preview}")
else:
logger.error(f"❌ LinkedIn post generation failed: {response.error}")
return False
logger.info("🎉 LinkedIn service test completed successfully!")
return True
except Exception as e:
logger.error(f"❌ LinkedIn service test failed: {str(e)}")
return False
async def main():
"""Main test function."""
logger.info("🚀 Starting LinkedIn Service Test")
logger.info("=" * 50)
success = await test_linkedin_service()
if success:
logger.info("\n🎉 SUCCESS: LinkedIn service is working correctly!")
logger.info("✅ Service initialization successful")
logger.info("✅ Post generation working")
logger.info("✅ Ready for production use")
else:
logger.error("\n❌ FAILURE: LinkedIn service test failed")
sys.exit(1)
if __name__ == "__main__":
# Configure logging
logger.remove()
logger.add(
sys.stderr,
format="<green>{time:YYYY-MM-DD HH:mm:ss}</green> | <level>{level: <8}</level> | <cyan>{name}</cyan>:<cyan>{function}</cyan>:<cyan>{line}</cyan> - <level>{message}</level>",
level="INFO"
)
# Run the test
asyncio.run(main())

239
backend/test_native_grounding.py Normal file
View File

@@ -0,0 +1,239 @@
#!/usr/bin/env python3
"""
Test script for native Google Search grounding implementation.
This script tests the new GeminiGroundedProvider that uses native Google Search
grounding instead of custom search implementation.
Usage:
python test_native_grounding.py
"""
import asyncio
import os
import sys
from pathlib import Path
# Add the backend directory to the Python path
backend_dir = Path(__file__).parent
sys.path.insert(0, str(backend_dir))
from loguru import logger
from services.llm_providers.gemini_grounded_provider import GeminiGroundedProvider
async def test_native_grounding():
"""Test the native Google Search grounding functionality."""
try:
logger.info("🧪 Testing Native Google Search Grounding")
# Check if GEMINI_API_KEY is set
if not os.getenv('GEMINI_API_KEY'):
logger.error("❌ GEMINI_API_KEY environment variable not set")
logger.info("Please set GEMINI_API_KEY to test native grounding")
return False
# Initialize the grounded provider
logger.info("🔧 Initializing Gemini Grounded Provider...")
provider = GeminiGroundedProvider()
logger.info("✅ Provider initialized successfully")
# Test 1: Basic grounded content generation
logger.info("\n📝 Test 1: Basic LinkedIn Post Generation")
test_prompt = "Write a professional LinkedIn post about the latest AI trends in 2025"
result = await provider.generate_grounded_content(
prompt=test_prompt,
content_type="linkedin_post",
temperature=0.7,
max_tokens=500
)
if result and 'content' in result:
logger.info("✅ Content generated successfully")
logger.info(f"📊 Content length: {len(result['content'])} characters")
logger.info(f"🔗 Sources found: {len(result.get('sources', []))}")
logger.info(f"📚 Citations found: {len(result.get('citations', []))}")
# Display the generated content
logger.info("\n📄 Generated Content:")
logger.info("-" * 50)
logger.info(result['content'][:500] + "..." if len(result['content']) > 500 else result['content'])
logger.info("-" * 50)
# Display sources if available
if result.get('sources'):
logger.info("\n🔗 Sources:")
for i, source in enumerate(result['sources']):
logger.info(f" {i+1}. {source.get('title', 'Unknown')}")
logger.info(f" URL: {source.get('url', 'N/A')}")
# Display search queries if available
if result.get('search_queries'):
logger.info(f"\n🔍 Search Queries Used: {result['search_queries']}")
# Display grounding metadata info
if result.get('grounding_metadata'):
logger.info("✅ Grounding metadata found")
else:
logger.warning("⚠️ No grounding metadata found")
else:
logger.error("❌ Content generation failed")
if 'error' in result:
logger.error(f"Error: {result['error']}")
return False
# Test 2: Article generation
logger.info("\n📝 Test 2: LinkedIn Article Generation")
article_prompt = "Create a comprehensive article about sustainable business practices in tech companies"
article_result = await provider.generate_grounded_content(
prompt=article_prompt,
content_type="linkedin_article",
temperature=0.7,
max_tokens=1000
)
if article_result and 'content' in article_result:
logger.info("✅ Article generated successfully")
logger.info(f"📊 Article length: {len(article_result['content'])} characters")
logger.info(f"🔗 Sources: {len(article_result.get('sources', []))}")
# Check for article-specific processing
if 'title' in article_result:
logger.info(f"📰 Article title: {article_result['title']}")
if 'word_count' in article_result:
logger.info(f"📊 Word count: {article_result['word_count']}")
else:
logger.error("❌ Article generation failed")
return False
# Test 3: Content quality assessment
logger.info("\n📝 Test 3: Content Quality Assessment")
if result.get('content') and result.get('sources'):
quality_metrics = provider.assess_content_quality(
content=result['content'],
sources=result['sources']
)
logger.info("✅ Quality assessment completed")
logger.info(f"📊 Overall score: {quality_metrics.get('overall_score', 'N/A')}")
logger.info(f"🔗 Source coverage: {quality_metrics.get('source_coverage', 'N/A')}")
logger.info(f"🎯 Tone score: {quality_metrics.get('tone_score', 'N/A')}")
logger.info(f"📝 Word count: {quality_metrics.get('word_count', 'N/A')}")
logger.info(f"🏆 Quality level: {quality_metrics.get('quality_level', 'N/A')}")
# Test 4: Citation extraction
logger.info("\n📝 Test 4: Citation Extraction")
if result.get('content'):
citations = provider.extract_citations(result['content'])
logger.info(f"✅ Extracted {len(citations)} citations")
for i, citation in enumerate(citations):
logger.info(f" Citation {i+1}: {citation.get('reference', 'Unknown')}")
logger.info("\n🎉 All tests completed successfully!")
return True
except ImportError as e:
logger.error(f"❌ Import error: {str(e)}")
logger.info("💡 Make sure to install required dependencies:")
logger.info(" pip install google-genai loguru")
return False
except Exception as e:
logger.error(f"❌ Test failed with error: {str(e)}")
return False
async def test_individual_components():
"""Test individual components of the native grounding system."""
try:
logger.info("🔧 Testing Individual Components")
# Test 1: Provider initialization
logger.info("\n📋 Test 1: Provider Initialization")
if not os.getenv('GEMINI_API_KEY'):
logger.warning("⚠️ Skipping provider test - no API key")
return False
provider = GeminiGroundedProvider()
logger.info("✅ Provider initialized successfully")
# Test 2: Prompt building
logger.info("\n📋 Test 2: Prompt Building")
test_prompt = "Test prompt for LinkedIn post"
grounded_prompt = provider._build_grounded_prompt(test_prompt, "linkedin_post")
if grounded_prompt and len(grounded_prompt) > len(test_prompt):
logger.info("✅ Grounded prompt built successfully")
logger.info(f"📊 Original length: {len(test_prompt)}")
logger.info(f"📊 Enhanced length: {len(grounded_prompt)}")
else:
logger.error("❌ Prompt building failed")
return False
# Test 3: Content processing methods
logger.info("\n📋 Test 3: Content Processing Methods")
# Test post processing
test_content = "This is a test LinkedIn post #AI #Technology"
post_processing = provider._process_post_content(test_content)
if post_processing:
logger.info("✅ Post processing works")
logger.info(f"🔖 Hashtags found: {len(post_processing.get('hashtags', []))}")
# Test article processing
test_article = "# Test Article\n\nThis is test content for an article."
article_processing = provider._process_article_content(test_article)
if article_processing:
logger.info("✅ Article processing works")
logger.info(f"📊 Word count: {article_processing.get('word_count', 'N/A')}")
logger.info("✅ All component tests passed")
return True
except Exception as e:
logger.error(f"❌ Component test failed: {str(e)}")
return False
async def main():
"""Main test function."""
logger.info("🚀 Starting Native Grounding Tests")
logger.info("=" * 60)
# Test individual components first
component_success = await test_individual_components()
if component_success:
# Test the full integration
integration_success = await test_native_grounding()
if integration_success:
logger.info("\n🎉 SUCCESS: All tests passed!")
logger.info("✅ Native Google Search grounding is working correctly")
logger.info("✅ Gemini API integration successful")
logger.info("✅ Grounding metadata processing working")
logger.info("✅ Content generation with sources successful")
else:
logger.error("\n❌ FAILURE: Integration tests failed")
sys.exit(1)
else:
logger.error("\n❌ FAILURE: Component tests failed")
sys.exit(1)
if __name__ == "__main__":
# Configure logging
logger.remove()
logger.add(
sys.stderr,
format="<green>{time:YYYY-MM-DD HH:mm:ss}</green> | <level>{level: <8}</level> | <cyan>{name}</cyan>:<cyan>{function}</cyan>:<cyan>{line}</cyan> - <level>{message}</level>",
level="INFO"
)
# Run the tests
asyncio.run(main())

73
backend/test_simple_grounding.py Normal file
View File

@@ -0,0 +1,73 @@
#!/usr/bin/env python3
"""
Simple test script to verify basic grounding functionality.
This script tests the core components without triggering API overload.
"""
import asyncio
import sys
import os
from pathlib import Path
# Add the backend directory to the Python path
backend_dir = Path(__file__).parent
sys.path.insert(0, str(backend_dir))
from loguru import logger
from services.llm_providers.gemini_grounded_provider import GeminiGroundedProvider
async def test_basic_functionality():
"""Test basic grounding functionality."""
try:
logger.info("🧪 Testing Basic Grounding Functionality")
# Initialize provider
provider = GeminiGroundedProvider()
logger.info("✅ Provider initialized successfully")
# Test prompt building
prompt = "Write a short LinkedIn post about AI trends"
grounded_prompt = provider._build_grounded_prompt(prompt, "linkedin_post")
logger.info(f"✅ Grounded prompt built: {len(grounded_prompt)} characters")
# Test content processing
test_content = "AI is transforming industries #AI #Technology"
processed = provider._process_post_content(test_content)
logger.info(f"✅ Content processed: {len(processed.get('hashtags', []))} hashtags found")
logger.info("🎉 Basic functionality test completed successfully!")
return True
except Exception as e:
logger.error(f"❌ Basic functionality test failed: {str(e)}")
return False
async def main():
"""Main test function."""
logger.info("🚀 Starting Simple Grounding Test")
logger.info("=" * 50)
success = await test_basic_functionality()
if success:
logger.info("\n🎉 SUCCESS: Basic grounding functionality is working!")
logger.info("✅ Provider initialization successful")
logger.info("✅ Prompt building working")
logger.info("✅ Content processing working")
logger.info("✅ Ready for API integration")
else:
logger.error("\n❌ FAILURE: Basic functionality test failed")
sys.exit(1)
if __name__ == "__main__":
# Configure logging
logger.remove()
logger.add(
sys.stderr,
format="<green>{time:YYYY-MM-DD HH:mm:ss}</green> | <level>{level: <8}</level> | <cyan>{name}</cyan>:<cyan>{function}</cyan>:<cyan>{line}</cyan> - <level>{message}</level>",
level="INFO"
)
# Run the test
asyncio.run(main())