Chapter 13 · Capstone 2: multi-agent ticket routing

Combine chapter 8 subgraphs with chapter 3 conditional routing into a supervisor plus two domain agents: tickets are routed by keyword to the billing or engineering subgraph, then return to the supervisor to finish. No model is required.

Mechanics and design choices

The supervisor does no domain work; it only writes the routing field. Each worker node invokes a compiled subgraph and returns only what it needs, then sets routing to done so the supervisor can finish. Three tickets cover both paths, and the assertions lock down the queue verdict and the subject marker.

Example

import operator
from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, START, END

class BillingState(TypedDict):
    subject: str
    finding: str

def billing_validate(state: BillingState):
    return {"finding": "billing queue: " + ("valid" if state["subject"].strip() else "missing detail")}

def billing_close(state: BillingState):
    return {"subject": state["subject"] + " [processed]"}

billing = (
    StateGraph(BillingState)
    .add_node("validate", billing_validate)
    .add_node("close", billing_close)
    .add_edge(START, "validate")
    .add_edge("validate", "close")
    .add_edge("close", END)
    .compile()
)

class EngState(TypedDict):
    subject: str
    finding: str

def eng_triage(state: EngState):
    severity = "high" if len(state["subject"].split()) > 25 else "low"
    return {"finding": "engineering queue: severity=" + severity}

def eng_assign(state: EngState):
    return {"subject": state["subject"] + " [assigned]"}

eng = (
    StateGraph(EngState)
    .add_node("triage", eng_triage)
    .add_node("assign", eng_assign)
    .add_edge(START, "triage")
    .add_edge("triage", "assign")
    .add_edge("assign", END)
    .compile()
)

class TicketState(TypedDict):
    subject: str
    routing: str
    notes: Annotated[list[str], operator.add]

def supervise(state: TicketState):
    if state["routing"] in ("billing", "engineering", "done"):
        return {}
    text = state["subject"].lower()
    routing = "billing" if ("invoice" in text or "refund" in text) else "engineering"
    return {"routing": routing}

def run_billing(state: TicketState):
    result = billing.invoke({"subject": state["subject"], "finding": ""})
    return {"subject": result["subject"], "notes": [result["finding"]], "routing": "done"}

def run_engineering(state: TicketState):
    result = eng.invoke({"subject": state["subject"], "finding": ""})
    return {"subject": result["subject"], "notes": [result["finding"]], "routing": "done"}

supervisor = (
    StateGraph(TicketState)
    .add_node("supervise", supervise)
    .add_node("run_billing", run_billing)
    .add_node("run_engineering", run_engineering)
    .add_edge(START, "supervise")
    .add_conditional_edges(
        "supervise",
        lambda s: s["routing"],
        {"billing": "run_billing", "engineering": "run_engineering", "done": END},
    )
    .add_edge("run_billing", "supervise")
    .add_edge("run_engineering", "supervise")
    .compile()
)

for subject, expected in [
    ("Refund for invoice #1042", "billing queue: valid"),
    ("Crash when saving a long file, please help", "engineering queue: severity=low"),
]:
    result = supervisor.invoke({"subject": subject, "routing": "", "notes": []})
    assert result["notes"] == [expected], result
    assert result["subject"].endswith("[processed]") or result["subject"].endswith("[assigned]")
    print(result["subject"], "|", result["notes"])
print("routing graph ok")

Run and verify

Run it directly: it should print two ticket lines and routing graph ok. The second ticket has 6 words, below the 25-word threshold, so it is judged low severity, confirming the deterministic rule rather than random behavior.

Pitfalls

Workers clear routing so the supervisor can re-decide; wiring a worker straight to END would skip the supervisor for other ticket types. With several workers, every worker must return to the supervisor. Keep subgraph internals out of the parent state, as shown in chapter 12.

🧪 随堂测验

点击你认为正确的选项。答错时会展示正确答案与原因解析。

1. What does the supervisor own?

2. Where does a worker go after it finishes?

3. How does a subgraph result reach the parent state?

4. Does this example need a model?

Hands-on exercises

1. Add a third security queue and its routing keywords.
2. Add a conditional edge in the engineering subgraph: high severity takes the urgent branch.
3. Parametrize tests just above and below the 25-word threshold.

Official references

• use-subgraphs
• workflows-agents
• graph-api

Verified on 2026-09-17; examples were run locally without network access.