| """ |
| LangGraph Orchestrator Graph — Multi-Agent Suite |
| ================================================= |
| Assembles all specialist agent nodes into a unified StateGraph with conditional routing edges. |
| |
| Graph Flow: |
| [START] → planner_node |
| │ |
| ├─► (routing == "retrieval") ──► retrieval_node ──┐ |
| ├─► (routing == "sql") ────────► sql_node ────────┼─► synthesizer_node ──► [END] |
| ├─► (routing == "analytics") ──► analytics_node ──┤ |
| ├─► (routing == "code") ───────► code_node ───────┤ |
| ├─► (routing == "web") ────────► web_node ────────┤ |
| │ │ |
| └─► (routing == "direct") ────────────────────────┘ |
| """ |
| from __future__ import annotations |
|
|
| import time |
| from typing import Dict, Any, Literal |
|
|
| import structlog |
| from langgraph.graph import StateGraph, START, END |
|
|
| from agents.state import CopilotState |
| from agents.planner import planner_node |
| from agents.retrieval_agent import retrieval_node |
| from agents.sql_agent import sql_node |
| from agents.analytics_agent import analytics_node |
| from agents.code_agent import code_node |
| from agents.web_agent import web_node |
| from agents.synthesizer import synthesizer_node |
|
|
| logger = structlog.get_logger(__name__) |
|
|
|
|
| def route_next(state: CopilotState) -> Literal["retrieval", "sql", "analytics", "code", "web", "synthesizer"]: |
| """ |
| Conditional edge function evaluated after supervisor/planner. |
| Routes to specialist node based on planner decision. |
| """ |
| decision = state.get("routing_decision", "direct") |
| logger.info("Routing conditional edge evaluated", decision=decision) |
|
|
| if decision == "retrieval": |
| return "retrieval" |
| elif decision == "sql": |
| return "sql" |
| elif decision == "analytics": |
| return "analytics" |
| elif decision == "code": |
| return "code" |
| elif decision == "web": |
| return "web" |
| else: |
| return "synthesizer" |
|
|
|
|
| def build_copilot_graph(): |
| """ |
| Constructs and compiles the multi-agent executable LangGraph. |
| """ |
| builder = StateGraph(CopilotState) |
|
|
| |
| builder.add_node("planner", planner_node) |
| builder.add_node("retrieval", retrieval_node) |
| builder.add_node("sql", sql_node) |
| builder.add_node("analytics", analytics_node) |
| builder.add_node("code", code_node) |
| builder.add_node("web", web_node) |
| builder.add_node("synthesizer", synthesizer_node) |
|
|
| |
| builder.add_edge(START, "planner") |
|
|
| |
| builder.add_conditional_edges( |
| "planner", |
| route_next, |
| { |
| "retrieval": "retrieval", |
| "sql": "sql", |
| "analytics": "analytics", |
| "code": "code", |
| "web": "web", |
| "synthesizer": "synthesizer", |
| }, |
| ) |
|
|
| |
| builder.add_edge("retrieval", "synthesizer") |
| builder.add_edge("sql", "synthesizer") |
| builder.add_edge("analytics", "synthesizer") |
| builder.add_edge("code", "synthesizer") |
| builder.add_edge("web", "synthesizer") |
| builder.add_edge("synthesizer", END) |
|
|
| |
| return builder.compile() |
|
|
|
|
| |
| copilot_graph = build_copilot_graph() |
|
|
|
|
| async def run_copilot_pipeline( |
| query: str, |
| tenant_id: str, |
| user_id: str, |
| user_role: str = "employee", |
| session_id: str = "default", |
| history: list = None, |
| ) -> Dict[str, Any]: |
| """ |
| High-level runner interface for the LangGraph Copilot Pipeline. |
| """ |
| start_time = time.perf_counter() |
|
|
| initial_state: CopilotState = { |
| "query": query, |
| "tenant_id": tenant_id, |
| "user_id": user_id, |
| "user_role": user_role, |
| "session_id": session_id, |
| "history": history or [], |
| "agent_outputs": [], |
| "retrieved_chunks": [], |
| "citations": [], |
| } |
|
|
| logger.info( |
| "Invoking Copilot LangGraph pipeline", |
| query=query, |
| tenant_id=tenant_id, |
| user_id=user_id, |
| ) |
|
|
| final_state = await copilot_graph.ainvoke(initial_state) |
|
|
| duration_ms = (time.perf_counter() - start_time) * 1000 |
|
|
| return { |
| "response": final_state.get("final_response", ""), |
| "routing_decision": final_state.get("routing_decision", "direct"), |
| "agent_type": final_state.get("routing_decision", "synthesizer"), |
| "citations": final_state.get("citations", []), |
| "sources": final_state.get("citations", []), |
| "agent_outputs": final_state.get("agent_outputs", []), |
| "tokens_used": final_state.get("tokens_used", 0), |
| "latency_ms": round(duration_ms, 2), |
| "error": final_state.get("error"), |
| } |
|
|
