12 LangGraph 图式 Agent

上一篇拿 Claude Agent SDK 与手写版逐项对完了账，系列的最后一位选手再用另一种世界观重写 Agent：不是循环，是图。本文是「从零手写 Agent」系列第 12 篇收官之作：LangGraph 的状态图范式、十行重写的写法、手动建图与手写循环的对应关系，最后给出三种实现的选型收束，并回放整个系列的来路。

本文按"图式世界观 → 十行重写 → 手动建图 → checkpointer 与选型 → 总结"组织。

1、图式世界观

LangGraph 的核心抽象是三个词：State（一个共享字典，对应手写版的历史列表）、Node（一个函数，收到状态返回更新——"调用模型"与"执行工具"各是一个节点）、Edge（节点间的转移，其中条件边按状态决定去向）。你手写的 while 循环，在 LangGraph 里被显式画成了一张图。

解读：图里"模型到工具再回模型"的环，就是 while stop_reason == "tool_use" 的可视化；条件边上那两个标签，正是循环退出判断的两个分支。范式换了，控制流一模一样——这是走完前十篇后应有的直觉。相关资料见 LangGraph 文档。

1.1、State 的追加语义

三件套里最值得多看一眼的是 State。手写版直接往 history 列表 append；LangGraph 的节点不修改状态，只返回一份更新，由框架按每个字段的 reducer 合并进共享字典。MessagesState 内置的追加语义等价于自己声明一个带 reducer 的状态：

from typing import Annotated, TypedDict
from langgraph.graph.message import add_messages

class AgentState(TypedDict):
    messages: Annotated[list, add_messages]   # 追加合并，不是整体覆盖

解读：Annotated 把"这个字段怎么合并"写进了类型声明——节点只管返回新消息，框架负责追加，这正是手写版两处 history.append 隐含的"追加而非覆盖"约定，在图式范式里成了显式、可更换的策略。把 reducer 换成自定义函数，同一张图就能改写记忆行为，比如追加前去重、超长时滚动截断——第 7 篇的压缩器在图范式下正是从这个接入口挂进去的。

2、十行重写

LangGraph 提供了现成的 ReAct 循环图，配合装饰器声明工具，重写工作量骤降：

from langchain_anthropic import ChatAnthropic
from langchain_core.tools import tool
from langgraph.prebuilt import create_react_agent

@tool
def read_file_tool(path: str) -> str:
    """读取项目内的文本文件。"""
    return read_file(path)          # 直接复用手写版的工具函数

llm = ChatAnthropic(model=MODEL, max_tokens=2048)
agent = create_react_agent(
    llm, [read_file_tool, search_files_tool],
    prompt="你是代码调研助手，回答要引用文件和行号。",
)
result = agent.invoke({"messages": [("user", task)]})

解读：@tool 装饰器把函数签名与文档字符串自动转成工具说明书——与手写 schema 是同一件事的两种工效学；result["messages"] 就是手写版的历史列表，LangGraph 管它叫状态。工具函数层完全复用，框架替换的只是循环与协议层。

2.1、被替换的只有循环层

分清哪一层被换掉、哪一层原样保留，是对照的关键。原样保留的是工具函数本体：read_file 这些第 6 篇写好的函数一个字不用改。被替换的有三样：工具说明书从手写 JSON schema 变成装饰器从签名与 docstring 生成；循环从 while 变成图引擎驱动；消息从裸字典变成 LangChain 的消息对象。连系统提示词都有对应物——create_react_agent 的 prompt 参数，就是第 5 篇 system 参数的别名。

顺带一个选型注脚：LangGraph 1.x 起官方文档力推的新入口是 langchain.agents 里的 create_agent，create_react_agent 仍是文档完整的经典入口，两者心智模型一致，本文沿用后者。

3、手动建图

预制的循环图不够用时，手动建图把控制流完全显式化。骨架如下：

from langgraph.graph import StateGraph, MessagesState, START, END

def call_model(state):              # 节点一：调用模型
    return {"messages": [llm_with_tools.invoke(state["messages"])]}

def run_tools(state):               # 节点二：执行工具并回填结果
    ...

def should_continue(state):         # 条件边：手写循环的退出判断
    last = state["messages"][-1]
    return "tools" if getattr(last, "tool_calls", None) else END

g = StateGraph(MessagesState)
g.add_node("model", call_model)
g.add_node("tools", run_tools)
g.add_edge(START, "model")
g.add_conditional_edges("model", should_continue)
g.add_edge("tools", "model")
agent = g.compile()

解读：run_tools 节点里要做的事，正是系列第 4 篇的往返核心——取出调用块、按 id 配对回填结果消息，各家框架无一例外。手动建图的价值在复杂流程编排：固定的多阶段流水线、插入人工审批节点、失败走不同重试分支——这些画在图上一目了然，写在循环里就是面条。

3.1、补全 run_tools 与首次调用

骨架里省略的 run_tools 是唯一还需要动手写的部分，补全如下：

from langchain_core.messages import ToolMessage

TOOLS = {t.name: t for t in [read_file_tool, search_files_tool]}

def run_tools(state):
    last = state["messages"][-1]
    results = []
    for call in last.tool_calls:            # 取出本轮全部调用块
        fn = TOOLS[call["name"]]            # 第 4 篇的函数表，原样复用
        try:
            out = fn.invoke(call["args"])   # @tool 函数直接调用
            results.append(ToolMessage(content=str(out),
                                       tool_call_id=call["id"]))
        except Exception as e:              # 第 6 篇铁律：失败要大声
            results.append(ToolMessage(content=f"工具失灵：{e}",
                                       tool_call_id=call["id"]))
    return {"messages": results}

解读：这正是系列第 4 篇往返核心的逐条落地——遍历调用块、查函数表执行、按 tool_call_id 配对回填，连异常转文字结果都遵守第 6 篇"失败要大声"的原则。骨架里的 llm_with_tools 则来自 llm.bind_tools(tools)，绑定工具说明书的那次调用，与手写版 tools=self.tools_spec 同位。建图收尾与首次调用如下：

agent = g.compile()                                  # 节点与边编译成可执行图
result = agent.invoke({"messages": [("user", task)]})

for m in result["messages"]:                         # 历史就在状态里
    print(getattr(m, "type", "?"))                   # 逐条打印，verbose 同款
answer = result["messages"][-1].content

解读：compile 把节点与边编译成可执行图，invoke 一次跑完；返回值里的 messages 就是手写版的 history——同一个数组，从循环的私有变量变成了图的公共状态。要过程可观测，把 invoke 换成 stream 即可逐事件输出，对应手写版的 verbose 开关。API 细节见 Graph API 参考。

3.2、与手写循环逐行对照

至此可以把第 5 篇的循环与这张图逐行对上，对照如下：

解读：两列的差距比想象小——范式换了，五脏六腑全是从前那些零件。差异的方向更有启发：手写版的退出判断嵌在循环体内部，加一条分支就要动心脏；图式把分支升格为"边"这种一等公民，多一条路径只是多画一条线，不改任何现有节点。审批流就是一个现成例子：在模型与工具之间插一个人工节点，条件边改成"通过走工具、驳回回模型"，生产里最常见的人机协同 Agent 就是这么一张图——下一节说的复杂编排，价值正源于此。

4、checkpointer 与三强选型

LangGraph 的 checkpointer 把状态每步自动落盘：同一会话连续提问它记得你，换成 SQLite 存储后跨进程也记得，还能回退到任意一步做时间旅行调试。它与系列第 7 篇持久化的差异是本质性的注脚：手写版存的是"结果"，checkpointer 存的是"轨迹"。

4.1、checkpointer 上手

两个会话之间"记得你"的最小验证如下：

from langgraph.checkpoint.memory import InMemorySaver

agent = g.compile(checkpointer=InMemorySaver())     # 每步自动存档
cfg = {"configurable": {"thread_id": "user-42"}}    # 会话身份证

agent.invoke({"messages": [("user", "我叫小齐，喜欢蓝色")]}, cfg)
agent.invoke({"messages": [("user", "我叫什么、喜欢什么颜色？")]}, cfg)

解读：thread_id 是会话身份证：同一个 id 的多次 invoke 共享一份状态，第二问不带任何历史直接答对——第 2 篇要靠宿主程序自己持有 history 才能做到的事，这里一个参数就内置了。生产用法是把 InMemorySaver 换成 SqliteSaver（独立安装 langgraph-checkpoint-sqlite，用 SqliteSaver.from_conn_string 建存储），落盘后跨进程也记得；更妙的是每一步都是一帧存档，get_state_history 能拉出全部历史帧，任意一帧都可作回放起点——第 7 篇的"存结果"升级成"存轨迹"，时间旅行调试从设想变成内置功能。持久化细节见 LangGraph 持久化文档。

三种实现的选型收束如下：

解读：简单 Agent 三者皆可；要工程保障选 Agent SDK；流程复杂、需要审批与分支重试，图式才发力。模型与端点配置沿环境变量注入，与手写版共用同一套接入层。

5、总结

LangGraph 用 State、Node、Edge 三件套把 Agent 显式建模为图，条件边就是手写循环的退出判断，checkpointer 把第 7 篇的存档升级为可回溯轨迹；三种实现没有王者，只有场景匹配。回到系列第 0 篇的公式——Agent 等于大模型加工具加循环——走完十二篇后应有一层新的理解：公式里的每一项都平常无奇，真正的功夫在工具面怎么划、上下文放什么、如何证明它干得好。模型只是心脏，你才是设计师。后续方向建议从 MCP 接入与真实 embedding 开始，把手写版继续养大。LangChain 生态见 LangChain 官网。

连同第 0 篇的路线图，十三篇至此收官。把来路铺开看：

解读：三个阶段各有主旋律——前三篇打地基（协议、记忆、流式），中间五篇造机器（循环、工具、边界、压缩、分身），后四篇补工程化再对照框架（检索、评估、SDK、图式）。回头看，每一篇其实都在回答同两个问题：它怎么工作，以及它失控了怎么办——第二问的答案从 max_turns 一路长到权限链与 checkpointer，这条"防跑飞"的暗线比任何单个知识点都值得带走。

给准备走这条路的读者三条建议。其一，第 1 到 5 篇必须亲手敲，一步别跳——裸 HTTP、往返、循环是所有框架的解码器，抄近道的人永远只能在文档里捡咒语。其二，第 6 到 8 篇的边界意识决定你的 Agent 能不能见生产，权限、沙箱、压缩永远先于炫技。其三，第 9、10 篇的评估闭环至少完整跑一次——没有标尺，后面所有调优都是玄学；两篇框架对照放在最后，正是因为对账需要资本。学完不是终点：挑一个你自己的真实小项目，把 miniagent 养成日常工具，这门手艺才算真正毕业。