Pattern Router

Capsule overview

A router classifies the incoming request and sends it to the right specialist. Unlike a supervisor (which coordinates multiple agents across multiple steps), a router makes ONE decision: "who should handle this?" Think of the difference as a receptionist vs a project manager. The receptionist tells you "go to floor 3, accounting office" and their job is done. The project manager assigns tasks, reviews results, asks for corrections, and coordinates everyone's work.

In capsules 02-04 you built systems with Supervisor, Handoffs, and Subagents. In every one of those patterns, there's an agent that orchestrates: it decides, delegates, reviews, and delegates again. The router is deliberately simpler — it's the pattern you reach for when a supervisor's complexity isn't justified. If your system only needs to classify and delegate, a router is faster, cheaper, and easier to debug.

This capsule starts with deterministic routers (no LLM, pure logic), moves on to LLM-based routers with structured output, and finishes with advanced patterns like multi-level routing and fallbacks.

Router vs Supervisor: the key distinction

Before writing any code, you need to be clear on the difference:

The rule: if the work is done once you've delegated to the specialist, use a router. If you need to review the result and possibly delegate to another agent, you need a supervisor.

Deterministic router: rules without an LLM

The simplest router doesn't use an LLM at all. It classifies with rules: keywords, regex, message length, metadata. It's the fastest, the cheapest, and the most predictable.

Basic example: keyword matching

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langchain.chat_models import init_chat_model
from langchain_core.messages import AnyMessage, HumanMessage, SystemMessage
from IPython.display import Image, display

class RouterState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
    routed_to: str

def deterministic_router(state: RouterState) -> dict:
    text = state["messages"][-1].content.lower()

    if any(kw in text for kw in ["code", "program", "bug", "function", "code error"]):
        return {"routed_to": "code_agent"}
    elif any(kw in text for kw in ["write", "draft", "article", "blog", "text"]):
        return {"routed_to": "writer_agent"}
    else:
        return {"routed_to": "general_agent"}

def route_decision(state: RouterState) -> str:
    return state["routed_to"]

def make_specialist(system_prompt: str):
    def specialist(state: RouterState) -> dict:
        model = init_chat_model("openai:gpt-4.1-mini")
        msgs = [SystemMessage(content=system_prompt)] + state["messages"]
        response = model.invoke(msgs)
        return {"messages": [response]}
    return specialist

graph = StateGraph(RouterState)
graph.add_node("router", deterministic_router)
graph.add_node("code_agent", make_specialist(
    "You are a programming expert. Respond in English, with code when it's relevant."
))
graph.add_node("writer_agent", make_specialist(
    "You are a professional writer. Respond in English with clear, structured prose."
))
graph.add_node("general_agent", make_specialist(
    "You are a general assistant. Respond in English in a friendly way."
))

graph.add_edge(START, "router")
graph.add_conditional_edges("router", route_decision, {
    "code_agent": "code_agent",
    "writer_agent": "writer_agent",
    "general_agent": "general_agent",
})
for agent in ["code_agent", "writer_agent", "general_agent"]:
    graph.add_edge(agent, END)

app = graph.compile()
display(Image(app.get_graph().draw_mermaid_png()))

result = app.invoke({
    "messages": [HumanMessage(content="I have a bug in my Python function")],
    "routed_to": "",
})
print(f"Routed to: {result['routed_to']}")
print(result["messages"][-1].content[:120])
# Output:
# Routed to: code_agent
# Sure, I'll help you track down that bug. Could you share the code for your function...

The flow is linear: START → router → specialist → END. The router doesn't use an LLM — it's pure Python logic. That means zero latency on the routing decision and zero token cost.

Deterministic variants

Keyword matching is the simplest form, but it's not the only one:

import re

def regex_router(state: RouterState) -> dict:
    text = state["messages"][-1].content

    if re.search(r"def\s+\w+|class\s+\w+|import\s+\w+", text):
        return {"routed_to": "code_agent"}
    elif re.search(r"\d+[\+\-\*/]\d+|calculate|percent", text.lower()):
        return {"routed_to": "math_agent"}
    else:
        return {"routed_to": "general_agent"}

def length_router(state: RouterState) -> dict:
    text = state["messages"][-1].content
    word_count = len(text.split())

    if word_count > 200:
        return {"routed_to": "summarizer_agent"}
    elif word_count < 10:
        return {"routed_to": "clarification_agent"}
    else:
        return {"routed_to": "general_agent"}

def metadata_router(state: RouterState) -> dict:
    """Router based on message metadata, not on content."""
    last_msg = state["messages"][-1]
    if hasattr(last_msg, "additional_kwargs"):
        lang = last_msg.additional_kwargs.get("language", "es")
        if lang == "en":
            return {"routed_to": "english_agent"}
    return {"routed_to": "spanish_agent"}

LLM-based router: classification with a model

When rules aren't enough — because the inputs are ambiguous, users phrase things in many ways, or the categories are subtle — you use an LLM to classify. The key is structured output: the model returns a specific category, not free text.

Classification with structured output

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated, Literal
import operator
from pydantic import BaseModel, Field
from langgraph.graph import StateGraph, START, END
from langchain.chat_models import init_chat_model
from langchain_core.messages import AnyMessage, HumanMessage, SystemMessage
from IPython.display import Image, display

class IntentClassification(BaseModel):
    """Classification of the user's intent."""
    intent: Literal["research", "analysis", "creative", "code", "general"] = Field(
        description="The category that best describes the user's request"
    )
    reasoning: str = Field(
        description="Brief explanation of why this category was chosen"
    )

class LLMRouterState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
    intent: str
    routing_reasoning: str

classifier = init_chat_model("openai:gpt-4.1-mini")

def llm_router(state: LLMRouterState) -> dict:
    structured_model = classifier.with_structured_output(IntentClassification)
    result = structured_model.invoke([
        SystemMessage(content=(
            "Classify the user's request into one of these categories:\n"
            "- research: look up information, investigate a topic, compare options\n"
            "- analysis: analyze data, find patterns, evaluate metrics\n"
            "- creative: write content, generate ideas, create narratives\n"
            "- code: write code, debug, explain programming concepts\n"
            "- general: anything that doesn't fit the categories above"
        )),
        state["messages"][-1],
    ])
    return {
        "intent": result.intent,
        "routing_reasoning": result.reasoning,
    }

def route_by_intent(state: LLMRouterState) -> str:
    return state["intent"]

def make_specialist(name: str, system_prompt: str):
    def specialist(state: LLMRouterState) -> dict:
        model = init_chat_model("openai:gpt-4.1-mini")
        msgs = [SystemMessage(content=system_prompt)] + state["messages"]
        return {"messages": [model.invoke(msgs)]}
    return specialist

specialists = {
    "research": ("You are a researcher. Find accurate information and cite sources. Respond in English."),
    "analysis": ("You are a data analyst. Identify patterns and present conclusions. Respond in English."),
    "creative": ("You are a creative writer. Generate original, engaging content. Respond in English."),
    "code": ("You are a software engineer. Write clean code with explanations. Respond in English."),
    "general": ("You are a general assistant. Answer clearly and warmly. Respond in English."),
}

graph = StateGraph(LLMRouterState)
graph.add_node("router", llm_router)

for intent_name, system_prompt in specialists.items():
    graph.add_node(f"{intent_name}_agent", make_specialist(intent_name, system_prompt))

graph.add_edge(START, "router")
graph.add_conditional_edges("router", route_by_intent, {
    name: f"{name}_agent" for name in specialists
})
for name in specialists:
    graph.add_edge(f"{name}_agent", END)

app = graph.compile()
display(Image(app.get_graph().draw_mermaid_png()))

result = app.invoke({
    "messages": [HumanMessage(content="What are the differences between React and Vue for large projects?")],
    "intent": "",
    "routing_reasoning": "",
})
print(f"Intent: {result['intent']}")
print(f"Reasoning: {result['routing_reasoning']}")
print(result["messages"][-1].content[:150])
# Output:
# Intent: research
# Reasoning: The user wants to compare two frameworks, which is a research task
# React and Vue are JavaScript frameworks with different approaches. React uses a model...

The LLM classifies the request into one of 5 categories and explains why. with_structured_output guarantees the result is one of the valid categories — there's no risk of the model inventing a category that doesn't exist.

Router with the Functional API

If your flow is simple (classify → delegate → respond), the Functional API is more direct:

from dotenv import load_dotenv
load_dotenv()

from langgraph.func import entrypoint, task
from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage, SystemMessage

model = init_chat_model("openai:gpt-4.1-mini")

def classify_intent(text: str) -> str:
    text_lower = text.lower()
    if any(kw in text_lower for kw in ["code", "program", "bug", "function"]):
        return "code"
    elif any(kw in text_lower for kw in ["write", "draft", "article", "blog"]):
        return "creative"
    elif any(kw in text_lower for kw in ["research", "compare", "differences", "what is"]):
        return "research"
    return "general"

SPECIALIST_PROMPTS = {
    "code": "You are a programming expert. Respond in English.",
    "creative": "You are a professional writer. Respond in English.",
    "research": "You are a researcher. Respond in English, with sources.",
    "general": "You are a general assistant. Respond in English.",
}

@task
def ask_specialist(question: str, system_prompt: str) -> str:
    response = model.invoke([
        SystemMessage(content=system_prompt),
        HumanMessage(content=question),
    ])
    return response.content

@entrypoint()
def router_agent(question: str) -> dict:
    intent = classify_intent(question)
    prompt = SPECIALIST_PROMPTS[intent]
    answer = ask_specialist(question, prompt).result()
    return {"intent": intent, "answer": answer}

result = router_agent.invoke("Write a short article about artificial intelligence")
print(f"Intent: {result['intent']}")
print(result["answer"][:120])
# Output:
# Intent: creative
# Artificial intelligence has stopped being science fiction and turned into something...

No StateGraph, no edges, no nodes. The classification is an if/elif/else and the delegation is a dict lookup. For simple routers with 3-5 destinations, this approach is cleaner.

Multi-level routing

When you have many possible destinations (10+), a single router gets imprecise. The fix: hierarchical routing. First classify by domain, then by subtask within that domain.

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated, Literal
import operator
from pydantic import BaseModel, Field
from langgraph.graph import StateGraph, START, END
from langchain.chat_models import init_chat_model
from langchain_core.messages import AnyMessage, HumanMessage, SystemMessage
from IPython.display import Image, display

class DomainClassification(BaseModel):
    domain: Literal["engineering", "business", "support"] = Field(
        description="Main domain of the request"
    )

class EngineeringSubtask(BaseModel):
    subtask: Literal["frontend", "backend", "devops"] = Field(
        description="Subtask within engineering"
    )

class MultiLevelState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
    domain: str
    subtask: str

classifier = init_chat_model("openai:gpt-4.1-mini")

def domain_router(state: MultiLevelState) -> dict:
    structured = classifier.with_structured_output(DomainClassification)
    result = structured.invoke([
        SystemMessage(content=(
            "Classify into: engineering (code, infrastructure, tech), "
            "business (sales, metrics, strategy), "
            "support (user help, FAQ, troubleshooting)"
        )),
        state["messages"][-1],
    ])
    return {"domain": result.domain}

def route_domain(state: MultiLevelState) -> str:
    return state["domain"]

def engineering_subrouter(state: MultiLevelState) -> dict:
    structured = classifier.with_structured_output(EngineeringSubtask)
    result = structured.invoke([
        SystemMessage(content=(
            "Classify the engineering task into: "
            "frontend (UI, React, CSS), backend (API, DB, Python), devops (deploy, CI/CD, Docker)"
        )),
        state["messages"][-1],
    ])
    return {"subtask": result.subtask}

def route_engineering(state: MultiLevelState) -> str:
    return state["subtask"]

def make_agent(prompt: str):
    def agent(state: MultiLevelState) -> dict:
        model = init_chat_model("openai:gpt-4.1-mini")
        return {"messages": [model.invoke(
            [SystemMessage(content=prompt)] + state["messages"]
        )]}
    return agent

graph = StateGraph(MultiLevelState)

graph.add_node("domain_router", domain_router)
graph.add_node("eng_subrouter", engineering_subrouter)
graph.add_node("business_agent", make_agent("Business expert. Respond in English."))
graph.add_node("support_agent", make_agent("Support agent. Respond in English."))
graph.add_node("frontend_agent", make_agent("Frontend expert (React, CSS). Respond in English."))
graph.add_node("backend_agent", make_agent("Backend expert (Python, APIs). Respond in English."))
graph.add_node("devops_agent", make_agent("DevOps expert (Docker, CI/CD). Respond in English."))

graph.add_edge(START, "domain_router")
graph.add_conditional_edges("domain_router", route_domain, {
    "engineering": "eng_subrouter",
    "business": "business_agent",
    "support": "support_agent",
})
graph.add_conditional_edges("eng_subrouter", route_engineering, {
    "frontend": "frontend_agent",
    "backend": "backend_agent",
    "devops": "devops_agent",
})
for agent in ["business_agent", "support_agent", "frontend_agent", "backend_agent", "devops_agent"]:
    graph.add_edge(agent, END)

app = graph.compile()
display(Image(app.get_graph().draw_mermaid_png()))

result = app.invoke({
    "messages": [HumanMessage(content="How do I set up a multi-stage Dockerfile for my Python app?")],
    "domain": "",
    "subtask": "",
})
print(f"Domain: {result['domain']}, Subtask: {result['subtask']}")
print(result["messages"][-1].content[:150])
# Output:
# Domain: engineering, Subtask: devops
# To set up a multi-stage Dockerfile in Python, you need to define two stages...

The first router classifies into 3 domains. If the domain is engineering, a second router narrows it down to 3 subtasks. That way you cover 5 final destinations with 3-option decisions each — more precise than a single 5-option router.

Router with fallback

What happens when the router isn't sure? Instead of guessing, it asks for clarification.

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated, Literal
import operator
from pydantic import BaseModel, Field
from langgraph.graph import StateGraph, START, END
from langchain.chat_models import init_chat_model
from langchain_core.messages import AnyMessage, HumanMessage, AIMessage, SystemMessage
from IPython.display import Image, display

class ConfidentClassification(BaseModel):
    intent: Literal["code", "writing", "research", "unclear"] = Field(
        description="The user's intent. Use 'unclear' if you can't classify it confidently."
    )
    confidence: float = Field(
        description="Confidence in the classification, from 0.0 to 1.0"
    )

class FallbackState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
    intent: str
    confidence: float

classifier = init_chat_model("openai:gpt-4.1-mini")
CONFIDENCE_THRESHOLD = 0.7

def classify_with_confidence(state: FallbackState) -> dict:
    structured = classifier.with_structured_output(ConfidentClassification)
    result = structured.invoke([
        SystemMessage(content=(
            "Classify the request. If it's ambiguous or you could confuse two categories, "
            "mark it 'unclear' and lower the confidence."
        )),
        state["messages"][-1],
    ])
    return {"intent": result.intent, "confidence": result.confidence}

def route_with_fallback(state: FallbackState) -> str:
    if state["confidence"] < CONFIDENCE_THRESHOLD or state["intent"] == "unclear":
        return "clarification"
    return state["intent"]

def clarification_node(state: FallbackState) -> dict:
    return {"messages": [AIMessage(content=(
        "I'm not sure how to best help you. Could you tell me whether you need:\n"
        "1. Help with code\n"
        "2. Me to write content\n"
        "3. Me to research a topic"
    ))]}

def make_agent(prompt: str):
    def agent(state: FallbackState) -> dict:
        model = init_chat_model("openai:gpt-4.1-mini")
        return {"messages": [model.invoke(
            [SystemMessage(content=prompt)] + state["messages"]
        )]}
    return agent

graph = StateGraph(FallbackState)
graph.add_node("classify", classify_with_confidence)
graph.add_node("clarification", clarification_node)
graph.add_node("code", make_agent("Code expert. Respond in English."))
graph.add_node("writing", make_agent("Professional writer. Respond in English."))
graph.add_node("research", make_agent("Researcher. Respond in English."))

graph.add_edge(START, "classify")
graph.add_conditional_edges("classify", route_with_fallback, {
    "code": "code",
    "writing": "writing",
    "research": "research",
    "clarification": "clarification",
})
for node in ["code", "writing", "research", "clarification"]:
    graph.add_edge(node, END)

app = graph.compile()
display(Image(app.get_graph().draw_mermaid_png()))

result = app.invoke({
    "messages": [HumanMessage(content="Python")],
    "intent": "",
    "confidence": 0.0,
})
print(f"Intent: {result['intent']}, Confidence: {result['confidence']}")
print(result["messages"][-1].content[:200])
# Output (example with an ambiguous input):
# Intent: unclear, Confidence: 0.3
# I'm not sure how to best help you. Could you tell me whether you need:
# 1. Help with code
# 2. Me to write content
# 3. Me to research a topic

The model returns a confidence alongside the classification. If the confidence is below the threshold (0.7), the system asks for clarification instead of sending the user to the wrong specialist. That matters especially when routing mistakes are costly (e.g. sending a complaint to the wrong department).

Comparison table: deterministic vs LLM

When to graduate from deterministic to LLM

Always start with a deterministic router. Move up to an LLM only when:

• ❌ Users phrase things in ways your keywords don't cover ("help me with this chunk of script" → is that code?)
• ❌ You have more than 5-6 categories and the rules get brittle
• ❌ The inputs are multi-intent ("research Docker and write me a summary")
• ❌ Your team doesn't want to maintain an ever-growing keyword list

And stay deterministic when:

• ✅ The inputs are structured (forms, commands, metadata)
• ✅ You have 2-3 well-differentiated categories
• ✅ Latency and cost matter (high volume)
• ✅ You need 100% reproducible behavior for testing

The natural progression is: deterministic → deterministic + LLM fallback → full LLM. Don't jump to the end without a reason.

Troubleshooting

Problem 1: The deterministic router sends things to the wrong agent

Symptom: The user says "Can you write me a function?" and the router sends it to the writer_agent instead of the code_agent because "write" matched first. Cause: The order the keywords are evaluated in creates conflicts. "Write" shows up in both coding and writing contexts. Fix: Put the more specific keywords first, or use keyword combinations instead of individual keywords:

def improved_router(state: RouterState) -> dict:
    text = state["messages"][-1].content.lower()

    if any(kw in text for kw in ["function", "bug", "code", "program", "variable"]):
        return {"routed_to": "code_agent"}
    elif any(kw in text for kw in ["article", "blog", "draft", "paragraph"]):
        return {"routed_to": "writer_agent"}
    return {"routed_to": "general_agent"}

Problem 2: The LLM router returns a category that doesn't exist in the edges

Symptom: ValueError: Expected one of ['code', 'writing', 'research'] when you run the graph. Cause: The model produced a category outside the defined options (e.g. "programming" instead of "code"). Fix: Use Literal in your Pydantic model to constrain the options, and with_structured_output to guarantee conformance:

class Classification(BaseModel):
    intent: Literal["code", "writing", "research"]  # Only these 3 options

Problem 3: The LLM router is too slow for the user experience

Symptom: The user waits 1-2 seconds just for the system to decide who to send the request to, before the specialist even starts working. Cause: LLM classification adds latency before the real answer. Fix: Use a faster model for classification (e.g. gpt-4.1-nano) and save the more capable models for the specialists:

fast_classifier = init_chat_model("openai:gpt-4.1-nano")
smart_specialist = init_chat_model("openai:gpt-4.1")

Problem 4: Multi-level routing makes too many LLM calls

Symptom: With 2 levels of LLM routing, total classification latency goes over 2 seconds before any specialist does any work. Cause: Each routing level is a sequential call to the LLM. Fix: Make the first level deterministic (fast, no LLM) and only use an LLM for the second level:

def hybrid_routing(state):
    text = state["messages"][-1].content.lower()
    if any(kw in text for kw in ["code", "api", "deploy", "docker"]):
        return {"domain": "engineering"}
    elif any(kw in text for kw in ["sales", "revenue", "customer"]):
        return {"domain": "business"}
    return {"domain": "support"}

Exercises

Exercise 1: Deterministic router for support (Easy)

Build a deterministic router with 3 destinations: billing, technical (technical problems), and general. Use keyword matching. Test it with 3 different messages.

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langchain.chat_models import init_chat_model
from langchain_core.messages import AnyMessage, HumanMessage, SystemMessage

class SupportState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
    department: str

def support_router(state: SupportState) -> dict:
    text = state["messages"][-1].content.lower()
    if any(kw in text for kw in ["invoice", "billing", "payment", "price", "plan"]):
        return {"department": "billing"}
    elif any(kw in text for kw in ["error", "not working", "down", "bug", "slow"]):
        return {"department": "technical"}
    return {"department": "general"}

def route_dept(state: SupportState) -> str:
    return state["department"]

def make_agent(prompt: str):
    def agent(state: SupportState) -> dict:
        model = init_chat_model("openai:gpt-4.1-mini")
        return {"messages": [model.invoke(
            [SystemMessage(content=prompt)] + state["messages"]
        )]}
    return agent

graph = StateGraph(SupportState)
graph.add_node("router", support_router)
graph.add_node("billing", make_agent("Billing agent. Respond in English."))
graph.add_node("technical", make_agent("Technical support. Respond in English."))
graph.add_node("general", make_agent("General assistant. Respond in English."))

graph.add_edge(START, "router")
graph.add_conditional_edges("router", route_dept, {
    "billing": "billing", "technical": "technical", "general": "general"
})
for node in ["billing", "technical", "general"]:
    graph.add_edge(node, END)

app = graph.compile()

tests = [
    "How much does the premium plan cost?",
    "My app is not working, it started yesterday",
    "What are your support hours?",
]
for msg in tests:
    result = app.invoke({"messages": [HumanMessage(content=msg)], "department": ""})
    print(f"'{msg[:40]}...' → {result['department']}")
# Output:
# 'How much does the premium plan cost?...' → billing
# 'My app is not working, it started yester...' → technical
# 'What are your support hours?...' → general

Exercise 2: LLM router with structured output (Easy)

Convert the deterministic router from exercise 1 into an LLM-based router using with_structured_output. Compare the results with the input "I was charged twice this month" (which a keyword router might send to general because it doesn't literally contain "invoice").

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated, Literal
import operator
from pydantic import BaseModel, Field
from langgraph.graph import StateGraph, START, END
from langchain.chat_models import init_chat_model
from langchain_core.messages import AnyMessage, HumanMessage, SystemMessage

class DeptClassification(BaseModel):
    department: Literal["billing", "technical", "general"] = Field(
        description="Department that should handle this request"
    )

class SupportState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
    department: str

classifier = init_chat_model("openai:gpt-4.1-mini")

def llm_support_router(state: SupportState) -> dict:
    structured = classifier.with_structured_output(DeptClassification)
    result = structured.invoke([
        SystemMessage(content=(
            "Classify the user's request:\n"
            "- billing: anything about payments, charges, invoices, plans, prices\n"
            "- technical: technical problems, errors, performance, bugs\n"
            "- general: anything else"
        )),
        state["messages"][-1],
    ])
    return {"department": result.department}

def route_dept(state: SupportState) -> str:
    return state["department"]

def make_agent(prompt: str):
    def agent(state: SupportState) -> dict:
        model = init_chat_model("openai:gpt-4.1-mini")
        return {"messages": [model.invoke(
            [SystemMessage(content=prompt)] + state["messages"]
        )]}
    return agent

graph = StateGraph(SupportState)
graph.add_node("router", llm_support_router)
graph.add_node("billing", make_agent("Billing agent. Respond in English."))
graph.add_node("technical", make_agent("Technical support. Respond in English."))
graph.add_node("general", make_agent("General assistant. Respond in English."))

graph.add_edge(START, "router")
graph.add_conditional_edges("router", route_dept, {
    "billing": "billing", "technical": "technical", "general": "general"
})
for node in ["billing", "technical", "general"]:
    graph.add_edge(node, END)

app = graph.compile()

result = app.invoke({
    "messages": [HumanMessage(content="I was charged twice this month")],
    "department": "",
})
print(f"Department: {result['department']}")
print(result["messages"][-1].content[:120])
# Output:
# Department: billing
# I'm really sorry about the duplicate charge. Let me help you get that resolved...

Exercise 3: Router with the Functional API (Medium)

Implement a router with the Functional API that classifies questions into 4 categories: math, history, science, other. Use a deterministic router. Test it with 4 different inputs.

from dotenv import load_dotenv
load_dotenv()

from langgraph.func import entrypoint, task
from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage, SystemMessage

model = init_chat_model("openai:gpt-4.1-mini")

def classify_subject(text: str) -> str:
    text_lower = text.lower()
    if any(kw in text_lower for kw in ["calculate", "equation", "sum", "percent", "number"]):
        return "math"
    elif any(kw in text_lower for kw in ["history", "war", "century", "revolution", "ancient"]):
        return "history"
    elif any(kw in text_lower for kw in ["cell", "atom", "planet", "chemistry", "physics"]):
        return "science"
    return "other"

PROMPTS = {
    "math": "You are a math teacher. Explain step by step. Respond in English.",
    "history": "You are a historian. Give context and dates. Respond in English.",
    "science": "You are a scientist. Explain precisely. Respond in English.",
    "other": "You are a general tutor. Respond in English, clearly.",
}

@task
def answer_question(question: str, system_prompt: str) -> str:
    return model.invoke([
        SystemMessage(content=system_prompt),
        HumanMessage(content=question),
    ]).content

@entrypoint()
def tutor_router(question: str) -> dict:
    subject = classify_subject(question)
    answer = answer_question(question, PROMPTS[subject]).result()
    return {"subject": subject, "answer": answer}

tests = [
    "What is 15 percent of 240?",
    "What caused the French Revolution?",
    "How does an atom work?",
    "What is the capital of Japan?",
]
for q in tests:
    result = tutor_router.invoke(q)
    print(f"'{q[:35]}...' → {result['subject']}")
# Output:
# 'What is 15 percent of 240?...' → math
# 'What caused the French Revolution?...' → history
# 'How does an atom work?...' → science
# 'What is the capital of Japan?...' → other

Exercise 4: Multi-level router (Medium)

Build a 2-level router. Level 1: classify into tech or business (deterministic). Level 2: if it's tech, use an LLM to sub-classify into frontend, backend, or data. Implement it with StateGraph.

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated, Literal
import operator
from pydantic import BaseModel, Field
from langgraph.graph import StateGraph, START, END
from langchain.chat_models import init_chat_model
from langchain_core.messages import AnyMessage, HumanMessage, SystemMessage

class TechSubclass(BaseModel):
    area: Literal["frontend", "backend", "data"] = Field(
        description="Specific technical area"
    )

class TwoLevelState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
    level1: str
    level2: str

classifier = init_chat_model("openai:gpt-4.1-mini")

def level1_router(state: TwoLevelState) -> dict:
    text = state["messages"][-1].content.lower()
    if any(kw in text for kw in ["code", "api", "database", "react", "deploy", "python"]):
        return {"level1": "tech"}
    return {"level1": "business"}

def route_level1(state: TwoLevelState) -> str:
    return state["level1"]

def level2_tech_router(state: TwoLevelState) -> dict:
    structured = classifier.with_structured_output(TechSubclass)
    result = structured.invoke([
        SystemMessage(content=(
            "Classify this technical question into:\n"
            "- frontend: UI, React, CSS, JavaScript, web design\n"
            "- backend: APIs, servers, Python, databases, authentication\n"
            "- data: data, analytics, ML, pandas, visualization"
        )),
        state["messages"][-1],
    ])
    return {"level2": result.area}

def route_level2(state: TwoLevelState) -> str:
    return state["level2"]

def make_agent(prompt: str):
    def agent(state: TwoLevelState) -> dict:
        model = init_chat_model("openai:gpt-4.1-mini")
        return {"messages": [model.invoke(
            [SystemMessage(content=prompt)] + state["messages"]
        )]}
    return agent

graph = StateGraph(TwoLevelState)
graph.add_node("level1", level1_router)
graph.add_node("level2_tech", level2_tech_router)
graph.add_node("business_agent", make_agent("Business expert. Respond in English."))
graph.add_node("frontend_agent", make_agent("Frontend expert. Respond in English."))
graph.add_node("backend_agent", make_agent("Backend expert. Respond in English."))
graph.add_node("data_agent", make_agent("Data and ML expert. Respond in English."))

graph.add_edge(START, "level1")
graph.add_conditional_edges("level1", route_level1, {
    "tech": "level2_tech",
    "business": "business_agent",
})
graph.add_conditional_edges("level2_tech", route_level2, {
    "frontend": "frontend_agent",
    "backend": "backend_agent",
    "data": "data_agent",
})
for node in ["business_agent", "frontend_agent", "backend_agent", "data_agent"]:
    graph.add_edge(node, END)

app = graph.compile()

result = app.invoke({
    "messages": [HumanMessage(content="How do I optimize a SQL query that takes 30 seconds?")],
    "level1": "",
    "level2": "",
})
print(f"Level 1: {result['level1']}, Level 2: {result['level2']}")
print(result["messages"][-1].content[:120])
# Output:
# Level 1: tech, Level 2: backend
# To optimize a slow SQL query, you first need to identify the bottleneck...

Exercise 5: Router with fallback and confidence (Advanced)

Implement an LLM router that returns a confidence. If the confidence is < 0.6, the system must ask for clarification. If it's between 0.6 and 0.8, it should route but add a disclaimer. If it's > 0.8, it routes normally. Use StateGraph.

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated, Literal
import operator
from pydantic import BaseModel, Field
from langgraph.graph import StateGraph, START, END
from langchain.chat_models import init_chat_model
from langchain_core.messages import AnyMessage, HumanMessage, AIMessage, SystemMessage

class SmartClassification(BaseModel):
    intent: Literal["code", "writing", "research", "unclear"] = Field(
        description="The user's intent"
    )
    confidence: float = Field(description="Confidence from 0.0 to 1.0")

class SmartRouterState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
    intent: str
    confidence: float
    confidence_tier: str

classifier = init_chat_model("openai:gpt-4.1-mini")

def smart_classifier(state: SmartRouterState) -> dict:
    structured = classifier.with_structured_output(SmartClassification)
    result = structured.invoke([
        SystemMessage(content=(
            "Classify the request. If it's ambiguous, use 'unclear' with low confidence. "
            "Be honest about your confidence level."
        )),
        state["messages"][-1],
    ])
    if result.confidence < 0.6:
        tier = "low"
    elif result.confidence < 0.8:
        tier = "medium"
    else:
        tier = "high"
    return {
        "intent": result.intent,
        "confidence": result.confidence,
        "confidence_tier": tier,
    }

def route_by_confidence(state: SmartRouterState) -> str:
    if state["confidence_tier"] == "low":
        return "clarification"
    return state["intent"]

def clarification_node(state: SmartRouterState) -> dict:
    return {"messages": [AIMessage(content=(
        "I'm not sure I understand your request. Could you be more specific? "
        "Do you need help with code, writing, or research?"
    ))]}

def make_agent(prompt: str, add_disclaimer: bool = False):
    def agent(state: SmartRouterState) -> dict:
        model = init_chat_model("openai:gpt-4.1-mini")
        response = model.invoke(
            [SystemMessage(content=prompt)] + state["messages"]
        )
        content = response.content
        if add_disclaimer and state["confidence_tier"] == "medium":
            content = (
                "⚠️ *Note: I read your request as a "
                f"{state['intent']} question. If you need something else, just tell me.*\n\n"
                + content
            )
        return {"messages": [AIMessage(content=content)]}
    return agent

graph = StateGraph(SmartRouterState)
graph.add_node("classify", smart_classifier)
graph.add_node("clarification", clarification_node)
graph.add_node("code", make_agent("Code expert. Respond in English.", add_disclaimer=True))
graph.add_node("writing", make_agent("Professional writer. Respond in English.", add_disclaimer=True))
graph.add_node("research", make_agent("Researcher. Respond in English.", add_disclaimer=True))
graph.add_node("unclear", clarification_node)

graph.add_edge(START, "classify")
graph.add_conditional_edges("classify", route_by_confidence, {
    "code": "code", "writing": "writing",
    "research": "research", "unclear": "unclear",
    "clarification": "clarification",
})
for node in ["code", "writing", "research", "unclear", "clarification"]:
    graph.add_edge(node, END)

app = graph.compile()

for msg in ["Optimize this loop in Python", "hmm", "something about data and creative writing"]:
    result = app.invoke({
        "messages": [HumanMessage(content=msg)],
        "intent": "", "confidence": 0.0, "confidence_tier": "",
    })
    print(f"'{msg}' → intent={result['intent']}, "
          f"confidence={result['confidence']:.2f}, tier={result['confidence_tier']}")
    print(f"  Response: {result['messages'][-1].content[:80]}...")
    print()
# Output (example):
# 'Optimize this loop in Python' → intent=code, confidence=0.95, tier=high
#   Response: To optimize a loop in Python, there are several strategies you can...
#
# 'hmm' → intent=unclear, confidence=0.15, tier=low
#   Response: I'm not sure I understand your request. Could you be more specific?...
#
# 'something about data and creative writing' → intent=writing, confidence=0.65, tier=medium
#   Response: ⚠️ *Note: I read your request as a writing question. If you need...

Exercise 6: Benchmark deterministic vs LLM (Advanced)

Build a test suite with 10 sample messages. Implement a deterministic router and an LLM one for the same 3 categories. Compare accuracy (against manual labels) and report how many the deterministic one gets wrong but the LLM gets right.

from dotenv import load_dotenv
load_dotenv()

from typing import Literal
from pydantic import BaseModel, Field
from langchain.chat_models import init_chat_model

test_cases = [
    ("How much does the enterprise plan cost?", "billing"),
    ("My app crashes on startup", "technical"),
    ("Do you have offices in Mexico?", "general"),
    ("I was charged twice this month", "billing"),
    ("The login button doesn't respond", "technical"),
    ("How do I cancel my subscription?", "billing"),
    ("The page loads very slowly", "technical"),
    ("Can I talk to a human?", "general"),
    ("I need a refund", "billing"),
    ("Which programming languages do you support?", "general"),
]

def deterministic_classify(text: str) -> str:
    text_lower = text.lower()
    if any(kw in text_lower for kw in ["invoice", "billing", "payment", "price", "plan", "cost"]):
        return "billing"
    elif any(kw in text_lower for kw in ["error", "not working", "crashes", "slow", "doesn't respond"]):
        return "technical"
    return "general"

class DeptClass(BaseModel):
    department: Literal["billing", "technical", "general"] = Field(
        description="billing=payments/costs/subscriptions, technical=errors/performance, general=anything else"
    )

classifier = init_chat_model("openai:gpt-4.1-mini")
structured = classifier.with_structured_output(DeptClass)

def llm_classify(text: str) -> str:
    result = structured.invoke(f"Classify this support request: {text}")
    return result.department

det_correct = 0
llm_correct = 0
llm_wins = []

for text, label in test_cases:
    det_result = deterministic_classify(text)
    llm_result = llm_classify(text)

    det_ok = det_result == label
    llm_ok = llm_result == label
    det_correct += det_ok
    llm_correct += llm_ok

    if llm_ok and not det_ok:
        llm_wins.append((text, label, det_result))

    status = "✅" if det_ok else "❌"
    status_llm = "✅" if llm_ok else "❌"
    print(f"  Det {status} LLM {status_llm} | '{text[:40]}' → det={det_result}, llm={llm_result}, label={label}")

print(f"\nDeterministic: {det_correct}/{len(test_cases)} ({det_correct/len(test_cases)*100:.0f}%)")
print(f"LLM:           {llm_correct}/{len(test_cases)} ({llm_correct/len(test_cases)*100:.0f}%)")
print(f"\nCases where the LLM was right and the deterministic one failed ({len(llm_wins)}):")
for text, label, det_result in llm_wins:
    print(f"  '{text}' → label={label}, det said={det_result}")
# Output (example):
# Deterministic: 7/10 (70%)
# LLM:           10/10 (100%)
#
# Cases where the LLM was right and the deterministic one failed (3):
#   'I was charged twice this month' → label=billing, det said=general
#   'How do I cancel my subscription?' → label=billing, det said=general
#   'I need a refund' → label=billing, det said=general

Summary

In this capsule you learned:

• A router makes ONE decision ("who handles this?") and delegates — unlike a supervisor, which coordinates multiple steps
• Deterministic routers (keywords, regex, metadata) are fast, cheap, and predictable — always start here
• LLM routers with with_structured_output handle ambiguous inputs that rules don't cover
• You implemented routers with StateGraph (nodes + conditional edges) and with the Functional API (if/elif/else + dict lookup)
• Multi-level routing breaks complex decisions into hierarchical steps (domain → subtask)
• Fallback with confidence keeps the router from guessing when it shouldn't — it asks for clarification instead of getting it wrong
• The natural progression is: deterministic → deterministic + LLM fallback → full LLM

Next capsule: Shared vs Isolated State — the hardest design decision in multi-agent: what each agent can see and modify.

Further resources

1. Multi-Agent Architectures — LangGraph Docs — Official documentation of multi-agent patterns, routers included
2. How to build a multi-agent network — Practical tutorial on routing between agents
3. Structured Output — LangChain Docs — Reference for with_structured_output for classification
4. Conditional Edges — LangGraph Docs — Documentation for add_conditional_edges
5. How to create a router — Official guide to creating routers in LangGraph
6. Pydantic Models — Docs — Pydantic reference for structured output schemas

Module 10 — LangChain & LangGraph: From Chains to Agents