1. 从零理解 LangGraph它到底解决了什么问题1.1 为什么传统链式调用不够用了如果你之前用过大模型的应用开发框架大概率接触过“链”这个概念——把提示词模板、模型调用、输出解析器串成一条线输入进去结果出来。这种模式处理简单任务没问题比如“翻译一段话”“总结一篇文章”一条路走到黑就行。但一旦任务变得复杂比如需要根据中间结果决定下一步做什么、需要回头重新处理、需要同时跑多个分支再汇总链式结构就开始捉襟见肘了。我举个实际场景你就明白了。假设你要做一个“智能客服工单处理”的智能体用户发来一个问题系统先判断这是咨询类还是投诉类如果是咨询类直接查知识库回答如果是投诉类还要判断紧急程度紧急的转人工不紧急的走自动回复模板。这中间还有可能用户补充信息需要回到判断环节重新走一遍。这种“带条件分支、带循环、带状态记忆”的流程用传统的链式写法会写得非常拧巴代码里全是 if-else 嵌套维护起来很痛苦。LangGraph 就是冲着这类问题来的。它把智能体的工作流建模成一张图图里有节点和边节点代表一个处理步骤边代表步骤之间的流转关系。你可以定义条件边让流程根据当前状态动态选择走哪条路也可以定义环让流程回到之前的节点重新处理。更关键的是它内置了状态管理机制整个流程共享一份状态数据每个节点都能读取和修改它这样多步骤之间的信息传递就变得非常自然。1.2 LangGraph 的核心概念拆解在动手写代码之前有几个核心概念必须先搞清楚不然后面看代码会一头雾水。我用生活化的类比来解释。状态State你可以把它想象成一个“共享文件夹”。整个图在运行过程中所有节点都可以往这个文件夹里放东西、拿东西。比如第一个节点放进去“用户问题”第二个节点放进去“问题分类结果”第三个节点根据这两个信息决定下一步。状态通常用一个字典或者带类型注解的类来表示。节点Node节点就是图里的一个“工位”每个工位负责干一件具体的事。它可以是一次大模型调用可以是一段数据处理逻辑也可以是一个工具函数的执行。节点接收当前状态作为输入处理后返回状态的更新部分。边Edge边定义了工位之间的“传送带”。普通边就是无条件地从 A 传到 B条件边则像一个“分拣员”根据当前状态里的某个值决定下一步去 B 还是去 C。图Graph把节点和边组装起来就形成了完整的流程图。LangGraph 提供了编译机制编译后的图可以像普通函数一样被调用传入初始状态返回最终状态。注意LangGraph 和 LangChain 不是替代关系。LangGraph 更底层它管的是流程编排和状态流转LangChain 提供的是模型封装、提示词模板、工具定义这些“零件”。实际开发中两者经常搭配使用用 LangChain 的组件作为 LangGraph 节点的内部实现。1.3 哪些场景适合用 LangGraph不是所有任务都需要上 LangGraph。如果你的流程是一条直线没有分支没有循环那用简单的链式调用就够了引入图结构反而增加复杂度。但以下几类场景LangGraph 的优势会非常明显多轮对话系统需要记住对话历史根据用户意图切换不同处理逻辑还可能回到之前的环节补充信息。复杂决策流程比如风控审核、工单路由、多级审批模拟每一步都要根据上一步的结果决定下一步走向。多智能体协作多个智能体各司其职有的负责规划、有的负责执行、有的负责审核它们之间需要传递消息和状态。需要人工介入的流程某些关键节点需要暂停等待人工确认确认后再继续执行LangGraph 的中断机制可以支持这种模式。我个人的经验是当你发现自己的业务流程用文字描述时出现了“如果……就……否则……”“重复直到……”“同时进行……”这类词汇就应该考虑用 LangGraph 来建模了。2. 环境搭建与第一个可运行的图2.1 依赖安装与版本选择动手之前先把环境弄干净。我强烈建议用虚拟环境避免和系统里其他项目的依赖打架。Python 版本建议 3.10 以上因为 LangGraph 用到了一些较新的类型注解特性。python -m venv langgraph-env source langgraph-env/bin/activate # Windows 用 langgraph-env\Scripts\activate pip install langgraph langchain langchain-openai这里解释一下这几个包的分工。langgraph是核心库提供图的定义和运行机制langchain提供基础组件抽象langchain-openai是模型接入层如果你用其他模型服务换成对应的包就行。版本方面LangGraph 迭代比较快建议用pip install langgraph --upgrade装最新稳定版老版本有些 API 差异比较大网上教程可能对不上。实操心得我踩过一个坑就是 LangGraph 和 LangChain 的版本不匹配导致导入报错。建议装完之后用pip list | grep lang看一下两个库的版本如果差得太多手动指定一个兼容的组合。一般来说同一个月发布的版本兼容性比较好。2.2 最小可运行示例一个两节点的图先不搞复杂的写一个最简单的图跑通流程建立信心。这个图只有两个节点第一个节点把输入的数字加一第二个节点把结果乘以二。from typing import TypedDict from langgraph.graph import StateGraph, END # 定义状态结构 class MyState(TypedDict): number: int result: int # 定义节点函数 def add_one(state: MyState) - dict: return {number: state[number] 1} def multiply_two(state: MyState) - dict: return {result: state[number] * 2} # 构建图 builder StateGraph(MyState) builder.add_node(add_one, add_one) builder.add_node(multiply_two, multiply_two) # 定义边 builder.set_entry_point(add_one) builder.add_edge(add_one, multiply_two) builder.add_edge(multiply_two, END) # 编译 graph builder.compile() # 运行 output graph.invoke({number: 5}) print(output) # {number: 6, result: 12}这段代码虽然简单但把 LangGraph 的核心要素都包含了状态定义、节点函数、边的连接、入口点设置、编译运行。你把这几个概念对应到上一节讲的类比上应该能对上号。2.3 状态设计的几个关键决策状态设计是 LangGraph 开发中最容易出问题的地方我单独拎出来讲。状态本质上是一个 TypedDict 或者 Pydantic 模型它决定了整个流程能传递哪些信息。第一个决策是状态粒度。状态里放太多东西会让每个节点都变得臃肿放太少又会导致节点之间信息不够用。我的经验是按照“流程中需要共享的最小信息集”来设计。比如一个问答流程状态里至少要有用户原始问题、当前处理阶段、已收集的信息、最终答案。中间过程的临时变量尽量放在节点内部不要塞进状态。第二个决策是状态更新方式。LangGraph 默认的更新是“合并”语义节点返回的字典会和现有状态合并。但有些场景你需要覆盖而不是合并比如一个列表字段你希望新值替换旧值而不是追加。这时候可以用Annotated配合自定义的 reducer 函数来控制。from typing import Annotated from operator import add class MyState(TypedDict): messages: Annotated[list, add] # 新消息会追加到列表 count: int # 默认覆盖语义第三个决策是状态的不可变性。虽然技术上你可以直接修改传入的 state 字典但强烈建议不要这么做。始终返回新的字典片段让 LangGraph 去处理合并。这样做的好处是流程可追溯调试的时候能清楚看到每个节点对状态做了什么改动。3. 条件分支与循环让流程真正“活”起来3.1 条件边的实现方式条件边是 LangGraph 区别于普通工作流引擎的核心特性。它的作用是在某个节点执行完之后根据当前状态决定下一步走哪个节点。实现条件边需要一个路由函数它接收当前状态返回一个字符串这个字符串对应接下来要去的节点名称。然后在添加边的时候用add_conditional_edges把路由函数和节点映射关系传进去。def route_by_category(state: MyState) - str: if state[category] 咨询: return answer_question elif state[category] 投诉: return handle_complaint else: return fallback builder.add_conditional_edges( classify, # 从这个节点出发 route_by_category, # 用这个函数决定去向 { answer_question: answer_question, handle_complaint: handle_complaint, fallback: fallback } )这里有个细节要注意路由函数返回的字符串必须和映射字典里的键对应上否则运行时会报错。我建议把节点名称定义成常量路由函数和映射字典都引用常量避免手写字符串拼错。3.2 循环的构建与终止条件循环在 LangGraph 里就是一条边指回了之前的节点。但循环必须有终止条件否则会无限执行下去。终止条件通常放在条件边里判断。举个实际例子一个“信息收集”流程智能体不断向用户提问直到收集齐所有必要信息才进入下一步。def should_continue(state: MyState) - str: required_fields [name, issue, contact] collected state.get(collected, {}) if all(field in collected for field in required_fields): return process else: return ask_more builder.add_conditional_edges( ask_more, should_continue, { process: process, ask_more: ask_more # 指回自己形成循环 } )注意循环节点指回自己是一种常见写法但要确保状态在每次循环中都有变化否则判断条件永远不变就成了死循环。我一般会在状态里加一个计数器循环超过一定次数就强制跳出作为兜底保护。3.3 并行分支与结果汇总有些任务可以拆成多个子任务并行处理最后汇总结果。LangGraph 支持从一个节点出发连多条边到不同节点这些节点会依次执行注意默认是顺序执行不是真正的并行计算但逻辑上是并行的分支。builder.add_edge(split, branch_a) builder.add_edge(split, branch_b) builder.add_edge(branch_a, merge) builder.add_edge(branch_b, merge)汇总节点需要能够处理来自多个分支的状态更新。如果两个分支都往同一个列表字段里追加数据用Annotated[list, add]就能自动合并。如果两个分支写的是同一个标量字段后执行的会覆盖先执行的这时候就要在汇总节点里做特殊处理比如把两个分支的结果分别存在不同字段里。我实际做过的项目中并行分支特别适合“多角度分析”场景。比如一个内容审核流程同时从“敏感词检测”“情感倾向分析”“事实一致性校验”三个角度分析最后汇总成一个综合判断。每个分支独立处理互不干扰汇总节点负责整合。4. 状态管理与持久化让智能体记住上下文4.1 内存状态与检查点机制默认情况下LangGraph 的状态只在单次调用期间存在调用结束就没了。但很多场景需要跨调用保持状态比如多轮对话。这时候就需要用到**检查点Checkpoint**机制。检查点的工作原理是每次图执行到一个节点结束时把当前状态快照保存下来。下次用同一个会话 ID 调用时可以从上次的快照恢复继续往下走。from langgraph.checkpoint.memory import MemorySaver memory MemorySaver() graph builder.compile(checkpointermemory) # 第一次调用 config {configurable: {thread_id: conversation-001}} graph.invoke({messages: [你好]}, config) # 第二次调用会自动恢复上次的状态 graph.invoke({messages: [我想问个问题]}, config)thread_id是会话标识同一个 ID 的多次调用共享状态。这个机制让多轮对话的实现变得非常简单你不需要手动管理对话历史LangGraph 帮你存了。4.2 持久化后端的选型考量MemorySaver把状态存在内存里进程重启就丢了只适合开发和测试。生产环境需要持久化到外部存储。LangGraph 支持多种后端选型时考虑这几个因素后端类型适用场景优点注意事项内存开发调试零配置速度快进程重启丢失SQLite单机小规模文件存储无需额外服务并发写入有限制PostgreSQL生产环境并发好可靠需要维护数据库Redis高并发场景读写极快数据可能丢失需配合其他存储我个人的建议是开发阶段用内存上线初期用 SQLite 过渡用户量上来之后换 PostgreSQL。切换后端只需要改检查点对象的初始化代码业务逻辑不用动迁移成本很低。4.3 状态更新的调试技巧状态在多个节点之间流转出了问题往往很难定位是哪个节点改错了。我总结了一个调试方法在每个节点函数入口和出口打印状态的关键字段。def my_node(state: MyState) - dict: print(f[my_node] 输入状态: {state}) result {field: new_value} print(f[my_node] 输出更新: {result}) return result更高级的做法是用 LangGraph 提供的流式输出功能它可以实时返回每个节点的执行结果配合可视化工具能看到状态在每一步的变化。这个在排查复杂流程问题时特别有用。5. 工具调用与外部能力接入5.1 把工具封装成节点智能体之所以“智能”很大程度体现在它能调用外部工具——查数据库、调 API、执行计算等等。在 LangGraph 里工具调用通常封装成一个节点。from langchain_core.tools import tool tool def search_knowledge(query: str) - str: 根据关键词搜索知识库 # 实际实现可能是查向量数据库 return f关于{query}的搜索结果... def tool_node(state: MyState) - dict: # 从状态里取出需要调用的工具和参数 tool_name state[next_tool] tool_args state[tool_args] if tool_name search_knowledge: result search_knowledge.invoke(tool_args) else: result 未知工具 return {tool_result: result}这种写法的好处是工具调用的逻辑和流程控制分离了。流程控制由边来决定什么时候进入工具节点工具节点只负责执行。5.2 模型决策与工具选择的衔接实际场景中通常是模型来决定调用哪个工具。流程是这样的模型节点输出一个“工具调用意图”条件边根据这个意图路由到对应的工具节点工具执行完再回到模型节点模型根据工具结果生成最终回复。def model_node(state: MyState) - dict: # 调用大模型传入对话历史和可用工具列表 response llm.invoke(state[messages]) if response.tool_calls: return { next_tool: response.tool_calls[0][name], tool_args: response.tool_calls[0][args], messages: [response] } else: return {messages: [response], next_tool: none} def route_after_model(state: MyState) - str: if state.get(next_tool) none: return end else: return tool这个模式在官方文档里叫 ReAct 循环——推理、行动、观察、再推理。用 LangGraph 实现起来非常直观每个环节对应一个节点循环通过条件边指回模型节点来实现。5.3 工具调用的错误处理工具调用失败是常态网络超时、参数错误、返回格式不对都可能发生。如果不处理整个流程就崩了。我的做法是在工具节点里包一层异常捕获把错误信息也当作一种结果返回给模型让模型决定怎么应对。def tool_node(state: MyState) - dict: try: result execute_tool(state[next_tool], state[tool_args]) return {tool_result: result, tool_error: None} except Exception as e: return {tool_result: None, tool_error: str(e)}然后在模型节点里如果发现tool_error不为空就在提示词里告诉模型“上次工具调用失败了错误信息是XXX请决定是重试还是换一种方式”。这样智能体就有了基本的容错能力。实操心得工具描述的质量直接影响模型选择工具的准确率。我踩过的坑是工具描述写得太简略模型经常选错工具。后来把每个工具的功能、适用场景、参数含义都写清楚准确率明显提升。工具描述本身就是提示词工程的一部分值得花时间打磨。6. 常见问题与排查技巧实录6.1 状态字段丢失或类型错误这是新手最常遇到的问题。表现是运行时报 KeyError或者某个字段的值类型不对。根本原因通常是状态定义不完整或者节点返回了状态里没有的字段。排查步骤第一检查 TypedDict 里是否定义了所有用到的字段第二检查每个节点返回的字典键是否都在状态定义里第三如果用了 Annotated reducer确认 reducer 函数的输入输出类型匹配。我一般会在开发阶段加一个校验函数在每个节点执行前检查状态完整性早发现早修复。6.2 条件边路由不生效条件边不生效通常有三种原因路由函数返回值与映射字典的键不匹配、路由函数抛异常导致默认走了某条边、条件边添加的位置不对。排查时先在路由函数里加打印确认它被调用了以及返回值是什么。然后检查映射字典的键是否和返回值完全一致大小写、空格都要注意。最后确认add_conditional_edges的第一个参数是正确的源节点名称。6.3 循环无法终止循环不终止的根因是终止条件永远为假。可能是状态更新逻辑有问题导致判断依据的字段一直没变也可能是判断条件写错了比如用了or而不是and。我的兜底方案是在状态里加一个step_count字段每次循环加一判断条件里加上step_count max_steps。这样即使逻辑有问题也不会无限跑下去。6.4 检查点恢复后状态不对用检查点做多轮对话时有时候恢复出来的状态和预期不符。常见原因是thread_id用错了或者状态更新没有正确写入检查点。排查方法调用graph.get_state(config)查看当前检查点里存的状态是什么和预期对比。如果发现状态是旧的检查一下是不是在调用时忘了传 config或者 thread_id 变了。问题现象可能原因排查方法KeyError状态字段未定义检查 TypedDict 定义路由不跳转返回值不匹配打印路由函数返回值死循环终止条件恒假加计数器兜底状态不恢复thread_id 不一致检查 config 参数工具调用失败参数格式错误打印工具输入输出6.5 性能优化的一点经验图跑得慢通常瓶颈在模型调用上。优化方向有几个一是减少不必要的模型调用能用规则判断的就不要调模型二是把可以并行的节点拆开虽然 LangGraph 默认顺序执行但你可以用异步节点来实现真正的并行三是缓存重复的模型调用结果同样的输入直接返回缓存。我在一个项目里把“意图分类”从模型调用改成了关键词匹配加模型兜底的混合方案整体响应时间降了差不多一半。不是所有判断都需要大模型该用规则的时候就用规则这是实战中很重要的取舍。7. 从 Demo 到可用一些工程化建议7.1 图的模块化拆分当流程变复杂节点数量超过十个之后把所有代码写在一个文件里会很难维护。我的做法是按功能拆分成多个模块每个模块定义自己的子图然后在主图里把子图当作节点来用。LangGraph 支持子图嵌套编译后的子图可以直接add_node到父图里。这样每个子图可以独立测试独立演进主图只负责高层编排。这个思路和微服务架构很像每个子图是一个“服务”主图是“API 网关”。7.2 日志与可观测性生产环境必须要有日志。我建议在每个节点函数里记录节点名称、输入状态的关键字段、输出更新、执行耗时。这些日志汇总起来就能还原出每次调用的完整执行路径。如果条件允许可以把这些日志接入可视化面板按 thread_id 分组展示这样排查问题时能直观看到流程走了哪些节点、在每个节点停留了多久、状态怎么变化的。7.3 测试策略LangGraph 的测试分两层节点级测试和图级测试。节点级测试就是单独测每个节点函数给一个输入状态断言输出更新是否符合预期。图级测试是端到端测试给一个初始状态断言最终状态是否正确。我习惯用 mock 替换模型调用这样测试不依赖外部服务跑得快也稳定。对于条件边要专门测试每个分支是否都能正确到达。对于循环要测试正常终止和异常终止两种情况。7.4 版本兼容与升级LangGraph 还在快速迭代API 偶尔会有破坏性变更。我的建议是锁定版本号升级前先看变更日志确认影响范围。升级后在测试环境跑一遍完整的回归测试没问题再上生产。另外把图的定义和业务逻辑分离这样即使 LangGraph 的 API 变了只需要改图定义那部分代码业务逻辑不受影响。这个原则在任何框架的使用中都适用。8. 一个完整的实战案例拆解8.1 需求描述与流程设计假设要做一个“技术文档问答助手”用户用自然语言提问助手从文档库中检索相关内容并生成回答。如果检索不到相关内容助手要主动告知用户并建议换个问法。流程设计如下入口节点接收用户问题分类节点判断问题类型概念解释、操作步骤、故障排查检索节点根据问题类型选择不同的检索策略生成节点根据检索结果生成回答质量检查节点判断回答是否合格不合格则回到检索节点重新检索最多重试两次最后输出节点返回最终回答。8.2 状态结构设计class QAState(TypedDict): question: str question_type: str search_results: list answer: str quality_score: float retry_count: int final_output: str这个状态覆盖了流程中所有需要传递的信息。retry_count用于控制循环次数quality_score用于质量检查节点的判断依据。8.3 关键节点的实现要点分类节点用模型判断问题类型提示词里给出三类定义和示例。检索节点根据类型选择不同的检索策略概念解释类用语义检索操作步骤类用关键词加语义混合检索故障排查类优先匹配错误码。生成节点的提示词里包含检索到的文档片段要求模型基于这些片段回答不要编造。质量检查节点用模型给回答打分低于阈值就触发重试。8.4 条件边与循环控制分类节点之后的条件边根据question_type路由到不同的检索策略节点。质量检查节点之后的条件边判断如果quality_score达标或者retry_count超过两次走向输出节点否则retry_count加一回到检索节点。这个案例把前面讲的所有概念都用上了状态管理、条件分支、循环、工具调用、模型决策。你把这个案例跑通LangGraph 的基本用法就掌握得差不多了。后续遇到更复杂的场景无非是节点更多、分支更细、状态更丰富核心思路是一样的。