条件路由与循环:让图能"思考"

条件边:根据 state 决定下一步去哪

最小例子:意图分流

from typing import TypedDict, Literal
from langgraph.graph import StateGraph, START, END

class State(TypedDict):
    query: str
    intent: str
    answer: str

def classify(s: State):
    # 实际会调 LLM,这里简化
    q = s["query"]
    if "退款" in q: return {"intent": "refund"}
    if "物流" in q: return {"intent": "shipping"}
    return {"intent": "other"}

def refund_node(s):   return {"answer": "退款流程..."}
def shipping_node(s): return {"answer": "物流查询..."}
def other_node(s):    return {"answer": "已转人工"}

def route(s) -> Literal["refund", "shipping", "other"]:
    return s["intent"]    # 返回"下一个节点名"

g = StateGraph(State)
g.add_node("classify", classify)
g.add_node("refund", refund_node)
g.add_node("shipping", shipping_node)
g.add_node("other", other_node)

g.add_edge(START, "classify")
g.add_conditional_edges("classify", route)   # 分流
for n in ["refund", "shipping", "other"]:
    g.add_edge(n, END)

app = g.compile()

显式映射:路由函数返回"标签",map 决定目的地

g.add_conditional_edges(
    "classify",
    route,
    {
        "refund":   "refund_flow",     # 标签 → 节点名
        "shipping": "shipping_flow",
        "other":    END,                 # 可以直接连 END
    }
)

循环:ReAct 的灵魂

循环 = 一条边指回之前的节点。条件边天生支持。

class State(TypedDict):
    messages: Annotated[list, add_messages]
    n_steps: int

def think(s):
    rsp = llm.invoke(s["messages"], tools=TOOLS)
    return {"messages": [rsp], "n_steps": s.get("n_steps", 0) + 1}

def act(s):
    last = s["messages"][-1]
    outs = [run_tool(c) for c in last.tool_calls]
    return {"messages": outs}

def should_continue(s) -> Literal["act", END]:
    last = s["messages"][-1]
    if getattr(last, "tool_calls", None):
        return "act"
    return END

g = StateGraph(State)
g.add_node("think", think)
g.add_node("act", act)
g.add_edge(START, "think")
g.add_conditional_edges("think", should_continue)
g.add_edge("act", "think")   # 回到 think → 循环
app = g.compile()

图示

循环终止:3 种保险

别让 Agent 跑飞,必须设边界:

① 业务终止:靠 LLM 自己决定

上面例子里,should_continue 返回 END 就停。这是"正常终止"。

② 全局 recursion_limit:防死循环

app.invoke(
    {"messages": [...]},
    config={"recursion_limit": 25},   # 默认 25
)
# 超过会抛 GraphRecursionError

③ state 里自己计步数

def should_continue(s):
    if s.get("n_steps", 0) >= 10:
        return END   # 硬停
    if not s["messages"][-1].tool_calls:
        return END
    return "act"

Command:节点里直接指定下一步

除了在 edge 上决定,节点内部也能返回 Command 同时更新 state + 跳转:

from langgraph.graph import StateGraph, START, END
from langgraph.types import Command
from typing import Literal

def classify_and_route(s) -> Command[Literal["refund", "shipping", END]]:
    intent = detect_intent(s["query"])
    return Command(
        update={"intent": intent},       # 改 state
        goto="refund" if intent == "refund" else END,  # 跳下一个
    )

g.add_node("classify", classify_and_route)
g.add_edge(START, "classify")
# 注意: 用了 Command 就不需要 add_conditional_edges

Command vs conditional_edges:什么时候用哪个?

场景	推荐	原因
纯路由,无 state 变化	conditional_edges	路由逻辑独立,易复用
节点产出结果 + 需要跳多处	Command	少一次 state 读取,逻辑紧凑
Multi-Agent handoff	Command	"我处理完了,交给 researcher"一句话
子图之间跳转	Command(graph="parent")	能跨越子图边界

并行分支:fan-out 与 Send API

静态并行:简单 fan-out

g.add_edge("init", "search_web")
g.add_edge("init", "search_db")
# 两者并行,都跑完 → merge
g.add_edge("search_web", "merge")
g.add_edge("search_db", "merge")

动态并行:Send 启动 N 个实例

比如"把文档切 10 段,每段并行喂给 summarizer":

from langgraph.types import Send

def dispatch(s):
    # 返回一个 Send 列表,每个 Send 独立跑一次 summarize
    return [Send("summarize", {"chunk": c}) for c in s["chunks"]]

g.add_conditional_edges("split", dispatch, ["summarize"])
g.add_edge("summarize", "reduce")

Send 是 LangGraph 的 map-reduce 底层原语,见 Ch8 多 agent 再展开。

路由函数的 7 个实战模式

1. LLM 判断下一步:把 state 喂给小模型,让它输出"下一个节点标签"
2. 有工具调用就循环,否则结束:最经典 ReAct 路由
3. 检索结果不足就追问:召回 < 3 条 → 回到 rewrite_query
4. 置信度分流:high → 直接回复,low → 升级人工
5. 预算用完强停:total_cost > 预算 → END
6. 按角色 handoff:当前 agent 说"我搞不定" → 转 supervisor
7. 失败重试:节点报错 → 回到自己,但用 n_retries 控次数

常见坑

坑	症状	解法
路由函数返回了不存在的节点名	KeyError / 报 "no edge"	Literal 类型标注 + 明确 map
条件边 + 静态边并存	同时跳两处,意外并行	一个节点只用一种出边
循环无终止	GraphRecursionError	业务终止 + 步数上限双保险
Command 里 goto 了未注册节点	运行时报错	Literal 提示 + 节点先 add_node
fan-out 后 state 冲突	两个并行节点都改 count	列表字段用 reducer;标量字段拆开

本章小结