AI 原生开放教材

外观

AI 工程从零到一（中文）源码研读·中文原文 ↗原件 ↓仓库 ↗

Agent 状态机——图、节点与检查点 ​

手写的 ReAct 循环是一个 while True。把同一个循环写成显式的图，你就能给它打 checkpoint、中断它、给它分支、在它里面时间旅行。agent 没变。变的是包在它外面的那套框架。

类型： Build 语言： Python 前置要求： 阶段 11 · 09（Function Calling）、阶段 11 · 14（Model Context Protocol） 预计时间： ~75 分钟

问题背景 ​

你交付一个 function-calling agent。它跑了三轮，然后出了岔子：模型试了一个返回 500 的工具、用户在任务中途改了主意，或者 agent 没等人签字就决定给一笔订单退款。while True: 循环没有钩子。你没法暂停它，没法回退它，也没法分叉去看"要是模型当时挑了另一个工具会怎样"。你把它发到 demo 之外的那一刻，agent 就成了一个要么成功要么失败的黑箱。

一旦你看清了，下一步就显而易见。agent 本来就是一个状态机——system prompt 加消息历史加待执行的 tool call 加下一个动作。把这个状态机显式化：为"模型思考""一个工具运行""一个人批准"设节点，为它们之间的条件转移设边。一旦图显式了，框架就白白得到四样东西：checkpoint（在步骤之间保存状态）、中断（为人暂停）、流式（流式吐 token 和中间事件）和时间旅行（回退到先前的状态、试一个不同的分支）。

LangGraph 是这层抽象的参考实现。它不是 LangChain 那种意义上的 agent 框架（"给你一个 AgentExecutor，自求多福"）。它是一个图运行时，有一等的状态、一等的持久化和一等的中断。agent 循环是你画出来的，不是你手写的。

核心概念 ​

LangGraph StateGraph：节点、边和 checkpointer

一个 StateGraph 有三样东西。

  1. State（状态）。 一个流过整张图的有类型字典（TypedDict 或 Pydantic 模型）。每个节点收到完整状态、返回一个部分更新，LangGraph 用每个字段的 reducer 来合并它——对应该累积的列表用 operator.add，默认是覆盖。
  2. Nodes（节点）。 Python 函数 state -> partial_state。每个都是一个离散步骤："调用模型""运行工具""做摘要"。
  3. Edges（边）。 节点之间的转移。静态边只去一个地方。条件边接受一个路由函数 state -> next_node_name，让图能根据模型输出分支。

你编译这张图。编译绑定拓扑、附上一个 checkpointer（可选，但对生产至关重要）、返回一个 runnable。你用一个初始状态和一个 thread_id 来调用它。执行的每一步都持久化一个以 (thread_id, checkpoint_id) 为键的 checkpoint。

四种超能力 ​

Checkpoint。 每次节点转移都把新状态写进一个存储（测试用内存，生产用 Postgres/Redis/SQLite）。用同一个 thread_id 再次调用这张图就能恢复。图从它暂停的地方接着跑。

中断。 给一个节点标上 interrupt_before=["human_review"]，执行就在那个节点运行之前停下。状态持久化。你的 API 回复用户"等待批准"。之后对同一个 thread_id 带 Command(resume=...) 的请求恢复执行。

流式。 graph.stream(state, mode="updates") 在状态增量发生时把它们 yield 出来。mode="messages" 流式吐模型节点里的 LLM token。mode="values" yield 完整快照。你挑要在 UI 里呈现什么。

时间旅行。 graph.get_state_history(thread_id) 返回完整的 checkpoint 日志。把任意一个先前的 checkpoint_id 传给 graph.invoke，你就从那个点分叉。对调试（"要是模型当时挑了工具 B 会怎样？"）和回放生产 trace 的回归测试都很棒。

Reducer 才是关键 ​

每个状态字段都有一个 reducer。大多数默认值没问题——新值覆盖旧值。但消息列表需要 operator.add，这样新消息是追加而非替换。并行边通过 reducer 合并它们的更新。如果两个节点都更新 messages、而你忘了 Annotated[list, add_messages]，第二个会悄悄胜出，你就丢了半轮对话。reducer 是这个库里唯一微妙的东西；把它弄对，其余的都能组合好。

四个节点构成的 ReAct 图 ​

一个生产级 ReAct agent 是四个节点和两条边：

  1. agent——用当前消息历史调用 LLM。返回 assistant 消息（可能含 tool_calls）。
  2. tools——执行最后那条 assistant 消息里的任何 tool_calls，把工具结果作为 tool 消息追加进去。
  3. 一条从 agent 出发的条件边，如果最后一条消息有 tool_calls 就路由到 tools，否则到 END。
  4. 一条从 tools 回到 agent 的静态边。

就这样。你用大约 40 行代码就得到了完整的 ReAct 循环（Thought → Action → Observation → Thought → …），带 checkpoint、中断和流式。

StateGraph vs Send（扇出） ​

Send(node_name, state) 让一个节点派发并行子图。例子：agent 决定一次查询三个检索器。每个 Send 都生成目标节点的一次并行执行；它们的输出通过状态 reducer 合并。这就是 LangGraph 不靠线程原语来表达 orchestrator-workers 模式的方式。

子图 ​

一个编译好的图能成为另一张图里的一个节点。外层图看到的是单个节点；内层图有它自己的状态和自己的 checkpoint。这就是团队如何构建 supervisor-worker agent：supervisor 图把用户意图路由给一个按领域划分的 worker 子图。

figure
l5-state-graph-ledger

动手构建 ​

第 1 步：状态与节点 ​

python
from typing import Annotated, TypedDict
from langchain_core.messages import AnyMessage, HumanMessage, AIMessage
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode
from langgraph.checkpoint.memory import MemorySaver

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

def agent_node(state: State) -> dict:
    response = llm.invoke(state["messages"])
    return {"messages": [response]}

def should_continue(state: State) -> str:
    last = state["messages"][-1]
    return "tools" if getattr(last, "tool_calls", None) else END

tool_node = ToolNode(tools=[search_web, read_file])

graph = StateGraph(State)
graph.add_node("agent", agent_node)
graph.add_node("tools", tool_node)
graph.set_entry_point("agent")
graph.add_conditional_edges("agent", should_continue, {"tools": "tools", END: END})
graph.add_edge("tools", "agent")

app = graph.compile(checkpointer=MemorySaver())

add_messages 是让消息列表累积而非覆盖的那个 reducer。忘了它是最常见的 LangGraph bug。

第 2 步：带一个 thread 运行 ​

python
config = {"configurable": {"thread_id": "user-42"}}
for event in app.stream(
    {"messages": [HumanMessage("find the Anthropic headquarters address")]},
    config,
    stream_mode="updates",
):
    print(event)

每个更新都是一个字典 \{node_name: state_delta\}。你的前端可以把它们流式传到 UI，让用户看到"agent 在思考……调用 search_web……拿到结果……作答。"

第 3 步：加一个人在环中断 ​

给一个节点打标记，让执行在它运行之前暂停。

python
app = graph.compile(
    checkpointer=MemorySaver(),
    interrupt_before=["tools"],  # 在每次 tool call 之前暂停
)

state = app.invoke({"messages": [HumanMessage("delete the production database")]}, config)
# state["__interrupt__"] 被设上了。检视被提议的 tool call。
# 如果批准：
from langgraph.types import Command
app.invoke(Command(resume=True), config)
# 如果拒绝：写一条拒绝消息再恢复
app.update_state(config, {"messages": [AIMessage("Blocked by human reviewer.")]})

状态、checkpoint 和 thread 在中断期间全都持久化。除了执行过程中，没有东西留在内存里。

第 4 步：用时间旅行做调试 ​

python
history = list(app.get_state_history(config))
for snapshot in history:
    print(snapshot.values["messages"][-1].content[:80], snapshot.config)

# 从一个先前的 checkpoint 分叉
target = history[3].config  # 回退三步
for event in app.stream(None, target, stream_mode="values"):
    pass  # 从那个点往后回放

把 None 作为输入传入会从给定的 checkpoint 回放；传入一个值会在恢复之前把它作为更新追加到那个 checkpoint 的状态上。这就是你如何在不重跑整段对话的情况下复现一次糟糕的 agent 运行。

第 5 步：为生产换掉 checkpointer ​

python
from langgraph.checkpoint.postgres import PostgresSaver

with PostgresSaver.from_conn_string("postgresql://...") as checkpointer:
    checkpointer.setup()
    app = graph.compile(checkpointer=checkpointer)

SQLite、Redis 和 Postgres 都已自带。MemorySaver 是给测试用的。任何要跨重启持久化的东西都想要一个真正的存储。

这项 skill ​

你把 agent 当图来构建，而不是当 while True 循环。

在你伸手去拿 LangGraph 之前，做一个 60 秒的设计：

  1. 给节点命名。 每一个离散决策或有副作用的动作都是一个节点。"agent 思考""工具运行""reviewer 批准""响应流式"。如果你列不出来，这个任务还不是 agent 形状的。
  2. 声明状态。 一个最小的 TypedDict，每个列表字段都有 reducer。别把一切都塞进 messages；把任务专属的字段（一个工作中的 plan、一个 budget 计数器、一个 retrieved_docs 列表）提升到顶层。
  3. 画出边。 除非下一步取决于模型输出，否则用静态边。每条条件边都需要一个带命名分支的路由函数。
  4. 提前选好 checkpointer。 测试用 MemorySaver，其他一切用 Postgres/Redis/SQLite。没有它就别上线——没有 checkpointer 就没有恢复、没有中断、没有时间旅行。
  5. 在工具运行之前、而非之后决定中断。 审批放在进入有副作用节点的那条边上，这样你能在造成伤害之前取消；校验放在从模型出来的那条边上，这样你能廉价地拒绝糟糕的调用。
  6. 默认流式。 UI 用 mode="updates"，模型节点里的 token 级流式用 mode="messages"，评估期间的完整快照用 mode="values"。

拒绝交付任何没有 checkpointer 的 LangGraph agent。拒绝交付任何在副作用之后才中断的。拒绝交付任何 messages 字段没把 add_messages 作为 reducer 的。

练习 ​

  1. 简单。 用一个计算器工具和一个网页搜索工具实现上面那个四节点 ReAct 图。核验对一段两轮对话，list(app.get_state_history(config)) 至少返回四个 checkpoint。
  2. 中等。 加一个在 agent 之前运行的 planner 节点，把一个结构化的 plan: list[str] 写进状态。让 agent 把计划步骤标记为完成。如果 plan 在一次 checkpoint 恢复后丢失（reducer 错了），就让测试失败。
  3. 困难。 构建一个 supervisor 图，用 Send 在三个子图（researcher、writer、reviewer）之间路由。每个子图有它自己的状态和 checkpointer。在外层图上加一个 interrupt_before=["writer"]，让人能批准研究简报。确认从一个先前 checkpoint 的时间旅行只重跑那个分叉出去的分支。

关键术语 ​

术语大家怎么说它实际是什么
StateGraph"LangGraph 那张图"编译之前你往上加节点和边的那个构建器对象。
Reducer"字段怎么合并"当一个节点为某字段返回更新时应用的函数 (old, new) -> merged；默认是覆盖，add_messages 是追加。
Thread"一个对话 ID"一个 thread_id 字符串，圈定一个会话的所有 checkpoint。
Checkpoint"一个暂停的状态"节点转移后完整图状态的一份持久化快照，以 (thread_id, checkpoint_id) 为键。
中断"为人暂停"interrupt_before / interrupt_after 在节点边界停止执行；用 Command(resume=...) 恢复。
时间旅行"从先前一步分叉"graph.invoke(None, config_with_old_checkpoint_id) 从那个 checkpoint 往后回放。
Send"并行子图派发"一个节点可以返回的构造子，用来生成目标节点的 N 个并行执行。
子图"把编译好的图当节点"一个编译好的 StateGraph 被当作另一张图里的节点；保留它自己的状态作用域。

延伸阅读 ​