我第一次在项目里引入LangGraph时，同事看到代码的第一反应是：这不就是给Agent外面套了个while循环吗？当时我辩解了几句，但说实话自己也没完全想清楚“状态图编排”到底比循环强在哪。直到后来一个多轮调研Agent项目被迫支持人工审核、断点续跑和失败重试时，我才真正体会到，状态图这套东西解决的不是“能不能跑”，而是“跑乱了以后能不能收拾”。

这篇文章不聊概念定义，直接拆LangGraph状态图编排的核心机制：State怎么管、Node和Edge怎么搭、循环和人工介入怎么落地，最后放一个完整的多轮调研Agent搭建过程和我踩过的坑。适合已经写过LangChain基础链、正准备做复杂Agent的开发者。你不需要用过LangGraph，从第一节开始跟着建图，基本就能上手。

1. 链式调用的死穴，就是状态图登场的理由

1.1 一个让我放弃链式调用的实际场景

我最早写的版本用的是链式调用：先检索，再生成草稿，再校验，发现问题就手动回去重新调用一次检索函数。看起来也不复杂，函数就是那三五个，逻辑都直来直去。但真正跑起来以后，问题一个接一个冒出来。

第一个问题是状态没法自然地在步骤间传递。校验节点发现“资料偏旧”，希望检索节点用这个原因重新构造关键词。在链式写法里，这通常要靠一个全局字典或者层层传参来做。一开始只有两三个字段还好，等我把“历史检索词、上一轮结论、错误原因、重试次数”都塞进去以后，那个全局字典已经变成了谁都不敢动的黑盒。改一处字段，另外两处可能就漏了同步。

第二个问题是流程规则一变，代码结构就得跟着重排。产品后来提了需求：校验失败以后，先基于失败原因精化检索词，最多重试三次，如果第三次还是不行，就走人工兜底。用链式代码写这个分支逻辑，意味着外层要套上多层的if/else，每加一个分支，分支之间的耦合就高一层。这类代码不是不能写，是真的难以维护，也没法可靠测试。

第三个问题最致命，是没法中断和恢复。链式代码一旦开始跑，要么一口气跑到结束，要么自己实现复杂的暂停机制。可到了真实业务里，人工审核可能要等好几分钟甚至更久，一个调用卡在那里，整个流程都得跟着停摆。

1.2 图编排到底给了我们什么

LangGraph是LangChain生态里针对Agent执行过程的状态编排框架。它不负责选模型、不负责调工具，干的事只有一件：把你定义的每一步业务逻辑组织成一张有状态的图，然后按图执行。和链式调用相比，它提供的核心能力可以归结成四点：

• 显式状态：所有节点读写同一个State对象，不再有藏在函数参数里的隐式流转。
• 条件路由：边可以根据节点返回的结果，决定下一步去哪个节点。
• 循环与重试：图允许存在环，反思、修正、重写这类Agent行为变成结构内天然支持的路径。
• 暂停与恢复：通过检查点机制，一次运行可以被挂起，之后从断点继续。

可以简单理解成：链式调用是一条固定的流水线，工人A做完给工人B，工人B做完给工人C，中途想改流程就得把线拆了重装；而状态图更像一个带着任务面板的调度中心，每个工人只负责自己那一道工序，状态都写在共享面板上，换工序顺序只改接线，不用重写工人。

当然，图编排不是银弹。如果你的Agent流程就是检索→回答这种一次性调用，用链式反而更轻。这也是我后来给团队定的选型标准：只有流程中包含“循环、分支、人工介入、断点续跑”四件事中至少两件时，才值得上状态图。

2. 先把State讲透：它相当于整个编排系统的共享数据库

2.1 从TypedDict声明开始，但真正的魔法藏在Annotated里

在LangGraph里定义一个图的起点是声明State。最常用的方式是用TypedDict描述字段，配合Annotated声明字段的更新方式。看一个例子：

from typing import TypedDict, Annotated, Literal import operator from langgraph.graph import StateGraph, START, END class AgentState(TypedDict): topic: str # 调研主题 notes: Annotated[list, operator.add] # 检索笔记，自动追加 feedback: str # 上一轮校验反馈 retry_count: int # 已经重试了几次 draft: str # 当前草稿 final_report: str # 最终报告

这里有两层含义。TypedDict只是给状态定义了一个“外壳”，告诉你和LangGraph这个状态会包含哪些字段，每个字段是什么类型。真正影响行为的是Annotated的第二参数。LangGraph把notes: Annotated[list, operator.add]里的operator.add当成一个reducer：当某个节点返回新notes时，LangGraph不会直接覆盖旧值，而是把旧列表和新列表拼在一起。

用一句话概括State的作用：它是一个贯穿整张图的共享数据对象，每个节点都接收它的当前版本，业务代码处理完以后，把要更新的字段返回出来，LangGraph按字段上的reducer合并出下一版本。

2.2 Reducer机制：覆盖、追加、还有自定义合并策略

Reducer是我觉得理解LangGraph的分水岭。就好比多人协作修改一份文档，如果每个人都直接覆盖同一个段落，后写的就会把先写的顶掉；如果事先约定新内容只能追加到文末，就不会互相挤掉。Reducer就是这种约定。

LangGraph里Reducer的默认规则是覆盖，也就是节点返回某个字段的新值，就直接替换旧值。比如search_node里返回{"retry_count": 3}，上一次的count会被直接换成3。这个规则适用于topic、draft这种“只需要最新版本”的字段。

第二种是追加，典型写法就是Annotated[list, operator.add]。检索笔记、消息历史这种会持续累积的字段，最好都声明成追加型。这样每次节点产生新笔记，旧笔记不会丢。

第三种是自定义Reducer。你完全可以自己写一个小函数，LangGraph只要求它接收两个参数——当前旧值和新值，返回合并后的结果。我写过的一个用法是保留最新的N条消息：

def keep_last_n(old_messages: list, new_messages: list) -> list: return (old_messages + new_messages)[-10:] class AgentState(TypedDict): messages: Annotated[list, keep_last_n]

为什么需要Reducer？因为状态图里可能存在并行分支。两条分支同时想要写同一个字段，如果不定义合并策略，LangGraph就会采用默认覆盖，后执行完的分支把先执行完的分支的结果丢掉了。Reducer就是给这类冲突场景提供的“冲突消解规则”。

2.3 返回值约定：部分更新、空返回、和千万别改传入对象

节点函数的返回值是它和LangGraph之间的通信协议，约定很简单：

• 返回一个dict，表示对State做部分更新，只更新dict里出现的key，其余字段保持不变。
• 返回None，表示这个节点不修改任何状态。
• 返回空dict效果等同于None。

三条约定之外，还有一条非常容易踩的经验：不要在节点内部直接修改传入的state对象。比如state["notes"].append("...")再返回空dict，看起来也能达到目的，但这样做绕过了reducer机制，在并行分支和检查点恢复时会出现状态对不上的玄学问题。正确做法永远是构造新值返回，让LangGraph去走reducer。

注意：绝对不要在节点内部直接修改传入的state对象。构造新值返回，让LangGraph走reducer，是唯一稳妥的写法。

还有个细节：节点读state时不要假设字段一定存在。图的第一轮里，feedback可能压根没写过。习惯用state.get("feedback", "")这种写法，能省掉很多“KeyError”现场。

3. Node和Edge的本质：状态机上最小粒度的业务单元

3.1 Node就是一个“输入State、输出State变更”的普通函数

Node是图里真正干活的单元。它可以调用大模型、调工具、跑计算，但在LangGraph眼里，Node就是一个接收State并返回State变更的函数。它的签名统一是(state) -> dict | None。

我实际写节点时，最舒服的一点是它完全不用关心自己在图里的位置。一个检索节点不需要知道上游是哪个节点，只要state里有所需字段（比如topic），它就能独立运行。这让单元测试变得非常轻：构造一份state字典，直接调用节点函数，就能验证逻辑。

写一个示例：

def search_node(state: AgentState) -> dict: query = state["topic"] if state.get("feedback"): query = f"{query}，重点针对反馈: {state['feedback']}" results = fake_search(query) # 真实项目里替换成检索服务 return { "notes": [f"[第{state.get('retry_count', 0) + 1}轮] {r}" for r in results[:3]], "retry_count": state.get("retry_count", 0) + 1, }

这个节点同时更新了notes和retry_count两个字段，分别走追加和覆盖两种reducer，一次调用可以混着用。节点内部只干一件事：基于已有状态算出新状态。

3.2 边的三种接法：顺序、条件、并行

节点定义好了之后，剩下的就是接线。LangGraph支持三类边。

第一类是普通顺序边。workflow.add_edge(node_a, node_b)表示node_a执行完就执行node_b。它适合没有分叉的路径。

第二类是条件边。它由三部分组成：来源节点、路由函数、路由映射表。路由函数接收当前State，返回一个字符串；映射表把这个字符串对应到目标节点。前面已有的should_retry就是这个思路。

第三类是并行边。在LangGraph里，从一个节点同时连到多个下游节点，就形成了并行Fan-out；这些并行分支最终汇聚到一个下游节点时，就是Fan-in。图引擎会等所有并行分支执行完，再继续跑汇总节点。并行分支共享同一个State，好处是数据不用单独传，坏处是如果没有好的Reducer，多个分支同时写一个字段就会互相覆盖。实际做法一般是：并行分支各自写自己独立命名的字段，最后汇聚节点统一读取并汇总。

3.3 compile之后，你写的拓扑结构才变成一个可执行程序

把节点和边注册完之后，调用compile()会做两件事：一是做结构校验，检查节点之间是否连通、边的指向是否存在、条件边返回的节点名是否合法；二是把这套图结构打包成一个可调用的执行器。

workflow = StateGraph(AgentState) workflow.add_node("search", search_node) workflow.add_node("draft", draft_node) workflow.add_edge(START, "search") workflow.add_edge("search", "draft") workflow.add_edge("draft", END) app = workflow.compile() result = app.invoke({"topic": "低代码平台的数据权限设计"}) print(result["draft"])

我刚开始用的时候，觉得StateGraph(AgentState)这一行有点多余：State已经写在TypedDict里了，为什么编译时还要再传一次？后来才明白，这一步是告诉执行引擎“这张图的State结构长什么样”，好让它在运行时统一处理Reducer和检查点。没有这层声明，LangGraph不知道哪些字段该合并、哪些该覆盖。

到这里为止，一套最小的图已经能跑了。但真正让LangGraph和普通链式代码拉开差距的，是接下来要说的循环和人工介入。那是Agent场景里最硬的需求，也是链式代码几乎做不优雅的部分。

4. 循环、断点恢复与人工介入：状态图真正拉开差距的地方

4.1 条件边指向历史节点，就形成了循环

在链式代码里，循环靠 while 和 for 实现；在状态图里，循环就是一条“往回指”的条件边。比如check_node校验完草稿，条件边返回 “retry” 时，路由目标直接指回search_node，图执行器就会带着更新后的State再跑一遍检索节点。每跑一轮，retry_count加1，直到条件边返回“finish”跳出。

这个设计把“重试”从代码逻辑变成了图结构，好处是显而易见的：所有节点仍然是纯函数式的输入输出，不需要自己维护循环变量；循环次数、跳出条件全部集中在一个路由函数里，想改规则只改一处。

循环路径上最容易出的问题是忘记更新退出条件。如果路由函数只判断retry_count >= 3，但整个图的任何节点都没有把retry_count往上加，这个判断就永远是False，图就会在循环里一直跑下去。更稳妥的做法是在进入循环的那个节点里统一累加计数器，同时把路由函数的判断条件作为唯一出口，两者配合才不会出现死循环。

4.2 Checkpointer：把一次运行变成可以被暂停和恢复的会话

LangGraph的另一个关键机制是检查点。当你用compile(checkpointer=...)编译图时，每次节点执行完，都会把当前State快照写入检查点存储。再次执行时，你传入一个thread_id，执行器就从这个thread_id对应的状态继续跑，而不是从零开始。

from langgraph.checkpoint.memory import InMemorySaver memory = InMemorySaver() app = workflow.compile(checkpointer=memory) config = {"configurable": {"thread_id": "task-001"}} # 第一次运行，跑到某个节点（比如人工节点）停下来 app.invoke({"topic": "某个主题"}, config) # 第二次运行，从上次停止的位置继续执行 app.invoke({"topic": "某个主题"}, config)

对真实业务来说，这个能力比想象中有用得多。进程崩溃了，状态没丢；一个人工审核等了好几个小时，确认之后还能准确回到原来的位置继续跑；甚至同一套图可以同时跑多个任务，每个任务用不同thread_id隔离状态。InMemorySaver在单机、单进程的小型服务里够用，生产环境一般要换成持久化的检查点实现，但接口用法是一致的。

4.3 interrupt()：把图挂起，等一个人来拍板

真正让我朋友从“while循环论”里闭嘴的，是interrupt这个能力。在不带状态编排的Agent代码里，想实现“等人工确认后再继续”非常别扭，通常要拆成两个流程，还得自己维护待确认任务的持久化。LangGraph里只需要在节点里调用interrupt。

from langgraph.types import interrupt, Command def review_node(state: AgentState) -> dict: decision = interrupt({"draft": state["draft"]}) if decision == "approve": return {"final_report": state["draft"]} return {"draft": f"根据反馈修订: {decision}"}

图执行到interrupt时会停下来，把中间数据（这里是一份草稿）交给调用方；你拿着这份草稿给人工看，等人工写下决定，再用Command把决定喂回图，图会从interrupt点继续执行。这个机制让“机器先干活，人来拍板”成为流程的一等公民，而不是事后补救。

我个人的使用经验是：interrupt节点尽量设计成“只暴露必要信息”。真实项目里，人工审核看到的内容越精简越好，不要把检索笔记、中间日志全塞进去。我在调研报告助手里只让审核人看到草稿和三条核心摘要，人看得快，也更容易判断。

5. 一个多轮调研Agent的完整搭建：从State定义到逐轮跑通

5.1 需求拆解与State设计

前面的概念都拆开讲了，这一节把它们拼在一起，做一个能用的小系统。假设需求是这样的：用户输入一个调研主题，Agent要多次检索、整理出草稿、校验草稿质量；校验不过就带着反馈回到检索节点精化搜索词，最多循环3轮；校验通过之后进入人工审核，审核通过才输出最终报告。

这个需求包含了条件路由、循环、人工介入三类特性，正好覆盖前面所有机制。先定义State：

from typing import TypedDict, Annotated, Literal import operator from langgraph.graph import StateGraph, START, END from langgraph.checkpoint.memory import InMemorySaver from langgraph.types import interrupt, Command class ReportState(TypedDict): topic: str notes: Annotated[list, operator.add] feedback: str retry_count: int draft: str final_report: str

5.2 节点实现：让每个业务动作独立成函数

然后定义五个节点。检索节点负责根据topic和feedback生成查询词并追加笔记，同时把retry_count加一；整理节点把notes汇总成一份草稿；校验节点判断草稿是否可用，并返回feedback；人工审核节点interrupt等待确认；收尾节点写final_report。

def search_node(state: ReportState) -> dict: query = state["topic"] if state.get("feedback"): query += f"，并按反馈修正: {state['feedback']}" results = ["相关资料A", "相关资料B", "相关资料C"] # 实际项目替换为真实检索 new_notes = [f"[第{state.get('retry_count', 0) + 1}轮] {r}" for r in results] return {"notes": new_notes, "retry_count": state.get("retry_count", 0) + 1} def draft_node(state: ReportState) -> dict: content = "\n".join(state["notes"]) draft = f"关于{state['topic']}的调研草稿：\n{content}" return {"draft": draft} def check_node(state: ReportState) -> dict: # 真实项目里让大模型判断；这里简化成前两轮必失败，第三轮通过 quality = "pass" if state["retry_count"] >= 3 else "fail" return {"feedback": "" if quality == "pass" else "信息不够新，需要补查最新进展"} def review_node(state: ReportState) -> dict: decision = interrupt({"draft": state["draft"]}) if decision == "approve": return {"final_report": state["draft"]} return {"draft": f"按审核意见修订：{decision}"} def finalize_node(state: ReportState) -> dict: return {"final_report": state["draft"]}

5.3 条件路由、编译与逐轮状态验证

接下来连边。这里有一条条件边，从check_node出发，根据校验结果选择回检索节点还是进人工审核；如果校验失败但重试次数已经到上限，也进人工审核兜底。

def should_retry(state: ReportState) -> Literal["retry", "finish"]: if state["feedback"] == "": return "finish" if state["retry_count"] >= 3: return "finish" return "retry" workflow = StateGraph(ReportState) workflow.add_node("search", search_node) workflow.add_node("draft", draft_node) workflow.add_node("check", check_node) workflow.add_node("review", review_node) workflow.add_node("finalize", finalize_node) workflow.add_edge(START, "search") workflow.add_edge("search", "draft") workflow.add_edge("draft", "check") workflow.add_conditional_edges( "check", should_retry, {"retry": "search", "finish": "review"}, ) workflow.add_edge("review", "finalize") workflow.add_edge("finalize", END) app = workflow.compile(checkpointer=InMemorySaver())

最后用stream模式逐轮看状态变化。第一次invoke会停在interrupt，拿到草稿后，再以Command(resume="approve")继续跑，最终拿到final_report。下面是前两轮跑失败、第三轮进人工审核的状态快照：

代码到这里为止。给你留个小小的挑战：把should_retry里的条件从“retry_count >= 3”改成“notes长度超过20就强制退出”，看看整个流程的行为变化。这种规则改动在链式代码里要拆好几个函数，在状态图里只需要改一条边和一个路由函数。

6. 我踩过的坑：状态污染、路由失配与调试三板斧

6.1 并行分支同写一个字段，后写把先写顶掉了

我第一次做并行检索时，把两个并行分支各自返回的notes都定义成了普通字段，结果跑完经常只剩其中一路的结果。原因就是默认Reducer是覆盖，两个分支都写notes，后执行完的覆盖先执行完的。后来把notes改成Annotated[list, operator.add]才解决。凡是会被多个节点写入的累积型字段，优先考虑追加型Reducer；如果并行分支要各自保留独立结果，就给每个分支单独命名一个字段，最后汇聚节点再统一处理。这个原则写在代码注释里都不为过。

6.2 条件边返回了未注册的节点名

还有一次，编译时一切正常，运行到条件边才报错。原因是路由函数里返回的字符串是 “search”，而我注册节点时用的是 “search_node”。LangGraph编译时不会检查条件边映射里的目标名是否真实存在，这个问题直到运行到那一步才暴露。后来我习惯把节点名定义成模块级常量，路由映射和节点注册共用同一组常量，从源头避免手写字符串不一致。再配合类型注解返回值的Literal，编辑器也能提前帮你标红。

6.3 大对象堆在State里，循环越多性能越差

LangGraph为了隔离状态，会对写入State的数据做拷贝处理。如果你的节点把每次检索的全文都追加进notes，循环三轮后state里会攒下一大批重复文本，内存和反序列化开销都会上去。我的优化习惯是：检索节点只存摘要或与问题最相关的几个片段，原始长文本存在外部存储里，用id引用；需要完整内容时再临时读取。State尽量轻量——它是用来流转任务信息的，不是用来当数据库的。

6.4 调试三板斧：先看图、再单测、最后stream逐步看

排查状态图问题，我一般按这个顺序来。

第一板斧是可视化。LangGraph本身提供了图结构导出接口，可以看到节点和边的完整拓扑，检查条件边连的到底是不是想去的节点。

第二板斧是节点单测。不管整张图有多复杂，每个节点本身就是个普通函数，直接构造一份固定的State调用它，看返回值和State变更是否符合预期。绝大多数逻辑bug在这一步就能定位。

第三板斧是用stream模式按步骤观察。invoke只给你终态，stream能让你看到每一步节点执行完后的State变化。哪一步状态没更新、哪一步updated到了多余字段，一眼就能看出来。

调试Agent和调试普通后端服务的思路不一样。普通后端看日志就行，状态图这种“每一步都在改变状态快照”的程序，最重要的是盯住状态本身的变化。把这套三板斧用熟，比翻着文档猜强得多。

最后再分享一个我在实际项目里的习惯：给节点命名的时候，按“动词+对象”统一风格（比如search、draft、check），不要混用名词和动词；同时把图里所有节点名集中放在一个常量模块里，节点注册、条件边映射、测试用例都引用同一份常量。这样图变复杂以后，改名字不会连带出一堆字符串失配。状态图编排看起来把简单的事变复杂了，但它把“流程状态”变成了一等公民，一旦你的Agent开始需要反思、重试、人工介入，这个复杂度就是值得的。