1. 本章目标 #

第 23 章让图学会了分叉，但所有的边仍然是「往前走」。这一章加上最后一块拼图：往回走。

一条边指回已经跑过的节点，图就成了环。而环一旦成立，一个新问题立刻冒出来：

它什么时候停？

这不是学术问题。你从第 2 章就在用的 create_agent，本质就是一个 model ↔ tools 环：模型说要调工具，工具跑完把结果塞回去，模型再看一眼，再决定要不要接着调。这个环转几圈、怎么停，一直是 LangChain 替你管的。这一章把它拆开，自己写一遍。

学完这一章，你应该能做到：

• 写出带退出条件的循环，并说清 recursion_limit 到底数的是什么（实测过六种拓扑）
• 手写 model ↔ tools 环，理解 ToolNode 和 tools_condition 各自替你做了什么
• 说清 ToolNode 默认「接参数错误、不接业务异常」的机制，并按场景选对 handle_tool_errors
• 给 Agent 环装上三重终止保险，并在触发上限时优雅降级而不是抛异常
• 避开四个循环场景专属的坑，其中两个会让程序跑到天荒地老，或者直接 400
• 产出：一个受控 Agent 环，能干活、能自己停、停不下来也能体面收场

顺序图和分支图有一个共同的好处：它们天然会停。边是有限的，走完就到 END，最坏情况也就是走错分支。

环把这个保障拿掉了。于是「终止」从一个自动成立的性质，变成一件你必须自己负责的事。更麻烦的是，循环里的失控有三种完全不同的样子：

失控形态	症状	谁来兜底
退出条件写错	一直转，直到撞 recursion_limit 抛异常	recursion_limit（§3）
模型自己停不下来	一圈接一圈调工具，每圈都在烧钱	业务轮次上限 + 降级（§6）
护栏本身坏了	不报错、结果看着正常，只是上限从此无效	只能靠自检（§7.1）

第三种最危险，因为它没有任何征兆。写出一个环只需要一行 add_edge，所以本章的重点不在这里，而在怎么保证它一定会停，以及停下来时给用户留个体面的交代。

参考文档：

• Graph API - Loops
• Recursion limit
• ToolNode

2. 循环就是一条指回去的边 #

2.1. 最小的环 #

循环在 LangGraph 里没有专门的 API。你不需要 add_loop，只要让一条边指回前面的节点就行。最短的写法是让节点连回它自己：

# operator.add 待会儿给 log 字段当 reducer
import operator
# Annotated 挂 reducer，Literal 声明路由出口，TypedDict 定义状态结构
from typing import Annotated, Literal, TypedDict

# END 是终止哨兵，START 是入口哨兵，StateGraph 是图的构建器
from langgraph.graph import END, START, StateGraph

# 状态：一个计数器 + 一份日志
class CountState(TypedDict):
    # 没有 reducer，每次写入都是覆盖
    n: int
    # 配了 operator.add，每次写入都是追加
    log: Annotated[list[str], operator.add]

# 循环体：每转一圈把 n 加一
def tick(state: CountState) -> dict:
    """每转一圈把 n 加一。"""
    # 返回值是「增量」：n 覆盖成新值，log 追加一条
    return {"n": state["n"] + 1, "log": [f"tick -> {state['n'] + 1}"]}

# 退出条件：这是循环的灵魂，先写它再写边
def should_continue(state: CountState) -> Literal["tick", "__end__"]:
    """退出条件：n 到 3 就停，否则回到 tick。"""
    # 没到 3 就返回节点名继续转，到了就返回 END
    return "tick" if state["n"] < 3 else END

# 开始搭图，未编译的图纸统一叫 builder
builder = StateGraph(CountState)
# 只有一个节点
builder.add_node("tick", tick)
# 入口指向它
builder.add_edge(START, "tick")
# 关键就这一行：条件边的一个出口指回 tick 自己
builder.add_conditional_edges("tick", should_continue, ["tick", END])
# 编译成可执行图，编译后的对象统一叫 graph
graph = builder.compile()

# 跑一次，看最终状态
print(graph.invoke({"n": 0, "log": []}))

运行输出：

{'n': 3, 'log': ['tick -> 1', 'tick -> 2', 'tick -> 3']}

这段代码里最值得细看的是 should_continue 的返回类型 Literal["tick", "__end__"]。第 23 章 §4.4 实测过：Literal 和 add_conditional_edges 第三个参数（path_map）都能声明出口，两者任一存在，图纸就画得对、非法返回值也会当场 KeyError。这里两个都写了，属于「带包带保险」：Literal 给 IDE 和读者看，path_map 给运行时看。

注意 END 在 Literal 里必须写成字符串 "__end__"，因为 Literal 只接受字面量，不接受 END 这个变量名。这是唯一一个需要你手写魔法字符串的地方。

画出来能看得更清楚，tick -.-> tick 那条自指的虚线就是环：

import operator
from typing import Annotated, TypedDict

from langgraph.graph import END, START, StateGraph

class CountState(TypedDict):
    n: int
    log: Annotated[list[str], operator.add]

builder = StateGraph(CountState)
builder.add_node("tick", lambda s: {"n": s["n"] + 1, "log": [f"tick -> {s['n'] + 1}"]})
builder.add_edge(START, "tick")
builder.add_conditional_edges("tick", lambda s: "tick" if s["n"] < 3 else END, ["tick", END])
graph = builder.compile()

# draw_mermaid 把图渲染成 Mermaid 文本，可以直接贴到支持 Mermaid 的地方看图
print(graph.get_graph().draw_mermaid())

运行输出（省略了前后的样式定义，只留主体）：

graph TD;
    __start__([<p>__start__</p>]):::first
    tick(tick)
    __end__([<p>__end__</p>]):::last
    __start__ --> tick;
    tick -.-> __end__;
    tick -.-> tick;

（真实输出前面还有一段 config: flowchart: curve: linear 的 front matter，末尾还有三行 classDef 样式。本章后面的图纸都只保留 graph TD; 主体。）

三条边对应三件事：实线 __start__ --> tick 是入口固定边，两条虚线是条件边的两个出口。「循环」= 条件边 + 一个指回去的出口，第 23 章学的东西这里全都能用，只是路由函数的返回值里多了一个已经跑过的节点名。

2.2. 环里的状态是累积的 #

循环最容易让人迷糊的地方是：转了三圈，状态里到底攒了什么？答案完全取决于你给字段配了什么 reducer，这正是第 21 章的内容，在循环里才真正显出分量。顺序图里 reducer 配错了顶多丢条日志，循环里配错了会让护栏失效（§7.1）。

用 stream 的默认 updates 模式看每一圈的增量最直观：

import operator
from typing import Annotated, TypedDict

from langgraph.graph import END, START, StateGraph

class CountState(TypedDict):
    n: int
    log: Annotated[list[str], operator.add]

builder = StateGraph(CountState)
builder.add_node("tick", lambda s: {"n": s["n"] + 1, "log": [f"tick -> {s['n'] + 1}"]})
builder.add_edge(START, "tick")
builder.add_conditional_edges("tick", lambda s: "tick" if s["n"] < 3 else END, ["tick", END])
graph = builder.compile()

# updates 模式：每个超步产出一个 {节点名: 该节点的返回值}
for chunk in graph.stream({"n": 0, "log": []}):
    # 打印这一圈 tick 返回了什么
    print(chunk)

运行输出：

{'tick': {'n': 1, 'log': ['tick -> 1']}}
{'tick': {'n': 2, 'log': ['tick -> 2']}}
{'tick': {'n': 3, 'log': ['tick -> 3']}}

updates 看到的是「这一圈写了什么」。想看「写完之后攒成了什么」，换 values：

import operator
from typing import Annotated, TypedDict

from langgraph.graph import END, START, StateGraph

class CountState(TypedDict):
    n: int
    log: Annotated[list[str], operator.add]

builder = StateGraph(CountState)
builder.add_node("tick", lambda s: {"n": s["n"] + 1, "log": [f"tick -> {s['n'] + 1}"]})
builder.add_edge(START, "tick")
builder.add_conditional_edges("tick", lambda s: "tick" if s["n"] < 3 else END, ["tick", END])
graph = builder.compile()

# values 模式：每个超步产出一份合并后的完整状态
for chunk in graph.stream({"n": 0, "log": []}, stream_mode="values"):
    # 第一条是初始状态，后面每条是一圈之后的全量状态
    print(chunk)

运行输出：

{'n': 0, 'log': []}
{'n': 1, 'log': ['tick -> 1']}
{'n': 2, 'log': ['tick -> 1', 'tick -> 2']}
{'n': 3, 'log': ['tick -> 1', 'tick -> 2', 'tick -> 3']}

两个字段的行为完全不同：

• n 没有 reducer，每圈覆盖，从 0 → 1 → 2 → 3
• log 配了 operator.add，每圈追加，长度从 0 涨到 3

循环里的「累积」全靠 reducer，不靠循环本身。 你在 tick 里写 state["n"] + 1 是自己算的累加，log 的累加则是 reducer 替你做的。第一种写法在环里可行，是因为节点读到的一定是上一圈合并后的状态；但一旦有并行分支同时写 n 就会冲突（第 21 章的 InvalidUpdateError），所以计数类字段一律用 reducer 更稳。

顺便记一个排查技巧：debug 模式能看到每一圈的超步号，这个数字和 §3 的 recursion_limit 是同一个口径：

import operator
from typing import Annotated, TypedDict

from langgraph.graph import END, START, StateGraph

class CountState(TypedDict):
    n: int
    log: Annotated[list[str], operator.add]

builder = StateGraph(CountState)
builder.add_node("tick", lambda s: {"n": s["n"] + 1, "log": [f"tick -> {s['n'] + 1}"]})
builder.add_edge(START, "tick")
builder.add_conditional_edges("tick", lambda s: "tick" if s["n"] < 3 else END, ["tick", END])
graph = builder.compile()

# debug 模式的事件里带 step 字段，也就是超步号
for chunk in graph.stream({"n": 0, "log": []}, stream_mode="debug"):
    # 只看任务开始的事件，避免重复
    if chunk["type"] == "task":
        # step 是超步号，payload["name"] 是本超步要跑的节点
        print(f"step={chunk['step']} node={chunk['payload']['name']}")

运行输出：

step=1 node=tick
step=2 node=tick
step=3 node=tick

2.3. 两个节点也能组成环 #

自指只是最简形式。更常见的是两个（或更多）节点首尾相接：

# operator.add 给 log 字段当 reducer
import operator
# Annotated 挂 reducer，TypedDict 定义状态结构
from typing import Annotated, TypedDict

# 图的三件套
from langgraph.graph import END, START, StateGraph

# 和 CountState 结构一样，换个名字方便区分
class PingState(TypedDict):
    # 计数器，没有 reducer
    n: int
    # 日志，追加
    log: Annotated[list[str], operator.add]

# 环上的第一个节点：负责推进计数
def ping(state: PingState) -> dict:
    # n 加一，日志记一条
    return {"n": state["n"] + 1, "log": ["ping"]}

# 环上的第二个节点：只记日志，负责做决定
def pong(state: PingState) -> dict:
    # 不动 n，只追加日志
    return {"log": ["pong"]}

# 搭图
builder = StateGraph(PingState)
# 注册推进节点
builder.add_node("ping", ping)
# 注册决策节点
builder.add_node("pong", pong)
# 入口进 ping
builder.add_edge(START, "ping")
# ping 之后固定走 pong
builder.add_edge("ping", "pong")
# pong 决定：回到 ping 再转一圈，还是结束
builder.add_conditional_edges("pong", lambda s: "ping" if s["n"] < 3 else END, ["ping", END])
# 编译并跑
graph = builder.compile()

# 看最终状态
print(graph.invoke({"n": 0, "log": []}))

运行输出：

{'n': 3, 'log': ['ping', 'pong', 'ping', 'pong', 'ping', 'pong']}

日志里 ping、pong 交替出现三轮，说明环真的在转。图纸上，环由「一条实线去 + 一条虚线回」构成：

graph TD;
    __start__([<p>__start__</p>]):::first
    ping(ping)
    pong(pong)
    __end__([<p>__end__</p>]):::last
    __start__ --> ping;
    ping --> pong;
    pong -.-> __end__;
    pong -.-> ping;

有两个设计细节值得留意，它们在 §4 会原样复现：

• 判断放在环的哪一侧是你选的。 这里放在 pong，所以 ping → pong 是固定边。放在 ping 也行，只是语义变成「每次推进前先看要不要继续」。
• 计数由 ping 负责，判断由 pong 负责。 「谁改变量、谁做决定」分开，比塞在一个节点里更好审计。这个分工是被「路由函数只读不写」（第 23 章 §3.3）逼出来的。

记住这个形状。 把 ping 换成「调模型」、pong 换成「跑工具」，就是 §4 要手写的 Agent 环。

3. recursion_limit：兜底，不是设计 #

3.1. 忘了退出条件会怎样 #

把上面的条件边换成固定边，环就没有出口了。这不是刻意写出来的 bug，真实项目里更常见的形态是「路由函数漏了一个 return」或者「退出条件永远不成立」（§7.1 就是后者）：

import operator
from typing import Annotated, TypedDict

from langgraph.graph import START, StateGraph

class CountState(TypedDict):
    n: int
    log: Annotated[list[str], operator.add]

def tick(state: CountState) -> dict:
    return {"n": state["n"] + 1, "log": [f"tick -> {state['n'] + 1}"]}

# 循环没有出口时抛的就是这个异常
from langgraph.errors import GraphRecursionError

# 重新搭一张图，这次故意不给出口
builder = StateGraph(CountState)
# 还是同一个 tick 节点
builder.add_node("tick", tick)
# 入口
builder.add_edge(START, "tick")
# 无条件回到自己，永远出不去
builder.add_edge("tick", "tick")
# 注意：compile() 完全不会报错，它只查拓扑连通性，不查「能不能停」
graph = builder.compile()

# 设一个很小的 limit，免得真跑一万圈
try:
    # recursion_limit 是 config 的顶层键，不是 configurable 里的键（第 22 章 §5.2）
    graph.invoke({"n": 0, "log": []}, {"recursion_limit": 8})
# 撞上限时抛 GraphRecursionError
except GraphRecursionError as e:
    # 打印类名和完整信息
    print(type(e).__name__, "|", str(e))

运行输出：

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

两件事值得记住：

第一，compile() 不检查死循环。 第 22 章 §2.3 说过 compile() 只保证「图能跑」不保证「跑得对」，死循环就是最典型的例子：一条 tick → tick 的边在拓扑上完全合法。没有任何静态检查能救你，退出条件只能靠自己写对。

第二，异常信息在建议你「把限制调大」。 这个建议在少数情况下是对的（比如你的任务真的需要几十轮），但绝大多数时候撞上限意味着退出条件有问题。请把 recursion_limit 理解成保险丝，不是开关：保险丝烧了说明电路有问题，正确做法是修电路，而不是换个更粗的保险丝。

第 22 章提过，这个值在当前版本（LangGraph 1.2.11）默认是 10007，可以被环境变量 LANGGRAPH_DEFAULT_RECURSION_LIMIT 覆盖。默认值大到没有实际保护意义：一个失控的 Agent 环，在撞上 10007 之前能烧掉你几千次模型调用的钱。所以真正的退出条件必须自己写，这是 §6 的主题。

3.2. 它数的是超步，不是节点执行次数 #

这个细节很多人搞错，它直接决定了你该把 recursion_limit 设成多少。与其背规则，不如实测。下面这段代码对同一张图从小到大试 limit，找出最小可用值：

import operator
from typing import Annotated, TypedDict

from langgraph.errors import GraphRecursionError
from langgraph.graph import END, START, StateGraph

class CountState(TypedDict):
    n: int
    log: Annotated[list[str], operator.add]

builder = StateGraph(CountState)
builder.add_node("tick", lambda s: {"n": s["n"] + 1, "log": [f"tick -> {s['n'] + 1}"]})
builder.add_edge(START, "tick")
builder.add_conditional_edges("tick", lambda s: "tick" if s["n"] < 3 else END, ["tick", END])
graph = builder.compile()

# 对这个三圈循环，从 1 试到 4
for lim in [1, 2, 3, 4]:
    try:
        # 能跑通就打印最终的 n
        print(f"limit={lim}: OK -> {graph.invoke({'n': 0, 'log': []}, {'recursion_limit': lim})['n']}")
    # 跑不通就记一笔
    except GraphRecursionError:
        # 说明这个 limit 太小
        print(f"limit={lim}: GraphRecursionError")

运行输出：

limit=1: GraphRecursionError
limit=2: GraphRecursionError
limit=3: GraphRecursionError
limit=4: OK -> 3

跑 3 圈需要 limit=4。换 §2.3 的两节点环（ping → pong 各跑 3 次）再测一遍：

import operator
from typing import Annotated, TypedDict

from langgraph.errors import GraphRecursionError
from langgraph.graph import END, START, StateGraph

class PingState(TypedDict):
    n: int
    log: Annotated[list[str], operator.add]

builder = StateGraph(PingState)
builder.add_node("ping", lambda s: {"n": s["n"] + 1, "log": ["ping"]})
builder.add_node("pong", lambda s: {"log": ["pong"]})
builder.add_edge(START, "ping")
builder.add_edge("ping", "pong")
builder.add_conditional_edges("pong", lambda s: "ping" if s["n"] < 3 else END, ["ping", END])
graph = builder.compile()

# ping/pong 交替，一共 6 次节点执行
for lim in [4, 5, 6, 7]:
    try:
        # 能跑通就打印最终的 n
        print(f"limit={lim}: OK -> {graph.invoke({'n': 0, 'log': []}, {'recursion_limit': lim})['n']}")
    # 这个 limit 太小
    except GraphRecursionError:
        # 记一笔继续试下一个
        print(f"limit={lim}: GraphRecursionError")

运行输出：

limit=4: GraphRecursionError
limit=5: GraphRecursionError
limit=6: GraphRecursionError
limit=7: OK -> 3

把六种拓扑都测一遍，规律就清楚了：

图的形状	节点执行次数	超步数	最小可用 limit
单节点 → END	1	1	2
两节点串行	2	2	3
三节点串行	3	3	4
一个节点扇出到三个节点	4	2	3
tick 自指转 3 圈	3	3	4
ping/pong 各跑 3 次	6	6	7

两条结论：

1. 最小可用 limit = 超步数 + 1。 多出来的那 1 步是图确认「没有下一个节点了」所需的收尾。可以把 limit 理解成执行引擎的「迭代预算」：跑 N 个超步花掉 N 份预算，最后那次「确认没有下一个节点」还要再花 1 份。
2. 它数超步，不数节点。 看加粗那一行：4 次节点执行，但因为三个节点在同一个超步里并行（第 20 章），只需要 limit=3。按节点数估算会算多，在有扇出的图里会多算不少。

反过来说，死循环的图会实实在在跑掉 limit 个超步才报错。这就是为什么默认值 10007 危险：一个 model ↔ tools 环撞上默认上限，意味着五千次模型调用已经发生了。下一节的现场快照正好能印证这一点。

想知道自己的图有几个超步，不用推公式，stream 数一遍最准：

import operator
from typing import Annotated, TypedDict

from langgraph.graph import END, START, StateGraph

class CountState(TypedDict):
    n: int
    log: Annotated[list[str], operator.add]

builder = StateGraph(CountState)
builder.add_node("tick", lambda s: {"n": s["n"] + 1, "log": [f"tick -> {s['n'] + 1}"]})
builder.add_edge(START, "tick")
builder.add_conditional_edges("tick", lambda s: "tick" if s["n"] < 3 else END, ["tick", END])
graph = builder.compile()

# updates 模式每个超步产出一条，长度就是超步数
steps = [list(chunk.keys()) for chunk in graph.stream({"n": 0, "log": []})]
# 打印每个超步跑了哪些节点，以及总数
print(steps, len(steps))

运行输出：

[['tick'], ['tick'], ['tick']] 3

实用建议： 给带环的图显式设一个小的 recursion_limit。经验公式是 2 × 最大轮次 + 5，其中 model ↔ tools 一轮占两个超步，+5 留给入口、收尾和降级节点。它不负责正常的业务终止，只负责在你的退出条件写错时尽快报错，而不是烧一晚上的 token。§8 的实战里把这个公式写成了 limit_for() 函数。

3.3. 撞上限时，现场就没了 #

GraphRecursionError 是抛出来的，不是返回的。这意味着 invoke 没有返回值，前面几圈攒下的中间结果全部拿不到：

import operator
from typing import Annotated, TypedDict

from langgraph.errors import GraphRecursionError
from langgraph.graph import START, StateGraph

class CountState(TypedDict):
    n: int
    log: Annotated[list[str], operator.add]

builder = StateGraph(CountState)
builder.add_node("tick", lambda s: {"n": s["n"] + 1, "log": [f"tick -> {s['n'] + 1}"]})
builder.add_edge(START, "tick")
builder.add_edge("tick", "tick")
graph = builder.compile()

# 用这张没有出口的图，limit 设 5
try:
    # 前 5 圈其实都成功跑完了
    graph.invoke({"n": 0, "log": []}, {"recursion_limit": 5})
# 但异常一抛，返回值就没了
except GraphRecursionError:
    # 想拿中间结果，没有任何办法
    print("没有返回值，中间结果全丢")

运行输出：

没有返回值，中间结果全丢

除非你挂了 checkpointer。第 22 章讲过，节点异常时状态会存在失败之前的位置，recursion_limit 同理：

import operator
from typing import Annotated, TypedDict

from langgraph.errors import GraphRecursionError
from langgraph.graph import START, StateGraph

class CountState(TypedDict):
    n: int
    log: Annotated[list[str], operator.add]

builder = StateGraph(CountState)
builder.add_node("tick", lambda s: {"n": s["n"] + 1, "log": [f"tick -> {s['n'] + 1}"]})
builder.add_edge(START, "tick")
builder.add_edge("tick", "tick")

# 内存版持久化，够演示用；跨进程要用 SqliteSaver / PostgresSaver
from langgraph.checkpoint.memory import InMemorySaver

# 同一份图纸，这次编译时挂上 checkpointer
graph = builder.compile(checkpointer=InMemorySaver())
# 有 checkpointer 就必须给 thread_id（第 22 章 §5.1）
cfg = {"configurable": {"thread_id": "loop-1"}}
try:
    # 注意 recursion_limit 和 configurable 是同级的，别塞进 configurable 里
    graph.invoke({"n": 0, "log": []}, {**cfg, "recursion_limit": 5})
# 照样撞上限
except GraphRecursionError:
    # 这次不打印，直接去看现场
    pass

# 从 checkpointer 里取出最后一次快照
snap = graph.get_state(cfg)
# 撞限前攒下的状态全在
print("values:", snap.values)
# next 指着「本来要跑的下一个节点」
print("next:", snap.next)
# 顺带看看历史攒了几条
print("历史条数:", len(list(graph.get_state_history(cfg))))

运行输出：

values: {'n': 5, 'log': ['tick -> 1', 'tick -> 2', 'tick -> 3', 'tick -> 4', 'tick -> 5']}
next: ('tick',)
历史条数: 7

五圈的结果都在，next 还指着 tick。注意 n 正好是 5，和 recursion_limit=5 对得上，上一节说的「死循环会跑掉 limit 个超步」，这就是证据：钱是真花了，异常只是事后通知你。

这份快照在排查 Agent 为什么停不下来时极其有用，你能看到它在第几圈开始打转、每圈都调了什么工具（§7.3 有真实案例）。

还有一个附带好处：改对退出条件之后，可以从断点接着跑，不用从头开始。

import operator
from typing import Annotated, TypedDict

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.errors import GraphRecursionError
from langgraph.graph import END, START, StateGraph

class CountState(TypedDict):
    n: int
    log: Annotated[list[str], operator.add]

builder = StateGraph(CountState)
builder.add_node("tick", lambda s: {"n": s["n"] + 1, "log": [f"tick -> {s['n'] + 1}"]})
builder.add_edge(START, "tick")
builder.add_conditional_edges("tick", lambda s: "tick" if s["n"] < 3 else END, ["tick", END])

# 这张图有正常出口，编译时挂上 checkpointer
graph = builder.compile(checkpointer=InMemorySaver())
# 新的会话 id
cfg = {"configurable": {"thread_id": "loop-2"}}
try:
    # 先用一个过小的 limit 故意撞一次
    graph.invoke({"n": 0, "log": []}, {**cfg, "recursion_limit": 2})
except GraphRecursionError:
    # 此时 checkpoint 里已经存了跑到一半的状态
    print("先用 limit=2 撞一次")

# 传 None 表示「不给新输入，从上次断点继续」（第 22 章 §7.3）
print("再用 limit=10 续跑:", graph.invoke(None, {**cfg, "recursion_limit": 10}))

运行输出：

先用 limit=2 撞一次
再用 limit=10 续跑: {'n': 3, 'log': ['tick -> 1', 'tick -> 2', 'tick -> 3']}

注意 log 里只有三条、没有重复：撞限那一下没有产生额外写入，续跑是从断点接着走的，不是重跑。

给任何带环的图挂 checkpointer，这是本章最省事的一条建议：排查成本从「只有一行异常」降到「一眼看穿」，还顺便获得了续跑能力。

4. 手写 model ↔ tools 环 #

现在把 §2.3 的 ping/pong 换成真货。这一节是本章的骨架，也是回答「create_agent 到底是什么」的地方。

4.1. 环的形状 #

Agent 环只有两个节点和一条判断：

START → model → 模型提了 tool_calls 吗？
                 ├─ 提了  → tools → 回到 model
                 └─ 没提  → END

「没提工具需求」就是自然退出条件。模型觉得信息够了、可以直接回答用户了，它就不会再返回 tool_calls，环自然解开。

对照 §2.3 的形状：model 相当于 ping（推进状态），tools 相当于 pong（干活），只是判断挪到了 model 这一侧，因为「要不要继续」这件事只有看了模型最新的回复才知道。

这个环有个容易被忽略的特点：转几圈完全由模型决定，你无法预先知道。 顺序图和分支图的最坏执行时间是可以静态算出来的，Agent 环不行。这就是 §6 要装护栏的根本原因。

4.2. 完整实现 #

# Literal 用来声明路由函数的出口
from typing import Literal

# 从 .env 读 API Key
from dotenv import load_dotenv
# 统一的模型入口
from langchain.chat_models import init_chat_model
# 用来判断最后一条消息是不是「模型提了工具需求」
from langchain_core.messages import AIMessage
# @tool 把普通函数变成模型可调用的工具
from langchain_core.tools import tool
# MessagesState 是自带 add_messages reducer 的预制状态
from langgraph.graph import END, START, MessagesState, StateGraph
# ToolNode 是执行 tool_calls 的预制节点
from langgraph.prebuilt import ToolNode

# 加载环境变量，override=True 让 .env 覆盖同名系统变量
load_dotenv(override=True)

# 工具一：查库存
@tool
def get_stock(sku: str) -> str:
    """查询某个 SKU 的库存数量。"""
    # 打印一行，方便在输出里确认工具真被调用了
    print(f"   [tool] get_stock({sku})")
    # 用字典模拟数据库，查不到就如实说
    return {"A-100": "库存 12 件", "B-200": "库存 0 件"}.get(sku, "查无此 SKU")

# 工具二：查价格
@tool
def get_price(sku: str) -> str:
    """查询某个 SKU 的单价（元）。"""
    # 同样打印调用痕迹
    print(f"   [tool] get_price({sku})")
    # 模拟价目表
    return {"A-100": "单价 39 元", "B-200": "单价 58 元"}.get(sku, "查无此 SKU")

# 工具清单，绑模型和建 ToolNode 都要用
TOOLS = [get_stock, get_price]
# bind_tools 把工具的名字和参数结构告诉模型，模型才可能返回 tool_calls
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0).bind_tools(TOOLS)

# 节点一：调模型
def call_model(state: MessagesState) -> dict:
    """节点一：把完整消息历史丢给模型，把模型回复追加进状态。"""
    # 只返回新增的那一条，历史由 add_messages reducer 负责累积
    return {"messages": [model.invoke(state["messages"])]}

# 路由：环的退出条件
def should_continue(state: MessagesState) -> Literal["tools", "__end__"]:
    """路由：最后一条 AIMessage 带 tool_calls 就去跑工具，否则收工。"""
    # 取最新一条消息，也就是模型刚说的话
    last = state["messages"][-1]
    # 是 AIMessage 且带 tool_calls，说明模型要用工具
    if isinstance(last, AIMessage) and last.tool_calls:
        # 去工具节点
        return "tools"
    # 否则模型已经能回答了，收工
    return END

# 搭图
builder = StateGraph(MessagesState)
# 注册调模型节点
builder.add_node("model", call_model)
# 注册工具节点，ToolNode 直接当节点用
builder.add_node("tools", ToolNode(TOOLS))
# 入口进 model
builder.add_edge(START, "model")
# model 之后按 should_continue 分流，两个出口都显式声明
builder.add_conditional_edges("model", should_continue, ["tools", END])
# 就是这条边构成了环：工具跑完永远回到模型
builder.add_edge("tools", "model")
# 编译
graph = builder.compile()

# 问一个需要两个工具才能答完的问题
out = graph.invoke({"messages": [{"role": "user", "content": "A-100 还有货吗？多少钱？"}]})
# 逐条打印消息轨迹
for m in out["messages"]:
    # 用类名区分消息类型
    kind = type(m).__name__
    # 带 tool_calls 的 AIMessage 只打印调了哪些工具
    if kind == "AIMessage" and m.tool_calls:
        # 打印工具名列表
        print(f"{kind}: 调用 {[c['name'] for c in m.tool_calls]}")
    else:
        # 其它消息打印截断后的正文
        print(f"{kind}: {str(m.content)[:70]}")

运行输出（模型的措辞每次会有出入）：

   [tool] get_stock(A-100)
   [tool] get_price(A-100)
HumanMessage: A-100 还有货吗？多少钱？
AIMessage: 调用 ['get_stock', 'get_price']
ToolMessage: 库存 12 件
ToolMessage: 单价 39 元
AIMessage: A-100 还有货，库存 **12 件**，单价 **39 元**。

这就是 create_agent 的核心。 三十行代码，你已经复刻了前面十几章一直在用的东西。

有几点值得注意：

• 用的是 MessagesState（第 21 章），它自带 add_messages reducer，所以每个节点只返回新增的消息，历史自动累积。这是环能正常工作的前提，模型每圈都能看到之前所有的工具结果。把 MessagesState 换成一个没有 reducer 的 messages: list，环立刻废掉：每圈都覆盖历史，模型永远看不到工具结果，只会一直重复要求调工具。
• 模型这一圈同时提了两个 tool_calls，ToolNode 一次全跑完，所以只转了一圈就拿到全部信息。用 stream 数一遍就是三个超步：

from typing import Literal

from dotenv import load_dotenv
from langchain.chat_models import init_chat_model
from langchain_core.messages import AIMessage
from langchain_core.tools import tool
from langgraph.graph import END, START, MessagesState, StateGraph
from langgraph.prebuilt import ToolNode

load_dotenv(override=True)

@tool
def get_stock(sku: str) -> str:
    """查询某个 SKU 的库存数量。"""
    print(f"   [tool] get_stock({sku})")
    return {"A-100": "库存 12 件", "B-200": "库存 0 件"}.get(sku, "查无此 SKU")

@tool
def get_price(sku: str) -> str:
    """查询某个 SKU 的单价（元）。"""
    print(f"   [tool] get_price({sku})")
    return {"A-100": "单价 39 元", "B-200": "单价 58 元"}.get(sku, "查无此 SKU")

TOOLS = [get_stock, get_price]
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0).bind_tools(TOOLS)

def call_model(state: MessagesState) -> dict:
    return {"messages": [model.invoke(state["messages"])]}

def should_continue(state: MessagesState) -> Literal["tools", "__end__"]:
    last = state["messages"][-1]
    return "tools" if isinstance(last, AIMessage) and last.tool_calls else END

builder = StateGraph(MessagesState)
builder.add_node("model", call_model)
builder.add_node("tools", ToolNode(TOOLS))
builder.add_edge(START, "model")
builder.add_conditional_edges("model", should_continue, ["tools", END])
builder.add_edge("tools", "model")
graph = builder.compile()

# 数一下这次运行经过了哪些超步
for chunk in graph.stream({"messages": [{"role": "user", "content": "A-100 还有货吗？多少钱？"}]}, stream_mode="updates"):
    # chunk 的键是节点名
    for node, upd in chunk.items():
        # 取这个节点新增的最后一条消息
        msg = upd["messages"][-1]
        # 带 tool_calls 就打印工具名，否则打印正文
        desc = f"调用 {[c['name'] for c in msg.tool_calls]}" if getattr(msg, "tool_calls", None) else str(msg.content)[:40]
        # 打印节点名和它干了什么
        print(f"[{node}] {type(msg).__name__}: {desc}")

运行输出：

[model] AIMessage: 调用 ['get_stock', 'get_price']
   [tool] get_stock(A-100)
   [tool] get_price(A-100)
[tools] ToolMessage: 单价 39 元
[model] AIMessage: A-100 目前还有货，库存 12 件，单价 39 元。

（工具自己的 print 会插在 [tools] 那行之前，因为它们是在节点执行期间打印的。[tools] 那行只显示了最后一条 ToolMessage。）

• tools → model 是固定边：工具跑完永远回到模型。判断在 model 这一侧做，不在 tools 这一侧。

4.3. 不需要工具时，环转零圈 #

同一张图问一句闲聊：

from typing import Literal

from dotenv import load_dotenv
from langchain.chat_models import init_chat_model
from langchain_core.messages import AIMessage
from langchain_core.tools import tool
from langgraph.graph import END, START, MessagesState, StateGraph
from langgraph.prebuilt import ToolNode

load_dotenv(override=True)

@tool
def get_stock(sku: str) -> str:
    """查询某个 SKU 的库存数量。"""
    return {"A-100": "库存 12 件", "B-200": "库存 0 件"}.get(sku, "查无此 SKU")

@tool
def get_price(sku: str) -> str:
    """查询某个 SKU 的单价（元）。"""
    return {"A-100": "单价 39 元", "B-200": "单价 58 元"}.get(sku, "查无此 SKU")

TOOLS = [get_stock, get_price]
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0).bind_tools(TOOLS)

def call_model(state: MessagesState) -> dict:
    return {"messages": [model.invoke(state["messages"])]}

def should_continue(state: MessagesState) -> Literal["tools", "__end__"]:
    last = state["messages"][-1]
    return "tools" if isinstance(last, AIMessage) and last.tool_calls else END

builder = StateGraph(MessagesState)
builder.add_node("model", call_model)
builder.add_node("tools", ToolNode(TOOLS))
builder.add_edge(START, "model")
builder.add_conditional_edges("model", should_continue, ["tools", END])
builder.add_edge("tools", "model")
graph = builder.compile()

# 一句用不着工具的话
out = graph.invoke({"messages": [{"role": "user", "content": "你好"}]})
# 看消息条数
print("消息条数:", len(out["messages"]))
# 看消息类型
print("类型:", [type(m).__name__ for m in out["messages"]])

运行输出：

消息条数: 2
类型: ['HumanMessage', 'AIMessage']

只有 HumanMessage 和 AIMessage，tools 节点没执行。模型没提工具需求，should_continue 第一次被调用就返回了 END。

环不是「至少转一圈」，是「按需转」。 这一点很重要：同一张图既能处理「需要查三轮资料」的问题，也能处理「你好」，不需要为简单问题另开一条链路。第 19 章说的「用图统一表达控制流」，好处就在这里。

4.4. 和 create_agent 对比一眼 #

from dotenv import load_dotenv
from langchain_core.tools import tool

load_dotenv(override=True)

@tool
def get_stock(sku: str) -> str:
    """查询某个 SKU 的库存数量。"""
    return {"A-100": "库存 12 件", "B-200": "库存 0 件"}.get(sku, "查无此 SKU")

@tool
def get_price(sku: str) -> str:
    """查询某个 SKU 的单价（元）。"""
    return {"A-100": "单价 39 元", "B-200": "单价 58 元"}.get(sku, "查无此 SKU")

TOOLS = [get_stock, get_price]

# 内置的预制 Agent
from langchain.agents import create_agent

# 一行就搭出和上面等价的东西，返回的同样是 CompiledStateGraph，所以也叫 graph
graph = create_agent(model="deepseek:deepseek-v4-flash", tools=TOOLS)
# 把它的图纸打出来，和我们手写的对比
print(graph.get_graph().draw_mermaid())

运行输出（只保留主体）：

graph TD;
    __start__([<p>__start__</p>]):::first
    model(model)
    tools(tools)
    __end__([<p>__end__</p>]):::last
    __start__ --> model;
    model -.-> __end__;
    model -.-> tools;
    tools -.-> model;

结构和我们手写的几乎一样，节点名都一致（model、tools）。把边逐条列出来，唯一的区别就现形了：

from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain_core.tools import tool

load_dotenv(override=True)

@tool
def get_stock(sku: str) -> str:
    """查询某个 SKU 的库存数量。"""
    return {"A-100": "库存 12 件", "B-200": "库存 0 件"}.get(sku, "查无此 SKU")

@tool
def get_price(sku: str) -> str:
    """查询某个 SKU 的单价（元）。"""
    return {"A-100": "单价 39 元", "B-200": "单价 58 元"}.get(sku, "查无此 SKU")

TOOLS = [get_stock, get_price]
graph = create_agent(model="deepseek:deepseek-v4-flash", tools=TOOLS)

# 逐条打印边，看哪些是条件边
for e in graph.get_graph().edges:
    # conditional=True 表示这是条件边（图上画虚线）
    print(f"{e.source} -> {e.target} conditional={e.conditional}")

运行输出：

__start__ -> model conditional=False
model -> __end__ conditional=True
model -> tools conditional=True
tools -> model conditional=True

tools -> model 在官方版里是条件边（虚线），我们写的则是固定边（实线）。官方版在 tools 这一侧也留了判断口子，用来支持「工具返回后直接结束」的场景，比如结构化输出、Command(goto=END) 这类提前收尾。功能上，对我们这个例子来说两者等价。

所以第 19 章那句话在这里落地了：create_agent 就是一个预制的 LangGraph。 你现在有能力自己搭一个，也就有能力在它不够用的时候改造它。§8 的实战就是在这个骨架上加了两个出口节点。

5. ToolNode 与 tools_condition #

上一节用了两个预制件，值得单独拆开讲，尤其是 ToolNode 的错误处理，它的行为和大多数人的预期不一样。这一节全部不调模型：tool_calls 可以手写，ToolNode 照样能跑。

5.1. tools_condition：把判断写好了 #

should_continue 那段逻辑太常见，LangGraph 直接提供了现成的。§4.2 里自己写的那个函数可以整段删掉，换成一行：

之前： builder.add_conditional_edges("model", should_continue, ["tools", END])
之后： builder.add_conditional_edges("model", tools_condition, {"tools": "tools", END: END})

它从 langgraph.prebuilt 导入：

# 预制的路由函数，签名和我们手写的 should_continue 一致
from langgraph.prebuilt import tools_condition

它的核心逻辑就是我们手写那几行的通用版。下面这个函数是把源码去掉文档字符串之后的等价实现，可以直接跑：

# 等价于 tools_condition 的实现（源码去掉注释和文档字符串后就是这样）
def tools_condition_core(state, messages_key: str = "messages") -> str:
    """判断最后一条消息有没有 tool_calls。"""
    # 支持三种输入形态之一：直接传消息列表
    if isinstance(state, list):
        # 取最后一条
        ai_message = state[-1]
    # 字典状态就按 messages_key 取
    elif (isinstance(state, dict) and (messages := state.get(messages_key, []))) or (
        # 取不到再当属性取一次，兼容 BaseModel 形态的状态
        messages := getattr(state, messages_key, [])
    ):
        # 取最后一条消息
        ai_message = messages[-1]
    else:
        # 什么都取不到就报错，避免静默返回错误分支
        raise ValueError(f"No messages found in input state to tool_edge: {state}")
    # 有 tool_calls 就去工具节点
    if hasattr(ai_message, "tool_calls") and len(ai_message.tool_calls) > 0:
        # 注意返回的是硬编码的字符串 "tools"
        return "tools"
    # 否则结束
    return "__end__"

实测三种输入形态都能用：

# 手造一条带 tool_calls 的 AIMessage
from langchain_core.messages import AIMessage
# 预制的路由函数
from langgraph.prebuilt import tools_condition

# tool_calls 是一个字典列表，字段固定为 name / args / id / type
call = {"name": "get_stock", "args": {"sku": "A-100"}, "id": "m1", "type": "tool_call"}
# 字典状态 + 有工具需求
print("带 tool_calls:", tools_condition({"messages": [AIMessage(content="", tool_calls=[call])]}))
# 字典状态 + 没有工具需求
print("不带 tool_calls:", tools_condition({"messages": [AIMessage(content="好了")]}))
# 直接传消息列表也行
print("list 输入:", tools_condition([AIMessage(content="好了")]))
# 自定义状态键就传 messages_key
print("自定义键:", tools_condition({"chat": [AIMessage(content="好了")]}, messages_key="chat"))

运行输出：

带 tool_calls: tools
不带 tool_calls: __end__
list 输入: __end__
自定义键: __end__

在图里用自定义状态键，包一层 lambda 即可：lambda s: tools_condition(s, messages_key="chat_history")。

关于「工具节点必须叫 tools」，说法要更准确一点。 tools_condition 返回的是硬编码的字符串 "tools"，但那只是路由函数的返回值，不一定是节点名。用字典形式的 path_map 就能把它映射到任意节点。下面这段用一个假模型节点验证，不花钱：

# 手造带 tool_calls 的 AIMessage
from langchain_core.messages import AIMessage
# 演示用的小工具
from langchain_core.tools import tool
# 图的三件套 + 自带 add_messages 的预制状态
from langgraph.graph import END, START, MessagesState, StateGraph
# 工具节点和预制路由函数
from langgraph.prebuilt import ToolNode, tools_condition

# 一个最简单的工具
@tool
def echo(text: str) -> str:
    """把输入原样返回。"""
    # 直接回显
    return text

# 假模型节点：第一次要求调工具，之后直接收工
def fake_model_node(state: MessagesState) -> dict:
    # 历史里已经有 ToolMessage 就说明工具跑过了，直接结束
    if any(type(m).__name__ == "ToolMessage" for m in state["messages"]):
        # 返回一条没有 tool_calls 的普通回复
        return {"messages": [AIMessage(content="收到工具结果，收工")]}
    # 否则造一条工具调用请求
    return {"messages": [AIMessage(content="", tool_calls=[{"name": "echo", "args": {"text": "hi"}, "id": "e1", "type": "tool_call"}])]}

# 把工具节点起名叫 run_tools（故意不叫 tools）
builder = StateGraph(MessagesState)
# 注册假模型节点
builder.add_node("model", fake_model_node)
# 节点名是 run_tools
builder.add_node("run_tools", ToolNode([echo]))
# 入口
builder.add_edge(START, "model")
# 用字典 path_map 把返回值 "tools" 映射到实际节点 run_tools
builder.add_conditional_edges("model", tools_condition, {"tools": "run_tools", END: END})
# 工具跑完回到模型，构成环
builder.add_edge("run_tools", "model")
# 编译
graph = builder.compile()
# 跑通了：改名不影响，因为 path_map 做了映射
print("字典 path_map 改名:", [type(m).__name__ for m in graph.invoke({"messages": [{"role": "user", "content": "试试"}]})["messages"]])

运行输出：

字典 path_map 改名: ['HumanMessage', 'AIMessage', 'ToolMessage', 'AIMessage']

但如果你用列表形式的 path_map，列表里没有 "tools" 这个名字，运行时就会当场报错：

from langchain_core.messages import AIMessage
from langchain_core.tools import tool
from langgraph.graph import END, START, MessagesState, StateGraph
from langgraph.prebuilt import ToolNode, tools_condition

@tool
def echo(text: str) -> str:
    """把输入原样返回。"""
    return text

def fake_model_node(state: MessagesState) -> dict:
    if any(type(m).__name__ == "ToolMessage" for m in state["messages"]):
        return {"messages": [AIMessage(content="收到工具结果，收工")]}
    return {"messages": [AIMessage(content="", tool_calls=[{"name": "echo", "args": {"text": "hi"}, "id": "e1", "type": "tool_call"}])]}

# 同一张图，只把 path_map 换成列表形式
builder = StateGraph(MessagesState)
# 假模型节点照旧
builder.add_node("model", fake_model_node)
# 工具节点还是叫 run_tools
builder.add_node("run_tools", ToolNode([echo]))
# 入口照旧
builder.add_edge(START, "model")
# 列表里只有 run_tools，没有 tools_condition 实际返回的 "tools"
builder.add_conditional_edges("model", tools_condition, ["run_tools", END])
# 环照旧
builder.add_edge("run_tools", "model")
# 编译
graph = builder.compile()
try:
    # 路由函数返回 "tools"，但出口清单里没有它
    graph.invoke({"messages": [{"role": "user", "content": "试试"}]})
# 第 23 章 §4.2 讲过：有 path_map 时非法返回值会当场 KeyError
except KeyError as e:
    # 打印异常
    print("报错: KeyError", e)

运行输出：

报错: KeyError 'tools'

结论：节点不是非得叫 tools，但你必须用字典 path_map 显式做映射。省掉 path_map 又改了名字，就是第 23 章 §4.2 那个坑的翻版：静默走 END，工具永远不执行。

另外两点：

• 它硬编码了「有 tool_calls 就继续」，没有任何轮次限制。想加护栏就得自己写 should_continue，这正是 §6 要做的事。
• 它只看最后一条消息。如果你在 model 和路由之间插了别的节点（比如写日志的节点），最后一条就不是 AIMessage 了，tools_condition 会直接返回 __end__，静默走错分支。

5.2. ToolNode：批量执行 + 生成 ToolMessage #

ToolNode 做三件事：从最后一条 AIMessage 里取出所有 tool_calls、并行执行、把每个结果包成带 tool_call_id 的 ToolMessage 追加进状态。

想单独观察它的行为，把它放进一张只有一个节点的图里就行：

# 手造 AIMessage 用
from langchain_core.messages import AIMessage
# 两个用来演示的工具
from langchain_core.tools import tool

# 图的三件套 + 自带 add_messages 的预制状态
from langgraph.graph import END, START, MessagesState, StateGraph
# 本节的主角
from langgraph.prebuilt import ToolNode

# 一个正常的工具
@tool
def add(a: int, b: int) -> int:
    """两个整数相加。"""
    # 直接返回结果
    return a + b

# 另一个正常的工具
@tool
def mul(a: int, b: int) -> int:
    """两个整数相乘。"""
    # 直接返回结果
    return a * b

# 把 ToolNode 单独跑起来的小工具函数
def run_tool_node(node, msgs):
    """把 ToolNode 放进一张一节点的图里跑，返回新增的消息。"""
    # 用 MessagesState 就够了
    builder = StateGraph(MessagesState)
    # ToolNode 直接当节点
    builder.add_node("tools", node)
    # 入口直接进工具节点
    builder.add_edge(START, "tools")
    # 跑完就结束
    builder.add_edge("tools", END)
    # 编译后跑一次
    graph = builder.compile()
    # 只返回新增的部分，方便观察
    return graph.invoke({"messages": msgs})["messages"][len(msgs):]

# 一条 AIMessage 里塞两个 tool_calls
ai = AIMessage(
    # content 可以是空字符串，工具调用信息在 tool_calls 里
    content="",
    # 两个调用，id 必须唯一
    tool_calls=[
        # 第一个：1 + 2
        {"name": "add", "args": {"a": 1, "b": 2}, "id": "m1", "type": "tool_call"},
        # 第二个：5 × 6
        {"name": "mul", "args": {"a": 5, "b": 6}, "id": "m2", "type": "tool_call"},
    ],
)
# 跑一遍，看生成了什么
for m in run_tool_node(ToolNode([add, mul]), [ai]):
    # tool_call_id 对应回请求，name 是工具名，status 标记成功还是失败
    print(f"{type(m).__name__}(id={m.tool_call_id}) name={m.name} status={m.status}: {m.content!r}")

运行输出：

ToolMessage(id=m1) name=add status=success: '3'
ToolMessage(id=m2) name=mul status=success: '30'

三个细节：

• tool_call_id 的对应关系必须严格保持。 模型 API 会校验每个 tool_calls 都有对应的应答，这个约束在 §7.2 会咬人。
• 返回值被转成了字符串：add 返回 int 3，ToolMessage 里是 '3'。工具返回复杂对象时记得自己序列化成模型读得懂的文本。
• status 字段是可观测性的抓手：批量跑完之后 [m.status for m in msgs] 一眼就能看出这一轮有没有工具失败。

两个边界情况的行为也值得知道：

from langchain_core.messages import AIMessage, HumanMessage
from langchain_core.tools import tool
from langgraph.graph import END, START, MessagesState, StateGraph
from langgraph.prebuilt import ToolNode

@tool
def add(a: int, b: int) -> int:
    """两个整数相加。"""
    return a + b

@tool
def mul(a: int, b: int) -> int:
    """两个整数相乘。"""
    return a * b

def run_tool_node(node, msgs):
    """把 ToolNode 放进一张一节点的图里跑，返回新增的消息。"""
    builder = StateGraph(MessagesState)
    builder.add_node("tools", node)
    builder.add_edge(START, "tools")
    builder.add_edge("tools", END)
    graph = builder.compile()
    return graph.invoke({"messages": msgs})["messages"][len(msgs):]

# 最后一条 AIMessage 没有 tool_calls
print("空 tool_calls:", run_tool_node(ToolNode([add, mul]), [AIMessage(content="没有工具需求")]))
try:
    # 最后一条根本不是 AIMessage，直接扔一条 HumanMessage 进去
    run_tool_node(ToolNode([add, mul]), [HumanMessage(content="你好")])
# 这种情况会响亮地失败
except ValueError as e:
    # 打印异常信息
    print("报错:", e)

运行输出：

空 tool_calls: []
报错: No AIMessage found in input

前者返回空列表（什么也不做），后者直接 ValueError。所以路由必须保证只在有 tool_calls 时才进 tools 节点，tools_condition 和我们手写的 should_continue 都是在做这件事。

最后一个容易踩的实现细节：ToolNode 不能脱离图单独 invoke。

from langchain_core.messages import AIMessage
from langchain_core.tools import tool
from langgraph.prebuilt import ToolNode

@tool
def add(a: int, b: int) -> int:
    """两个整数相加。"""
    return a + b

@tool
def mul(a: int, b: int) -> int:
    """两个整数相乘。"""
    return a * b

ai = AIMessage(content="", tool_calls=[{"name": "add", "args": {"a": 1, "b": 2}, "id": "m1", "type": "tool_call"}])

# 试着直接调用
try:
    # 不在图的执行上下文里
    ToolNode([add, mul]).invoke({"messages": [ai]})
# 会因为缺少运行时上下文而报错
except ValueError as e:
    # 打印异常信息
    print("单独 invoke 报错:", e)

运行输出：

单独 invoke 报错: Missing required config key 'N/A' for 'tools'.

它需要图提供的运行时上下文。这解释了一件事：§6.2 会在自己的节点函数里写 TOOL_NODE.invoke(state)，那样能跑是因为节点函数本身就在图的执行上下文里；同样一行代码搬到图外面就会报这个错。

ToolNode.invoke() 在编译图之外被调用时，在调用 invoke 时，手动传入一个空的 Runtime 对象：

from langgraph.runtime import Runtime

# 修复：传入 runtime=Runtime()
ToolNode([add, mul]).invoke({"messages": [ai]}, runtime=Runtime())

5.3. 错误处理：默认行为和你想的不一样 #

这里有个反直觉的设计，也是本章最值得记住的 API 细节。用一个参数会校验失败的工具和一个内部抛业务异常的工具测四种情况：

from langchain_core.messages import AIMessage
from langchain_core.tools import tool
from langgraph.graph import END, START, MessagesState, StateGraph
from langgraph.prebuilt import ToolNode

@tool
def add(a: int, b: int) -> int:
    """两个整数相加。"""
    return a + b

@tool
def mul(a: int, b: int) -> int:
    """两个整数相乘。"""
    return a * b

def run_tool_node(node, msgs):
    """把 ToolNode 放进一张一节点的图里跑，返回新增的消息。"""
    builder = StateGraph(MessagesState)
    builder.add_node("tools", node)
    builder.add_edge(START, "tools")
    builder.add_edge("tools", END)
    graph = builder.compile()
    return graph.invoke({"messages": msgs})["messages"][len(msgs):]

# 一个总会抛业务异常的工具
@tool
def boom(x: int) -> str:
    """一个总是抛业务异常的工具。"""
    # 模拟数据库连不上、下游 500 之类的故障
    raise ValueError(f"工具炸了: {x}")

# 工具清单加上 boom
TOOLS3 = [add, mul, boom]

# 造一条只调一个工具的 AIMessage，方便反复用
def one_call(name, args):
    """构造一条只有一个 tool_call 的 AIMessage。"""
    # id 固定为 m1，本节不关心 id
    return AIMessage(content="", tool_calls=[{"name": name, "args": args, "id": "m1", "type": "tool_call"}])

# 确认工具清单装好了
print("工具清单:", [t.name for t in TOOLS3])

运行输出：

工具清单: ['add', 'mul', 'boom']

下面四段都复用这套工具清单。为了让每段都能单独运行，公共前置会重复出现，重点只看注释密集的那几行。

5.3.1 情况一：模型填错参数 #

默认就会被接住，转成 status="error" 的 ToolMessage 回灌给模型：

# ↓ §5.3 的公共前置：add / mul / boom 三个工具、run_tool_node 和 one_call
from langchain_core.messages import AIMessage
from langchain_core.tools import tool
from langgraph.graph import END, START, MessagesState, StateGraph
from langgraph.prebuilt import ToolNode

@tool
def add(a: int, b: int) -> int:
    """两个整数相加。"""
    return a + b

@tool
def mul(a: int, b: int) -> int:
    """两个整数相乘。"""
    return a * b

@tool
def boom(x: int) -> str:
    """一个总是抛业务异常的工具。"""
    raise ValueError(f"工具炸了: {x}")

TOOLS3 = [add, mul, boom]

def run_tool_node(node, msgs):
    builder = StateGraph(MessagesState)
    builder.add_node("tools", node)
    builder.add_edge(START, "tools")
    builder.add_edge("tools", END)
    graph = builder.compile()
    return graph.invoke({"messages": msgs})["messages"][len(msgs):]

def one_call(name, args):
    return AIMessage(content="", tool_calls=[{"name": name, "args": args, "id": "m1", "type": "tool_call"}])

# a 给了字符串，b 干脆没给
for m in run_tool_node(ToolNode(TOOLS3), [one_call("add", {"a": "不是数字"})]):
    # 看 status 和正文
    print(f"{type(m).__name__} status={m.status}: {m.content!r}")

运行输出（用 !r 打印，所以换行显示成了 \n）：

ToolMessage status=error: "Error invoking tool 'add' with kwargs {'a': '不是数字'} with error:\n a: Input should be a valid integer, unable to parse string as an integer\nb: Field required\n Please fix the error and try again."

注意错误信息里连「哪个参数错了、错在哪」都写清楚了，还附了一句 Please fix the error and try again。模型看到这条会在下一圈自己改参数重试，这正是环的价值：自我纠错不需要你写任何代码。

5.3.2 情况二：模型调不存在的工具 #

同样被接住，还会把可用工具列表告诉模型：

# ↓ §5.3 的公共前置：add / mul / boom 三个工具、run_tool_node 和 one_call
from langchain_core.messages import AIMessage
from langchain_core.tools import tool
from langgraph.graph import END, START, MessagesState, StateGraph
from langgraph.prebuilt import ToolNode

@tool
def add(a: int, b: int) -> int:
    """两个整数相加。"""
    return a + b

@tool
def mul(a: int, b: int) -> int:
    """两个整数相乘。"""
    return a * b

@tool
def boom(x: int) -> str:
    """一个总是抛业务异常的工具。"""
    raise ValueError(f"工具炸了: {x}")

TOOLS3 = [add, mul, boom]

def run_tool_node(node, msgs):
    builder = StateGraph(MessagesState)
    builder.add_node("tools", node)
    builder.add_edge(START, "tools")
    builder.add_edge("tools", END)
    graph = builder.compile()
    return graph.invoke({"messages": msgs})["messages"][len(msgs):]

def one_call(name, args):
    return AIMessage(content="", tool_calls=[{"name": name, "args": args, "id": "m1", "type": "tool_call"}])

# 编一个不存在的工具名
for m in run_tool_node(ToolNode(TOOLS3), [one_call("不存在", {})]):
    # 看它怎么提示模型
    print(f"{type(m).__name__} status={m.status}: {m.content!r}")

运行输出：

ToolMessage status=error: 'Error: 不存在 is not a valid tool, try one of [add, mul, boom].'

5.3.3 情况三：工具内部抛业务异常 #

默认不接，直接把异常抛出去中断整张图：

# ↓ §5.3 的公共前置：add / mul / boom 三个工具、run_tool_node 和 one_call
from langchain_core.messages import AIMessage
from langchain_core.tools import tool
from langgraph.graph import END, START, MessagesState, StateGraph
from langgraph.prebuilt import ToolNode

@tool
def add(a: int, b: int) -> int:
    """两个整数相加。"""
    return a + b

@tool
def mul(a: int, b: int) -> int:
    """两个整数相乘。"""
    return a * b

@tool
def boom(x: int) -> str:
    """一个总是抛业务异常的工具。"""
    raise ValueError(f"工具炸了: {x}")

TOOLS3 = [add, mul, boom]

def run_tool_node(node, msgs):
    builder = StateGraph(MessagesState)
    builder.add_node("tools", node)
    builder.add_edge(START, "tools")
    builder.add_edge("tools", END)
    graph = builder.compile()
    return graph.invoke({"messages": msgs})["messages"][len(msgs):]

def one_call(name, args):
    return AIMessage(content="", tool_calls=[{"name": name, "args": args, "id": "m1", "type": "tool_call"}])

# 调那个必炸的工具
try:
    # 默认配置
    run_tool_node(ToolNode(TOOLS3), [one_call("boom", {"x": 9})])
# 原始异常直接穿透到调用方，这里接住只是为了让脚本能跑完
except ValueError as e:
    # 类型和信息都是工具里抛的那个
    print("报错:", type(e).__name__, e)

运行输出：

报错: ValueError 工具炸了: 9

为什么会有这个区分？ 看默认处理器的实现就明白了，它只有四行：

# 这个异常类型是分界线所在
from langgraph.prebuilt.tool_node import ToolInvocationError

# 等价于 langgraph/prebuilt/tool_node.py 里的 _default_handle_tool_errors
def default_handle_tool_errors(e: Exception) -> str:
    """默认错误处理器：只接住「调用本身不合法」这一类。"""
    # 参数校验失败、工具不存在都属于 ToolInvocationError
    if isinstance(e, ToolInvocationError):
        # 把校验信息作为 ToolMessage 正文回灌给模型
        return e.message
    # 其它异常一律原样抛出，中断整张图
    raise e

# 手造一个「参数校验失败」的异常（真实场景里由 ToolNode 内部抛出）
invocation_err = ToolInvocationError("add", source=ValueError("a: 不是整数"), tool_kwargs={"a": "不是数字"})
# 验证分界：一个可恢复的、一个不可恢复的，分别过一遍
for err in (invocation_err, ValueError("数据库连不上")):
    try:
        # 被接住就打印它回灌给模型的文本
        print(f"{type(err).__name__} -> 接住了: {default_handle_tool_errors(err)}")
    # 没被接住就说明它被原样抛出了
    except Exception as e:
        # 打印异常类型
        print(f"{type(err).__name__} -> 抛出去了: {type(e).__name__}")

运行输出：

ToolInvocationError -> 接住了: Error invoking tool 'add' with kwargs {'a': '不是数字'} with error:
 a: 不是整数
 Please fix the error and try again.
ValueError -> 抛出去了: ValueError

顺便看清了一件事：情况一那段又长又细的报错文本，就是 ToolInvocationError.message 拼出来的，内容依次是 tool_name、tool_kwargs、原始校验错误，再加一句 Please fix the error and try again。

ToolInvocationError 涵盖的正是情况一和情况二，也就是参数校验失败、工具不存在这两类「模型自己填错了」的问题。这类错误模型看一眼就能改。而数据库连不上、下游服务 500 这类问题，模型再试一百次也没用，不如让它响亮地失败，交给上层重试或告警。

5.3.4 情况四：想让业务异常也回灌给模型 #

用 handle_tool_errors。它有四种写法：

# ↓ §5.3 的公共前置：add / mul / boom 三个工具、run_tool_node 和 one_call
from langchain_core.messages import AIMessage
from langchain_core.tools import tool
from langgraph.graph import END, START, MessagesState, StateGraph
from langgraph.prebuilt import ToolNode

@tool
def add(a: int, b: int) -> int:
    """两个整数相加。"""
    return a + b

@tool
def mul(a: int, b: int) -> int:
    """两个整数相乘。"""
    return a * b

@tool
def boom(x: int) -> str:
    """一个总是抛业务异常的工具。"""
    raise ValueError(f"工具炸了: {x}")

TOOLS3 = [add, mul, boom]

def run_tool_node(node, msgs):
    builder = StateGraph(MessagesState)
    builder.add_node("tools", node)
    builder.add_edge(START, "tools")
    builder.add_edge("tools", END)
    graph = builder.compile()
    return graph.invoke({"messages": msgs})["messages"][len(msgs):]

def one_call(name, args):
    return AIMessage(content="", tool_calls=[{"name": name, "args": args, "id": "m1", "type": "tool_call"}])

# 写法一，True：用内置模板包装异常
print(run_tool_node(ToolNode(TOOLS3, handle_tool_errors=True), [one_call("boom", {"x": 9})])[0].content)
# 写法二，字符串：固定提示，不暴露内部异常细节
print(run_tool_node(ToolNode(TOOLS3, handle_tool_errors="工具暂时不可用，请换个方式"), [one_call("boom", {"x": 9})])[0].content)
# 写法三，函数：按异常类型动态决定提示词
print(run_tool_node(ToolNode(TOOLS3, handle_tool_errors=lambda e: f"调用失败({type(e).__name__})，请改用其他工具"), [one_call("boom", {"x": 9})])[0].content)
# 写法四，异常类型（或类型元组）：只接住这几类，其余照旧抛出
print(run_tool_node(ToolNode(TOOLS3, handle_tool_errors=(ValueError, TimeoutError)), [one_call("boom", {"x": 9})])[0].content)

运行输出：

Error: ValueError('工具炸了: 9')
 Please fix your mistakes.
工具暂时不可用，请换个方式
调用失败(ValueError)，请改用其他工具
Error: ValueError('工具炸了: 9')
 Please fix your mistakes.

写法四的价值在于分类处理：(ValueError, TimeoutError) 会被接住回灌，其它异常（比如权限错误、配置错误）仍然会中断图。换一个不在名单里的异常类型试试：

from langchain_core.messages import AIMessage
from langchain_core.tools import tool
from langgraph.graph import END, START, MessagesState, StateGraph
from langgraph.prebuilt import ToolNode

@tool
def boom(x: int) -> str:
    """一个总是抛业务异常的工具。"""
    raise ValueError(f"工具炸了: {x}")

def run_tool_node(node, msgs):
    builder = StateGraph(MessagesState)
    builder.add_node("tools", node)
    builder.add_edge(START, "tools")
    builder.add_edge("tools", END)
    graph = builder.compile()
    return graph.invoke({"messages": msgs})["messages"][len(msgs):]

def one_call(name, args):
    return AIMessage(content="", tool_calls=[{"name": name, "args": args, "id": "m1", "type": "tool_call"}])

# 再加一个抛 TimeoutError 的工具
@tool
def boom_io(x: int) -> str:
    """总是抛 TimeoutError 的工具。"""
    # 模拟下游超时
    raise TimeoutError(f"超时: {x}")

# ValueError 被接住
print("boom:", run_tool_node(ToolNode([boom, boom_io], handle_tool_errors=ValueError), [one_call("boom", {"x": 1})])[0].status)
# TimeoutError 不在名单里，原样抛出
try:
    # 调那个抛超时的工具
    run_tool_node(ToolNode([boom, boom_io], handle_tool_errors=ValueError), [one_call("boom_io", {"x": 1})])
except TimeoutError as e:
    # 说明它确实没被接住
    print("boom_io 报错:", e)

运行输出：

boom: error
boom_io 报错: 超时: 1

反过来，handle_tool_errors=False 会把连参数错误都不接，情况一也变成抛异常：

# ↓ §5.3 的公共前置：add 工具、run_tool_node 和 one_call
from langchain_core.messages import AIMessage
from langchain_core.tools import tool
from langgraph.graph import END, START, MessagesState, StateGraph
from langgraph.prebuilt import ToolNode
# 参数校验失败时抛的异常类型
from langgraph.prebuilt.tool_node import ToolInvocationError

@tool
def add(a: int, b: int) -> int:
    """两个整数相加。"""
    return a + b

def run_tool_node(node, msgs):
    builder = StateGraph(MessagesState)
    builder.add_node("tools", node)
    builder.add_edge(START, "tools")
    builder.add_edge("tools", END)
    graph = builder.compile()
    return graph.invoke({"messages": msgs})["messages"][len(msgs):]

def one_call(name, args):
    return AIMessage(content="", tool_calls=[{"name": name, "args": args, "id": "m1", "type": "tool_call"}])

# 关掉所有错误接管
try:
    # 参数依然填错
    run_tool_node(ToolNode([add], handle_tool_errors=False), [one_call("add", {"a": "x"})])
# 这次连校验错误也直接抛
except ToolInvocationError as e:
    # 异常类型是 ToolInvocationError
    print("报错:", type(e).__name__)

运行输出：

报错: ToolInvocationError

选型建议：

场景	推荐写法	理由
演示 / 内部工具	默认（不传）	模型能自己改的错自己改，真故障响亮失败
生产环境	字符串或函数	完全控制回灌给模型的文本
需要分类处理	异常类型元组	可恢复的接住，不可恢复的中断
想让模型看到原始异常	True	⚠️ 只在内部环境用

True 的风险要说清楚：它会把 repr(e) 塞进消息历史，而异常信息里可能带着连接串、内部路径、SQL 片段。这些既不该让模型看见，更不该出现在最终回复里，因为模型很可能把它复述给用户。

5.4. wrap_tool_call：给工具加重试 #

当前版本的 ToolNode 还有一对钩子 wrap_tool_call / awrap_tool_call，可以拦截每一次工具执行。最常见的用途是重试：

from langchain_core.messages import AIMessage
from langchain_core.tools import tool
from langgraph.graph import END, START, MessagesState, StateGraph
from langgraph.prebuilt import ToolNode

def run_tool_node(node, msgs):
    builder = StateGraph(MessagesState)
    builder.add_node("tools", node)
    builder.add_edge(START, "tools")
    builder.add_edge("tools", END)
    graph = builder.compile()
    return graph.invoke({"messages": msgs})["messages"][len(msgs):]

def one_call(name, args):
    return AIMessage(content="", tool_calls=[{"name": name, "args": args, "id": "m1", "type": "tool_call"}])

# 一个计数器，记录工具被真正调用了几次
CALLS = {"n": 0}

# 前两次失败、第三次成功的工具，用来验证重试真的发生了
@tool
def flaky(x: int) -> str:
    """前两次失败、第三次成功的工具。"""
    # 每次调用都计数
    CALLS["n"] += 1
    # 前两次抛异常
    if CALLS["n"] < 3:
        # 模拟偶发故障
        raise ValueError(f"第 {CALLS['n']} 次失败")
    # 第三次成功
    return f"第 {CALLS['n']} 次成功"

# 拦截器：接到「请求」和「执行函数」，自己决定怎么调
def retry_wrapper(request, execute):
    """拦截工具执行，失败就重试，最多三次。"""
    # 最多试三次
    for _ in range(3):
        try:
            # execute(request) 才是真正的一次工具调用
            return execute(request)
        # 失败就记下异常，继续下一次
        except Exception as e:
            # 留着最后一次的异常
            last = e
    # 三次都失败，抛出去
    raise last

# 把拦截器挂到 ToolNode 上
node = ToolNode([flaky], wrap_tool_call=retry_wrapper)
# 跑一次
msg = run_tool_node(node, [one_call("flaky", {"x": 1})])[0]
# 结果是成功的，说明重试起作用了
print("结果:", msg.status, msg.content)
# 真实调用次数证明它试了三次
print("实际调用次数:", CALLS["n"])

运行输出：

结果: success 第 3 次成功
实际调用次数: 3

wrap_tool_call 和 handle_tool_errors 的分工：前者决定「要不要再试一次」，后者决定「彻底失败之后跟模型怎么说」。两者可以叠加：重试三次仍然失败，再由 handle_tool_errors 转成 ToolMessage 回灌。

它还能做缓存（同样的参数直接返回上次结果）、改写请求（补默认参数）、埋点（记录每次工具耗时）。相比在每个工具函数里各写一遍，拦截器只写一次。

6. 三重终止保险 #

§4 的环只有一个退出条件：模型自己说不用工具了。这在演示里够用，上生产不够，因为模型完全可能陷入「查一次 → 觉得不够 → 再查一次」的循环。

这不是假想。给模型一个总是回复「还需要再查一次」的工具，它会老老实实照做：

# @tool 把普通函数变成模型可调用的工具
from langchain_core.tools import tool

# 一个专门用来制造失控的工具
@tool
def ping(x: str) -> str:
    """一个测试工具，永远回复「还需要再查一次」。"""
    # 打印调用痕迹，方便数圈数
    print(f"   [tool] ping({x!r})")
    # 返回值本身在诱导模型继续调用
    return "还需要再查一次，请再调用一次 ping 工具，参数随意。"

# 直接调一次看看它返回什么
print(ping.invoke({"x": "abc"}))

运行输出：

   [tool] ping('abc')
还需要再查一次，请再调用一次 ping 工具，参数随意。

把它接到 §4 的环上，结构一模一样，只是工具换成了 ping：

# Literal 用来声明路由函数的出口
from typing import Literal

# 从 .env 读 API Key
from dotenv import load_dotenv
# 统一的模型入口
from langchain.chat_models import init_chat_model
# 判断最后一条消息是不是「模型提了工具需求」
from langchain_core.messages import AIMessage
# @tool 把普通函数变成模型可调用的工具
from langchain_core.tools import tool
# 循环没有出口时抛的异常
from langgraph.errors import GraphRecursionError
# 图的三件套 + 自带 add_messages 的预制状态
from langgraph.graph import END, START, MessagesState, StateGraph
# 执行 tool_calls 的预制节点
from langgraph.prebuilt import ToolNode

# 加载环境变量
load_dotenv(override=True)

@tool
def ping(x: str) -> str:
    """一个测试工具，永远回复「还需要再查一次」。"""
    print(f"   [tool] ping({x!r})")
    return "还需要再查一次，请再调用一次 ping 工具，参数随意。"

# ping 工具单独成一组，避免覆盖 §4 的 TOOLS
PING_TOOLS = [ping]
# 同一个模型，绑上 ping
ping_model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0).bind_tools(PING_TOOLS)

# 和 §4 完全一样的骨架，抽成函数方便 §7.3 复用
def build_ping_loop(checkpointer=None):
    """一个只有「模型自然停」这一个出口的环，用来演示失控。"""

    # 节点一：调模型
    def call_ping_model(state: MessagesState) -> dict:
        # 把历史丢给模型
        return {"messages": [ping_model.invoke(state["messages"])]}

    # 路由：和 §4 的 should_continue 一样，没有任何轮次限制
    def route(state: MessagesState) -> Literal["tools", "__end__"]:
        # 看最后一条消息
        last = state["messages"][-1]
        # 有 tool_calls 就继续跑工具
        return "tools" if isinstance(last, AIMessage) and last.tool_calls else END

    # 搭图
    builder = StateGraph(MessagesState)
    # 模型节点
    builder.add_node("model", call_ping_model)
    # 工具节点
    builder.add_node("tools", ToolNode(PING_TOOLS))
    # 入口
    builder.add_edge(START, "model")
    # 两个出口
    builder.add_conditional_edges("model", route, ["tools", END])
    # 构成环
    builder.add_edge("tools", "model")
    # checkpointer 可选，§7.3 会用到
    return builder.compile(checkpointer=checkpointer)

# 这张图除了「模型自然停」没有任何护栏
graph = build_ping_loop()
# limit 只给 4，两圈就撞上，演示不需要真烧钱
try:
    # 模型会老老实实按工具的要求一圈接一圈调
    graph.invoke({"messages": [{"role": "user", "content": "请用 ping 工具查一下 abc"}]}, {"recursion_limit": 4})
# 撞上限
except GraphRecursionError:
    # 只能拿到一行异常
    print("抛错: GraphRecursionError")
    # 前面几圈的结果一个都拿不到
    print("invoke 没有返回值，中间跑出来的东西全拿不到")

运行输出（第二圈传什么参数由模型决定，不一定还是 abc）：

   [tool] ping('abc')
   [tool] ping('abc')
抛错: GraphRecursionError
invoke 没有返回值，中间跑出来的东西全拿不到

没有护栏就是这个下场：要么烧钱到 limit，要么抛异常给用户看。两个结果都不能接受。注意这里每一圈都是一次真实的模型调用，limit=10007 的默认值意味着一次失控能刷掉几千次调用的费用。

顺便说一个实测中的有趣现象：把 recursion_limit 放宽到 8，有时候模型自己会醒过来：

AIMessage: 工具已经连续返回了相同的结果，我在这里停止循环，向您汇报情况。

同样的输入、同样的 temperature=0，有的运行会一直转到撞限，有的会在第三四圈自己停下。这恰恰说明了问题所在：模型能不能停下来是概率性的，而你的账单是确定性的。护栏不能指望模型自觉。

6.1. 三层保险各管什么 #

正确做法是叠三层，各自管不同的失控程度：

保险	触发条件	结果	定位
一、模型自然停	最后一条 AIMessage 没有 tool_calls	正常回复	99% 的情况走这里
二、轮次上限	tool_rounds >= max_tool_rounds	优雅降级，仍然给用户一个回复	业务护栏，主力
三、recursion_limit	前两层都失效	抛 GraphRecursionError	代码写错时的保险丝

关键在第二层。它和第三层的区别不是数值大小，而是结果的性质：

• recursion_limit 抛异常、丢现场，用户看到的是 500
• 轮次上限走一个专门的降级节点，用户看到的是「我查到了这些，剩下的没查到」

也就是说，第二层是业务行为，第三层是故障兜底。很多人只设了第三层（甚至只用默认值），等于把「模型话多」这种日常情况当成故障处理。

三层之间的数值关系也别搞反：recursion_limit 必须大于轮次上限所需的超步数，否则第三层会先触发，第二层永远轮不到（降级节点根本跑不到）。§3.2 的公式 2 × 最大轮次 + 5 就是在保证这个顺序。

6.2. 轮次计数器怎么写 #

轮次要记在状态里，而且必须配 reducer：

# operator.add 当 reducer
import operator
# Annotated 用来挂 reducer
from typing import Annotated

# 继承 MessagesState，白拿 add_messages
from langgraph.graph import MessagesState

# 在消息状态基础上加两个控制槽位
class AgentState(MessagesState):
    # 必须有 reducer：节点返回 1 的语义是「加一」
    tool_rounds: Annotated[int, operator.add]
    # 普通字段，后写覆盖先写
    stopped_by: str

然后在工具节点里顺手 +1。因为要加这一下，就不能直接把 ToolNode 当节点用了，包一层：

import operator
from typing import Annotated

from langchain_core.tools import tool
from langgraph.graph import MessagesState
from langgraph.prebuilt import ToolNode

@tool
def get_stock(sku: str) -> str:
    """查询某个 SKU 的库存数量。"""
    return {"A-100": "库存 12 件", "B-200": "库存 0 件"}.get(sku, "查无此 SKU")

@tool
def get_price(sku: str) -> str:
    """查询某个 SKU 的单价（元）。"""
    return {"A-100": "单价 39 元", "B-200": "单价 58 元"}.get(sku, "查无此 SKU")

TOOLS = [get_stock, get_price]

class AgentState(MessagesState):
    tool_rounds: Annotated[int, operator.add]
    stopped_by: str

# ToolNode 是无状态的，建一次就够，别在节点函数里反复 new
TOOL_NODE = ToolNode(TOOLS, handle_tool_errors="工具暂时不可用，请换个思路或如实说明查不到。")

# 包一层，把「跑工具」和「记轮次」放一起
def run_tools(state: AgentState) -> dict:
    """包一层 ToolNode，顺便把轮次 +1。"""
    # ToolNode 返回 {"messages": [ToolMessage, ...]}
    result = TOOL_NODE.invoke(state)
    # 有 operator.add，这个 1 是「加一」不是「设为一」
    return {**result, "tool_rounds": 1}

# 这段只是定义，真正跑起来要等它被注册进图（下一段就会用上）
print("工具节点已就绪:", [t.name for t in TOOLS])

运行输出：

工具节点已就绪: ['get_stock', 'get_price']

三个注意事项：

• tool_rounds 数的是「模型调工具的轮次」，不是工具调用的总次数。 模型一圈提六个 tool_calls，ToolNode 并行跑完，只算一轮。这个区分在 §7.4 展开，也是很多人把上限设得太松的原因。
• TOOL_NODE 建在函数外面。 ToolNode 本身没有状态，每次调用都 ToolNode(TOOLS, ...) 重新构造纯属浪费（它要遍历工具列表、解析每个工具的 schema）。
• 为什么这里能 TOOL_NODE.invoke(state)？ 因为 run_tools 是被图调用的节点函数，本身就跑在图的执行上下文里。§5.2 演示过，同样一行放到图外面会报 Missing required config key。

写完计数器，务必空跑验证它真的在涨。 不用调模型，拿一个假模型就能测，这几行代码能帮你避开本章头号坑（§7.1）：

# operator.add 给轮次计数器当 reducer
import operator
# Annotated 挂 reducer，Literal 声明路由出口
from typing import Annotated, Literal

# 手造带 tool_calls 的 AIMessage
from langchain_core.messages import AIMessage
# @tool 把普通函数变成工具
from langchain_core.tools import tool
# 图的三件套 + 自带 add_messages 的预制状态
from langgraph.graph import END, START, MessagesState, StateGraph
# 执行 tool_calls 的预制节点
from langgraph.prebuilt import ToolNode

class AgentState(MessagesState):
    tool_rounds: Annotated[int, operator.add]
    stopped_by: str

# 假工具：不打印、不碰真数据，只返回固定结果
@tool
def mock_tool(sku: str) -> str:
    """自检用的假工具，返回固定结果。"""
    # 内容无关紧要，只要是个合法的 ToolMessage 正文
    return f"{sku} 的假结果"

# 假工具的 ToolNode，同样建一次
MOCK_NODE = ToolNode([mock_tool])

# 用假模型搭一张结构相同的图，专门验证计数器
def build_mock(max_rounds: int = 3):
    """假模型 + 真 ToolNode，专门验证轮次计数和护栏。"""

    # 假模型：一直假装还想调工具，把上限逼出来
    def fake_model(state: AgentState) -> dict:
        # 轮次还没到 5 就继续「要求」调工具
        if state["tool_rounds"] < 5:
            # 造一条带 tool_calls 的假 AIMessage
            return {"messages": [AIMessage(content="", tool_calls=[{"name": "mock_tool", "args": {"sku": "A-100"}, "id": "c1", "type": "tool_call"}])]}
        # 到了就假装答完
        return {"messages": [AIMessage(content="查完了")]}

    # 工具节点用假工具，自检时不该碰真的下游系统
    def run_tools_mock(state: AgentState) -> dict:
        # 仍然走真的 ToolNode，才能顺带验证 ToolMessage 序列合法
        return {**MOCK_NODE.invoke(state), "tool_rounds": 1}

    # 路由：到上限就直接结束（这里只验计数，不验降级）
    def route(state: AgentState) -> Literal["tools", "done"]:
        # 看最后一条
        last = state["messages"][-1]
        # 模型不要工具了就结束
        if not (isinstance(last, AIMessage) and last.tool_calls):
            # 走结束标记节点
            return "done"
        # 轮次用完了也结束
        if state["tool_rounds"] >= max_rounds:
            # 同样走结束标记节点
            return "done"
        # 否则继续
        return "tools"

    # 搭图
    builder = StateGraph(AgentState)
    # 假模型节点
    builder.add_node("model", fake_model)
    # 工具节点
    builder.add_node("tools", run_tools_mock)
    # 结束标记节点
    builder.add_node("done", lambda s: {"stopped_by": "上限或自然结束"})
    # 入口
    builder.add_edge(START, "model")
    # 两个出口
    builder.add_conditional_edges("model", route, ["tools", "done"])
    # 构成环
    builder.add_edge("tools", "model")
    # 结束
    builder.add_edge("done", END)
    # 编译
    return builder.compile()

# 造一份完整初始状态（TypedDict 没有默认值）
mock_input = {"messages": [{"role": "user", "content": "x"}], "tool_rounds": 0, "stopped_by": ""}
# 编译出自检用的图
graph = build_mock(3)
# 用 stream 看每一圈的增量
for chunk in graph.stream(mock_input, stream_mode="updates"):
    # 遍历这一超步的节点
    for node, upd in chunk.items():
        # 工具节点要额外看轮次增量
        if node == "tools":
            # 增量应该恒为 1
            print(f"   [{node}] tool_rounds 增量: {upd.get('tool_rounds')}")
        else:
            # 其它节点只打印名字
            print(f"   [{node}]")
# 再跑一次看最终值
print("最终 tool_rounds:", graph.invoke(mock_input)["tool_rounds"], "（上限 3，符合预期）")

运行输出：

   [model]
   [tools] tool_rounds 增量: 1
   [model]
   [tools] tool_rounds 增量: 1
   [model]
   [tools] tool_rounds 增量: 1
   [model]
   [done]
最终 tool_rounds: 3 （上限 3，符合预期）

上限设 3，实际跑了 3 轮就停，计数器和护栏都是好的。不花一分钱，几秒钟跑完。 把 AgentState 里的 Annotated[int, operator.add] 改成 int 再跑这段，你会立刻看到「增量恒为 1、最终值也是 1」，然后一路跑到 recursion_limit。

6.3. 降级节点：停下来也要给个交代 #

路由函数现在有三个出口，对应「继续 / 降级 / 自然结束」：

# operator.add 给轮次计数器当 reducer
import operator
# Annotated 挂 reducer，Literal 声明路由出口
from typing import Annotated, Literal

# 判断有没有 tool_calls 要用 AIMessage
from langchain_core.messages import AIMessage
# 自带 add_messages reducer 的预制状态
from langgraph.graph import MessagesState

class AgentState(MessagesState):
    tool_rounds: Annotated[int, operator.add]
    stopped_by: str

# 轮次上限，实战里由 build_graph 的参数传进来（§8.2）
max_tool_rounds = 4

# 三个出口全部写进 Literal，图纸和运行时校验都靠它
def should_continue(state: AgentState) -> Literal["tools", "force_finish", "done"]:
    # 看模型最新那条回复
    last = state["messages"][-1]
    # 保险一：模型没提工具需求，自然结束
    if not (isinstance(last, AIMessage) and last.tool_calls):
        # 走标记节点，把终止原因写进状态
        return "done"
    # 保险二：还想调工具，但轮次用完了，走降级出口
    if state["tool_rounds"] >= max_tool_rounds:
        # 注意：此时模型已经提了 tool_calls，但我们不打算执行
        return "force_finish"
    # 否则正常跑工具
    return "tools"

# 手造三种状态，把三个出口各走一遍
want_tool = AIMessage(content="", tool_calls=[{"name": "get_stock", "args": {"sku": "A-100"}, "id": "m1", "type": "tool_call"}])
# 没提工具需求
print("模型答完了:", should_continue({"messages": [AIMessage(content="查完了")], "tool_rounds": 1, "stopped_by": ""}))
# 提了工具需求且轮次还够
print("还想调且没超限:", should_continue({"messages": [want_tool], "tool_rounds": 1, "stopped_by": ""}))
# 提了工具需求但轮次用完了
print("还想调但已超限:", should_continue({"messages": [want_tool], "tool_rounds": 4, "stopped_by": ""}))

运行输出：

模型答完了: done
还想调且没超限: tools
还想调但已超限: force_finish

为什么专门加一个 done 节点，而不是直接返回 END？因为调用方需要知道这次是怎么停的。done 只做一件事：把 stopped_by 标成「模型自然结束」。有了它，返回值里永远有一个明确的终止原因，不用靠「轮次是不是等于上限」去猜。

force_finish 的思路是把工具收走，逼模型用已有信息作答：

# operator.add 给轮次计数器当 reducer
import operator
# Annotated 挂 reducer
from typing import Annotated

# 从 .env 读 API Key
from dotenv import load_dotenv
# 统一的模型入口
from langchain.chat_models import init_chat_model
# AIMessage 判断 tool_calls，ToolMessage 补「已取消」回执
from langchain_core.messages import AIMessage, ToolMessage
# 自带 add_messages reducer 的预制状态
from langgraph.graph import MessagesState

load_dotenv(override=True)

class AgentState(MessagesState):
    tool_rounds: Annotated[int, operator.add]
    stopped_by: str

# 轮次上限
max_tool_rounds = 1
# 降级时用的是**没绑工具**的模型，理由见下文第二点
base = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)

# 降级节点：不执行工具，直接让模型收尾
def force_finish(state: AgentState) -> dict:
    """超限降级：把工具收走，逼模型用已有信息作答。"""
    # 最后一条一定是「提了 tool_calls」的 AIMessage
    last = state["messages"][-1]
    # 关键：这些 tool_calls 没人应答，
    # 直接发给模型会被 API 判为非法消息序列，必须先补「已取消」回执
    cancels = [
        # 每个 tool_call_id 都要有一条对应的 ToolMessage
        ToolMessage(
            # 如实告诉模型为什么没执行
            content=f"已取消：达到 {max_tool_rounds} 轮工具调用上限，本次未执行。",
            # 必须和 AIMessage 里的 id 一一对应
            tool_call_id=c["id"],
            # 带上工具名，便于模型对齐
            name=c["name"],
            # 标成 error，语义上这是一次失败的调用
            status="error",
        )
        # 遍历这一轮所有被取消的调用
        for c in last.tool_calls
    ]
    # 再补一条用户视角的说明，明确「不要再要工具了」
    note = {
        # 用 user 角色，模型对用户指令的服从度最高
        "role": "user",
        # 说清约束和期望的输出形式
        "content": (
            # 第一句：告诉它发生了什么
            f"（系统）已达到 {max_tool_rounds} 轮工具调用上限，"
            # 第二句：明确禁止继续要工具，并要求给结论
            "不要再请求任何工具。请基于已经查到的信息给出结论，"
            # 第三句：要求如实交代缺口，避免它编造
            "并明确说明哪些信息尚未查到。"
        ),
    }
    # 用没绑工具的 base：模型根本没有工具可调，这是结构性保证
    answer = base.invoke([*state["messages"], *cancels, note])
    # 取消回执和最终回复都写进历史，并标注终止原因
    return {
        # 顺序很重要：先回执，再回复
        "messages": [*cancels, answer],
        # 让调用方知道这次是被护栏拦下的
        "stopped_by": f"达到 {max_tool_rounds} 轮工具上限",
    }

# 手造一份「模型刚提了工具需求、但轮次已经用完」的状态，直接调这个节点看效果
state = {
    # 前两条模拟已经查到的信息，最后一条是没人应答的工具请求
    "messages": [
        {"role": "user", "content": "帮我查一下所有保温杯的库存和价格"},
        AIMessage(content="", tool_calls=[{"name": "get_stock", "args": {"sku": "A-100"}, "id": "m1", "type": "tool_call"}]),
    ],
    # 轮次已经到上限
    "tool_rounds": 1,
    # 终止原因待填
    "stopped_by": "",
}
# 直接调用降级节点（它是普通函数，不依赖图的上下文）
out = force_finish(state)
# 先看补出来的取消回执
print("取消回执:", out["messages"][0].content)
# 再看模型基于已有信息给出的收尾回复
print("最终回复:", out["messages"][-1].content[:60])
# 终止原因也一并写好了
print("终止原因:", out["stopped_by"])

运行输出（模型的措辞每次会有出入）：

取消回执: 已取消：达到 1 轮工具调用上限，本次未执行。
最终回复: 抱歉，刚才的查询被系统取消，实际上**没有成功获取到任何保温杯的库存或价格数据**。

因此，以下信息目前均**尚未查到
终止原因: 达到 1 轮工具上限

这段代码里有三个决定，每个都有理由：

一、补 cancels 不是可选的。 漏了直接 400，§7.2 详说。

二、用不绑工具的 base，而不是绑了工具的模型。 这比在提示词里写「不要调用工具」更可靠：模型根本没有工具可以调，这是结构性保证，不是概率性约束。第 19 章那句「能用结构保证的，就不要靠提示词约束」在这里又应验了一次。

三、note 用 user 角色而不是 system。 消息历史里通常已经有一条 system 提示了，再插一条 system 容易和原有指令打架，而模型对「最新的用户指令」服从度最高。另外注意 note 只传给这一次模型调用（在 base.invoke 的参数里），没有写进返回的 messages：它是给模型的临时指令，不该污染对话历史。

7. 四个坑 #

这四个坑都是循环场景专属的：顺序图和分支图里它们要么不存在，要么无害。按危险程度排，第一个最值得记。

7.1. 计数器忘了 reducer，护栏就静默失效了 #

这是本章最坑的一个，因为代码看起来完全正确：

# operator.add 当 reducer
import operator
# Annotated 挂 reducer，TypedDict 定义状态结构
from typing import Annotated, TypedDict

# 撞上限时抛的异常
from langgraph.errors import GraphRecursionError
# 图的三件套
from langgraph.graph import END, START, StateGraph

# 一个「看起来对」的状态定义
class BadGuard(TypedDict):
    # 少了 Annotated[int, operator.add]
    steps: int
    # 终止原因
    stopped_by: str
    # 日志有 reducer，所以它是对的
    log: Annotated[list[str], operator.add]

# 循环体：意图是「每次 +1」
def work_bad(state: BadGuard) -> dict:
    # 返回 1 的意图是「加一」，但没有 reducer 就是「设为一」
    return {"steps": 1, "log": [f"steps 读到 {state['steps']}"]}

# 搭一张最简单的「计数到 3 就停」的图
builder = StateGraph(BadGuard)
# 循环体
builder.add_node("work", work_bad)
# 到上限后的收尾节点
builder.add_node("finish", lambda s: {"stopped_by": "达到步数上限"})
# 入口
builder.add_edge(START, "work")
# 退出条件：到 3 就走 finish，否则回到自己
builder.add_conditional_edges("work", lambda s: "finish" if s["steps"] >= 3 else "work", ["work", "finish"])
# 收尾即结束
builder.add_edge("finish", END)
# 编译
graph = builder.compile()
try:
    # limit 设小一点，别真跑一万圈
    graph.invoke({"steps": 0, "stopped_by": "", "log": []}, {"recursion_limit": 5})
# 退出条件永远不成立，只能靠保险丝
except GraphRecursionError as e:
    # 打印第一句话就够了
    print("死循环:", str(e).split(".")[0] + ".")

逻辑看着没毛病：每次 +1，到 3 就停。但 steps 没有 reducer，{"steps": 1} 是覆盖不是累加，它永远是 1，永远不满足 >= 3：

运行输出：

死循环: Recursion limit of 5 reached without hitting a stop condition.

加上 reducer 就对了，其它一个字都不用改：

# operator.add 当 reducer
import operator
# Annotated 挂 reducer，TypedDict 定义状态结构
from typing import Annotated, TypedDict

# 图的三件套
from langgraph.graph import END, START, StateGraph

# 唯一的改动：给 steps 挂上 operator.add
class GoodGuard(TypedDict):
    # 现在 {"steps": 1} 的语义变成「在原值上加一」
    steps: Annotated[int, operator.add]
    # 终止原因
    stopped_by: str
    # 日志照旧
    log: Annotated[list[str], operator.add]

# 循环体：顺手把「第几圈」记进日志
def work_good(state: GoodGuard) -> dict:
    # state["steps"] 是上一圈之后的累加值
    return {"steps": 1, "log": [f"work#{state['steps'] + 1}"]}

# 结构和上面完全一样
builder = StateGraph(GoodGuard)
# 循环体
builder.add_node("work", work_good)
# 收尾节点，顺手记一条日志
builder.add_node("finish", lambda s: {"stopped_by": "达到步数上限", "log": ["finish: 达到步数上限"]})
# 入口
builder.add_edge(START, "work")
# 同一个退出条件
builder.add_conditional_edges("work", lambda s: "finish" if s["steps"] >= 3 else "work", ["work", "finish"])
# 收尾即结束
builder.add_edge("finish", END)
# 编译
graph = builder.compile()
# 这次不用设 limit，它自己会停
print(graph.invoke({"steps": 0, "stopped_by": "", "log": []}))

运行输出：

{'steps': 3, 'stopped_by': '达到步数上限', 'log': ['work#1', 'work#2', 'work#3', 'finish: 达到步数上限']}

但在 Agent 环里，症状比死循环更隐蔽。 把 §8 实战的 Annotated[int, operator.add] 改回 int 再跑同一个问题：

跑完了, tool_rounds = 1  终止: 模型自然结束
实际轮次（带 tool_calls 的 AIMessage 条数）: 2
→ 汇报 1，实际 2，护栏已失效

没有死循环，没有报错，结果看起来完全正常。 因为 Agent 环有两个出口，模型这次自己停了，坏掉的计数器根本没机会暴露。但它已经埋了两颗雷：

• 汇报的 tool_rounds = 1 是错的，实际跑了 2 轮，你的监控数据从此不可信
• max_tool_rounds 这道护栏已经完全失效，只是还没遇到会失控的问题

死循环至少会立刻报错，静默失效则会一直潜伏到某天模型真的停不下来。而那一天通常是流量高峰，或者某个工具的返回值刚被人改成了「请再查一次」。

自检方法： 就是 §6.2 那个假模型，不接真模型空跑几圈，确认计数器真的在涨。或者拿一个已知需要 N 轮的问题去跑，核对汇报的轮次对不对。三行代码的事，能省掉一次账单事故。

顺便记住 reducer 字段的另一个性质，它在 §10 的练习里会考你：有 reducer 的字段，入口传的初始值也是「加」不是「设」。

import operator
from typing import Annotated, TypedDict

from langgraph.graph import END, START, StateGraph

class GoodGuard(TypedDict):
    steps: Annotated[int, operator.add]
    stopped_by: str
    log: Annotated[list[str], operator.add]

builder = StateGraph(GoodGuard)
builder.add_node("work", lambda s: {"steps": 1, "log": [f"work#{s['steps'] + 1}"]})
builder.add_node("finish", lambda s: {"stopped_by": "达到步数上限", "log": ["finish: 达到步数上限"]})
builder.add_edge(START, "work")
builder.add_conditional_edges("work", lambda s: "finish" if s["steps"] >= 3 else "work", ["work", "finish"])
builder.add_edge("finish", END)
graph = builder.compile()

# 同一张图，入口把 steps 给成 5 而不是 0
print("入口给 5:", graph.invoke({"steps": 5, "stopped_by": "", "log": []}))

运行输出：

入口给 5: {'steps': 6, 'stopped_by': '达到步数上限', 'log': ['work#6', 'finish: 达到步数上限']}

5 + 1 = 6，只跑了一圈就超限。所以在有 checkpointer 的多轮会话里，传 {"tool_rounds": 0} 不会把计数器清零，它的语义是「加 0」。

7.2. 在「想调工具」时切断，会留下未应答的 tool_calls #

§6.3 里补 cancels 那段，是被实际报错逼出来的。不补的话：

openai.BadRequestError: Error code: 400 - {'error': {'message': "An assistant message with
'tool_calls' must be followed by tool messages responding to each 'tool_call_id'.
(insufficient tool messages following tool_calls message)", 'type': 'invalid_request_error',
'param': None, 'code': 'invalid_request_error'}}

原因是轮次上限的判断发生在模型已经提了 tool_calls、工具还没跑的那个时刻。此时消息历史的最后一条是一个「提了要求但没人应答」的 AIMessage。OpenAI 兼容协议（DeepSeek、通义等都遵循）明确要求每个 tool_call_id 都得有对应的 ToolMessage，这个序列非法。

画出来是这样：

合法：  AIMessage(tool_calls=[a, b]) → ToolMessage(a) → ToolMessage(b) → 下一次调模型
非法：  AIMessage(tool_calls=[a, b]) →                                 下一次调模型  ← 400
补救：  AIMessage(tool_calls=[a, b]) → ToolMessage(a, 已取消) → ToolMessage(b, 已取消) → 下一次调模型

规则：一旦决定不执行模型请求的工具，就必须给每个 tool_call_id 补一条 ToolMessage 说明原因。 这条规则不只适用于轮次上限，人工审批拒绝（第 26 章）、权限校验拦截、超时放弃等场景同样适用。凡是「模型要调工具但你不给它调」的地方，都要补。

补完之后模型的回复很体面：

保温杯目前共有 A-100、A-101、A-102 三个 SKU，但受查询次数限制，库存和价格均未能获取到。

它准确说明了查到什么、没查到什么。这比抛 500 强太多。而且注意：回执内容会影响回复质量。写「已取消：达到 N 轮工具调用上限」，模型就知道是被限流了；如果只写一句「失败」，模型可能会猜成商品不存在，然后给用户一个错误结论。

（这段 400 报错是刻意复现出来的，不是本章示例代码的正常行为。§6.3 和 §8.2 的 force_finish 都补了回执，跑起来不会报错。想亲眼看到它，把构造 cancels 的那段注释掉即可，这也是 §10 的第 2 题。）

7.3. 没挂 checkpointer，环炸了就什么都不剩 #

§3.3 演示过机制，这里强调它在排查真实问题上的价值。给 §6 那个失控的 ping 环挂上 checkpointer，撞限之后能完整看到打转的轨迹：

from typing import Literal

from dotenv import load_dotenv
from langchain.chat_models import init_chat_model
from langchain_core.messages import AIMessage
from langchain_core.tools import tool
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.errors import GraphRecursionError
from langgraph.graph import END, START, MessagesState, StateGraph
from langgraph.prebuilt import ToolNode

load_dotenv(override=True)

@tool
def ping(x: str) -> str:
    """一个测试工具，永远回复「还需要再查一次」。"""
    print(f"   [tool] ping({x!r})")
    return "还需要再查一次，请再调用一次 ping 工具，参数随意。"

PING_TOOLS = [ping]
ping_model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0).bind_tools(PING_TOOLS)

def build_ping_loop(checkpointer=None):
    def call_ping_model(state: MessagesState) -> dict:
        return {"messages": [ping_model.invoke(state["messages"])]}

    def route(state: MessagesState) -> Literal["tools", "__end__"]:
        last = state["messages"][-1]
        return "tools" if isinstance(last, AIMessage) and last.tool_calls else END

    builder = StateGraph(MessagesState)
    builder.add_node("model", call_ping_model)
    builder.add_node("tools", ToolNode(PING_TOOLS))
    builder.add_edge(START, "model")
    builder.add_conditional_edges("model", route, ["tools", END])
    builder.add_edge("tools", "model")
    return builder.compile(checkpointer=checkpointer)

# 同一个 build_ping_loop，这次编译时挂 checkpointer
graph = build_ping_loop(InMemorySaver())
# 有 checkpointer 就必须给 thread_id
cfg = {"configurable": {"thread_id": "runaway-1"}}
# 这次问题的输入
inputs = {"messages": [{"role": "user", "content": "请用 ping 工具查一下 abc"}]}
try:
    # limit 给 4，保证在模型可能自己醒过来之前就撞上
    graph.invoke(inputs, {**cfg, "recursion_limit": 4})
# 异常照样抛，但现场留在 checkpointer 里
except GraphRecursionError:
    # 不用管异常，直接去看现场
    pass

# 撞限之后照样能取快照
snap = graph.get_state(cfg)
# 消息条数是最直观的进度指标
print("消息条数:", len(snap.values["messages"]))
# next 指着「本来要跑的下一个节点」
print("next:", snap.next)
# 逐条打印，看它在重复什么
for m in snap.values["messages"]:
    # 用类名区分类型
    kind = type(m).__name__
    # 带 tool_calls 的只打印工具名
    if kind == "AIMessage" and m.tool_calls:
        # 打印这一圈调了哪些工具
        print(f"   {kind}: 调用 {[c['name'] for c in m.tool_calls]}")
    else:
        # 其它打印截断正文
        print(f"   {kind}: {str(m.content)[:60]}")

   [tool] ping('abc')
   [tool] ping('abc')
消息条数: 5
next: ('model',)
   HumanMessage: 请用 ping 工具查一下 abc
   AIMessage: 调用 ['ping']
   ToolMessage: 还需要再查一次，请再调用一次 ping 工具，参数随意。
   AIMessage: 调用 ['ping']
   ToolMessage: 还需要再查一次，请再调用一次 ping 工具，参数随意。

（模型每次转几圈会有波动，你跑出来的消息条数可能不是 5。）

三条线索一次拿全：

• next: ('model',) 说明它不是跑完了，是工具刚跑完、正要回到模型时被拦下的
• 两条一模一样的 ToolMessage 说明是工具返回的内容在诱导模型反复调用
• 两条 AIMessage 调的是同一个工具，坐实了这是在打转而不是在推进

问题在工具的返回值设计，不在图。 没有 checkpointer 的话你只有一个 GraphRecursionError，以上三条线索一条都拿不到。

（这里 limit 只给了 4，所以轨迹里只有两圈。放宽到 8 会看到更多圈，但也可能碰上 §6 提到的「模型自己醒过来」，那时快照里的 next 会是空元组 ()，最后一条 AIMessage 是正常回复。next 是不是空的，正是区分「被拦下」和「自己停了」的判据，第 22 章 §6.2 也用过这一招。）

排查循环问题的另一个趁手工具是 stream，好处是不用等撞限、边跑边看：

from typing import Literal

from dotenv import load_dotenv
from langchain.chat_models import init_chat_model
from langchain_core.messages import AIMessage
from langchain_core.tools import tool
from langgraph.errors import GraphRecursionError
from langgraph.graph import END, START, MessagesState, StateGraph
from langgraph.prebuilt import ToolNode

load_dotenv(override=True)

@tool
def ping(x: str) -> str:
    """一个测试工具，永远回复「还需要再查一次」。"""
    return "还需要再查一次，请再调用一次 ping 工具，参数随意。"

PING_TOOLS = [ping]
ping_model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0).bind_tools(PING_TOOLS)

def build_ping_loop(checkpointer=None):
    def call_ping_model(state: MessagesState) -> dict:
        return {"messages": [ping_model.invoke(state["messages"])]}

    def route(state: MessagesState) -> Literal["tools", "__end__"]:
        last = state["messages"][-1]
        return "tools" if isinstance(last, AIMessage) and last.tool_calls else END

    builder = StateGraph(MessagesState)
    builder.add_node("model", call_ping_model)
    builder.add_node("tools", ToolNode(PING_TOOLS))
    builder.add_edge(START, "model")
    builder.add_conditional_edges("model", route, ["tools", END])
    builder.add_edge("tools", "model")
    return builder.compile(checkpointer=checkpointer)

graph = build_ping_loop()
inputs = {"messages": [{"role": "user", "content": "请用 ping 工具查一下 abc"}]}

# updates 模式实时打印每个超步
try:
    # 故意用很小的 limit，几圈就停
    for chunk in graph.stream(inputs, {"recursion_limit": 4}, stream_mode="updates"):
        # 遍历这一超步涉及的节点
        for node, upd in chunk.items():
            # 取该节点新增的最后一条消息
            msg = upd["messages"][-1]
            # 带 tool_calls 打印工具名，否则打印正文
            desc = f"调用 {[c['name'] for c in msg.tool_calls]}" if getattr(msg, "tool_calls", None) else str(msg.content)[:50]
            # 打印节点和内容
            print(f"   [{node}] {type(msg).__name__}: {desc}")
# 撞限时 stream 也会抛异常，但前面已经打印出来的东西还在
except GraphRecursionError:
    # 提示一下是被限制打断的
    print("   （撞上 limit=4）")

运行输出：

   [model] AIMessage: 调用 ['ping']
   [tools] ToolMessage: 还需要再查一次，请再调用一次 ping 工具，参数随意。
   [model] AIMessage: 调用 ['ping']
   [tools] ToolMessage: 还需要再查一次，请再调用一次 ping 工具，参数随意。
   （撞上 limit=4）

两者的分工：stream 用于开发时观察，checkpointer 用于生产事后复盘。 生产环境里你没法「重跑一次看看」，只能靠 checkpoint 里存下来的现场。

7.4. 轮次 ≠ 工具调用次数 ≠ 超步数 #

三个数很容易混，设上限时搞错就会限得太松或太紧。用 stream 数一遍 §8 实战的实际运行：

import operator
from typing import Annotated, Literal

from dotenv import load_dotenv
from langchain.chat_models import init_chat_model
from langchain_core.messages import AIMessage, ToolMessage
from langchain_core.tools import tool
from langgraph.graph import END, START, MessagesState, StateGraph
from langgraph.prebuilt import ToolNode

load_dotenv(override=True)

CATALOG = {
    "A-100": {"name": "保温杯 500ml", "stock": 12, "price": 39},
    "A-101": {"name": "保温杯 750ml", "stock": 0, "price": 55},
    "A-102": {"name": "保温杯 1L", "stock": 7, "price": 68},
    "B-200": {"name": "滤芯 三个月装", "stock": 30, "price": 58},
}

@tool
def search_sku(keyword: str) -> str:
    """按关键词搜索商品，返回匹配的 SKU 编号列表。"""
    hits = [k for k, v in CATALOG.items() if keyword in v["name"]]
    return "、".join(hits) if hits else "没有匹配的商品"

@tool
def get_stock(sku: str) -> str:
    """查询单个 SKU 的库存数量。一次只能查一个。"""
    item = CATALOG.get(sku)
    return f"{sku} 库存 {item['stock']} 件" if item else f"{sku} 查无此货"

@tool
def get_price(sku: str) -> str:
    """查询单个 SKU 的单价（元）。一次只能查一个。"""
    item = CATALOG.get(sku)
    return f"{sku} 单价 {item['price']} 元" if item else f"{sku} 查无此货"

TOOLS = [search_sku, get_stock, get_price]
TOOL_NODE = ToolNode(TOOLS, handle_tool_errors="工具暂时不可用，请换个思路或如实说明查不到。")
base = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
model_with_tools = base.bind_tools(TOOLS)

class AgentState(MessagesState):
    tool_rounds: Annotated[int, operator.add]
    stopped_by: str

def build_graph(max_tool_rounds: int = 4):
    def call_model(state: AgentState) -> dict:
        return {"messages": [model_with_tools.invoke(state["messages"])]}

    def run_tools(state: AgentState) -> dict:
        return {**TOOL_NODE.invoke(state), "tool_rounds": 1}

    def force_finish(state: AgentState) -> dict:
        last = state["messages"][-1]
        cancels = [
            ToolMessage(
                content=f"已取消：达到 {max_tool_rounds} 轮工具调用上限，本次未执行。",
                tool_call_id=c["id"], name=c["name"], status="error",
            )
            for c in last.tool_calls
        ]
        note = {"role": "user", "content": (
            f"（系统）已达到 {max_tool_rounds} 轮工具调用上限，不要再请求任何工具。"
            "请基于已经查到的信息给出结论，并明确说明哪些信息尚未查到。"
        )}
        answer = base.invoke([*state["messages"], *cancels, note])
        return {"messages": [*cancels, answer], "stopped_by": f"达到 {max_tool_rounds} 轮工具上限"}

    def should_continue(state: AgentState) -> Literal["tools", "force_finish", "done"]:
        last = state["messages"][-1]
        if not (isinstance(last, AIMessage) and last.tool_calls):
            return "done"
        if state["tool_rounds"] >= max_tool_rounds:
            return "force_finish"
        return "tools"

    builder = StateGraph(AgentState)
    builder.add_node("model", call_model)
    builder.add_node("tools", run_tools)
    builder.add_node("force_finish", force_finish)
    builder.add_node("done", lambda s: {"stopped_by": "模型自然结束"})
    builder.add_edge(START, "model")
    builder.add_conditional_edges("model", should_continue, ["tools", "force_finish", "done"])
    builder.add_edge("tools", "model")
    builder.add_edge("force_finish", END)
    builder.add_edge("done", END)
    return builder.compile(name="controlled-agent")

def new_input(question: str) -> dict:
    return {"messages": [{"role": "user", "content": question}], "tool_rounds": 0, "stopped_by": ""}

# 那个必须多轮才能答完的问题
Q = "帮我查一下所有保温杯的库存和价格，最后汇总成一句话"
# 编译一张上限 4 轮的图
graph = build_graph(4)
# 每个超步一条记录，键就是节点名
steps = [list(chunk.keys()) for chunk in graph.stream(new_input(Q), {"recursion_limit": 20}, stream_mode="updates")]
# 打印超步序列和总数
print(steps, len(steps))

运行输出（模型分几轮查完会有波动）：

[['model'], ['tools'], ['model'], ['tools'], ['model'], ['done']] 6

同一次运行的三个数：

口径	值	说明
工具轮次	2	tools 节点执行了几次，也就是 tool_rounds
工具调用次数	7	1 次 search_sku + 3 次 get_stock + 3 次 get_price
超步数	6	model → tools → model → tools → model → done

差得最多的是前两个：max_tool_rounds=4 允许的实际工具调用可能是几十次，因为一轮里模型可以并行提任意多个 tool_calls。如果你的工具单次成本高（每次都要查外部 API、每次都要付费），只限轮次不够，还得在工具层面限并发或限总数。

设 recursion_limit 则要按超步算。实测这张图的最小可用值正好是 §3.2 的 N+1：

limit=4: GraphRecursionError
limit=5: GraphRecursionError
limit=6: GraphRecursionError
limit=7: OK  轮次=2 终止=模型自然结束

6 个超步，最小 limit 是 7。实战里 limit_for(4) 给的是 13，留了足够余量。余量是必须的，因为模型可能这次两轮、下次三轮。别把 limit 卡在刚好能跑通的值上，否则一次模型的正常波动就变成线上 500。

别去套公式，用 stream 数一遍最靠谱，尤其是图里有并行分支的时候。

8. 实战：受控 Agent 环 #

把前面讲过的内容收成一份能上手改的完整脚本。

8.1. 设计 #

START → model → should_continue
                 ├─ tools（有 tool_calls 且未超限）→ 回到 model
                 ├─ force_finish（有 tool_calls 但已超限）→ END
                 └─ done（无 tool_calls，自然结束）→ END

比 §4 多了两个出口节点，每个都对应本章的一个知识点：

组件	对应小节	解决什么问题
tools → model 固定边	§2、§4	环本身
tool_rounds + operator.add	§6.2、§7.1	计数器不能忘 reducer
force_finish 补 cancels	§6.3、§7.2	切断循环时不能留未应答的 tool_calls
force_finish 用不绑工具的 base	§6.3	结构性保证优于提示词约束
done 节点写 stopped_by	§6.3	调用方永远知道这次是怎么停的
limit_for() 算 recursion_limit	§3.2、§6.1	第三层保险，且不能比第二层先触发

工具设计上故意做了两个限制，好让环真的转起来：search_sku 只返回 SKU 列表不带详情，get_stock / get_price 一次只能查一个。模型必须先搜再查，至少两轮。

8.2. controlled_agent.py #

"""受控 Agent 环：能干活、能自己停、停不下来也能体面收场。"""

# 让类型注解延迟求值，避免前向引用问题
from __future__ import annotations

# operator.add 给轮次计数器当 reducer
import operator
# lru_cache 让模型客户端只构造一次
from functools import lru_cache
# Annotated 挂 reducer，Literal 声明路由出口
from typing import Annotated, Literal

# 从 .env 读 API Key
from dotenv import load_dotenv
# 统一的模型入口
from langchain.chat_models import init_chat_model
# AIMessage 用来判断有没有 tool_calls，ToolMessage 用来补「已取消」回执
from langchain_core.messages import AIMessage, ToolMessage
# @tool 装饰器把普通函数变成模型可调用的工具
from langchain_core.tools import tool
# MessagesState 自带 add_messages reducer，是消息累积的前提
from langgraph.graph import END, START, MessagesState, StateGraph
# ToolNode 负责批量执行 tool_calls 并生成 ToolMessage
from langgraph.prebuilt import ToolNode

# 加载环境变量，override=True 让 .env 覆盖同名的系统变量
load_dotenv(override=True)

# 假数据：一个小商品目录，工具从这里取数
CATALOG = {
    # 有货的保温杯
    "A-100": {"name": "保温杯 500ml", "stock": 12, "price": 39},
    # 缺货的保温杯，用来验证模型会不会如实转述
    "A-101": {"name": "保温杯 750ml", "stock": 0, "price": 55},
    # 第三个保温杯，凑够「必须多轮查询」的量
    "A-102": {"name": "保温杯 1L", "stock": 7, "price": 68},
    # 一个不相关的商品，用来验证搜索是按关键词过滤的
    "B-200": {"name": "滤芯 三个月装", "stock": 30, "price": 58},
}

# 工具一：只返回 SKU 列表，故意不带库存和价格
@tool
def search_sku(keyword: str) -> str:
    """按关键词搜索商品，返回匹配的 SKU 编号列表。"""
    # 按名字里是否包含关键词过滤
    hits = [k for k, v in CATALOG.items() if keyword in v["name"]]
    # 打印一行，方便在输出里看到工具真的被调用了
    print(f"   [tool] search_sku({keyword!r}) -> {hits}")
    # 只给编号，逼模型再调一轮查详情
    return "、".join(hits) if hits else "没有匹配的商品"

# 工具二：一次只查一个 SKU 的库存
@tool
def get_stock(sku: str) -> str:
    """查询单个 SKU 的库存数量。一次只能查一个。"""
    # 打印调用痕迹
    print(f"   [tool] get_stock({sku!r})")
    # 查目录
    item = CATALOG.get(sku)
    # 查到就报库存，查不到就如实说
    return f"{sku} 库存 {item['stock']} 件" if item else f"{sku} 查无此货"

# 工具三：一次只查一个 SKU 的单价
@tool
def get_price(sku: str) -> str:
    """查询单个 SKU 的单价（元）。一次只能查一个。"""
    # 打印调用痕迹
    print(f"   [tool] get_price({sku!r})")
    # 查目录
    item = CATALOG.get(sku)
    # 查到就报单价，查不到就如实说
    return f"{sku} 单价 {item['price']} 元" if item else f"{sku} 查无此货"

# 工具清单，绑模型和建 ToolNode 都用它
TOOLS = [search_sku, get_stock, get_price]

# ToolNode 只建一次：它是无状态的，每次调用都新建纯属浪费
TOOL_NODE = ToolNode(TOOLS, handle_tool_errors="工具暂时不可用，请换个思路或如实说明查不到。")

# 状态：在 MessagesState 基础上加两个控制槽位
class AgentState(MessagesState):
    """在 MessagesState 基础上加两个控制槽位。"""

    # 必须有 reducer，否则永远是 1，护栏静默失效（§7.1）
    tool_rounds: Annotated[int, operator.add]
    # 记录这次是怎么停的，普通字段（后写覆盖先写）
    stopped_by: str

# 模型客户端只构造一次
@lru_cache(maxsize=1)
def get_models():
    """返回 (不绑工具的模型, 绑了工具的模型)。"""
    # 温度 0，让多轮工具调用的行为尽量可复现
    base = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
    # 绑工具的版本给 model 节点用，不绑的版本给降级节点用
    return base, base.bind_tools(TOOLS)

# 第三层保险的建议值：轮次上限换算成超步预算
def limit_for(max_tool_rounds: int) -> int:
    """recursion_limit 建议值：一轮 = model + tools 两个超步，再留点余量。"""
    # 2 倍轮次覆盖 model/tools 交替，+5 覆盖首尾和降级节点
    return 2 * max_tool_rounds + 5

# 组装图
def build_graph(max_tool_rounds: int = 4):
    """max_tool_rounds 是「模型调工具」的最大轮次，不是工具调用总次数。"""
    # 取出两个模型客户端
    base, model_with_tools = get_models()

    # 节点一：调模型
    def call_model(state: AgentState) -> dict:
        # 把完整历史交给模型，只把新回复追加进状态
        return {"messages": [model_with_tools.invoke(state["messages"])]}

    # 节点二：跑工具，顺便把轮次 +1
    def run_tools(state: AgentState) -> dict:
        """包一层 ToolNode，顺便把轮次 +1。"""
        # ToolNode 返回 {"messages": [ToolMessage, ...]}
        result = TOOL_NODE.invoke(state)
        # 有 operator.add，这个 1 是「加一」不是「设为一」
        return {**result, "tool_rounds": 1}

    # 节点三：超限降级
    def force_finish(state: AgentState) -> dict:
        """超限降级：把工具收走，逼模型用已有信息作答。"""
        # 最后一条一定是「提了 tool_calls」的 AIMessage
        last = state["messages"][-1]
        # 关键：这些 tool_calls 没人应答，
        # 直接发给模型会被 API 判为非法消息序列，必须先补「已取消」回执（§7.2）
        cancels = [
            # 每个 tool_call_id 都要有一条对应的 ToolMessage
            ToolMessage(
                # 如实告诉模型为什么没执行
                content=f"已取消：达到 {max_tool_rounds} 轮工具调用上限，本次未执行。",
                # 必须和 AIMessage 里的 id 一一对应
                tool_call_id=c["id"],
                # 带上工具名，便于模型对齐
                name=c["name"],
                # 标成 error，语义上这是一次失败的调用
                status="error",
            )
            # 遍历这一轮所有被取消的调用
            for c in last.tool_calls
        ]
        # 再补一条用户视角的说明，明确「不要再要工具了」
        note = {
            # 用 user 角色，模型对用户指令的服从度最高
            "role": "user",
            # 说清约束和期望的输出形式
            "content": (
                # 第一句：告诉它发生了什么
                f"（系统）已达到 {max_tool_rounds} 轮工具调用上限，"
                # 第二句：明确禁止继续要工具，并要求给结论
                "不要再请求任何工具。请基于已经查到的信息给出结论，"
                # 第三句：要求如实交代缺口，避免它编造
                "并明确说明哪些信息尚未查到。"
            ),
        }
        # 用没绑工具的 base：模型根本没有工具可调，这是结构性保证
        answer = base.invoke([*state["messages"], *cancels, note])
        # 取消回执和最终回复都要写进历史，并标注终止原因
        return {
            # 顺序很重要：先回执，再回复
            "messages": [*cancels, answer],
            # 让调用方知道这次是被护栏拦下的
            "stopped_by": f"达到 {max_tool_rounds} 轮工具上限",
        }

    # 节点四：自然结束时打个标记
    def mark_natural_end(state: AgentState) -> dict:
        # 只写 stopped_by，不动消息历史
        return {"stopped_by": "模型自然结束"}

    # 路由函数：三个出口，对应三种终止路径中的前两种
    def should_continue(state: AgentState) -> Literal["tools", "force_finish", "done"]:
        # 看模型最新那条回复
        last = state["messages"][-1]
        # 保险一：模型没提工具需求，自然结束
        if not (isinstance(last, AIMessage) and last.tool_calls):
            # 走标记节点
            return "done"
        # 保险二：还想调工具，但轮次用完了，走降级出口
        if state["tool_rounds"] >= max_tool_rounds:
            # 走降级节点
            return "force_finish"
        # 否则正常执行工具，回到 model 继续转
        return "tools"

    # 开始搭图
    builder = StateGraph(AgentState)
    # 注册调模型节点
    builder.add_node("model", call_model)
    # 注册工具节点（包了一层的版本）
    builder.add_node("tools", run_tools)
    # 注册降级节点
    builder.add_node("force_finish", force_finish)
    # 注册自然结束标记节点
    builder.add_node("done", mark_natural_end)
    # 入口边
    builder.add_edge(START, "model")
    # 三个出口全部显式声明（第 23 章 §4）
    builder.add_conditional_edges("model", should_continue, ["tools", "force_finish", "done"])
    # 这条边构成环：工具跑完永远回到模型
    builder.add_edge("tools", "model")
    # 降级完就结束
    builder.add_edge("force_finish", END)
    # 自然结束也结束
    builder.add_edge("done", END)
    # 起个名字，LangSmith 里好认
    return builder.compile(name="controlled-agent")

# 造一份完整初始状态
def new_input(question: str) -> dict:
    """状态是 TypedDict，没有默认值，入口必须给全（第 20 章的坑）。"""
    # 三个字段一个都不能少
    return {
        # 用户问题作为第一条消息
        "messages": [{"role": "user", "content": question}],
        # 轮次从 0 开始（注意：这是「加 0」，不是「设为 0」，§7.1）
        "tool_rounds": 0,
        # 终止原因待填
        "stopped_by": "",
    }

# 把一次运行的结果打印成人能看的样子
def report(title: str, out: dict) -> None:
    """打印一次运行的轮次、终止原因、消息轨迹和最终回复。"""
    # 场景标题
    print(f"\n--- {title} ---")
    # 两个控制字段：这是「可观测」的核心
    print(f"工具轮次: {out['tool_rounds']}   终止原因: {out['stopped_by']}")
    # 消息轨迹标题
    print("消息轨迹:")
    # 逐条打印消息
    for m in out["messages"]:
        # 用类名区分消息类型
        kind = type(m).__name__
        # 带 tool_calls 的 AIMessage 只打印调了哪些工具
        if kind == "AIMessage" and m.tool_calls:
            # 打印工具名列表
            print(f"   AIMessage -> 调用 {[c['name'] for c in m.tool_calls]}")
        else:
            # 其它消息打印截断后的正文
            print(f"   {kind}: {str(m.content)[:60]}")
    # 最终回复标题
    print("最终回复:")
    # 最后一条消息就是给用户的答案
    print("  ", out["messages"][-1].content)

# 直接运行本文件时的演示入口
if __name__ == "__main__":
    # 一个必须多轮才能答完的问题：先搜 SKU，再逐个查库存和价格
    Q = "帮我查一下所有保温杯的库存和价格，最后汇总成一句话"

    # 先把图纸打出来，确认三个出口都在
    print(build_graph().get_graph().draw_mermaid())

    # 场景一：上限够用，模型自然停；第三层保险按 limit_for 给
    report(
        # 场景标题
        "上限 4 轮：够用",
        # 4 轮足够查完，模型会自己停
        build_graph(max_tool_rounds=4).invoke(new_input(Q), {"recursion_limit": limit_for(4)}),
    )
    # 场景二：上限压到 1 轮，强制触发降级
    report(
        # 场景标题
        "上限 1 轮：触发降级",
        # 1 轮不够，第二层保险会接管
        build_graph(max_tool_rounds=1).invoke(new_input(Q), {"recursion_limit": limit_for(1)}),
    )
    # 场景三：闲聊，环转零圈
    report(
        # 场景标题
        "不需要工具",
        # 一句闲聊，tools 节点根本不会执行
        build_graph().invoke(new_input("你好，你能做什么？"), {"recursion_limit": limit_for(4)}),
    )

8.3. 跑起来 #

图的形状：

graph TD;
    __start__([<p>__start__</p>]):::first
    model(model)
    tools(tools)
    force_finish(force_finish)
    done(done)
    __end__([<p>__end__</p>]):::last
    __start__ --> model;
    model -.-> done;
    model -.-> force_finish;
    model -.-> tools;
    tools --> model;
    done --> __end__;
    force_finish --> __end__;

三条虚线从 model 出发（三个出口），一条实线 tools --> model 是环，两个出口节点各自连到 __end__。:::first / :::last 都在，说明没有孤立节点（第 23 章 §4.3）。

下面三段都是同一次真实运行的输出。模型的措辞、分几轮查完、一轮里提几个 tool_calls，每次都可能不同，你跑出来的数字和文案对不上是正常的。

场景一：上限够用，模型自然停。

   [tool] search_sku('保温杯') -> ['A-100', 'A-101', 'A-102']
   [tool] get_stock('A-100')
   [tool] get_price('A-100')
   [tool] get_stock('A-101')
   [tool] get_price('A-101')
   [tool] get_price('A-102')
   [tool] get_stock('A-102')

--- 上限 4 轮：够用 ---
工具轮次: 2   终止原因: 模型自然结束
消息轨迹:
   HumanMessage: 帮我查一下所有保温杯的库存和价格，最后汇总成一句话
   AIMessage -> 调用 ['search_sku']
   ToolMessage: A-100、A-101、A-102
   AIMessage -> 调用 ['get_stock', 'get_price', 'get_stock', 'get_price', 'get_stock', 'get_price']
   ToolMessage: A-100 库存 12 件
   ToolMessage: A-100 单价 39 元
   ToolMessage: A-101 库存 0 件
   ToolMessage: A-101 单价 55 元
   ToolMessage: A-102 库存 7 件
   ToolMessage: A-102 单价 68 元
   AIMessage: 已为您查询到所有保温杯的信息，汇总如下：
最终回复:
   已为您查询到所有保温杯的信息，汇总如下：

**所有保温杯的库存和价格：A-100 库存12件、单价39元；A-101 库存0件（缺货）、单价55元；A-102 库存7件、单价68元。**

两轮，七次工具调用，§7.4 那三个数的差别在这里看得很清楚。注意第二轮模型一口气提了六个 tool_calls，ToolNode 并行跑完，所以工具自己打印的顺序和 tool_calls 的顺序不一致（上面 get_price('A-102') 跑在了 get_stock('A-102') 前面）。并行执行意味着你不能依赖工具之间的执行顺序，如果两个工具有先后依赖，得让模型分两轮提。

场景二：上限 1 轮，触发降级。

--- 上限 1 轮：触发降级 ---
工具轮次: 1   终止原因: 达到 1 轮工具上限
消息轨迹:
   HumanMessage: 帮我查一下所有保温杯的库存和价格，最后汇总成一句话
   AIMessage -> 调用 ['search_sku']
   ToolMessage: A-100、A-101、A-102
   AIMessage -> 调用 ['get_stock', 'get_stock', 'get_stock', 'get_price', 'get_price', 'get_price']
   ToolMessage: 已取消：达到 1 轮工具调用上限，本次未执行。
   ToolMessage: 已取消：达到 1 轮工具调用上限，本次未执行。
   ToolMessage: 已取消：达到 1 轮工具调用上限，本次未执行。
   ToolMessage: 已取消：达到 1 轮工具调用上限，本次未执行。
   ToolMessage: 已取消：达到 1 轮工具调用上限，本次未执行。
   ToolMessage: 已取消：达到 1 轮工具调用上限，本次未执行。
   AIMessage: 保温杯目前共有 A-100、A-101、A-102 三个 SKU，但受查询次数限制，库存和
最终回复:
   保温杯目前共有 A-100、A-101、A-102 三个 SKU，但受查询次数限制，库存和价格均未能获取到。

这就是「体面收场」：没有异常、没有 500，模型准确交代了查到什么、没查到什么。六条取消回执既满足了协议要求（§7.2），也如实告诉了模型发生了什么。

对比三种「停」的用户体验，第二层保险的价值就很清楚：

做法	用户看到什么
什么都不做	等很久，然后 500（GraphRecursionError）
直接 return END	一句话都没有（最后一条消息是 tool_calls，没有正文）
走 force_finish	一段说明查到什么、没查到什么的完整回复

场景三：不需要工具。

--- 不需要工具 ---
工具轮次: 0   终止原因: 模型自然结束
消息轨迹:
   HumanMessage: 你好，你能做什么？
   AIMessage: 你好！我是你的商品查询助手。我可以帮你完成以下操作：
最终回复:
   你好！我是你的商品查询助手。我可以帮你完成以下操作：
   1. **搜索商品**：通过关键词搜索，找到相关的 SKU 编号。
   ...

零轮，tools 节点没执行，force_finish 也没执行。同一张图覆盖了从「闲聊」到「多轮查询」到「被限流降级」三种差别很大的情况。

8.4. 验收清单 #

• 三种终止路径都跑通，stopped_by 都能准确反映实际情况
• tool_rounds 真的在涨：用 stream 数出来的轮次要和它对得上（把 Annotated[int, operator.add] 改回 int 复现 §7.1）
• 降级路径不报 400（把 cancels 那段注释掉复现 §7.2）
• draw_mermaid() 画出三个出口，:::first / :::last 都在，没有孤立节点
• limit_for() 给的 recursion_limit 明显大于实测最小值（用 stream 数超步核对）
• 挂上 InMemorySaver 后，撞 recursion_limit 还能 get_state 看到现场

9. 实用约定与坑 #

约定

1. 任何环都先想退出条件，再写边。 先写 should_continue 再写 add_edge，顺序反了容易漏。
2. 计数器字段一律 Annotated[int, operator.add]。 这是循环里最容易犯的错，而且不报错。
3. 带环的图一律挂 checkpointer。 排查成本从「没有线索」降到「一眼看穿」，还顺便获得续跑能力。
4. 显式设一个小的 recursion_limit。 建议 2 × 最大轮次 + 5，默认的 10007 等于没设。
5. 上限触发走降级节点，不要让它抛异常。 用户要的是一个回答，哪怕是「没查到」。
6. 降级时用不绑工具的模型，别指望提示词里写「不要调用工具」。
7. handle_tool_errors 用字符串或函数，别用 True，它会把 repr(e) 塞进消息历史。
8. ToolNode 建在节点函数外面。 它无状态，反复构造是白花开销。
9. 上线前用假模型空跑一遍循环，确认计数器在涨、圈数符合预期（§6.2）。
10. 给「自然结束」也留一个标记节点。 调用方不该靠猜来判断这次是怎么停的。

会当场报错的坑

坑	报错	见
环没有出口	GraphRecursionError（且丢失全部现场）	§3.1、§3.3
recursion_limit 按节点次数估算	GraphRecursionError（它数的是超步，N 步要 N+1）	§3.2
切断循环时没补 ToolMessage	模型 API 400「insufficient tool messages」	§7.2
tools_condition 配列表 path_map 且节点改了名	KeyError: 'tools'	§5.1
让 tools 节点收到没有 tool_calls 的历史	ValueError: No AIMessage found in input	§5.2
在图外面调 ToolNode.invoke	ValueError: Missing required config key	§5.2
工具内部抛业务异常且没配 handle_tool_errors	原始异常穿透，整张图中断	§5.3

不报错但结果不对的坑

坑	症状	见
计数器没配 reducer	单出口循环会死循环；Agent 环则不报错、结果看着正常，护栏静默失效。本章头号坑	§7.1
有 reducer 的字段，入口传 0 想清零	那是「加 0」，多轮会话里计数器根本没被重置	§7.1、§10
以为轮次 = 工具调用次数	一轮可以并行跑几十个工具，成本估算差一个数量级	§7.4
把 recursion_limit 当业务开关	它抛异常且丢现场，只能当保险丝	§3.3、§6.1
recursion_limit 设得比轮次上限还紧	第三层先触发，降级节点永远跑不到	§6.1
状态用普通 list 而不是带 add_messages 的	每圈覆盖历史，模型永远看不到工具结果	§4.2
工具返回值诱导模型反复调用	环停不下来。问题在工具的返回值设计，不在图	§6、§7.3
在 model 和路由之间插了别的节点	tools_condition 只看最后一条，静默走 __end__	§5.1
循环里状态无限膨胀	转十圈消息历史就爆了。配合第 11 章的裁剪 / 摘要或第 21 章的 RemoveMessage	—
在路由函数里数轮次（而不是在节点里累加）	路由函数会被重放，第 23 章 §3.4 实测过它的执行次数不确定	§7.1

容易误解为 bug 的正常行为

1. 同一个问题两次运行的轮次不一样。 模型每次决定提几个 tool_calls 都可能不同，temperature=0 也只是减少波动。
2. 工具打印的顺序和 tool_calls 的顺序不一致。 ToolNode 是并行执行的（§8.3）。
3. values 模式比节点数多一条输出。 第一条是初始状态（第 22 章 §4.2）。
4. 扇出的图里节点执行次数大于超步数。 并行的节点算一个超步（§3.2）。

10. 练习 #

1. 复现头号坑。 把 controlled_agent.py 里的 tool_rounds: Annotated[int, operator.add] 改成 tool_rounds: int 再跑。你会发现它照样跑完、还不报错。先说清楚为什么，再说清楚什么情况下这个 bug 会变成事故。
2. 复现 400。 注释掉 force_finish 里构造 cancels 的那段，用 max_tool_rounds=1 跑，看完整的错误信息。然后想一想：如果只补一条 ToolMessage（模型提了六个），还会不会报错？
3. 改成按工具调用次数限流。 把 run_tools 里的 "tool_rounds": 1 改成 len(state["messages"][-1].tool_calls)，对比同一个问题下两种限流方式的行为差异：同样的上限值，哪种更容易触发降级？
4. 加第四层保险：超时。 在状态里记录起始时间戳，should_continue 里判断超过 N 秒就走 force_finish。想一想为什么这个判断不能在路由函数里做副作用（提示：第 23 章 §3.4 实测过路由函数会被重放）。
5. 换成 tools_condition。 用 tools_condition 重写 §4 的 should_continue，然后想清楚：为什么 §8 的实战不能直接用它？（提示：它只有两个出口，而且没有任何轮次概念。）
6. 给 ToolNode 加重试。 用 §5.4 的 wrap_tool_call，让 get_stock 在抛异常时自动重试两次。再想一想：重试次数应该算进 tool_rounds 吗？
7. 挂上 checkpointer 做多轮对话（强烈建议做）。 给 build_graph 加 checkpointer 参数，设 max_tool_rounds=2，用同一个 thread_id 连问三个问题：

问：A-100 有货吗       tool_rounds=1  终止=模型自然结束
问：那 A-102 呢        tool_rounds=2  终止=模型自然结束
问：B-200 的价格呢     tool_rounds=2  终止=达到 2 轮工具上限

第三轮还没开口就被降级了：tool_rounds 存在 checkpoint 里，跨轮累加，第三轮开始时预算已经耗尽。想清楚三个问题：这是 bug 还是 feature？new_input 里的 "tool_rounds": 0 为什么没起作用？如果要改成「每轮对话重置」，改哪里？

（第二问的答案在 §7.1 末尾：有 reducer 的字段，入口传的值也走 reducer，0 的语义是「加 0」。至于怎么重置，一个能跑通的土办法是传 -当前值 把它抵消掉，实测三轮都是 tool_rounds=1；更干净的做法是把「预算」和「已用」拆成两个字段，或者干脆不给这个字段配 reducer、改在节点里自己读写。）

11. 本章小结 #

• 循环不需要专门的 API，一条指回去的边就是环。第 23 章的条件边直接复用，只是出口里多了个已经跑过的节点。
• recursion_limit 数的是超步，实测六种拓扑都是「超步数 + 1」；有扇出时节点执行次数会明显多于超步数。它默认 10007，是保险丝不是开关，真正的退出条件必须自己写。
• 撞上限会丢掉全部现场，除非挂了 checkpointer。挂上之后不但能看到打转的轨迹，还能改对条件后从断点续跑。
• model ↔ tools 环就是 create_agent 的核心，三十行手写一遍，第 19 章「create_agent 是预制 LangGraph」那句话就彻底落地了。官方版唯一的结构差别是 tools → model 也是条件边。
• ToolNode 的错误处理有个反直觉的分界：参数错误和未知工具（ToolInvocationError）默认回灌给模型自我纠正，业务异常默认直接中断图。这个设计是合理的，因为前者模型能改，后者改不了。要调整就用 handle_tool_errors，生产环境优先用字符串或函数，需要分类处理时传异常类型元组。
• 三重终止保险：模型自然停（正常）、轮次上限（业务护栏，走降级）、recursion_limit（代码写错时的保险丝）。三者的区别不是数值，是结果的性质；而且第三层必须比第二层松，否则降级节点永远跑不到。
• 优雅降级的关键是用不绑工具的模型。这是结构性保证，比在提示词里写「不要调用工具」可靠得多。
• 切断循环时必须给每个 tool_call_id 补 ToolMessage，否则 API 直接 400。回执的文案还会影响模型的最终回复质量。这条规则在人工审批拒绝、权限拦截等场景同样适用，第 26 章会再遇到。
• 循环里最容易犯的错是计数器忘配 reducer，代码看着完全正确，实际值永远是 1。在 Agent 环里它不会报错也不会死循环，只是让护栏静默失效、监控数据失真，比死循环更值得警惕。上线前用假模型空跑一遍数圈数，几秒钟省掉一次账单事故。

到这里，图的三大能力就齐了：顺序（第 20 章）、分支（第 23 章）、循环（本章）。理论上，任何控制流都能用图表达。

下一章处理一个更实际的问题：图里既有确定性的 Python 节点，又想复用前面十几章积累的 create_agent，两者怎么并存。