Create employer_branding_coordinator.py

insight_and_tasks/employer_branding_coordinator.py

@@ -0,0 +1,326 @@
+# coordinators/employer_branding_coordinator.py
+import json
+import logging
+from typing import Optional, Dict, Any # Added Dict, Any
+from dataclasses import asdict # For converting dataclasses (like AgentMetrics) to dict
+
+from google.adk.agents import LlmAgent
+from google.adk.runners import InMemoryRunner
+from google.genai import types as genai_types # For ADK agent inputs
+
+# Project-specific imports
+from agents.follower_agent import EnhancedFollowerAnalysisAgent
+from agents.post_agent import EnhancedPostPerformanceAgent
+from agents.mentions_agent import EnhancedMentionsAnalysisAgent
+from data_models.metrics import AgentMetrics # To type hint inputs
+from utils.retry_mechanism import RetryMechanism # If ADK calls need retry
+
+# Configure logger for this module
+logger = logging.getLogger(__name__)
+
+DEFAULT_COORDINATOR_MODEL = "gemini-2.5-flash-preview-05-20" # Use a more capable model for synthesis
+
+class EnhancedEmployerBrandingCoordinator:
+    """
+    Enhanced coordinator for synthesizing insights from multiple agent metrics,
+    identifying correlations, and generating integrated strategic recommendations.
+    """
+    COORDINATOR_AGENT_NAME = "employer_branding_coordinator"
+    COORDINATOR_AGENT_DESCRIPTION = (
+        "Strategic coordinator that analyzes metrics from Follower, Post Performance, and Mentions agents "
+        "to find correlations, suggest potential causal links, and generate integrated strategies."
+    )
+    COORDINATOR_AGENT_INSTRUCTION = """
+    You are the Enhanced Employer Branding Coordinator. Your primary mission is to synthesize analyses and 
+    structured metrics (TimeSeries, Aggregate, Categorical) from three specialized agents: Follower Analysis, 
+    Post Performance, and Mentions Analysis. Your goal is to provide a holistic, integrated understanding of 
+    the LinkedIn employer branding performance.
+
+    You MUST focus on:
+    1.  Cross-Agent Correlations: Analyze how metrics from different agents relate to each other over time. 
+        Pay close attention to the 'time_series_metrics' provided by each agent.
+        -   Identify positive or negative correlations (e.g., "Follower growth rate increased by X% when posts about 'company culture' (Post Agent) were published, coinciding with a Y% rise in positive mentions (Mentions Agent)").
+        -   Note any leading or lagging indicators (e.g., "A spike in negative mentions often preceded a dip in follower growth by approximately 2 weeks.").
+        -   Look for relationships between specific content types/topics (from Post Agent) and follower engagement/growth (Follower Agent) or brand sentiment (Mentions Agent).
+    2.  Potential Causal Insights & Hypotheses: Based on observed correlations and temporal sequences, suggest plausible causal relationships. 
+        These are hypotheses, not definitive conclusions.
+        -   Example: "The Q2 campaign focusing on 'employee testimonials' (Post Agent data) likely contributed to the observed 15% increase in organic follower acquisition (Follower Agent data) and the shift towards more positive sentiment in mentions (Mentions Agent data) during the same period."
+    3.  Root Cause Analysis (Conceptual): For significant performance changes (e.g., sudden engagement drops, unexpected follower spikes, sharp sentiment shifts), attempt to identify potential root causes by cross-referencing data and summaries from all three agents.
+    4.  Predictive Insights (High-Level): Based on established trends and correlations, what are potential future performance trajectories or risks?
+        -   Example: "If the current trend of declining engagement on text-only posts continues, overall reach may decrease by X% next quarter unless content strategy is diversified."
+    5.  Integrated Strategic Recommendations: Formulate actionable, strategic advice that leverages insights from ALL THREE data sources to optimize overall employer branding.
+        -   Recommendations should be specific (e.g., "Increase frequency of video posts related to 'Team Achievements' as this format shows high engagement and correlates with positive mention spikes.").
+        -   Prioritize recommendations based on their potential impact, supported by the cross-agent analysis.
+        -   Suggest A/B tests or further investigations where appropriate.
+
+    INPUT: You will receive structured 'AgentMetrics' data (JSON format) from each of the three agents. This includes their own analysis summaries, time-series data, aggregate figures, and categorical breakdowns.
+
+    OUTPUT: A comprehensive, well-structured report covering:
+        I.  Overall Executive Summary: A brief (2-3 paragraph) overview of the most critical findings and strategic implications derived from the integrated analysis.
+        II. Detailed Cross-Agent Correlation Analysis: Elaborate on specific correlations found, with examples.
+        III.Key Causal Hypotheses: Present the most compelling potential causal links.
+        IV. Noteworthy Performance Shifts & Potential Root Causes: Discuss any major changes and their likely drivers.
+        V.  Forward-Looking Predictive Insights: Offer high-level predictions.
+        VI. Actionable Integrated Strategic Recommendations: Provide clear, prioritized recommendations.
+
+    Your analysis must be grounded in the provided data. Refer to specific metrics and agent findings to support your conclusions.
+    Be insightful and strategic. The goal is to provide a unified view that is more valuable than the sum of the individual agent analyses.
+    """
+
+    def __init__(self, api_key: str, model_name: Optional[str] = None):
+        self.api_key = api_key # Stored for LlmAgent or if agents need it passed explicitly
+        self.model_name = model_name or DEFAULT_COORDINATOR_MODEL
+
+        # Initialize individual agents. The coordinator will use their output.
+        # These agents are internal to the coordinator's process of getting data to synthesize.
+        self.follower_agent = EnhancedFollowerAnalysisAgent(api_key=api_key, model_name=model_name) # Pass down model if needed
+        self.post_agent = EnhancedPostPerformanceAgent(api_key=api_key, model_name=model_name)
+        self.mentions_agent = EnhancedMentionsAnalysisAgent(api_key=api_key, model_name=model_name)
+        
+        # The LLM agent for the coordinator itself, responsible for synthesis
+        self.coordinator_llm_agent = LlmAgent(
+            name=self.COORDINATOR_AGENT_NAME,
+            model=self.model_name, # Use the coordinator's (potentially more powerful) model
+            description=self.COORDINATOR_AGENT_DESCRIPTION,
+            instruction=self.COORDINATOR_AGENT_INSTRUCTION
+        )
+        self.retry_mechanism = RetryMechanism()
+        logger.info(f"{self.COORDINATOR_AGENT_NAME} initialized with model {self.model_name}.")
+        logger.info(f"It internally uses: Follower Agent ({self.follower_agent.model_name}), "
+                    f"Post Agent ({self.post_agent.model_name}), Mentions Agent ({self.mentions_agent.model_name}).")
+
+
+    async def generate_comprehensive_analysis(
+        self, 
+        follower_metrics: AgentMetrics, 
+        post_metrics: AgentMetrics, 
+        mentions_metrics: AgentMetrics
+    ) -> str:
+        """
+        Generates a comprehensive analysis by synthesizing metrics from all specialized agents.
+
+        Args:
+            follower_metrics: Metrics from the EnhancedFollowerAnalysisAgent.
+            post_metrics: Metrics from the EnhancedPostPerformanceAgent.
+            mentions_metrics: Metrics from the EnhancedMentionsAnalysisAgent.
+
+        Returns:
+            A string containing the comprehensive analysis report.
+        """
+        
+        # Prepare the input prompt for the coordinator's LlmAgent
+        # Serialize the AgentMetrics objects (which are dataclasses) to dictionaries
+        # then to JSON strings for clean inclusion in the prompt.
+        try:
+            follower_metrics_dict = asdict(follower_metrics)
+            post_metrics_dict = asdict(post_metrics)
+            mentions_metrics_dict = asdict(mentions_metrics)
+        except Exception as e:
+            logger.error(f"Error converting AgentMetrics to dict: {e}", exc_info=True)
+            return "Error: Could not process input metrics for coordination."
+
+        # Truncate individual agent summaries if they are too long to avoid overly large prompts
+        max_summary_len = 500 # Max characters for individual agent summaries in the prompt
+        follower_metrics_dict['analysis_summary'] = follower_metrics_dict.get('analysis_summary', '')[:max_summary_len]
+        post_metrics_dict['analysis_summary'] = post_metrics_dict.get('analysis_summary', '')[:max_summary_len]
+        mentions_metrics_dict['analysis_summary'] = mentions_metrics_dict.get('analysis_summary', '')[:max_summary_len]
+
+
+        synthesis_prompt = f"""
+        Please synthesize the following LinkedIn analytics insights, which are structured as 'AgentMetrics' 
+        from three specialized agents. Your primary task is to identify cross-metric correlations, 
+        deduce potential causal relationships, and provide integrated strategic recommendations based on 
+        your core instructions.
+
+        DATA FROM SPECIALIZED AGENTS:
+
+        1. Follower Analysis Agent Metrics:
+           - Agent Name: {follower_metrics_dict.get('agent_name')}
+           - Agent's Analysis Summary: {follower_metrics_dict.get('analysis_summary')}
+           - Time Series Metrics: {json.dumps([asdict(m) for m in follower_metrics.time_series_metrics], indent=2, default=str)}
+           - Aggregate Metrics: {json.dumps(follower_metrics_dict.get('aggregate_metrics'), indent=2, default=str)}
+           - Categorical Metrics: {json.dumps(follower_metrics_dict.get('categorical_metrics'), indent=2, default=str)}
+           - Time Periods Covered: {json.dumps(follower_metrics_dict.get('time_periods_covered'), default=str)}
+           - Key Insights by Agent: {json.dumps(follower_metrics_dict.get('key_insights'), default=str)}
+
+        2. Post Performance Agent Metrics:
+           - Agent Name: {post_metrics_dict.get('agent_name')}
+           - Agent's Analysis Summary: {post_metrics_dict.get('analysis_summary')}
+           - Time Series Metrics: {json.dumps([asdict(m) for m in post_metrics.time_series_metrics], indent=2, default=str)}
+           - Aggregate Metrics: {json.dumps(post_metrics_dict.get('aggregate_metrics'), indent=2, default=str)}
+           - Categorical Metrics: {json.dumps(post_metrics_dict.get('categorical_metrics'), indent=2, default=str)}
+           - Time Periods Covered: {json.dumps(post_metrics_dict.get('time_periods_covered'), default=str)}
+           - Key Insights by Agent: {json.dumps(post_metrics_dict.get('key_insights'), default=str)}
+
+        3. Mentions Analysis Agent Metrics:
+           - Agent Name: {mentions_metrics_dict.get('agent_name')}
+           - Agent's Analysis Summary: {mentions_metrics_dict.get('analysis_summary')}
+           - Time Series Metrics: {json.dumps([asdict(m) for m in mentions_metrics.time_series_metrics], indent=2, default=str)}
+           - Aggregate Metrics: {json.dumps(mentions_metrics_dict.get('aggregate_metrics'), indent=2, default=str)}
+           - Categorical Metrics: {json.dumps(mentions_metrics_dict.get('categorical_metrics'), indent=2, default=str)}
+           - Time Periods Covered: {json.dumps(mentions_metrics_dict.get('time_periods_covered'), default=str)}
+           - Key Insights by Agent: {json.dumps(mentions_metrics_dict.get('key_insights'), default=str)}
+
+        COORDINATION TASK:
+        Based on ALL the data presented above from the three agents, generate a comprehensive synthesis report.
+        Follow your core instructions meticulously, focusing on cross-agent correlations (especially using the 
+        time_series_metrics), causal hypotheses, root cause considerations for major shifts, predictive insights, 
+        and actionable, integrated strategic recommendations.
+        Structure your output as a detailed report with the specified sections.
+        """
+        
+        user_input_content = genai_types.Content(
+            role="user",
+            parts=[genai_types.Part(text=synthesis_prompt)]
+        )
+        
+        runner = InMemoryRunner(agent=self.coordinator_llm_agent, app_name=f"{self.COORDINATOR_AGENT_NAME}Runner")
+        user_id = f"system_user_coordinator_{int(datetime.utcnow().timestamp())}" # Unique ID for the run
+        
+        session = await runner.session_service.create_session(
+            app_name=f"{self.COORDINATOR_AGENT_NAME}Runner",
+            user_id=user_id
+        )
+        
+        result_text_parts = []
+        try:
+            logger.info(f"Running {self.COORDINATOR_AGENT_NAME} for synthesis. User ID: {user_id}, Session ID: {session.id}")
+            # Using retry for the ADK runner execution part
+            async def run_adk_coordinator():
+                temp_result_parts = []
+                async for event in runner.run(
+                    user_id=user_id,
+                    session_id=session.id,
+                    new_message=user_input_content
+                ):
+                    if hasattr(event, 'content') and event.content and event.content.parts:
+                        for part in event.content.parts:
+                            if hasattr(part, 'text'):
+                                temp_result_parts.append(part.text)
+                if not temp_result_parts:
+                    # This could happen if the LLM returns no content or an error not caught by ADK
+                    logger.warning(f"{self.COORDINATOR_AGENT_NAME} produced no text output from ADK run.")
+                    # Consider raising a specific error or returning a default message
+                    # For now, it will result in an empty string if no parts are collected.
+                return "".join(temp_result_parts)
+
+            # The retry_with_backoff expects a synchronous function.
+            # For async, you'd typically handle retries within the async logic or use an async retry library.
+            # For simplicity here, we'll run it once. If retries are critical for ADK calls,
+            # the ADK runner itself might have retry mechanisms, or this part needs adjustment.
+            # The original code didn't show retry for this ADK call, so keeping it direct.
+            
+            # Direct call without retry for the async ADK runner:
+            async for event in runner.run(
+                user_id=user_id,
+                session_id=session.id,
+                new_message=user_input_content
+            ):
+                if hasattr(event, 'content') and event.content and event.content.parts:
+                    for part in event.content.parts:
+                        if hasattr(part, 'text'):
+                            result_text_parts.append(part.text)
+            
+            final_result_text = "".join(result_text_parts)
+            if not final_result_text.strip():
+                 logger.warning(f"{self.COORDINATOR_AGENT_NAME} synthesis resulted in an empty string.")
+                 final_result_text = "Coordinator analysis did not produce output. Please check logs."
+
+
+        except Exception as e:
+            logger.error(f"Error during {self.COORDINATOR_AGENT_NAME} LLM agent execution: {e}", exc_info=True)
+            final_result_text = f"Error in coordinator synthesis: {str(e)}"
+        finally:
+            try:
+                await runner.session_service.delete_session(
+                    app_name=f"{self.COORDINATOR_AGENT_NAME}Runner", user_id=user_id, session_id=session.id
+                )
+            except Exception as session_del_e:
+                logger.error(f"Error deleting coordinator session: {session_del_e}")
+                
+        return final_result_text
+
+if __name__ == '__main__':
+    import asyncio
+    import pandas as pd # For creating dummy data
+    from datetime import datetime # For dummy data AgentMetrics
+
+    try:
+        from utils.logging_config import setup_logging
+        setup_logging()
+        logger.info("Logging setup for EnhancedEmployerBrandingCoordinator test.")
+    except ImportError:
+        logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s")
+        logger.warning("logging_config.py not found, using basicConfig for logging.")
+
+    MOCK_API_KEY = os.environ.get("GOOGLE_API_KEY", "test_api_key_coordinator")
+    MODEL_NAME = DEFAULT_COORDINATOR_MODEL # Or a specific test model
+
+    # Create dummy AgentMetrics data for testing
+    dummy_ts_metric = TimeSeriesMetric(metric_name="dummy_visits", values=[10.0,20.0], timestamps=["2023-01","2023-02"])
+    
+    follower_metrics_data = AgentMetrics(
+        agent_name="follower_analyst_test",
+        analysis_summary="Followers grew steadily. Demographic: Young professionals.",
+        time_series_metrics=[dummy_ts_metric],
+        aggregate_metrics={"avg_growth_rate": 0.05},
+        categorical_metrics={"top_industry": "Tech"},
+        time_periods_covered=["2023-01", "2023-02"],
+        key_insights=["Organic growth is strong."]
+    )
+    post_metrics_data = AgentMetrics(
+        agent_name="post_analyst_test",
+        analysis_summary="Video posts performed best. Engagement rate is 3%.",
+        time_series_metrics=[TimeSeriesMetric(metric_name="dummy_engagement", values=[0.03,0.035], timestamps=["2023-01","2023-02"], unit="%")],
+        aggregate_metrics={"avg_engagement_rate_overall": 0.032},
+        categorical_metrics={"top_media_type": "VIDEO"},
+        time_periods_covered=["2023-01", "2023-02"],
+        key_insights=["Video content is key for engagement."]
+    )
+    mentions_metrics_data = AgentMetrics(
+        agent_name="mentions_analyst_test",
+        analysis_summary="Mentions are mostly neutral. Sentiment score avg 0.1.",
+        time_series_metrics=[TimeSeriesMetric(metric_name="dummy_sentiment_score", values=[0.1,0.12], timestamps=["2023-01","2023-02"])],
+        aggregate_metrics={"overall_avg_sentiment": 0.11},
+        categorical_metrics={"dominant_sentiment": "Neutral"},
+        time_periods_covered=["2023-01", "2023-02"],
+        key_insights=["Brand perception is stable but not overly positive."]
+    )
+
+    coordinator = EnhancedEmployerBrandingCoordinator(api_key=MOCK_API_KEY, model_name=MODEL_NAME)
+
+    async def run_coordination():
+        logger.info("Generating comprehensive analysis from dummy metrics...")
+        # For local tests without real API calls, the LlmAgent might behave as a mock.
+        if MOCK_API_KEY == "test_api_key_coordinator":
+             logger.warning("Using a mock API key. Coordinator LlmAgent behavior might be limited or mocked.")
+             # Mock the ADK runner for the coordinator's LLM agent if needed
+             class MockCoordinatorADKRunner:
+                def __init__(self, agent, app_name): self.agent = agent
+                async def session_service_create_session(self, app_name, user_id): 
+                    class MockSession: id = "mock_coord_session_id"
+                    return MockSession()
+                async def run(self, user_id, session_id, new_message):
+                    # Simulate a response from the coordinator LLM
+                    yield genai_types.Content(parts=[genai_types.Part(text="Mock Coordinator Synthesis Report: Blah blah correlation. Recommendation: Do X.")])
+                async def session_service_delete_session(self, app_name, user_id, session_id): pass
+            
+             global InMemoryRunner # Make sure we are modifying the correct InMemoryRunner
+             OriginalInMemoryRunnerCoord = InMemoryRunner
+             InMemoryRunner = MockCoordinatorADKRunner
+
+
+        report = await coordinator.generate_comprehensive_analysis(
+            follower_metrics_data,
+            post_metrics_data,
+            mentions_metrics_data
+        )
+        
+        if MOCK_API_KEY == "test_api_key_coordinator" and 'OriginalInMemoryRunnerCoord' in globals():
+            InMemoryRunner = OriginalInMemoryRunnerCoord # Restore
+
+        print("\n--- EnhancedEmployerBrandingCoordinator Report ---")
+        print(report)
+
+    if __name__ == '__main__': # Inner check
+        asyncio.run(run_coordination())