RSS

第 7 章 / 共 14 章

控制流:条件边、Command、Send 与递归上限

约 7 分钟 更新于

7.1 四种写法,各管一件事

到目前为止我们的图都是直线。现在把财务那条规则实现出来:查完订单,金额超 500 走人工路径,否则走自动路径。

LangGraph 给了四种表达控制流的方式,容易混,先用一张表定位:

写法表达什么什么时候用
add_edge(a, b)a 完了一定去 b固定顺序
add_conditional_edges(a, fn, [...])a 完了由 fn 决定去哪分支、循环出口
Command(goto=..., update=...)节点自己说「我改了这些,下一步去这里」更新和跳转是同一个决定时
Send("node", 私有输入)同一个节点并行跑 N 次扇出 / map-reduce

7.2 条件边:分支和循环用的是同一个零件

def route_by_amount(state: TicketState) -> str:
    return "manual_review" if state["amount"] > 500 else "auto_refund"

builder.add_conditional_edges("fetch_order", route_by_amount,
                              ["manual_review", "auto_refund"])

第三个参数是可能去往的节点列表。它不是必需的,但强烈建议写——它让 draw_mermaid() 能画出正确的虚线,也让图在编译期就能发现你写错了节点名。

循环用的是同一个零件,只是有一条边指回自己或者上游:

def should_continue(state) -> str:
    return "tools" if state["messages"][-1].tool_calls else "end"

builder.add_conditional_edges("model", should_continue, {"tools": "tools", "end": END})
builder.add_edge("tools", "model")     # 回边

这正是第 3 章那张 create_agent 图的真身。你现在有能力自己写它了。

第7章:控制流:条件边、Command、Send 与递归上限
图 7.1:四种控制流写法的分工。左边两种改变「去哪」,Command 同时改变「去哪」和「state 变成什么」,Send 改变「跑几份」。

7.3 Command:更新和跳转一次说完

有时候「这一步的结论」和「下一步去哪」是同一个判断的两面。分成节点函数 + 路由函数两处写,逻辑会被割裂。Command 让你一次说完:

from langgraph.types import Command

def triage(state) -> Command:
    if state["amount"] > 500:
        return Command(goto="manual_review", update={"log": ["超额,转人工"]})
    return Command(goto="auto_refund", update={"log": ["额度内,自动处理"]})

实测输出确认它两件事都做了:{'log': ['router', 'second']}。

用了 Command(goto=...) 就不需要再给这个节点加条件边。如果你希望 draw_mermaid() 也能画出这些跳转,在 add_node 时补一个 destinations=("manual_review", "auto_refund")。

7.4 Send:把一条工单拆成 N 份并行处理

工单里可能有多个商品,每个都要单独判断。数量运行时才知道,不能预先画好节点。

from langgraph.types import Send

def fan_out(state):
    return [Send("judge_item", {"item": i}) for i in state["items"]]

builder.add_conditional_edges(START, fan_out, ["judge_item"])

实测输出:

输入 {'items': ['a','b','c'], 'results': []}
输出 {'items': ['a','b','c'], 'results': ['A','B','C']}

7.5 它不肯停下来的时候

模型陷在「调工具 → 看结果 → 再调同一个工具」的循环里,是 agent 最常见的线上故障之一。LangGraph 的兜底是数步数:

graph.invoke(input, {"recursion_limit": 8})

超了之后的真实报错:

GraphRecursionError: Recursion limit of 8 reached without hitting a stop condition.
You can increase the limit by setting the `recursion_limit` config key.
For troubleshooting, visit: https://docs.langchain.com/oss/python/langgraph/errors/GRAPH_RECURSION_LIMIT
  • 想控制成本,用第 4 章的 ModelCallLimitMiddleware(数模型调用次数)。
  • 想控制时长,用节点上的 timeout(第 11 章)。
  • recursion_limit 是防死循环的最后一道保险,不是业务开关。

7.6 练习与检查点