先说一个我最近在处理的任务做一个研究型问答Agent输入一个主题它要先拆解成几个搜索方向再去检索资料然后判断当前信息够不够不够就换关键词再搜一轮最后把结论整理成文。刚开始我用链式Pipeline写写完发现别扭得要命——因为流程里有一个非常关键的“判断动作”决定是继续搜索还是生成答案这不是靠顺序执行能表达的。后来切到LangGraph的StateGraph把流程画成一张有向图所有控制流都放进节点和边里一下子就通透了。这篇是系列第4篇。前几篇聊了模型调用、工具封装、Agent循环的基本概念这一篇不讲虚的直接上一个可以跑通的手写StateGraph重点拆解三个东西State怎么设计、Node怎么写、条件路由怎么控制多步推理。所谓“手写”就是不依赖LangGraph高层封装出来的Agent类而是把StateGraph的底层API一个个用起来确保每一步都长在自己手里。1. 为什么链式调用会撞墙一个真实的多步推理任务1.1 用链式Pipeline写这个任务看起来也“能跑”很多人的第一直觉是把流程写成顺序代码。大致是这样plan llm.invoke(f把主题{topic}拆解成3个搜索方向) for q in plan: results search(q) context \n.join(results) answer llm.invoke(f基于以下资料回答{topic}\n{context})这个写法在演示Demo里完全没问题单线程、顺序执行、逻辑清晰。可一旦你的需求变成“根据搜索结果决定下一步动作”它就有点撑不住了。比如我想让Agent判断“这一轮资料是否足够回答用户问题如果不够重新提炼关键词再搜一次”。用链式代码怎么写要么固定循环两三次要么在一层循环里面套if-else检索结果往哪个变量放、循环结束条件是什么全部要靠临时变量和状态标志位维护。代码一长你分不清“哪个if负责退出、哪个if负责重试、状态到底在哪个作用域里”。更麻烦的是中间状态的追踪。链式Pipeline里plan、query、results、context散落在各个局部变量中一旦某一步结果不符合预期你想看看“第二步拿到的query到底是什么”就得打断点或者到处print调试成本很高。1.2 问题的本质多步推理需要运行时决定流向我后来想明白了这不是代码风格问题是模型本身的问题。多步推理的核心特征是下一步执行哪个操作取决于上一步的输出。用人话说流程不是从A到B到C那么直而是从A到B之后B要看当前的信息量决定去C还是绕回B继续做。这种“运行时动态分支”的能力顺序代码是天然不擅长的——它擅长的是把确定的路走完不擅长在半路上根据路况改道。有向图结构恰恰擅长表达这种逻辑。把每个操作建模成图里的节点把“从哪个操作去哪个操作”建模成有向边再把动态分支建模成“条件边”。这样一来循环、回溯、并行、提前终止全部都能在图上画出来代码只是把这张图翻译一遍而已。1.3 LangGraph的StateGraph为有状态Agent设计的图框架LangGraph框架里最核心的东西就是StateGraph。它把多步推理拆成了三个基础概念State整张图共享的数据结构所有节点都能读也能往里面写。Node一个节点就是一个操作单元输入State输出State的增量更新。Edge节点之间的连接关系可以是无条件边也可以是条件边。这三个概念展开看其实就是把“画流程图”这件事落到了代码上。有个挺准确的类比StateGraph是一张流水线图纸State是工人们共享的一块黑板Node是图纸上的每个工位Edge是工位之间的传送带条件路由则是传送带旁的自动分拣闸机。对比一下链式Pipeline和StateGraph的差异会更直观维度链式PipelineStateGraph流程控制顺序执行分支靠if-else有向图节点之间靠边连接状态管理局部变量手动传递统一的State对象所有节点共享动态分支困难层层嵌套条件路由函数分支映射循环/回溯硬编码循环次数条件边指向已访问节点即可可观测性需要打断点每次节点执行都能拿到状态更新业务扩展改动影响整条链路加一个节点、加一条边即可这也是我执意在项目里手写StateGraph的原因不是链式代码不够用是当推理链路开始出现“判断、循环、回溯”时图结构更接近真实流程改起来也更稳。2. 状态设计StateSchema就是节点之间的共享黑板2.1 为什么要显式定义State而不是拿字典乱传写StateGraph第一步不是写节点而是先把State定义清楚。StateGraph里的State相当于整张图的“内存模型”每个节点都靠它拿输入也靠它回传结果。我第一次写的时候图省事直接用普通的dict到处传结果节点多了之后字段名字拼错一个程序不报错只是某个节点拿到的值是空的排查起来非常痛苦。后来老老实实按LangGraph推荐的方式用TypedDict先把结构定下来。from typing import TypedDict, Annotated import operator class ResearchState(TypedDict): topic: str # 用户输入的主题 search_plan: list[str] # 搜索计划 search_query: str # 当前搜索关键词 search_results: Annotated[list[str], operator.add] # 所有搜索结果累加 enough: bool # 信息是否足够 iterations: int # 已经检索了几轮 final_answer: str # 最终答案用TypedDict的好处是每个节点函数的入参和返回值都能和这个结构对齐。LangGraph在编译图的时候也会拿这个结构做校验字段类型对不上会直接报错而不是静默地等到运行时才出问题。黑板类比在这里很准确每个工位的人都可以看黑板也可以往黑板上加内容但不能把别人的字擦掉除非你负责这块内容。StateGraph的State逻辑上也是这样节点是共享读、按需写的。2.2 Annotated与reducer字段是“覆盖”还是“累加”State里藏着一个特别容易踩坑的设计节点返回State更新时默认是整字段覆盖。也就是说如果你的State是普通的class ResearchState(TypedDict): search_results: list[str]searcher节点每次返回{search_results: [... ]}那么后写的就会把前一次的覆盖掉。在多轮检索场景里你辛辛苦苦搜了三轮结果只留下最后一轮的结果前面的全丢了。LangGraph的解决办法是reducer标注。用Annotated配合operator.add把字段的更新方式从“覆盖”改成“合并”search_results: Annotated[list[str], operator.add]这样每个节点返回的列表就不是替换整个字段而是追加到已有列表后面。上面的示例State里search_results字段我故意用上了这个注解就是让多轮搜索的结果能累积起来。这里有一个我实际踩过的坑当条件路由让同一个searcher节点被再次调用时它返回的结果会被再次追加到search_results里。如果节点不是幂等的同一份内容就被记了两次。所以节点设计要尽量保证返回结果是“这次搜索的新内容”而不是把历史组装一遍再整包返回。万一出现重复可以在reducer里做去重或者接受少量重复在后续节点再清洗。2.3 State设计的几个实操习惯除了reducerState字段本身怎么设计也很有讲究。我的习惯如下字段拆细别塞大杂烩。宁可多几个字段也不要搞一个data: dict把所有东西装进去。字段就是黑板上的一块块区域分得清楚别人维护起来才不头大。命名统一用snake_case字段名本身就是文档。比如search_query一眼就知道是搜索词enough一眼就知道是布尔判断。大对象别进State。State这个对象在每个节点之间传递如果中间结果特别大——比如几万字的文档、图片的base64编码——塞进State会让图跑得臃肿。我的做法是State里只存文件路径或对象引用真正的大体积数据放外部队列或对象存储。给字段设好初始值。TypedDict本身没有默认值概念所以我在invoke的时候会把整个初始状态都给全后面节点的返回再逐步填充。这样既能让编译期校验生效也方便出问题时用完整state做单步复现。一个明确的结构化State相当于先把“流程里有哪些信息要共享”想清楚了这比直接写节点函数要省事得多。后面节点怎么拆、边怎么连其实都受State结构约束。3. 手写StateGraph节点、边和条件路由的代码骨架3.1 节点就是普通函数关键是返回增量StateGraph里的Node没有黑魔法本质就是一个Python函数接收当前完整State返回一个字典字典里是要更新的字段。没出现在返回字典里的字段保持不变。def plan_node(state: ResearchState) - dict: topic state[topic] return { search_plan: [f{topic}核心概念, f{topic}状态存储, f{topic}条件路由实践], search_query: f{topic}核心概念, }这个设计最大的好处是节点可以被单独测试。我写图的时候习惯先脱离LangGraph直接调用plan_node({topic: LangGraph})看返回值对不对通了再接进图里。对于思考型Agent单个节点函数往往就几十行把“调模型、调工具、解析输出”都封装在里面图的组装反而很薄。3.2 三种边的用法入口边、普通边和条件边LangGraph的边有几种接法手写时要分清add_edge(START, node_name)定义图的入口指向哪个节点。add_edge(node_a, node_b)无条件边执行完A一定去B。add_conditional_edges(node_a, route_func, mapping)条件边A执行完之后调用route_func根据返回值决定下一步去哪个节点。条件路由的route_func是一个普通函数入参是当前State返回一个字符串或字符串列表用于并行分发。mapping参数则是“路由函数返回值到目标节点名”的映射关系。route_func本身不执行模型调用它只是一段判断逻辑。我通常把它设计成只读取State里的某个字段比如读enough返回answerer还是searcher。这样判断逻辑可读性好也好测。3.3 完整示例带条件回路的四节点问答Agent下面这个例子就是我开头说的研究型问答Agent用手写StateGraph完整搭一遍。模拟的知识库检索用本地字典代替方便你直接复制跑。from typing import TypedDict, Annotated, Literal import operator from langgraph.graph import StateGraph, START, END from langgraph.checkpoint.memory import MemorySaver KNOWLEDGE { langgraph: LangGraph是构建有状态Agent的图框架核心是StateGraph、节点和条件路由。, state: State是贯穿整个图的共享数据支持reducer合并字段节点可读可写。, routing: 条件路由根据当前状态动态决定下一个执行节点是实现循环和回溯的关键。, } def search_online(query: str) - str: for key, content in KNOWLEDGE.items(): if key in query: return content return f关于「{query}」的资料不足需要调整关键词后补充搜索。 class ResearchState(TypedDict): topic: str search_plan: list[str] search_query: str search_results: Annotated[list[str], operator.add] enough: bool iterations: int final_answer: str def plan_node(state: ResearchState) - dict: topic state[topic] return { search_plan: [f{topic}核心概念, f{topic}状态存储, f{topic}条件路由实践], search_query: f{topic}核心概念, } def search_node(state: ResearchState) - dict: query state[search_query] result search_online(query) next_query LangGraph 状态设计 return { search_results: [f[{query}] - {result}], search_query: next_query, iterations: state[iterations] 1, } def review_node(state: ResearchState) - dict: combined_text .join(state[search_results]) enough state[iterations] 2 or len(combined_text) 100 return {enough: enough} def answer_node(state: ResearchState) - dict: context \n.join(state[search_results]) final_answer f经过{state[iterations]}轮检索结论如下\n{context} return {final_answer: final_answer} def route_review(state: ResearchState) - Literal[searcher, answerer]: return answerer if state[enough] else searcher builder StateGraph(ResearchState) builder.add_node(planner, plan_node) builder.add_node(searcher, search_node) builder.add_node(reviewer, review_node) builder.add_node(answerer, answer_node) builder.add_edge(START, planner) builder.add_edge(planner, searcher) builder.add_edge(searcher, reviewer) builder.add_conditional_edges( reviewer, route_review, {searcher: searcher, answerer: answerer}, ) builder.add_edge(answerer, END) app builder.compile(checkpointerMemorySaver()) config {configurable: {thread_id: demo-research}} initial_state { topic: LangGraph 多步推理, search_plan: [], search_query: , search_results: [], enough: False, iterations: 0, final_answer: , }跑起来之后我们来看每一步的状态更新这里用stream模式输出节点级更新for chunk in app.stream(initial_state, config, stream_modeupdates): for node_name, update in chunk.items(): print(f--- 节点: {node_name} ---) print(update) print()输出大致会长这样--- 节点: planner --- {search_plan: [LangGraph 多步推理核心概念, LangGraph 多步推理状态存储, LangGraph 多步推理条件路由实践], search_query: LangGraph 多步推理核心概念} --- 节点: searcher --- {search_results: [[LangGraph 多步推理核心概念] - LangGraph是构建有状态Agent的图框架...], iterations: 1, search_query: LangGraph 状态设计} --- 节点: reviewer --- {enough: False} --- 节点: searcher --- {search_results: [[LangGraph 状态设计] - State是贯穿整个图的共享数据...], iterations: 2, search_query: LangGraph 状态设计} --- 节点: reviewer --- {enough: True} --- 节点: answerer --- {final_answer: 经过2轮检索结论如下...}整个过程清楚展示了StateGraph多步推理的路径planner先定方向searcher执行检索reviewer判断信息够不够不够就通过条件路由回到searcher再来一轮够了就跳到answerer生成最终答案。这就是“手写”的核心价值——每一条边的走向你都能在代码里精确看到而不是靠一个隐藏在框架里的高层Agent替你决定。4. 调试图结构死路、漏边与可视化检查4.1 构建期最容易报错的三个地方手写StateGraph和写普通Python函数不同代码跑不起来往往不是逻辑问题而是图没“画”对。我遇到的几类典型报错包括边引用了不存在的节点名。比如add_edge(planner, search)但实际注册的节点叫searcher编译或运行时直接报Node not found。节点返回了StateSchema里没有的字段。LangGraph在较新版本中会直接报Unexpected key比如节点返回{result: ...}但State里没有这个字段。图里存在不可达节点。注册了一个节点但没有边指向它图上出现孤立节点编译时同样会报错。我的解决办法很简单节点名不要手写字符串统一用常量收敛起来。class NODES: PLANNER planner SEARCHER searcher REVIEWER reviewer ANSWERER answerer这样在add_edge和mapping里引用节点时IDE有提示拼错马上暴露。虽然看起来多了一点仪式感但在图结构稍微复杂一点之后这几十行常量能省大量检查时间。4.2 条件路由的坑魔法值拼写和死循环条件路由最经典的错误是route函数返回值和mapping的key对不上。比如route返回search但mapping字典里的key是searcher运行时就抛KeyError。这类问题在字符串拼写错误时非常隐蔽因为报错发生在执行到条件边的那一刻不是编译期。所以我建议route函数也返回常量而不是手写字符串def route_review(state: ResearchState) - Literal[searcher, answerer]: if state[enough]: return NODES.ANSWERER return NODES.SEARCHER另一个更严重的问题是无意中的死循环。条件路由可以把流程导回之前的节点理论上这个循环可以是无限的。比如reviewer判断条件永远为Falsesearcher和reviewer就会无限交替执行。我应对的办法是在State里放计数器review的条件里加入最大迭代上限enough state[iterations] 3 or len(combined_text) 100这样即使前面的判断逻辑写砸了最多跑三轮就会强制跳出循环不会让整个Agent卡死在线程里。4.3 用可视化把整张图“打印”出来LangGraph给StateGraph提供了一个非常好用的调试接口编译前就能把图画出来。我常用的做法是print(graph.get_graph().draw_mermaid())这会输出一段Mermaid格式的文本描述整张图的节点和边。你可以把这段文本贴到任意支持Mermaid渲染的在线工具里看到一张清晰的结构图哪些节点是普通边连的哪些节点是条件边连的有没有孤立节点有没有从START到END的完整路径都一目了然。我现在的习惯是每写完一个StateGraph先把图打印出来扫一眼确认结构和我脑子里的流程图一致再开始实际跑数据。这比看代码查边要快得多。4.4 分步验证先测节点再测整图还有一个我强烈推荐的调试习惯不要一上来就整图invoke。先手动构造简单的State直接调用单个节点函数temp_state dict(initial_state) temp_state[search_query] LangGraph 状态设计 print(search_node(temp_state))这一步能快速确认节点函数本身的输入输出有没有问题。节点全测过之后再整图跑。实测下来这样可以把“节点内部逻辑问题”和“图结构连接问题”分开定位排查速度快非常多。5. 从手写StateGraph到更上层的API我的实践体会5.1 什么时候我会坚持手写图很多LangGraph用户一上来就用高层封装省心确实省心但我个人在下面这些场景会坚持手写StateGraph业务流程复杂分支、循环、并行都要精确控制不希望框架替我猜测流程意图。可观测性要求高线上出问题时要能回答“Agent为什么走了这条路”手写图配合State记录很容易做到。团队要长期维护这个Agent手写图的每一处逻辑都能原样看到换人接手也读得懂。手写图确实会多写几十行代码但这笔投入换回来的是控制权和透明性。对一个要跑在生产环境的多步推理流程来说这两点价值远高于少写几行代码。5.2 上层的Agent API其实是“预制好的图”有人可能会问LangGraph本身不是提供了类似create_agent的高层函数吗直接调那个不是更简单确实是。但拆开看高层Agent API内部依然是StateGraph它预先帮你注册了模型节点、工具节点、以及“模型决定是否继续调用工具”的条件路由。也就是说高层API只是把一套通用的图模板封装好了真正跑起来还是State Node EDGE那一套。理解了这一点你去看高层API的日志和报错就会从容很多。比如“工具节点被反复调用”这类现象本质上就是条件路由把流程不断导回工具节点而已。手写一遍StateGraph等于是把Agent框架的“主板电路图”看过一遍之后用品牌机也更知道哪里可能出毛病。5.3 长期实践下来的几条建议最后一个部分分享几条我在项目里沉淀下来的习惯不一定适合所有人但确实帮我少踩了很多坑第一先画图再写代码。代码只是图纸的翻译图纸画错了代码怎么优化都没用。我一般先在纸上画出节点和边特别是条件分支的所有出口确认没有死路再开始写StateGraph。第二State里加一个debug字段。我习惯在每个关键路由节点里把决策原因写进State比如加一个reviewer_note内容是“第X轮信息长度Y判定enoughTrue”。事后看日志就知道分支为什么这样走不用反推。第三配合checkpointer使用。上面示例里我用了MemorySaver它能让整个图保留每一步的执行历史。配合thread_id你在排查时可以拿到任意一条链路的完整状态这对多步推理Agent的线上问题复盘帮助非常大。第四不要把大体积中间结果塞进State。State是每个节点都会接触的对象装太多东西直接影响执行速度和序列化开销。我最常看到的新手问题就是把一整篇文档塞进state然后又在下个节点截断处理。正确做法是State里存文档的路径或ID节点需要时再按需加载。手写StateGraph这件事对我来说最大的收获不是“更底层”这个噱头而是它强制我把Agent的每一步都摆到明面上。多步推理本来就是一个需要精确控制的过程把控制权握在手里模型行为才真正可控。如果你也在用链式代码写类似的分支流程写到头疼建议把这个四节点的示例跑一遍感受一下把流程画成图之后整个思路会清晰很多。 
长河君 · 2026/10/21 3:27:41 · 阅读 1.2k 手写LangGraph StateGraph：多步推理Agent的状态管理与条件路由实战
先说一个我最近在处理的任务做一个研究型问答Agent输入一个主题它要先拆解成几个搜索方向再去检索资料然后判断当前信息够不够不够就换关键词再搜一轮最后把结论整理成文。刚开始我用链式Pipeline写写完发现别扭得要命——因为流程里有一个非常关键的“判断动作”决定是继续搜索还是生成答案这不是靠顺序执行能表达的。后来切到LangGraph的StateGraph把流程画成一张有向图所有控制流都放进节点和边里一下子就通透了。这篇是系列第4篇。前几篇聊了模型调用、工具封装、Agent循环的基本概念这一篇不讲虚的直接上一个可以跑通的手写StateGraph重点拆解三个东西State怎么设计、Node怎么写、条件路由怎么控制多步推理。所谓“手写”就是不依赖LangGraph高层封装出来的Agent类而是把StateGraph的底层API一个个用起来确保每一步都长在自己手里。1. 为什么链式调用会撞墙一个真实的多步推理任务1.1 用链式Pipeline写这个任务看起来也“能跑”很多人的第一直觉是把流程写成顺序代码。大致是这样plan llm.invoke(f把主题{topic}拆解成3个搜索方向) for q in plan: results search(q) context \n.join(results) answer llm.invoke(f基于以下资料回答{topic}\n{context})这个写法在演示Demo里完全没问题单线程、顺序执行、逻辑清晰。可一旦你的需求变成“根据搜索结果决定下一步动作”它就有点撑不住了。比如我想让Agent判断“这一轮资料是否足够回答用户问题如果不够重新提炼关键词再搜一次”。用链式代码怎么写要么固定循环两三次要么在一层循环里面套if-else检索结果往哪个变量放、循环结束条件是什么全部要靠临时变量和状态标志位维护。代码一长你分不清“哪个if负责退出、哪个if负责重试、状态到底在哪个作用域里”。更麻烦的是中间状态的追踪。链式Pipeline里plan、query、results、context散落在各个局部变量中一旦某一步结果不符合预期你想看看“第二步拿到的query到底是什么”就得打断点或者到处print调试成本很高。1.2 问题的本质多步推理需要运行时决定流向我后来想明白了这不是代码风格问题是模型本身的问题。多步推理的核心特征是下一步执行哪个操作取决于上一步的输出。用人话说流程不是从A到B到C那么直而是从A到B之后B要看当前的信息量决定去C还是绕回B继续做。这种“运行时动态分支”的能力顺序代码是天然不擅长的——它擅长的是把确定的路走完不擅长在半路上根据路况改道。有向图结构恰恰擅长表达这种逻辑。把每个操作建模成图里的节点把“从哪个操作去哪个操作”建模成有向边再把动态分支建模成“条件边”。这样一来循环、回溯、并行、提前终止全部都能在图上画出来代码只是把这张图翻译一遍而已。1.3 LangGraph的StateGraph为有状态Agent设计的图框架LangGraph框架里最核心的东西就是StateGraph。它把多步推理拆成了三个基础概念State整张图共享的数据结构所有节点都能读也能往里面写。Node一个节点就是一个操作单元输入State输出State的增量更新。Edge节点之间的连接关系可以是无条件边也可以是条件边。这三个概念展开看其实就是把“画流程图”这件事落到了代码上。有个挺准确的类比StateGraph是一张流水线图纸State是工人们共享的一块黑板Node是图纸上的每个工位Edge是工位之间的传送带条件路由则是传送带旁的自动分拣闸机。对比一下链式Pipeline和StateGraph的差异会更直观维度链式PipelineStateGraph流程控制顺序执行分支靠if-else有向图节点之间靠边连接状态管理局部变量手动传递统一的State对象所有节点共享动态分支困难层层嵌套条件路由函数分支映射循环/回溯硬编码循环次数条件边指向已访问节点即可可观测性需要打断点每次节点执行都能拿到状态更新业务扩展改动影响整条链路加一个节点、加一条边即可这也是我执意在项目里手写StateGraph的原因不是链式代码不够用是当推理链路开始出现“判断、循环、回溯”时图结构更接近真实流程改起来也更稳。2. 状态设计StateSchema就是节点之间的共享黑板2.1 为什么要显式定义State而不是拿字典乱传写StateGraph第一步不是写节点而是先把State定义清楚。StateGraph里的State相当于整张图的“内存模型”每个节点都靠它拿输入也靠它回传结果。我第一次写的时候图省事直接用普通的dict到处传结果节点多了之后字段名字拼错一个程序不报错只是某个节点拿到的值是空的排查起来非常痛苦。后来老老实实按LangGraph推荐的方式用TypedDict先把结构定下来。from typing import TypedDict, Annotated import operator class ResearchState(TypedDict): topic: str # 用户输入的主题 search_plan: list[str] # 搜索计划 search_query: str # 当前搜索关键词 search_results: Annotated[list[str], operator.add] # 所有搜索结果累加 enough: bool # 信息是否足够 iterations: int # 已经检索了几轮 final_answer: str # 最终答案用TypedDict的好处是每个节点函数的入参和返回值都能和这个结构对齐。LangGraph在编译图的时候也会拿这个结构做校验字段类型对不上会直接报错而不是静默地等到运行时才出问题。黑板类比在这里很准确每个工位的人都可以看黑板也可以往黑板上加内容但不能把别人的字擦掉除非你负责这块内容。StateGraph的State逻辑上也是这样节点是共享读、按需写的。2.2 Annotated与reducer字段是“覆盖”还是“累加”State里藏着一个特别容易踩坑的设计节点返回State更新时默认是整字段覆盖。也就是说如果你的State是普通的class ResearchState(TypedDict): search_results: list[str]searcher节点每次返回{search_results: [... ]}那么后写的就会把前一次的覆盖掉。在多轮检索场景里你辛辛苦苦搜了三轮结果只留下最后一轮的结果前面的全丢了。LangGraph的解决办法是reducer标注。用Annotated配合operator.add把字段的更新方式从“覆盖”改成“合并”search_results: Annotated[list[str], operator.add]这样每个节点返回的列表就不是替换整个字段而是追加到已有列表后面。上面的示例State里search_results字段我故意用上了这个注解就是让多轮搜索的结果能累积起来。这里有一个我实际踩过的坑当条件路由让同一个searcher节点被再次调用时它返回的结果会被再次追加到search_results里。如果节点不是幂等的同一份内容就被记了两次。所以节点设计要尽量保证返回结果是“这次搜索的新内容”而不是把历史组装一遍再整包返回。万一出现重复可以在reducer里做去重或者接受少量重复在后续节点再清洗。2.3 State设计的几个实操习惯除了reducerState字段本身怎么设计也很有讲究。我的习惯如下字段拆细别塞大杂烩。宁可多几个字段也不要搞一个data: dict把所有东西装进去。字段就是黑板上的一块块区域分得清楚别人维护起来才不头大。命名统一用snake_case字段名本身就是文档。比如search_query一眼就知道是搜索词enough一眼就知道是布尔判断。大对象别进State。State这个对象在每个节点之间传递如果中间结果特别大——比如几万字的文档、图片的base64编码——塞进State会让图跑得臃肿。我的做法是State里只存文件路径或对象引用真正的大体积数据放外部队列或对象存储。给字段设好初始值。TypedDict本身没有默认值概念所以我在invoke的时候会把整个初始状态都给全后面节点的返回再逐步填充。这样既能让编译期校验生效也方便出问题时用完整state做单步复现。一个明确的结构化State相当于先把“流程里有哪些信息要共享”想清楚了这比直接写节点函数要省事得多。后面节点怎么拆、边怎么连其实都受State结构约束。3. 手写StateGraph节点、边和条件路由的代码骨架3.1 节点就是普通函数关键是返回增量StateGraph里的Node没有黑魔法本质就是一个Python函数接收当前完整State返回一个字典字典里是要更新的字段。没出现在返回字典里的字段保持不变。def plan_node(state: ResearchState) - dict: topic state[topic] return { search_plan: [f{topic}核心概念, f{topic}状态存储, f{topic}条件路由实践], search_query: f{topic}核心概念, }这个设计最大的好处是节点可以被单独测试。我写图的时候习惯先脱离LangGraph直接调用plan_node({topic: LangGraph})看返回值对不对通了再接进图里。对于思考型Agent单个节点函数往往就几十行把“调模型、调工具、解析输出”都封装在里面图的组装反而很薄。3.2 三种边的用法入口边、普通边和条件边LangGraph的边有几种接法手写时要分清add_edge(START, node_name)定义图的入口指向哪个节点。add_edge(node_a, node_b)无条件边执行完A一定去B。add_conditional_edges(node_a, route_func, mapping)条件边A执行完之后调用route_func根据返回值决定下一步去哪个节点。条件路由的route_func是一个普通函数入参是当前State返回一个字符串或字符串列表用于并行分发。mapping参数则是“路由函数返回值到目标节点名”的映射关系。route_func本身不执行模型调用它只是一段判断逻辑。我通常把它设计成只读取State里的某个字段比如读enough返回answerer还是searcher。这样判断逻辑可读性好也好测。3.3 完整示例带条件回路的四节点问答Agent下面这个例子就是我开头说的研究型问答Agent用手写StateGraph完整搭一遍。模拟的知识库检索用本地字典代替方便你直接复制跑。from typing import TypedDict, Annotated, Literal import operator from langgraph.graph import StateGraph, START, END from langgraph.checkpoint.memory import MemorySaver KNOWLEDGE { langgraph: LangGraph是构建有状态Agent的图框架核心是StateGraph、节点和条件路由。, state: State是贯穿整个图的共享数据支持reducer合并字段节点可读可写。, routing: 条件路由根据当前状态动态决定下一个执行节点是实现循环和回溯的关键。, } def search_online(query: str) - str: for key, content in KNOWLEDGE.items(): if key in query: return content return f关于「{query}」的资料不足需要调整关键词后补充搜索。 class ResearchState(TypedDict): topic: str search_plan: list[str] search_query: str search_results: Annotated[list[str], operator.add] enough: bool iterations: int final_answer: str def plan_node(state: ResearchState) - dict: topic state[topic] return { search_plan: [f{topic}核心概念, f{topic}状态存储, f{topic}条件路由实践], search_query: f{topic}核心概念, } def search_node(state: ResearchState) - dict: query state[search_query] result search_online(query) next_query LangGraph 状态设计 return { search_results: [f[{query}] - {result}], search_query: next_query, iterations: state[iterations] 1, } def review_node(state: ResearchState) - dict: combined_text .join(state[search_results]) enough state[iterations] 2 or len(combined_text) 100 return {enough: enough} def answer_node(state: ResearchState) - dict: context \n.join(state[search_results]) final_answer f经过{state[iterations]}轮检索结论如下\n{context} return {final_answer: final_answer} def route_review(state: ResearchState) - Literal[searcher, answerer]: return answerer if state[enough] else searcher builder StateGraph(ResearchState) builder.add_node(planner, plan_node) builder.add_node(searcher, search_node) builder.add_node(reviewer, review_node) builder.add_node(answerer, answer_node) builder.add_edge(START, planner) builder.add_edge(planner, searcher) builder.add_edge(searcher, reviewer) builder.add_conditional_edges( reviewer, route_review, {searcher: searcher, answerer: answerer}, ) builder.add_edge(answerer, END) app builder.compile(checkpointerMemorySaver()) config {configurable: {thread_id: demo-research}} initial_state { topic: LangGraph 多步推理, search_plan: [], search_query: , search_results: [], enough: False, iterations: 0, final_answer: , }跑起来之后我们来看每一步的状态更新这里用stream模式输出节点级更新for chunk in app.stream(initial_state, config, stream_modeupdates): for node_name, update in chunk.items(): print(f--- 节点: {node_name} ---) print(update) print()输出大致会长这样--- 节点: planner --- {search_plan: [LangGraph 多步推理核心概念, LangGraph 多步推理状态存储, LangGraph 多步推理条件路由实践], search_query: LangGraph 多步推理核心概念} --- 节点: searcher --- {search_results: [[LangGraph 多步推理核心概念] - LangGraph是构建有状态Agent的图框架...], iterations: 1, search_query: LangGraph 状态设计} --- 节点: reviewer --- {enough: False} --- 节点: searcher --- {search_results: [[LangGraph 状态设计] - State是贯穿整个图的共享数据...], iterations: 2, search_query: LangGraph 状态设计} --- 节点: reviewer --- {enough: True} --- 节点: answerer --- {final_answer: 经过2轮检索结论如下...}整个过程清楚展示了StateGraph多步推理的路径planner先定方向searcher执行检索reviewer判断信息够不够不够就通过条件路由回到searcher再来一轮够了就跳到answerer生成最终答案。这就是“手写”的核心价值——每一条边的走向你都能在代码里精确看到而不是靠一个隐藏在框架里的高层Agent替你决定。4. 调试图结构死路、漏边与可视化检查4.1 构建期最容易报错的三个地方手写StateGraph和写普通Python函数不同代码跑不起来往往不是逻辑问题而是图没“画”对。我遇到的几类典型报错包括边引用了不存在的节点名。比如add_edge(planner, search)但实际注册的节点叫searcher编译或运行时直接报Node not found。节点返回了StateSchema里没有的字段。LangGraph在较新版本中会直接报Unexpected key比如节点返回{result: ...}但State里没有这个字段。图里存在不可达节点。注册了一个节点但没有边指向它图上出现孤立节点编译时同样会报错。我的解决办法很简单节点名不要手写字符串统一用常量收敛起来。class NODES: PLANNER planner SEARCHER searcher REVIEWER reviewer ANSWERER answerer这样在add_edge和mapping里引用节点时IDE有提示拼错马上暴露。虽然看起来多了一点仪式感但在图结构稍微复杂一点之后这几十行常量能省大量检查时间。4.2 条件路由的坑魔法值拼写和死循环条件路由最经典的错误是route函数返回值和mapping的key对不上。比如route返回search但mapping字典里的key是searcher运行时就抛KeyError。这类问题在字符串拼写错误时非常隐蔽因为报错发生在执行到条件边的那一刻不是编译期。所以我建议route函数也返回常量而不是手写字符串def route_review(state: ResearchState) - Literal[searcher, answerer]: if state[enough]: return NODES.ANSWERER return NODES.SEARCHER另一个更严重的问题是无意中的死循环。条件路由可以把流程导回之前的节点理论上这个循环可以是无限的。比如reviewer判断条件永远为Falsesearcher和reviewer就会无限交替执行。我应对的办法是在State里放计数器review的条件里加入最大迭代上限enough state[iterations] 3 or len(combined_text) 100这样即使前面的判断逻辑写砸了最多跑三轮就会强制跳出循环不会让整个Agent卡死在线程里。4.3 用可视化把整张图“打印”出来LangGraph给StateGraph提供了一个非常好用的调试接口编译前就能把图画出来。我常用的做法是print(graph.get_graph().draw_mermaid())这会输出一段Mermaid格式的文本描述整张图的节点和边。你可以把这段文本贴到任意支持Mermaid渲染的在线工具里看到一张清晰的结构图哪些节点是普通边连的哪些节点是条件边连的有没有孤立节点有没有从START到END的完整路径都一目了然。我现在的习惯是每写完一个StateGraph先把图打印出来扫一眼确认结构和我脑子里的流程图一致再开始实际跑数据。这比看代码查边要快得多。4.4 分步验证先测节点再测整图还有一个我强烈推荐的调试习惯不要一上来就整图invoke。先手动构造简单的State直接调用单个节点函数temp_state dict(initial_state) temp_state[search_query] LangGraph 状态设计 print(search_node(temp_state))这一步能快速确认节点函数本身的输入输出有没有问题。节点全测过之后再整图跑。实测下来这样可以把“节点内部逻辑问题”和“图结构连接问题”分开定位排查速度快非常多。5. 从手写StateGraph到更上层的API我的实践体会5.1 什么时候我会坚持手写图很多LangGraph用户一上来就用高层封装省心确实省心但我个人在下面这些场景会坚持手写StateGraph业务流程复杂分支、循环、并行都要精确控制不希望框架替我猜测流程意图。可观测性要求高线上出问题时要能回答“Agent为什么走了这条路”手写图配合State记录很容易做到。团队要长期维护这个Agent手写图的每一处逻辑都能原样看到换人接手也读得懂。手写图确实会多写几十行代码但这笔投入换回来的是控制权和透明性。对一个要跑在生产环境的多步推理流程来说这两点价值远高于少写几行代码。5.2 上层的Agent API其实是“预制好的图”有人可能会问LangGraph本身不是提供了类似create_agent的高层函数吗直接调那个不是更简单确实是。但拆开看高层Agent API内部依然是StateGraph它预先帮你注册了模型节点、工具节点、以及“模型决定是否继续调用工具”的条件路由。也就是说高层API只是把一套通用的图模板封装好了真正跑起来还是State Node EDGE那一套。理解了这一点你去看高层API的日志和报错就会从容很多。比如“工具节点被反复调用”这类现象本质上就是条件路由把流程不断导回工具节点而已。手写一遍StateGraph等于是把Agent框架的“主板电路图”看过一遍之后用品牌机也更知道哪里可能出毛病。5.3 长期实践下来的几条建议最后一个部分分享几条我在项目里沉淀下来的习惯不一定适合所有人但确实帮我少踩了很多坑第一先画图再写代码。代码只是图纸的翻译图纸画错了代码怎么优化都没用。我一般先在纸上画出节点和边特别是条件分支的所有出口确认没有死路再开始写StateGraph。第二State里加一个debug字段。我习惯在每个关键路由节点里把决策原因写进State比如加一个reviewer_note内容是“第X轮信息长度Y判定enoughTrue”。事后看日志就知道分支为什么这样走不用反推。第三配合checkpointer使用。上面示例里我用了MemorySaver它能让整个图保留每一步的执行历史。配合thread_id你在排查时可以拿到任意一条链路的完整状态这对多步推理Agent的线上问题复盘帮助非常大。第四不要把大体积中间结果塞进State。State是每个节点都会接触的对象装太多东西直接影响执行速度和序列化开销。我最常看到的新手问题就是把一整篇文档塞进state然后又在下个节点截断处理。正确做法是State里存文档的路径或ID节点需要时再按需加载。手写StateGraph这件事对我来说最大的收获不是“更底层”这个噱头而是它强制我把Agent的每一步都摆到明面上。多步推理本来就是一个需要精确控制的过程把控制权握在手里模型行为才真正可控。如果你也在用链式代码写类似的分支流程写到头疼建议把这个四节点的示例跑一遍感受一下把流程画成图之后整个思路会清晰很多。