File size: 5,123 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 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 | """
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)
# 1. Add Nodes
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)
# 2. Add Edges
builder.add_edge(START, "planner")
# Conditional branching from planner
builder.add_conditional_edges(
"planner",
route_next,
{
"retrieval": "retrieval",
"sql": "sql",
"analytics": "analytics",
"code": "code",
"web": "web",
"synthesizer": "synthesizer",
},
)
# All specialist nodes feed into final 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)
# Compile executable graph
return builder.compile()
# ── Executable Compiled Graph ─────────────────────────────────
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"),
}
|