一、引言：单轮对话的天花板在哪里
把 LLM 接进业务系统的第一版代码通常长这样：拼一个 prompt，调一次invoke，把字符串返回给前端。
这在 Demo 阶段完全够用，但一旦进入生产，问题会集中爆发：
- 状态无处安放。多轮对话的历史、检索到的资料、上一次工具调用的返回值，只能靠调用方自己维护一个不断膨胀的
messages列表。 - 流程不可中断。凡是需要“人工审批后再继续”的场景（发邮件、改数据库、下单），单轮调用只能把整个上下文序列化出去，等审批完了再重新拼一遍 prompt —— 中间的 LLM 调用被白白重跑。
- 失败即重来。第三步的 API 超时了，前两步的 token 已经烧掉，你只能从头再来一遍。
- 循环不可控。Agent 自我反思、工具重试这类天然带环的结构，在 LCEL 那种有向无环图（DAG）里写不出来，只能塞进一个 while 循环，于是可观测性彻底归零。
- 没有审计轨迹。出事之后你无法回答“当时模型看到了什么、在哪一步做了错误决策”。
LangChain 的 LCEL 解决的是编排问题，它本质上是一张静态 DAG。而真实的 Agent =带环的状态机 + 持久化 + 人机协同。LangGraph 正是为这个模型设计的：把工作流显式建模成一张图，节点是函数，边是控制流，共享状态由 reducer 驱动合并，每一步执行后都落盘成 checkpoint。
这篇文章会从原理讲到代码，搭一个“调研 → 起草 → 人工审批 → 修订 → 归档”的可中断、可恢复智能体。
1.1 先看一段“看起来能跑”的代码
为了让后面的对比足够直观，我们先把那个 Demo 版本完整写出来。它的典型形态是这样的：
fromlangchain_openaiimportChatOpenAI llm=ChatOpenAI(model="gpt-4o-mini")history=[]# 调用方自己维护的全局状态defchat(user_input:str)->str:history.append({"role":"user","content":user_input})resp=llm.invoke(history)history.append({"role":"assistant","content":resp.content})returnresp.content这段代码的每一个“缺点”都不是风格问题，而是结构问题：
history是进程内的一个 list，进程一重启就没了，多副本部署时更是各说各话；- 一旦要加“先检索资料再回答”，你就得在
chat里塞if；加了工具调用，就要再塞一个while循环判断模型是否还要调工具； - 想加“下单前需要人工确认”，你只能把整个
history序列化到数据库，审批通过后再反序列化重建 —— 而重建之后，模型为了走到同一个位置，还得把之前所有的 token 重新烧一遍； - 出问题时，你翻遍日志也只能看到
input和output，中间发生过什么完全黑盒。
把这五个问题抽象一下，其实是三个能力缺失：状态的持久化、控制流的显式化、执行过程的可观测。LangGraph 的三个核心抽象 —— State、Edge、Checkpointer —— 正好一一对应。
1.2 LangGraph 到底在解决什么问题
可以用一句话概括 LangGraph 的心智模型：
一张带环的有向图 + 一个可序列化的共享状态 + 一个每步落盘的检查点。
把这句话和几个相邻概念做个对比，定位会更清楚：
| 方案 | 控制流 | 状态管理 | 中断/恢复 | 循环支持 | 适用场景 |
|---|---|---|---|---|---|
| 手写 while 循环 | 隐式（藏在代码里） | 自己维护 | 自己实现 | 支持但难控 | 一次性脚本 |
| LCEL（Runnable 组合） | 静态 DAG | 靠管道传递 | 不支持 | 不支持 | 线性数据流水线 |
| AgentExecutor | 隐式 ReAct 循环 | 靠 memory | 弱 | 内置但不可定制 | 通用工具型 Agent |
| LangGraph | 显式图结构 | TypedDict + reducer | checkpoint 原生支持 | 一等公民 | 生产级多步骤工作流 |
注意表格里 LCEL 那一行 —— 它并不是“旧方案”，LCEL 依然是 LangChain 表达式组合的基础，LangGraph 的节点内部照样可以用 LCEL 写链。二者的关系是嵌套而不是替代：LCEL 负责“一段数据怎么流”，LangGraph 负责“多个步骤怎么调度、在哪停、从哪续”。
二、核心原理：StateGraph 的三要素
2.1 State 与 Reducer：并发写入的合并规则
LangGraph 的状态就是一个TypedDict。关键设计在于字段级别的 reducer：当两个并行节点同时往同一个 key 写值时，用哪个 reducer 决定合并结果。
classState(TypedDict):messages:Annotated[list,add_messages]# 追加而非覆盖，自动去重/合并消息sources:Annotated[list[str],operator.add]# 并行检索结果自动拼接draft:str# 无注解 = 后写覆盖先写没有给 list 字段加 reducer，是新手最常踩的坑 —— 并行分支一跑就抛InvalidUpdateError。记住一条规则：任何可能被多个节点写入的字段，都必须显式声明 reducer。
理解 reducer 的关键是理解“节点返回的是什么”。节点函数的返回值不是新状态，而是状态增量（partial update）。框架拿到这个增量后，对每一个 key 做如下判断：
- 有 reducer 的 key：调用
reducer(旧值, 新值)，把结果写回； - 没有 reducer 的 key：直接用新值覆盖旧值；
- 增量里没出现的 key：原封不动保留。
所以下面两种写法语义完全不同：
# 写法 A：覆盖。第二次执行后 messages 只剩 1 条defnode_a(state):return{"messages":[new_msg]}# 写法 B：追加。第二次执行后 messages 有 2 条（依赖 add_messages）defnode_b(state):return{"messages":[new_msg]}# 前提是 State 里 messages 带了 add_messages写法 A 和 B 的代码长得一模一样，差别完全隐藏在State的注解里。这正是 reducer 设计“隐蔽但强大”的地方，也是团队协作时最容易出事故的地方 —— 建议在State定义上方统一写注释，标注每个字段的合并语义。
几个常用 reducer 一览：
| 字段类型 | 推荐 reducer | 语义 |
|---|---|---|
list[消息] | add_messages | 按id去重合并，支持RemoveMessage删除 |
list[str]/list[dict] | operator.add | 简单拼接 |
int（累加） | operator.add | 计数、求和 |
bool（标记位） | 不加（覆盖） | 后写覆盖 |
dict（合并） | 自定义函数 | 通常做浅合并{**old, **new} |
下面是一个自定义 dict 合并 reducer 的例子，在“配置逐步补全”的场景里非常有用：
defmerge_dict(old:dict,new:dict)->dict:"""浅合并：新值覆盖同名 key，未出现的 key 保留。"""return{**old,**new}classState(TypedDict):config:Annotated[dict,merge_dict]还有一个必须知道的细节：返回未在 State 中声明的 key，会被静默忽略。这不像 Python 字典会报KeyError，它不会报错，只会让你的“精心构造的字段”永远消失。调试时如果发现某个字段永远是初始值，第一件事就是检查它有没有写进TypedDict。
2.2 节点与边：把控制流画出来
节点是纯函数(state) -> dict，返回的 dict 是状态增量（partial update），由框架用 reducer 合并回全局状态。边分两种：
add_edge("a", "b")：确定性跳转；add_conditional_edges("a", router, {"x": "b", "y": "c"})：router 函数读状态返回 key，实现分支与循环（把条件边指回上游节点即可成环）。
START和END是两个虚节点，标明入口与出口。
关于节点函数，有三条容易被忽略的约定：
- 签名可以不接收全部参数。你可以只写
def node(state: State)，也可以用Literal["messages", "draft"]声明只关心某些 key（LangGraph 会把该 key 的最新值传进来）。后者在状态很大时能减少无谓的数据传递。 - 可以额外接收
config。写成def node(state: State, config: RunnableConfig)，就能拿到config["configurable"]["thread_id"]之类的运行时信息，做多租户路由、日志打点时必备。 - 返回值必须是 dict 或 None。返回
None表示“本节点不修改状态”，合法但不推荐 —— 因为它会让图的语义难以阅读。
条件边里的router函数写法也有讲究。它只读状态、不做副作用，返回值必须是字典里声明过的 key，否则运行时会报错。如果分支很多，用Literal标注返回值类型能在 IDE 里获得自动补全：
fromtypingimportLiteraldefrouter(state:State)->Literal["search","answer","give_up"]:ifstate["retry"]>=3:return"give_up"ifstate.get("need_search"):return"search"return"answer"关于“循环”，再强调一遍：LangGraph 里没有专门的while原语，循环就是“条件边指回上游节点”。这也意味着你必须自己准备熔断条件（比如重试计数），否则图会一直转到触发recursion_limit抛异常为止。默认recursion_limit是 25 个 super-step，可以在config里调整：
config={"recursion_limit":50,"configurable":{"thread_id":"t1"}}2.3 Checkpointer：可追踪、可恢复的全部秘密
编译图时传入 checkpointer，LangGraph 会在每个 super-step（一次图执行批次）结束后把完整状态序列化落盘，主键是thread_id。这一个机制同时解决了四件事：
| 能力 | 依赖的 API |
|---|---|
| 断点续跑 | thread_id+ checkpointer，重启进程后继续 |
| 人工介入 | interrupt()挂起，Command(resume=...)恢复 |
| 状态回放/时间旅行 | get_state_history()列出所有历史 checkpoint |
| 故障重试 | 从最后一个成功的 checkpoint 重新调度 |
interrupt()在节点内部调用，会抛出特殊信号让图暂停，并把 payload 交给人。关键陷阱：恢复时该节点会从头重跑，所以interrupt()之前的副作用代码必须幂等。生产环境用PostgresSaver，本地开发用SqliteSaver，纯测试用InMemorySaver。
这里把“super-step”这个概念再展开一下，因为它直接决定了 checkpoint 的粒度。LangGraph 的执行是按批次推进的：
- 一个 super-step 里，所有可以并行执行的节点同时跑（比如
plan之后的两个检索节点）； - 等这一批全部执行完，状态统一合并、写一次 checkpoint；
- 然后才计算下一批要执行哪些节点。
所以 checkpoint 的数量不等于节点数量，而是等于“批次数”。在多分支并行的图里，一次 checkpoint 可能对应好几个节点的产出，这是正常的。
interrupt()的行为同样值得推敲。它的完整生命周期是：
- 节点执行到
interrupt(payload)，LangGraph 抛出一个内部信号，当前节点中断； - 图暂停，状态（包括中断前的所有变更）写入 checkpoint，
payload暴露给调用方； - 调用方用
Command(resume=value)重新invoke； - LangGraph 从断点所在的 checkpoint 恢复，重新执行该节点函数；
- 这一次
interrupt()不再抛异常，而是直接返回resume传入的值，函数继续往下跑。
第 4 步是重点：“重跑”是真的从头跑一遍函数体。所以下面的写法是危险的：
defbad_node(state):send_email(state["draft"])# ❌ 副作用在 interrupt 之前decision=interrupt({"draft":state["draft"]})return{"approved":decision["approved"]}审批恢复后，send_email会被再调用一次。正确做法是把副作用挪到interrupt()之后，或者用幂等键去重。
关于持久化后端，选型建议如下：
| 实现 | 包 | 适用场景 | 注意点 |
|---|---|---|---|
InMemorySaver | 内置 | 单元测试、一次性脚本 | 进程结束即丢失 |
SqliteSaver | langgraph-checkpoint-sqlite | 本地开发、单机部署 | 需要check_same_thread=False |
PostgresSaver | langgraph-checkpoint-postgres | 生产环境 | 首次使用需调.setup()建表 |
Postgres 的接法大致长这样：
fromlanggraph.checkpoint.postgresimportPostgresSaver DB_URI="postgresql://user:pwd@localhost:5432/agent?sslmode=disable"withPostgresSaver.from_conn_string(DB_URI)assaver:saver.setup()# 建表，只需执行一次graph=builder.compile(checkpointer=saver)setup()会创建checkpoints、checkpoint_blobs、checkpoint_writes等几张表。生产环境务必建索引并按时间分区/清理，否则高并发下这张表会迅速膨胀到几十 GB。
三、实战：可中断、可恢复的调研智能体
场景：用户提一个技术选型问题，Agent 规划检索词 → 检索 → 起草报告 →暂停等待人工审批→ 不通过则按意见修订。
3.1 定义状态：先想清楚“什么要持久化”
状态设计是整个实现里最重要的一步。经验法则是：凡是需要在中断后恢复时还能拿到的东西，都必须进 State；凡是进 State 的东西，都必须可序列化。
fromtypingimportAnnotated,Literalimportoperatorfromlanggraph.graphimportStateGraph,START,ENDfromlanggraph.graph.messageimportadd_messagesfromlanggraph.checkpoint.memoryimportInMemorySaverfromlanggraph.typesimportinterrupt,Commandfromlangchain_core.messagesimportHumanMessage,SystemMessage MAX_REVISIONS=3# 熔断阈值，防止无限修订循环classResearchState(TypedDict):# ---- 输入 ----topic:str# ---- 中间产物 ----queries:list[str]# plan 节点产出sources:Annotated[list[str],operator.add]# 两个检索节点并行写入 → 必须加 reducerdraft:str# 每轮修订覆盖# ---- 人机协同 ----approved:boolreview_notes:str# 人工填写的修改意见# ---- 熔断 ----revision_count:Annotated[int,operator.add]# 每次审批 +1# ---- 审计 ----messages:Annotated[list,add_messages]# 全链路消息日志几个设计决策值得说明：
sources用operator.add，因为两个检索节点会并行写入，必须拼接而非覆盖；revision_count也用operator.add，每次进入 review 节点就 +1，这样即使节点重跑，语义依然是“审批次数 +1”（注意：由于 review 节点的重跑发生在 interrupt 之后，实际只会 +1 一次）；draft不加 reducer，因为它是“覆盖式”字段，每次修订就是最新版本；messages用add_messages，它会按消息id去重，天然适配流式追加和中断恢复。
3.2 定义节点：每个函数只做一件事
规划节点——把用户的问题拆成检索词：
defplan_node(state:ResearchState)->dict:"""把 topic 拆成 2~3 个检索词。纯函数、无副作用，可安全重跑。"""prompt=(f"你是技术调研助手。请针对下面的问题，生成 3 个互补的检索关键词，"f"每行一个，不要编号，不要解释。\n\n问题：{state['topic']}")raw=llm.invoke([HumanMessage(content=prompt)]).content queries=[q.strip()forqinraw.splitlines()ifq.strip()][:3]return{"queries":queries,# 用 add_messages 追加，形成审计日志"messages":[HumanMessage(content=f"[plan] 生成的检索词：{queries}")],}检索节点——两个并行版本，分别打内部知识库和公网：
defsearch_internal(state:ResearchState)->dict:"""查内部知识库。返回 sources 增量，由 operator.add 自动拼接。"""hits=[]forqinstate["queries"]:hits.extend(vector_store.similarity_search(q,k=2))return{"sources":[f"[内部]{h.page_content[:200]}"forhinhits]}defsearch_web(state:ResearchState)->dict:"""查公网。失败时返回空增量，不阻断主流程。"""try:results=[f"[公网]{r['title']}:{r['snippet']}"forqinstate["queries"]forrinweb_search(q,limit=2)]exceptExceptionasexc:# 单点失败降级return{"messages":[HumanMessage(content=f"[search_web] 失败：{exc}")]}return{"sources":results}起草节点——汇总资料写初稿：
defdraft_node(state:ResearchState)->dict:"""根据 sources 和人工意见生成/修订草稿。"""notes=state.get("review_notes")or"（首轮，无人工意见）"context="\n".join(state["sources"])[:8000]# 截断，防止超出上下文prompt=f"""你是技术分析师，请撰写一份关于「{state['topic']}」的调研简报。 要求： 1. 结构为：结论 / 依据 / 风险 / 建议； 2. 只使用下面提供的资料，不要编造； 3. 如果在修订，必须逐条回应人工意见。 【参考资料】{context}【人工修改意见】{notes}"""draft=llm.invoke([SystemMessage(content=prompt)]).contentreturn{"draft":draft,"messages":[HumanMessage(content=f"[draft] 第{state['revision_count']+1}稿完成")],}审批节点——整个方案的核心，用interrupt()暂停：
defreview_node(state:ResearchState)->dict:"""把草稿交给人工审批。interrupt 会让图在此暂停并落盘。"""# ⚠️ interrupt 之前的代码在恢复时会被重跑，这里必须保持无副作用decision=interrupt({"type":"approval_request","topic":state["topic"],"draft":state["draft"],"instructions":"用 Command(resume={'approved': bool, 'notes': str}) 恢复",})return{"approved":bool(decision.get("approved")),"review_notes":decision.get("notes",""),"revision_count":1,# 累加到 state 里（operator.add）"messages":[HumanMessage(content=f"[review] approved={decision.get('approved')}, "f"notes={decision.get('notes','')[:100]}")],}归档节点：
deffinalize_node(state:ResearchState)->dict:"""审批通过后归档。这是第一个可以安全做副作用的节点。"""archive_id=save_to_knowledge_base(topic=state["topic"],draft=state["draft"],sources=state["sources"],)return{"messages":[HumanMessage(content=f"[finalize] 已归档，id={archive_id}")]}3.3 组装图与条件边
defroute_after_review(state:ResearchState)->Literal["revise","approve","abort"]:"""审批后的三岔路由：通过 / 打回重写 / 熔断放弃。"""ifstate["approved"]:return"approve"ifstate["revision_count"]>=MAX_REVISIONS:return"abort"# 防死循环的关键return"revise"builder=StateGraph(ResearchState)builder.add_node("plan",plan_node)builder.add_node("search_internal",search_internal)builder.add_node("search_web",search_web)builder.add_node("draft",draft_node)builder.add_node("review",review_node)builder.add_node("finalize",finalize_node)builder.add_edge(START,"plan")# fan-out：plan 之后两个检索节点并行执行（同一个 super-step）builder.add_edge("plan","search_internal")builder.add_edge("plan","search_web")# fan-in：两个检索都完成后才进 draft（框架自动等待全部前置边）builder.add_edge("search_internal","draft")builder.add_edge("search_web","draft")builder.add_edge("draft","review")# 条件边：既能分支，也能成环（revise → draft 就是一条回边）builder.add_conditional_edges("review",route_after_review,{"revise":"draft","approve":"finalize","abort":END},)builder.add_edge("finalize",END)graph=builder.compile(checkpointer=InMemorySaver())这里有一个容易被忽略的语义：fan-in 是自动的。你不需要写“等待两个检索节点都完成”的逻辑，只要draft的两个上游节点都在，LangGraph 就会等它们全部落定后才调度draft。这也正是sources必须带operator.add的原因 —— 两个节点在同一个 super-step 里各自返回增量，框架需要 reducer 才知道怎么合并。
3.4 第一次运行：走到审批就停
config={"configurable":{"thread_id":"research-001"},"recursion_limit":50}result=graph.invoke({"topic":"2025 年选型：Postgres 还是 MongoDB 作为主库？","revision_count":0,"sources":[]},config,)print(result["__interrupt__"])# [Interrupt(value={'type': 'approval_request', 'draft': '.**更多硬核网安与AI工具包，请扫码获取完整源码！**..',
