I have document and live data. Creating RAG. So need correct routing of user query like where to send the user query. Data is dynamic but it is of a specific domain. so how to design supervisor node. No hardcoding i need
Hi @Prateek
What does “no hardcoding” realistically mean for you?
You still must register what “places you can look” are (doc index, live DB/API tools, etc.).
You can make routing data-driven:
- Route based on evidence (e.g., quick retrieval previews + relevance grading), not keywords.
- Route based on a catalog of sources/tools (names + descriptions + freshness), which you can update automatically as data/tools change.
Supervisor design (evidence-based, minimal assumptions)
A cheap “probe → decide → execute” pattern:
- Probe docs: do a fast top‑k retrieval from your vector store and keep short snippets (or titles/metadata).
- Decide route: have a small LLM router output a constrained structured decision (e.g.,
docs_rag,live_data,hybrid,clarify) using those snippets + tool catalog. - Execute: conditional edges route to the correct subgraph.
In general, see this example langgraph/examples/rag/langgraph_adaptive_rag.ipynb at main · langchain-ai/langgraph · GitHub
Structured output matters for routing
For routers, you want predictable outputs (a small finite set), not free-form text.
LangGraph docs show routing by augmenting an LLM with a schema using .with_structured_output(...) and then branching based on the returned field.
LangGraph docs “LLM with Structured Output for Routing” (workflows-agents) at Workflows and agents - Docs by LangChain.
Conditional edges - the right LangGraph primitive
In LangGraph, the idiomatic way to route is add_conditional_edges(source_node, routing_fn, mapping?).
The routing function inspects state and returns the next node label(s).
Source: LangGraph Graph API docs add_conditional_edges at Graph API overview - Docs by LangChain.
Sketch (supervisor node + two subgraphs)
The key “no hardcoding” bit is that the router sees:
- a doc probe (snippets + similarity scores)
- a live tool catalog (names + descriptions, can be generated from your tools)
Treat is as a pseudo-code.
from __future__ import annotations
from typing_extensions import Literal, TypedDict
from pydantic import BaseModel, Field
from langgraph.graph import StateGraph, START, END
from langchain_core.messages import HumanMessage, SystemMessage
class State(TypedDict, total=False):
question: str
# Evidence for routing (kept small/cheap)
doc_probe: list[dict] # [{ "score": float, "title": str, "snippet": str, "source_id": str }, ...]
# Router output
route: str
# Outputs
answer: str
class RouteDecision(BaseModel):
next: Literal["docs_rag", "live_data", "hybrid", "clarify"] = Field(
description="Which path should handle this user query?"
)
reason: str = Field(description="Short justification, for debugging/telemetry.")
def build_supervisor_router(llm, *, source_catalog: list[dict]):
"""Create a stable router runnable with structured output.
source_catalog can be generated dynamically from your tools (name/description/freshness).
"""
router = llm.with_structured_output(RouteDecision)
def route_fn(state: State) -> dict:
question = state["question"]
doc_probe = state.get("doc_probe", [])
# Keep the prompt generic; avoid baking in domain details.
msgs = [
SystemMessage(
content=(
"You are a routing supervisor. Choose the best route.\n"
"- Prefer docs_rag when the doc_probe contains high-relevance evidence.\n"
"- Prefer live_data for questions that require current/transactional values "
"or need querying a system of record.\n"
"- Use hybrid if both are needed.\n"
"- Use clarify if the question is ambiguous and routing would be guesswork.\n\n"
"You will be given:\n"
"1) a doc_probe (top snippets + scores)\n"
"2) a catalog of live tools/sources (names + descriptions)\n"
"Return a structured decision."
)
),
HumanMessage(
content=(
f"question: {question}\n\n"
f"doc_probe: {doc_probe}\n\n"
f"live_source_catalog: {source_catalog}\n"
)
),
]
decision = router.invoke(msgs)
return {"route": decision.next}
return route_fn
def supervisor_node(state: State) -> State:
# If you prefer, put routing *inside* this node instead of the conditional edge fn.
return state
def choose_next(state: State) -> str:
return state["route"]
def docs_rag_node(state: State) -> State:
# Your RAG pipeline/subgraph goes here (retrieve -> synthesize -> cite).
# Use state["question"] and return {"answer": "..."}.
raise NotImplementedError
def live_data_node(state: State) -> State:
# Option A (simple): call a tool-calling agent that knows your live tools.
# In LangChain v1, the recommended constructor is create_agent.
# Source: https://docs.langchain.com/oss/python/releases/langchain-v1/index
#
# from langchain.agents import create_agent
# agent = create_agent(model=..., tools=[...], system_prompt="...")
# result = agent.invoke({"messages": [{"role": "user", "content": state["question"]}]})
# return {"answer": result["messages"][-1].content}
raise NotImplementedError
def build_graph(*, llm, source_catalog: list[dict], doc_probe_fn):
# Build graph
g = StateGraph(State)
g.add_node("supervisor", supervisor_node)
g.add_node("docs_rag", docs_rag_node)
g.add_node("live_data", live_data_node)
# (optional) hybrid node that runs both, merges, then answers
# g.add_node("hybrid", hybrid_node)
# g.add_node("clarify", clarify_node)
g.add_edge(START, "supervisor")
# Precompute doc probe before routing (you can make this its own node too)
def enrich_with_probe(state: State) -> State:
state["doc_probe"] = doc_probe_fn(state["question"])
return state
# Route decision computed via structured router (as a conditional function or node)
route_fn = build_supervisor_router(llm, source_catalog=source_catalog)
# Compose: supervisor -> enrich -> route
g.add_node("probe", enrich_with_probe)
g.add_edge("supervisor", "probe")
def route_after_probe(state: State) -> str:
state.update(route_fn(state))
return choose_next(state)
g.add_conditional_edges(
"probe",
route_after_probe,
{
"docs_rag": "docs_rag",
"live_data": "live_data",
# "hybrid": "hybrid",
# "clarify": "clarify",
},
)
g.add_edge("docs_rag", END)
g.add_edge("live_data", END)
return g.compile()
Making it truly robust
- Add routing telemetry: store
(question, route, reason, outcomes)so you can spot systematic misroutes. - Use relevance grading: instead of a fixed similarity threshold, run an LLM “doc relevance grader” on doc_probe to decide whether docs are actually answering the question (the adaptive RAG example does this as a separate grader step).
Source:langgraph_adaptive_rag.ipynbexample. - Keep your “source catalog” dynamic:
- Doc source: store an auto-updated corpus summary and key facets (generated nightly or on ingestion).
- Live source: generate descriptions from your tool docstrings / OpenAPI schema / DB schema introspection.
Fast alternative (no explicit supervisor node)
If your “routing” is simply “choose the right tool”, you can skip an explicit supervisor graph and instead:
- Build a single
create_agent(...)with tools likesearch_docs,query_db,call_api - Let tool selection be the router (tool descriptions become the dynamic routing policy)
This is often simpler, and still uses LangGraph under the hood for durable execution, but you lose some explicit visibility/control over branches unless you log tool calls.