File size: 2,703 Bytes
939c0c0 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 | """
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
|