Update Render build configuration: fix deps, force py3.11, add build script

backend/services/intelligence/agents/specialized/social_amplification.py

@@ -0,0 +1,140 @@
+"""
+Social Amplification Agent implementation.
+"""
+from typing import Dict, Any, List, Optional
+from datetime import datetime
+from loguru import logger
+from .base import SIFBaseAgent, TXTAI_AVAILABLE, Agent
+from services.intelligence.agents.core_agent_framework import BaseALwrityAgent, TaskProposal
+
+try:
+    from services.intelligence.sif_integration import SIFIntegrationService
+    SIF_AVAILABLE = True
+except ImportError:
+    SIF_AVAILABLE = False
+
+class SocialAmplificationAgent(BaseALwrityAgent):
+    """
+    Agent responsible for social media monitoring, content adaptation, and distribution.
+    """
+    
+    def __init__(self, user_id: str, shared_llm_name: str, llm: Any = None, **kwargs):
+        super().__init__(user_id, "social_media_manager", shared_llm_name, llm, **kwargs)
+        
+        self.sif_service = None
+        if SIF_AVAILABLE:
+            try:
+                self.sif_service = SIFIntegrationService(user_id)
+            except Exception as e:
+                logger.warning(f"Failed to initialize SIF service for SocialAmplificationAgent: {e}")
+
+    def _create_txtai_agent(self):
+        """Create a specialized txtai Agent for social media."""
+        if not TXTAI_AVAILABLE or Agent is None:
+            return None
+            
+        _llm_for_agent = getattr(self.llm, "llm", self.llm)
+        return Agent(
+            tools=[
+                {
+                    "name": "social_monitor",
+                    "description": "Monitors social trends and conversations",
+                    "target": self._social_monitor_tool
+                },
+                {
+                    "name": "content_adapter",
+                    "description": "Adapts long-form content for social platforms",
+                    "target": self._content_adapter_tool
+                },
+                {
+                    "name": "engagement_optimizer",
+                    "description": "Optimizes posts for engagement (hashtags, timing)",
+                    "target": self._engagement_optimizer_tool
+                },
+                {
+                    "name": "distribution_manager",
+                    "description": "Manages posting schedule",
+                    "target": self._distribution_manager_tool
+                }
+            ],
+            llm=_llm_for_agent,
+            max_iterations=10,
+            # Removed unsupported 'system' argument
+            # Instruction will be provided via orchestrator context or initial prompt
+            # Instruction should be provided during invocation or via orchestrator context
+        )
+    
+    # Tool Implementations
+    
+    def _social_monitor_tool(self, context: Dict[str, Any]) -> Dict[str, Any]:
+        """
+        Social monitoring tool using SIF.
+        
+        Args:
+            context: Dictionary containing monitoring criteria like 'topics' or 'platforms'.
+        """
+        # Stub implementation
+        return {
+            "trends": ["AI in marketing", "Content automation"],
+            "source": "stub",
+            "timestamp": datetime.utcnow().isoformat()
+        }
+
+    def _content_adapter_tool(self, context: Dict[str, Any]) -> Dict[str, Any]:
+        """
+        Adapts content for specific platforms.
+        
+        Args:
+            context: Dictionary containing 'content' and 'platform' (e.g., 'linkedin', 'twitter').
+        """
+        # Stub implementation
+        return {"adapted_content": "Social post"}
+
+    def _engagement_optimizer_tool(self, context: Dict[str, Any]) -> Dict[str, Any]:
+        """
+        Optimizes content for engagement (hashtags, timing, hook).
+        
+        Args:
+            context: Dictionary containing 'content' to optimize.
+        """
+        # Stub implementation
+        return {
+            "optimization_suggestions": ["Use questions"],
+            "estimated_engagement_score": 8.5,
+            "timestamp": datetime.utcnow().isoformat()
+        }
+
+    def _distribution_manager_tool(self, context: Dict[str, Any]) -> Dict[str, Any]:
+        """
+        Manages distribution (scheduling/posting).
+        
+        Args:
+            context: Dictionary containing 'post_content' and 'schedule_time'.
+        """
+        # Stub implementation
+        return {
+            "distribution_plan": [],
+            "status": "scheduled",
+            "timestamp": datetime.utcnow().isoformat()
+        }
+
+    async def propose_daily_tasks(self, context: Dict[str, Any]) -> List[TaskProposal]:
+        """
+        Propose social media tasks.
+        """
+        proposals = []
+        
+        # 1. Social Post Creation
+        proposals.append(TaskProposal(
+            title="Create LinkedIn Thread",
+            description="Summarize your latest blog post into a 5-tweet thread.",
+            pillar_id="distribute",
+            priority="medium",
+            estimated_time=20,
+            source_agent="SocialAmplificationAgent",
+            reasoning="Repurpose existing content.",
+            action_type="navigate",
+            action_url="/content-planning-dashboard"
+        ))
+        
+        return proposals