| """ |
| Analytics Specialist Agent Node |
| =============================== |
| Generates data visualizations, computes statistical summary metrics, |
| and prepares structured chart JSON objects for frontend Recharts. |
| """ |
| from __future__ import annotations |
|
|
| import json |
| import structlog |
| from app.services.llm_gateway import llm_gateway |
| from agents.state import CopilotState |
|
|
| logger = structlog.get_logger(__name__) |
|
|
| ANALYTICS_SYSTEM_PROMPT = """You are an Analytics and Data Visualization Specialist AI. |
| Your job is to analyze the user's data or request and produce a structured chart configuration along with a narrative summary. |
| |
| Return a JSON object with: |
| { |
| "summary": "<Textual analysis and trend summary>", |
| "chart": { |
| "type": "<bar | line | pie>", |
| "title": "<Chart Title>", |
| "x_key": "<Category key name>", |
| "y_keys": ["<Metric key name>"], |
| "data": [ |
| {"<x_key>": "Jan", "<y_keys[0]>": 100}, |
| {"<x_key>": "Feb", "<y_keys[0]>": 150} |
| ] |
| } |
| } |
| """ |
|
|
|
|
| async def analytics_node(state: CopilotState) -> CopilotState: |
| """ |
| Analytics Specialist Node. |
| Analyzes trends and generates chart structure for frontend Recharts. |
| """ |
| query = state.get("query", "") |
| logger.info("Analytics Agent executing", query=query) |
|
|
| try: |
| response = await llm_gateway.generate( |
| prompt=f"User Request: {query}\nProvide analytics summary & chart JSON:", |
| system_prompt=ANALYTICS_SYSTEM_PROMPT, |
| temperature=0.1, |
| ) |
|
|
| content = response.content.strip() |
| if content.startswith("```"): |
| content = content.split("\n", 1)[-1].rsplit("```", 1)[0].strip() |
|
|
| try: |
| data = json.loads(content) |
| summary = data.get("summary", content) |
| chart = data.get("chart") |
| except Exception: |
| summary = content |
| chart = None |
|
|
| state["retrieved_chunks"] = [{ |
| "document_id": "analytics_report", |
| "document_name": "Analytics Engine", |
| "text": summary, |
| "score": 1.0, |
| "doc_type": "analytics" |
| }] |
|
|
| outputs = state.get("agent_outputs", []) |
| outputs.append({ |
| "agent_name": "analytics", |
| "content": summary, |
| "metadata": {"chart": chart} |
| }) |
| state["agent_outputs"] = outputs |
| state["active_agent"] = "analytics" |
|
|
| except Exception as e: |
| logger.error("Analytics Agent error", error=str(e)) |
| state["error"] = f"Analytics Error: {str(e)}" |
| state["retrieved_chunks"] = [{ |
| "document_name": "Analytics Error", |
| "text": f"Could not perform analytics: {str(e)}" |
| }] |
|
|
| return state |
|
|
