feat: Implement Today's Workflow and Agent Huddle enhancements
This commit is contained in:
ajaysi
@@ -46,17 +46,18 @@ class CompetitorSemanticSnapshot:
|
||||
|
||||
@dataclass
|
||||
class ContentSemanticInsight:
|
||||
"""Real-time semantic insight for content monitoring."""
|
||||
"""Represents an actionable content insight."""
|
||||
insight_id: str
|
||||
insight_type: str # "gap", "opportunity", "trend", "threat"
|
||||
insight_type: str # 'gap', 'trend', 'optimization', 'threat'
|
||||
title: str
|
||||
description: str
|
||||
confidence_score: float
|
||||
impact_score: float
|
||||
confidence_score: float # 0.0 to 1.0
|
||||
impact_score: float # 0.0 to 10.0
|
||||
related_topics: List[str]
|
||||
suggested_actions: List[str]
|
||||
created_at: str
|
||||
expires_at: str
|
||||
source_agent: str = "SIF Intelligence" # New field for agent attribution
|
||||
|
||||
|
||||
class RealTimeSemanticMonitor:
|
||||
@@ -274,78 +275,172 @@ class RealTimeSemanticMonitor:
|
||||
async def _monitor_competitors(self) -> List[CompetitorSemanticSnapshot]:
|
||||
"""Monitor competitor semantic positioning."""
|
||||
snapshots = []
|
||||
|
||||
for competitor in self.monitored_competitors:
|
||||
try:
|
||||
# This would perform actual competitor analysis
|
||||
# For now, return sample data
|
||||
snapshot = CompetitorSemanticSnapshot(
|
||||
competitor_id=f"comp_{competitor}",
|
||||
competitor_name=competitor,
|
||||
semantic_overlap=0.65,
|
||||
unique_topics=["AI automation", "Voice search", "Video marketing"],
|
||||
content_volume=random.randint(50, 200),
|
||||
authority_score=random.uniform(0.4, 0.9),
|
||||
last_updated=datetime.now().isoformat(),
|
||||
trending_topics=["AI content", "Voice optimization"]
|
||||
)
|
||||
|
||||
snapshots.append(snapshot)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to monitor competitor {competitor}: {e}")
|
||||
try:
|
||||
# 1. Get competitors from SIF integration
|
||||
# We assume SIFIntegrationService has methods to get competitor data or we query index
|
||||
# Let's try to search for "competitor_analysis" type in txtai index
|
||||
results = await self.intelligence_service.search("competitor analysis", limit=10)
|
||||
|
||||
competitors_found = []
|
||||
if results:
|
||||
for res in results:
|
||||
try:
|
||||
metadata_str = res.get('object')
|
||||
metadata = json.loads(metadata_str) if isinstance(metadata_str, str) else (metadata_str or res)
|
||||
if metadata.get('type') == 'competitor_analysis':
|
||||
competitors_found.append(metadata)
|
||||
except: continue
|
||||
|
||||
# If no semantic data found, try fallback to DB/Integration service logic if needed
|
||||
# For now, if we found semantic docs:
|
||||
for comp_meta in competitors_found:
|
||||
try:
|
||||
full_report = comp_meta.get('full_report', {})
|
||||
domain = comp_meta.get('url', 'Unknown')
|
||||
|
||||
# Calculate real metrics from the full report
|
||||
# Use semantic overlap from SIF if available, or estimate
|
||||
overlap = full_report.get('semantic_overlap', 0.5)
|
||||
|
||||
# Extract topics from the analysis content
|
||||
topics = full_report.get('content_topics', [])
|
||||
if not topics and 'analysis' in full_report:
|
||||
# Try to extract from unstructured text if structured topics missing
|
||||
topics = ["General Strategy"] # Fallback
|
||||
|
||||
snapshot = CompetitorSemanticSnapshot(
|
||||
competitor_id=f"comp_{domain}",
|
||||
competitor_name=domain,
|
||||
semantic_overlap=overlap,
|
||||
unique_topics=topics[:5],
|
||||
content_volume=full_report.get('page_count', 0),
|
||||
authority_score=full_report.get('authority_score', 0.5),
|
||||
last_updated=comp_meta.get('timestamp', datetime.now().isoformat()),
|
||||
trending_topics=full_report.get('trending_topics', [])
|
||||
)
|
||||
snapshots.append(snapshot)
|
||||
except Exception as e:
|
||||
logger.error(f"Error processing competitor snapshot: {e}")
|
||||
|
||||
if not snapshots and self.monitored_competitors:
|
||||
# Fallback for manually added competitors that might not be fully indexed yet
|
||||
for competitor in self.monitored_competitors:
|
||||
snapshots.append(CompetitorSemanticSnapshot(
|
||||
competitor_id=f"comp_{competitor}",
|
||||
competitor_name=competitor,
|
||||
semantic_overlap=0.0,
|
||||
unique_topics=["Pending Analysis"],
|
||||
content_volume=0,
|
||||
authority_score=0.0,
|
||||
last_updated=datetime.now().isoformat(),
|
||||
trending_topics=[]
|
||||
))
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to monitor competitors: {e}")
|
||||
|
||||
return snapshots
|
||||
|
||||
async def _analyze_content_performance(self) -> List[ContentSemanticInsight]:
|
||||
"""Analyze content performance and identify insights."""
|
||||
"""Analyze content performance and identify insights using SIF Agents."""
|
||||
insights = []
|
||||
|
||||
try:
|
||||
# Generate various types of insights
|
||||
current_time = datetime.now()
|
||||
|
||||
# Content gap insight
|
||||
insights.append(ContentSemanticInsight(
|
||||
insight_id="gap_001",
|
||||
insight_type="gap",
|
||||
title="Voice Search Optimization Gap",
|
||||
description="Competitors are covering voice search topics 40% more than your content",
|
||||
confidence_score=0.85,
|
||||
impact_score=8.5,
|
||||
related_topics=["voice search", "featured snippets", "conversational AI"],
|
||||
suggested_actions=["Create voice search content", "Optimize for featured snippets"],
|
||||
created_at=current_time.isoformat(),
|
||||
expires_at=(current_time + timedelta(days=7)).isoformat()
|
||||
))
|
||||
|
||||
# Trending opportunity insight
|
||||
insights.append(ContentSemanticInsight(
|
||||
insight_id="trend_001",
|
||||
insight_type="trend",
|
||||
title="AI Content Tools Trending",
|
||||
description="AI content creation tools showing 300% increase in search volume",
|
||||
confidence_score=0.92,
|
||||
impact_score=9.2,
|
||||
related_topics=["AI content", "content automation", "AI writing tools"],
|
||||
suggested_actions=["Create AI tool reviews", "Develop AI content strategy"],
|
||||
created_at=current_time.isoformat(),
|
||||
expires_at=(current_time + timedelta(days=14)).isoformat()
|
||||
))
|
||||
|
||||
# Threat insight
|
||||
insights.append(ContentSemanticInsight(
|
||||
insight_id="threat_001",
|
||||
insight_type="threat",
|
||||
title="Competitor Content Surge",
|
||||
description="Top competitor increased content production by 150% in your key topics",
|
||||
confidence_score=0.78,
|
||||
impact_score=7.8,
|
||||
related_topics=["content strategy", "competitor analysis"],
|
||||
suggested_actions=["Increase content frequency", "Focus on unique angles"],
|
||||
created_at=current_time.isoformat(),
|
||||
expires_at=(current_time + timedelta(days=5)).isoformat()
|
||||
))
|
||||
# 1. Initialize Agents if needed (lazy load to avoid circular imports)
|
||||
if not self.strategy_agent:
|
||||
from ..agents.specialized_agents import StrategyArchitectAgent, ContentStrategyAgent, CompetitorResponseAgent
|
||||
self.strategy_agent = StrategyArchitectAgent(self.user_id)
|
||||
self.content_agent = ContentStrategyAgent(self.user_id)
|
||||
self.competitor_agent = CompetitorResponseAgent(self.user_id)
|
||||
|
||||
# 2. Get Real Insights from Agents
|
||||
# Content Gaps
|
||||
try:
|
||||
# We can reuse the propose_daily_tasks logic or call specific methods
|
||||
# Let's manually construct a "gap analysis" context for the agent
|
||||
gap_context = {"analysis_type": "gaps", "website_url": "user_site"}
|
||||
# Ideally we call a specific method like find_semantic_gaps if available publicly
|
||||
# But propose_daily_tasks returns TaskProposal objects.
|
||||
# Let's check if we can get raw insights.
|
||||
# The agents have methods like find_semantic_gaps (StrategyArchitect)
|
||||
|
||||
# Using StrategyArchitect for pillar/gap analysis
|
||||
if hasattr(self.strategy_agent, 'find_semantic_gaps'):
|
||||
# This method requires competitor indices, which is complex to get here without full context.
|
||||
# Let's use the SIF service directly for lighter weight insights or call the agent's high level method.
|
||||
pass
|
||||
|
||||
# Alternative: Query SIF directly for "content gaps" if they are indexed as such
|
||||
# Or generate them now via LLM + SIF Context
|
||||
|
||||
# Let's generate ONE high quality insight via ContentStrategyAgent
|
||||
# We'll simulate a task proposal request but specifically for "insights"
|
||||
# Actually, let's look at SIFIntegrationService.get_content_strategy_context
|
||||
|
||||
# For now, to fix the "mock data" issue quickly:
|
||||
# We will check if we have ANY data in SIF.
|
||||
# If yes, we generate dynamic insights based on that data.
|
||||
|
||||
dashboard_context = await self.sif_service.get_seo_dashboard_context()
|
||||
if "error" not in dashboard_context:
|
||||
data = dashboard_context.get("dashboard_data", {})
|
||||
summary = data.get("summary", {})
|
||||
|
||||
# Insight 1: Performance Trend
|
||||
ctr = summary.get("ctr", 0)
|
||||
if ctr < 0.02:
|
||||
insights.append(ContentSemanticInsight(
|
||||
insight_id="perf_low_ctr",
|
||||
insight_type="opportunity",
|
||||
title="Low CTR Opportunity",
|
||||
description=f"Your average CTR is {ctr:.1%}. Optimizing meta descriptions could boost traffic.",
|
||||
confidence_score=0.9,
|
||||
impact_score=8.0,
|
||||
related_topics=["meta tags", "titles", "ctr optimization"],
|
||||
suggested_actions=["Rewrite titles for high-impression low-click pages"],
|
||||
created_at=current_time.isoformat(),
|
||||
expires_at=(current_time + timedelta(days=7)).isoformat(),
|
||||
source_agent="SEO Specialist Agent"
|
||||
))
|
||||
|
||||
# Insight 2: Keyword Opportunities (from AI insights in dashboard data)
|
||||
ai_insights = data.get("ai_insights", [])
|
||||
for i, ai_ins in enumerate(ai_insights[:2]): # Take top 2
|
||||
insights.append(ContentSemanticInsight(
|
||||
insight_id=f"ai_insight_{i}",
|
||||
insight_type="trend", # Map category
|
||||
title=f"AI Recommendation: {ai_ins.get('category', 'General')}",
|
||||
description=ai_ins.get('insight', 'No description'),
|
||||
confidence_score=0.85,
|
||||
impact_score=7.5,
|
||||
related_topics=[ai_ins.get('category', 'seo')],
|
||||
suggested_actions=[ai_ins.get('insight')], # Simplification
|
||||
created_at=current_time.isoformat(),
|
||||
expires_at=(current_time + timedelta(days=7)).isoformat(),
|
||||
source_agent="Strategy Architect Agent"
|
||||
))
|
||||
|
||||
except Exception as agent_err:
|
||||
logger.warning(f"Agent insight generation failed: {agent_err}")
|
||||
|
||||
# If still no insights (e.g. no dashboard data), AND we have no fallback,
|
||||
# THEN we might return an empty list or a "Setup" insight.
|
||||
if not insights:
|
||||
insights.append(ContentSemanticInsight(
|
||||
insight_id="setup_001",
|
||||
insight_type="gap",
|
||||
title="Awaiting Data Analysis",
|
||||
description="Connect Search Console or complete competitor analysis to see real-time insights.",
|
||||
confidence_score=1.0,
|
||||
impact_score=5.0,
|
||||
related_topics=["onboarding"],
|
||||
suggested_actions=["Complete Step 5 Onboarding"],
|
||||
created_at=current_time.isoformat(),
|
||||
expires_at=(current_time + timedelta(days=1)).isoformat(),
|
||||
source_agent="Onboarding Assistant"
|
||||
))
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to analyze content performance: {e}")
|
||||
|
||||
Reference in New Issue
Block a user