LangGraph · 实战入门 · 代码优先

别把它当流程图:LangGraph 是一台有状态的机器

它要解决的不是"怎么调模型",是"崩了怎么办、等你三天怎么办、你反悔了怎么办"。

2026-09-30实战教程 · 8 个可运行片段约 16 分钟

引言:先把 LangGraph 和 LangChain 的关系说清

这个问题劝退过不少人,所以放在最前面。

LangChain 是那层"最快把 agent 跑起来"的 API。给它一个模型、几个工具,它给你一个能调工具的循环。典型写法是一行 `create_agent(model, tools)`。

LangGraph 在它下面,是运行时。是这套 agent 真正跑起来的地方。

官方给的建议很直白:先用 LangChain 的高层 API 起步,需要更多控制权的时候,再往下沉到 LangGraph。两层不是替代关系,是嵌套关系——LangChain 的 agent 本来就构建在 LangGraph 运行时之上。

有个变化要提前说。LangChain 1.0(2025 年 10 月发布)之后,老的 langgraph.prebuilt.create_react_agent 被标记弃用,功能搬到了 langchain.agents.create_agent。网上大量教程还在用旧写法。你照着敲报 ImportError,不是你错,是文章过期了。

那为什么不直接用 LangChain 就好?

因为有个问题它答不了。想象一下:你的 agent 跑到第 14 步,第 3 步那个工具调用突然返回 429。整条链重来一遍,前面 13 步的 token 全白烧。再想象一下:流程跑到一半,需要财务同事审批一笔退款,他第二天早上才点同意——这中间服务器还重启过一次。

这两件事,普通 agent 循环处理不了。不是写得不够好,是它没有"状态"这个概念。每次调用都是全新的,过去不存在。

LangGraph 要解决的,就是"过去"这件事。它把流程明明白白画成一张图,每一步的输入输出都存下来。于是你能存档、能中断、能倒退回去重跑。

后面八节全是动手。先跑通最小的图,再一层层加上去。

01第一张图:State / Node / Edge,二十行跑通

LangGraph 只有三个概念,先把手感建立起来。

State 是一张共享的数据快照,通常是 TypedDict。所有节点都读它,也都可以往里写。Node 是一个普通函数,收 state、干活、返回要更新的那部分。Edge 决定下一步去哪个节点。

"节点干活,边决定下一步"。这句话记住,后面全通。

from typing import TypedDict from langgraph.graph import StateGraph, START, END class State(TypedDict): topic: str draft: str def write(state: State): return {"draft": f"{state['topic']} 的第一版草稿"} def polish(state: State): return {"draft": state["draft"] + "(已润色)"} builder = StateGraph(State) builder.add_node("write", write) builder.add_node("polish", polish) builder.add_edge(START, "write") builder.add_edge("write", "polish") builder.add_edge("polish", END) graph = builder.compile() print(graph.invoke({"topic": "LangGraph", "draft": ""}))

三个细节值得停一下。

节点返回的是增量,不是完整 state。`write` 只返回 `{"draft": ...}`,`topic` 自动保留。`START` 和 `END` 是两个特殊标记,一个入口一个出口。

还有 `compile()`。它不是可选项。编译期会做结构校验——比如你加了个节点却没有任何边指向它,这时候就报出来,而不是等线上跑到一半才发现。

跑完这段话,你可能会觉得:这不就是用复杂的方式写了个函数调用?

对。所以下一节讲的那个东西,才是第一道真正的门槛。

02状态为什么会被"吃掉":reducer 这一课不能跳

忘了配 reducer,程序不报错,只是悄悄丢数据。

把 state 里加一个消息列表,你会立刻踩坑。

class State(TypedDict): messages: list

节点 A 返回 `{"messages": [msg1]}`,节点 B 返回 `{"messages": [msg2]}`。跑完一看,只剩 msg2。msg1 没了。

原因是每个 state key 都配了一个 reducer——一个 `reducer(left, right)` 的二元函数,left 是当前值,right 是节点返回的更新。默认 reducer 直接忽略 left,用 right 覆盖。

对 `draft` 这种字段,覆盖很合理。对消息列表就是灾难:每轮只留最后一条,对话上下文全蒸发。

解法是显式声明 reducer:

from typing import Annotated from langgraph.graph import add_messages class State(TypedDict): messages: Annotated[list, add_messages] notes: Annotated[list, operator.add] # 只想简单追加时

`Annotated[T, reducer]` 的意思是"这个 key 用这个函数合并"。`add_messages` 是内置的,专门处理消息——它会追加,还会按消息 id 去重,把同 id 的消息改成更新而不是重复。

`operator.add` 也能追加,区别是它不认消息对象,只管把两个列表拼起来。

另外两个不常用、但值得知道的:Overwrite 可以绕过 reducer 直接覆盖,适合"我要把这个累积字段清空";UntrackedValue(JS 版)用来放不该进 checkpoint 的运行时字段,比如数据库连接、临时缓存。

这一节只有一句结论:state 里凡是要累积的字段,必须配 reducer。忘了配,不抛异常、不打日志,只是少数据。这是 LangGraph 里最难查的一类 bug。

03边才是主角:条件路由与 agent 循环

静态边只能表达直线,条件边才能表达"看情况"。

条件边挂一个函数。这个函数读 state,返回一个字符串,那个字符串决定下一个节点是谁。

def should_continue(state: State): last = state["messages"][-1] if getattr(last, "tool_calls", None): return "tools" return END builder.add_conditional_edges( "agent", should_continue, {"tools": "tools", END: END}, ) builder.add_edge("tools", "agent")

读一遍这段。`agent` 节点调用模型;模型如果要求调工具,就路由到 `tools`;工具执行完,最后一条边把流程送回 `agent`。这就是一个完整的 ReAct 循环,只不过它的每一步都在图上,每一步都能被存下来。

写到这里你大概看出来了:LangGraph 的"图",重点不在画得好看,在循环里的每一次跳跃都是一个可寻址的位置。普通 while 循环里,第几步是隐式的;图里,它是显式的。

那个路由函数有个纪律要守:只读 state,返回值纯。别在里面做 IO,别改 state。它的职责只有一个——回答"下一步去哪"。

顺带说一个迁移点。1.0 之后官方推荐用 langchain.agents.create_agent 起标准 agent,图和 middleware 都可以混着用。要自己控制循环节奏、要在循环里插中断点,就回到这里手写条件边。

04并行:Send 让一个节点变成 N 个

拿到 20 篇文档要分别摘要时,串行是浪费。

普通条件边只能路由到一个节点。要扇出成 N 个,得用 `Send`。

from langgraph.types import Send def fan_out(state: State): return [Send("summarize", {"doc": d}) for d in state["docs"]] builder.add_conditional_edges("split", fan_out, ["summarize"]) builder.add_edge("summarize", "merge")

`Send` 在条件边的位置返回一个列表,每个元素是"去哪个节点 + 带什么 state"。运行时并行起 20 个 `summarize` 实例,各自独立跑。

这里 reducer 又出现了一次。20 个实例如果把结果写进同一个 key,就必须用 `Annotated[list, operator.add]`,否则并发写 20 次,只剩最后 1 条。

还有个细节值得知道:LangGraph 的执行模型借鉴了 Google 的 Pregel。同一轮里并行跑的节点属于同一个 super-step,图会等这一轮全部结束才推进到下一步。

所以并行不是"谁快谁先走",是一轮一轮地走。理解这点,后面看 checkpoint 的粒度会顺很多。

05崩了怎么办:checkpointer 与 thread_id

前面四节讲结构,从这节开始讲"过去"。

挂一个 checkpointer,图就有记忆了。

from langgraph.checkpoint.memory import InMemorySaver graph = builder.compile(checkpointer=InMemorySaver()) config = {"configurable": {"thread_id": "user-42"}} graph.invoke({"messages": [("user", "帮我查下上周的订单")]}, config) graph.invoke({"messages": [("user", "那第二笔呢")]}, config) # 它记得上文

`thread_id` 就是那把钥匙。同一个 id 的多次 `invoke`,共享同一条状态线。

checkpointer 存的是"线程级"的图状态快照,每走一步存一次。所以引言里那个场景有了答案:第 14 步挂了,用同一个 `thread_id` 再 `invoke`,它从最后一个检查点继续,不是从零开始。

开发时用 `InMemorySaver`,进程一重启就没了。生产上要换持久实现:`PostgresSaver`、`AsyncPostgresSaver`、`SqliteSaver`。

那长期记忆呢?那是另一个东西,叫 Store。区别很干脆:

维度CheckpointerStore
作用范围单个 thread跨 thread、跨用户
存什么图的状态快照用户偏好、事实、共享知识
典型用途会话续接、HITL、时间旅行记住"这个人是谁"
开发实现InMemorySaverInMemoryStore
生产实现PostgresSaver / SqliteSaverPostgresStore / RedisStore / MongoDBStore

Store 用 tuple 命名空间组织数据,比如 `("user_42", "memories")`。配好 embedding(比如 `openai:text-embedding-3-small`)之后还能用自然语言检索,节点里通过 `runtime.store` 拿到它。

一句话区分:checkpointer 管"这次对话讲到哪了",Store 管"这个人是谁"。两个都要,就两个都挂,互不冲突。

06需要人点个头:interrupt() 与 Command(resume=)

审批流是所有 agent 都会撞上的需求。

LangGraph 的做法叫 `interrupt()`,写在节点里。

from langgraph.types import interrupt, Command def ask_approval(state: State): decision = interrupt({ "question": "批准这笔退款吗?", "amount": state["amount"], }) return {"status": "approved" if decision == "approve" else "rejected"}

第一次跑,图会在这一行停住。它用 persistence 层把当前状态存下来,然后无限期地等。这时候你把 `{"question": ..., "amount": ...}` 原样发给财务同事。

他第二天早上点了同意。你拿同一个 `thread_id` 再调一次——服务器中间重启过也没关系:

graph.invoke(Command(resume="approve"), config)

那个 `"approve"` 就成了 `interrupt()` 的返回值,函数接着往下走。

注意它和静态断点的区别:`interrupt()` 是动态的,可以放在节点任意位置,可以按业务逻辑决定触发不触发。断点写在配置里,`interrupt()` 写在代码里。

四条注意事项,每条都有代价。不要在 `interrupt()` 外面套 `try/except`,恢复时会把异常吃掉;不要重排同一个节点里多个 `interrupt()` 的顺序,恢复靠位置匹配;不要返回太复杂的值;`interrupt()` 之前的副作用必须幂等,因为那部分会重跑。

官方的常见用法也就那几类:审批、让人改一下 state 再放行、工具调用前确认、校验人工输入、多个并行 interrupt 用 id 映射恢复值。

07出错能退回去吗:时间旅行与容错

每一步都存了 checkpoint,"回到过去"就是两个普通操作。

Replay:拿历史 checkpoint 的 config 重新 `invoke`。它之前的节点不重跑,之后的会真的重新执行——包括 LLM 调用、API 请求、interrupt。所以结果可能和上一次不一样。

Fork:对某个历史 checkpoint 调 `update_state` 改一下 state,就分出了一条新分支,原来的线程不受影响。

history = list(graph.get_state_history(config)) # 时间倒序,history[0] 是当前 graph.update_state(history[2].config, {"status": "draft"}, as_node="review") graph.invoke(None, history[2].config)

`get_state_history` 给你所有检查点。`as_node` 指定这次修改算作哪个节点的产出,免得图分不清该走哪条边。

1.2 版本(2026 年 5 月)还补了三层安全网,值得知道:

同一个版本还带来 `DeltaChannel`(beta),只存每步增量而不是全量快照。长线程跑久了 checkpoint 会越来越胖,这个是冲着那个问题去的,用 `snapshot_frequency=K` 控制快照频率。

08从本机到线上:部署前的一张清单

demo 跑通和上线之间,还差四件事。

至于托管,有个改名要先知道:LangGraph Platform 现在叫 LangSmith Deployment。它把 persistence 那一层接管了,checkpointer 和 store 不用自己接,还提供把 agent 暴露成 MCP server、实时流式中间步骤、Assistants API 这些能力。

计划价格关键权益
Developer$0 / seat1 个 seat,5k base traces/月,社区支持,不含 Deployment
Plus$39 / seat/月10k traces/月,含 1 个 Serverless 部署,无限 seats
Enterprise定制自托管 / 混合部署,自定义 SSO、ABAC、RBAC,SLA

计量单位是 LCU(LangChain Compute Unit,$1.50/LCU)和 LSU(Storage Unit,$1.00/LSU)。规模化之后这笔账要单独算——这也是它被吐槽最多的地方之一。

想自己扛也行。`langgraph-cli` 加 Postgres 就能起来,只是上面那些能力得自己实现。

09收束:其他 Agent 框架,以及什么时候别用 LangGraph

先说别用的时候。

社区里有句话流传很广:do not graph what a function can do——能用函数解决的,就别上图。

如果你的流程是一条直线、单 agent、没有中断需求,那 LangGraph 带来的图编译、state schema、checkpoint 管理,全是净成本。这种场景,Pydantic AI 或者干脆手写一个循环,都更轻。

那其他选择各是什么定位?

框架状态编排模型适合谁
OpenAI Agents SDK仍 pre-1.0Handoff 式多 agent全栈在 OpenAI 上,要快速起多 agent
MS Agent Framework1.0 GA(2026-04)图式工作流.NET / Azure 企业环境;AutoGen 的官方继任者
CrewAI约 52.8K stars角色 crew + Flows上手最简单,但没有内置 checkpoint
Google ADK2.x,约 19.9k stars层级树 + WorkflowGCP 原生、多模态、要 A2A 互操作
Pydantic AI2.40.0类型安全 agent单 agent 场景;durable execution 支持 7 种引擎
Claude Agent SDK—subagent 委派只用 Claude 模型

选型其实就三个问题:要不要持久化?要不要人机协同?团队在哪个云上?

要 durable execution、要 HITL、要可审计、要能倒退回去重跑——这四个里中了一个,LangGraph 就是当前最成熟的那个答案。一个都不要,它大概率是过度设计。

同时说句实话:它的学习曲线被多个独立评测标成"陡"。要理解图编译、state schema、checkpoint、条件路由这一整套;复杂状态机调试起来不轻松;`langchain`、`langgraph`、`langsmith` 几个包的名字也确实容易让人晕。

"我在做意大利面。而且我讨厌这种必须让渡决策的黑盒。"—— 一位开发者在社区里对 LangGraph 的吐槽

这些吐槽是真的。但它们描述的是用一种重的方式解决一个复杂问题——而那些问题,轻的方式本来也解决不了。

所以入门的路子大概是:先在 LangChain 里把 agent 跑通,等你第一次真的需要"停三天再继续"的时候,再回来读第二节到第七节。

总结:四个可以带走的判断

参考资料