做 Agent 应用时最让人头疼的问题往往不是“模型不够聪明”而是“流程难以控制”。如果你之前用 LangChain 写过 Agent大概率遇到过这类情况Agent 的内部决策是一个黑盒想插入人工确认环节、想根据用户意图走不同分支、想持久化多轮对话状态都很难下手。LangGraph 正是为解决这类问题而诞生的图编排框架。它把一次 Agent 的执行过程建模成一张有向图每个业务环节是一个节点节点之间的流转关系就是边。这篇文章不会照着官方文档逐页翻译而是从零开始拆解 LangGraph 的三个核心组件State、Node、Edge最后通过一个可运行的完整案例演示如何用条件路由和多 Agent 协作搭建一个真实的业务工作流。本文适合有 Python 基础、想深入控制 Agent 流程的开发者。读完之后你能理解 LangGraph 的状态管理机制能独立编写节点函数和条件路由也能在项目里把多个 Agent 串成一条可维护的工作流。代码示例我会尽量完整方便你直接复制到本地运行。1. 背景与核心概念为什么需要 LangGraph1.1 LangChain 写 Agent 的痛点在 LangGraph 出现之前大部分开发者使用 LangChain 的 Agent 模块来构建大模型应用。LangChain Agent 的模式是“模型自主决定调用哪个工具”看起来很方便但在实际业务中会遇到几个很难绕开的问题。第一是流程不可控。模型可能跳过必须执行的校验步骤直接去调用工具也可能在同一个工具上反复循环浪费大量 token。第二是状态不透明。多轮对话中用户信息、中间结果、工具返回内容都堆在 memory 里缺少结构化的维护方式。第三是分支逻辑难以实现。业务上经常需要“根据用户问题类型走不同处理链”使用 LangChain 原生的 AgentExecutor 实现这种分支很别扭。你当然可以在 Prompt 里严格要求模型“先分类再执行”但 Prompt 约束属于软约束模型一旦出现幻觉整个流程就乱了。LangGraph 用图结构把流程变成了硬约束节点执行什么、节点之间怎么跳转都由代码决定模型只负责节点内部的具体生成任务。1.2 LangGraph 是什么LangGraph 是 LangChain 团队推出的低层编排框架专门用于构建有状态、可编排、可循环的 Agent 应用。官方定位是“LangChain 的编排层”它把 Agent 的工作流建模为一张图图中的节点可以是普通函数、工具函数、LLM 调用也可以是一个子图节点之间的边则定义了执行的先后与条件。LangGraph 和 LangChain 的关系可以这样理解LangChain 提供了模型封装、Prompt 模板、工具加载、向量存储等基础组件LangGraph 则在这些组件之上提供工作流控制能力。你完全可以脱离 LangChain只用 LangGraph 搭配原生 OpenAI SDK 写流程但更常见的做法是两者配合使用。LangGraph 有以下几个关键特性基于状态图StateGraph建模任何一次执行都可被追踪。支持条件路由可以根据节点返回值动态选择下一个节点。原生支持循环Agent 可以在循环中不断调用工具直到满足退出条件。支持并行节点执行多个 Agent 可以同时运行并汇总结果。支持子图嵌套适合把复杂流程拆成多个可复用小图。可插拔 Checkpointer方便实现持久化、断点续跑、时间旅行。1.3 核心三件套State、Node、EdgeLangGraph 的最小认知模型可以浓缩成三个词State、Node、Edge。State 是贯穿整个执行流程的共享状态对象。所有节点读取同一个 State节点执行完毕后把需要更新的字段返回LangGraph 负责合并和传递。Node 是图中的一个执行单元本质是一个 Python 函数输入是当前 State输出是更新后的 State 部分字段。Edge 是节点之间的连接线分为普通边和条件边普通边用于固定顺序流转条件边根据某个函数的结果动态决定下一步走向哪个节点。用一张简单的比喻来理解State 是流水线上的共享工单Node 是每个工位上的工人Edge 是工位之间的传送带。普通边是“做完 A 一定去 B”条件边是“做完 A 后根据工单上的标记决定去 B、C 还是 D”。2. 环境准备与版本说明2.1 安装 LangGraphLangGraph 的安装非常简单直接通过 pip 安装即可。建议在一个全新的虚拟环境中进行避免版本冲突。pip install langgraph如果需要配合 LangChain 使用可以一并安装pip install langchain langchain-openai如果你的项目里需要调用 OpenAI、通义千问等模型还需要安装对应的 Provider 包。以 OpenAI 为例pip install langchain-openai安装完成后可以通过下面这条命令检查版本pip show langgraph由于 LangGraph 迭代速度非常快API 会持续调整本文代码基于撰写时较新的 API 写法。如果你使用的版本较旧可能会遇到个别导入路径不一致的情况比如START、END的导入方式我会在代码注释和常见问题里解释。2.2 运行环境建议本文示例在以下环境验证通过操作系统Windows / macOS / Linux 均可Python 版本建议 3.10 及以上LangGraph 版本以较新版本为准包管理工具pip 或 poetry需要强调一点LangGraph 并不强制依赖 LangChain。如果你只是想体验图编排能力安装langgraph就够了。但在真实项目中我们往往需要使用现成的模型封装和工具包装所以两者通常会一起出现。2.3 本文示例代码结构为了让后面的内容更好理解我们先约定演示项目的文件结构。langgraph-demo/ ├── requirements.txt ├── 01_minimal_graph.py # 最小可运行示例 ├── 02_state_reducer.py # State 与 Reducer 示例 ├── 03_conditional_edge.py # 条件路由示例 ├── 04_multi_agent.py # 多 Agent 协作示例 └── README.md建议大家一步步手动创建这些文件不要直接复制整个项目。手写一遍能明显加深对图结构、状态传递的理解。3. 快速上手写一个最小可运行的 LangGraph 程序3.1 创建一张最简单的图先来看一个最基础的 LangGraph 程序。这个例子不涉及大模型只演示 State、Node、Edge 三个核心概念的最小组合。# 文件路径langgraph-demo/01_minimal_graph.py from typing import TypedDict from langgraph.graph import StateGraph, START, END class MyState(TypedDict): message: str def first_node(state: MyState): print(执行 first_node) return {message: state[message] 第一次处理} def second_node(state: MyState): print(执行 second_node) return {message: state[message] 第二次处理} graph StateGraph(MyState) graph.add_node(first, first_node) graph.add_node(second, second_node) graph.add_edge(START, first) graph.add_edge(first, second) graph.add_edge(second, END) app graph.compile() result app.invoke({message: Hello LangGraph}) print(result)运行效果如下执行 first_node 执行 second_node {message: Hello LangGraph第一次处理第二次处理}这里有几个关键点需要解释。TypedDict用于声明 State 的结构它告诉 LangGraph 这个图的状态包含哪些字段。每个节点函数接收一个字典类型的 state 参数返回一个字典返回内容只需要包含你希望更新的字段即可不需要返回全部 State。LangGraph 会把返回值自动合并到 State 中并传给下一个节点。StateGraph(MyState)创建了一张基于 MyState 类型的状态图。graph.add_node注册节点第一个参数是节点名称第二个参数是对应的处理函数。graph.add_edge定义边START是入口标记END是出口标记。3.2 invoke 执行流程内部到底做了什么当我们调用app.invoke({message: Hello LangGraph})时LangGraph 内部做了以下几件事。第一步接收初始字典作为 State 的初始值。第二步从START出发找到START指向的节点把当前 State 传给该节点函数。第三步节点函数执行完毕后拿到返回值与当前 State 做合并更新 State。第四步根据当前节点的出边找到下一个节点重复执行直到走到END。第五步把最终 State 返回给调用方。注意节点函数的返回值是“部分状态更新”不是完整覆盖。比如 first_node 返回{message: ...}LangGraph 会把 message 字段更新而其他字段如果存在也会保留。这个合并机制在下一步会展开讲。3.3 一个容易理解的执行顺序比喻你可以把这张图想象成一条消息处理流水线。初始消息是Hello LangGraphfirst 工位给消息追加了第一次处理second 工位又追了一句第二次处理最终从流水线出口拿到处理完成的最终消息。这个比喻能帮你理解 LangGraph 最简单的执行机制State 在节点之间按边流动每个节点只关心自己需要读取和修改的字段。4. State 状态管理深度拆解4.1 使用 TypedDict 定义状态结构在真实项目中State 往往不止一个字段。比如一个客服 Agent 可能需要以下状态用户消息列表、当前意图、检索到的知识片段、生成的最终回复、是否需要进行人工介入等。# 文件路径langgraph-demo/02_state_reducer.py from typing import TypedDict, Annotated from langgraph.graph.message import add_messages class AgentState(TypedDict): # 消息列表使用 add_messages reducer 做追加合并 messages: Annotated[list, add_messages] # 用户当前输入 user_input: str # 意图识别结果 intent: str # 中间标记 need_human_review: bool # 最终答案 final_answer: str这里最特殊的是messages字段的类型标注Annotated[list, add_messages]。Annotated是 Python 标准库typing提供的语法它允许你在类型之外附加额外元数据。LangGraph 会读取这个元数据来决定字段的合并方式。4.2 默认的更新规则覆盖合并如果字段没有使用Annotated标注 ReducerLangGraph 默认的更新规则是“每次节点返回新值就直接覆盖旧值”。这在大部分简单场景下是合理的。例如class SimpleState(TypedDict): count: int def increase_node(state: SimpleState): # 注意这里拿到的是旧值 new_count state[count] 1 return {count: new_count}当increase_node执行结束后count字段会被替换成new_count。4.3 Reducer自定义状态合并规则但有些场景下覆盖并非你想要的行为。比如多轮对话中messages 需要不断追加新的历史消息如果用覆盖逻辑每轮对话都会把旧消息冲掉这显然不对。这时就需要 Reducer。add_messages就是 LangGraph 内置的一个 Reducer它的作用是“合并新旧消息列表”。也就是说节点返回的新消息会被追加到 State 中已有消息列表的末尾而不是直接替换整个列表。from typing import TypedDict, Annotated from langgraph.graph.message import add_messages class ChatState(TypedDict): messages: Annotated[list, add_messages] def user_message_node(state: ChatState): return {messages: [{role: user, content: 你好}]} def ai_message_node(state: ChatState): return {messages: [{role: assistant, content: 有什么可以帮你}]}连续执行这两个节点后messages 会同时包含用户消息和助手回复。如果不用add_messages后一个节点返回的消息会把前一个节点的消息覆盖掉。4.4 在节点函数中修改 State 的状态值这是很多初学者会踩坑的地方节点函数到底怎样修改 State准确地说节点函数不应该直接修改传入的 state 参数。正确写法是返回一个字典字典包含你希望更新的字段LangGraph 之后会按照字段的 Reducer 规则把返回值合并进 State。def my_node(state: MyState): # 错误做法直接修改传入对象属性很多情况下不会生效且容易引入隐式 bug # state[final_answer] xxx # 正确做法返回需要更新的字段 return {final_answer: xxx}这样设计的好处是节点函数变成纯函数风格输入 state输出状态变更方便测试、追踪和并行执行。如果所有节点都直接改 state执行顺序一旦调整状态就会变得不可预测。5. Node 与 Edge 开发实战5.1 节点函数的基本要求节点函数是 LangGraph 的最小执行单元。函数签名必须满足第一个参数是当前 State 对象类型为你在StateGraph中声明的 TypedDict。返回值是一个字典包含需要更新的字段也可以返回None表示不更新任何状态。节点的定义方式有两种直接传函数或者传一个可调用对象。def call_llm(state: MyState): return {llm_response: 模拟回复} graph.add_node(llm, call_llm)节点内部通常可以做很多事情调用大模型、调用工具函数、查询数据库、调用外部 API、进行文件读写等。节点本身不关心图结构它只负责处理自己负责的事务然后把处理结果写回 State。5.2 普通边固定顺序流转普通边用于固定流程。add_edge方法接收两个参数分别是源节点和目标节点。graph.add_edge(START, node_a) graph.add_edge(node_a, node_b) graph.add_edge(node_b, END)上面这段代码意味着执行顺序固定为node_a 执行然后 node_b 执行最后结束。多个节点可以连接到同一个目标节点这叫做“汇合”。例如graph.add_edge(node_a, merge_node) graph.add_edge(node_b, merge_node) graph.add_edge(merge_node, END)执行完 node_a 和 node_b 后都会进入 merge_node。但要注意图是并行还是串行取决于add_edge是否处于并行分支中在默认串行情况下LangGraph 会按照拓扑顺序执行。5.3 条件边根据状态动态选择流程条件边是 LangGraph 最强大的能力之一。它的核心思路是通过一个路由函数根据当前 State 计算出下一个节点的名字然后跳转到对应节点。# 文件路径langgraph-demo/03_conditional_edge.py from typing import TypedDict from langgraph.graph import StateGraph, START, END import random class RouteState(TypedDict): user_input: str intent: str result: str def classify_node(state: RouteState): # 模拟意图识别 text state[user_input] if 天气 in text: intent weather elif 时间 in text: intent time else: intent chat print(识别意图, intent) return {intent: intent} def weather_node(state: RouteState): return {result: f【天气】你查询的内容是{state[user_input]}。模拟结果晴20℃。} def time_node(state: RouteState): return {result: f【时间】你查询的内容是{state[user_input]}。模拟结果2025-01-15 14:30。} def chat_node(state: RouteState): return {result: f【通用对话】你说{state[user_input]}。我收到了。} def route_by_intent(state: RouteState) - str: # 路由函数返回目标节点名称 if state[intent] weather: return weather elif state[intent] time: return time return chat graph StateGraph(RouteState) graph.add_node(classify, classify_node) graph.add_node(weather, weather_node) graph.add_node(time, time_node) graph.add_node(chat, chat_node) graph.add_edge(START, classify) graph.add_conditional_edges( classify, route_by_intent, { weather: weather, time: time, chat: chat, }, ) graph.add_edge(weather, END) graph.add_edge(time, END) graph.add_edge(chat, END) app graph.compile() for text in [今天北京天气怎么样, 现在几点了, 你好呀]: result app.invoke({user_input: text}) print(result[result])运行结果识别意图 weather 【天气】你查询的内容是今天北京天气怎么样。模拟结果晴20℃。 识别意图 time 【时间】你查询的内容是现在几点了。模拟结果2025-01-15 14:30。 识别意图 chat 【通用对话】你说你好呀。我收到了。add_conditional_edges接收三个参数第一个参数是源节点名称。第二个参数是路由函数这个函数接收当前 State返回一个字符串。第三个参数是映射字典key 是路由函数可能返回的字符串value 是目标节点名称。如果你的路由函数返回的字符串不在映射字典中LangGraph 会抛出一个运行时错误。因此保险的做法是路由函数里最后写一个兜底返回值作为默认分支。5.4 循环与回环Agent 的自主决策基础Agent 和普通链路的本质区别在于Agent 可以根据中间结果决定是否继续调用工具这需要图具有回环能力。LangGraph 天然支持循环你只需要让条件边从一个节点指回前面的节点。一个常见的“工具调用循环”模型如下开始节点接收用户问题。决策节点判断需要调用哪个工具或者是否直接给出最终答案。工具节点执行具体工具调用把结果写回状态。条件边从工具节点回到决策节点形成循环。问题的关键是要设置终止条件否则会无限循环。比如在 State 中维护一个max_rounds字段每轮循环加一当超过最大值时强制返回END。def should_continue(state: MyState) - str: if state[round] state[max_rounds]: return end return tool_node graph.add_conditional_edges( agent_node, should_continue, { tool_node: tool_node, end: END, }, )这里必须提到 LangGraph 的一个默认行为循环不会自动终止。如果你的代码里有循环边务必在路由函数中提供终止条件。6. 完整实战构建一个带条件路由的多 Agent 协作系统这一节我们把前面所有知识点串起来构建一个更接近真实业务的案例。该案例模拟一个“研发助手”系统里包含三个 Agent代码审查 Agent、单元测试 Agent、安全扫描 Agent。入口节点先判断用户请求属于哪一类然后派发给对应的 Agent最后把结果汇总输出。6.1 需求分析假设你有以下三个独立的人工智能任务代码审查 Agent接收代码片段返回代码规范问题和潜在 bug 建议。单元测试 Agent接收一个函数返回生成的测试用例思路。安全扫描 Agent接收依赖或配置文件返回安全风险提示。这三个 Agent 内部调用大模型但在本文中我们用普通函数模拟重点演示图编排而不是 Agent 内部实现。用户输入一段文本系统先通过“路由节点”判断输入类型然后决定让哪个 Agent 处理。最后无论哪个 Agent 先执行都要进入“汇总节点”生成一份统一格式的报告。6.2 定义状态# 文件路径langgraph-demo/04_multi_agent.py from typing import TypedDict class DevAssistantState(TypedDict): user_question: str task_type: str # code_review | test_generation | security_scan | unknown agent_result: str # 单个 Agent 的处理结果 final_report: str # 汇总报告我们把agent_result设计成公共字段无论哪个 Agent 执行都会把结果写进同一个字段后续的汇总节点只关心这个字段。6.3 编写节点函数def route_node(state: DevAssistantState): text state[user_question] if 审查 in text or review in text.lower(): task_type code_review elif 测试 in text or test in text.lower(): task_type test_generation elif 安全 in text or scan in text.lower(): task_type security_scan else: task_type unknown return {task_type: task_type} def code_review_agent(state: DevAssistantState): code state[user_question] result ( 【代码审查 Agent】\n f你提交的代码片段长度为 {len(code)} 个字符。\n 建议增加 try-except 异常处理避免空指针问题。\n 规范建议补充函数注释统一命名风格。 ) return {agent_result: result} def test_agent(state: DevAssistantState): content state[user_question] result ( 【单元测试 Agent】\n f针对你提供的函数{content[:30]}...\n 建议覆盖以下用例正常输入、空输入、超长输入、异常参数。 ) return {agent_result: result} def security_agent(state: DevAssistantState): content state[user_question] result ( 【安全扫描 Agent】\n f对配置或依赖内容{content[:30]}...\n 风险提示未检测到硬编码密钥但建议启用最小权限策略。 ) return {agent_result: result} def fallback_agent(state: DevAssistantState): result 【兜底 Agent】当前请求未匹配到任何专业 Agent建议提供更明确的需求描述。 return {agent_result: result} def report_node(state: DevAssistantState): report ( f 研发助手汇总报告 \n f需求类型{state[task_type]}\n f处理结果\n{state[agent_result]}\n f 报告结束 \n ) return {final_report: report}这里每个 Agent 节点都只负责更新agent_result字段整个设计是解耦的。如果以后要接入真实大模型只需要替换 Agent 节点内部的模拟实现即可图结构完全不需要变动。6.4 建立图与条件路由from langgraph.graph import StateGraph, START, END def route_by_task_type(state: DevAssistantState) - str: if state[task_type] code_review: return code_review elif state[task_type] test_generation: return test elif state[task_type] security_scan: return security return fallback graph StateGraph(DevAssistantState) graph.add_node(route, route_node) graph.add_node(code_review, code_review_agent) graph.add_node(test, test_agent) graph.add_node(security, security_agent) graph.add_node(fallback, fallback_agent) graph.add_node(report, report_node) graph.add_edge(START, route) graph.add_conditional_edges( route, route_by_task_type, { code_review: code_review, test: test, security: security, fallback: fallback, }, ) graph.add_edge(code_review, report) graph.add_edge(test, report) graph.add_edge(security, report) graph.add_edge(fallback, report) graph.add_edge(report, END) app graph.compile()6.5 运行与验证if __name__ __main__: inputs [ 请帮我审查这段代码\ndef foo(a, b):\n return a/b, 请为下面函数生成测试用例def add(x, y): return x y, 请扫描这个依赖配置requests2.28.0, 你好, ] for q in inputs: result app.invoke({user_question: q}) print(result[final_report]) print(----------------)运行效果类似 研发助手汇总报告 需求类型code_review 处理结果 【代码审查 Agent】 你提交的代码片段长度为 58 个字符。 建议增加 try-except 异常处理避免空指针问题。 规范建议补充函数注释统一命名风格。 报告结束 ----------------到这里你已经实现了一个最基本的“多 Agent 协作”系统。这里的“多 Agent 协作”体现在多个 Agent 节点被条件路由分发多个节点汇聚到同一个汇总节点最终输出统一格式的结果。7. 多 Agent 协作模式扩展7.1 并行执行模式上面的例子是“多选一”的派发模式。另一类常见的多 Agent 协作是“多个 Agent 同时执行然后汇总结果”。比如你可以同时让代码审查 Agent、测试 Agent、安全扫描 Agent 处理同一个任务最后把三份报告合并。LangGraph 中实现并行非常简单只需要从一个节点引出多条边指向不同节点然后再让这些节点都连向同一个汇总节点即可。graph.add_edge(START, split_node) graph.add_edge(split_node, code_review) graph.add_edge(split_node, test) graph.add_edge(split_node, security) graph.add_edge(code_review, merge_node) graph.add_edge(test, merge_node) graph.add_edge(security, merge_node) graph.add_edge(merge_node, END)此时merge_node会等待所有指向它的上游节点执行完毕后再运行从而读取到所有 Agent 的返回结果。并行模式适合子任务之间没有依赖关系、可以独立执行的场景。7.2 子图模式当业务越来越复杂一张图中的节点可能会超过几十个。这时候把图拆成多个子图会更清晰。LangGraph 支持把一张编译好的图作为另一个图的节点。# 子图专门处理代码审查流程 subgraph StateGraph(SubState) subgraph.add_node(...) subgraph_app subgraph.compile() # 主图把子图当作一个节点 main_graph StateGraph(MainState) main_graph.add_node(review_subgraph, subgraph_app)子图内部可以有自己的 State、Node、Edge主图只需要把子图当成一个黑盒即可。这样做的好处是子图可以独立测试、独立复用主图的逻辑也变得更清晰。7.3 监督者模式当存在多个独立 Agent 时另一种常见的组织方式是“监督者模式”。一个主控 Agent 负责监听用户请求并调度其他子 Agent 执行最后汇总结果。在 LangGraph 中监督者模式可以通过一个“分发节点”加条件路由来实现分发节点判断任务类型然后通过条件边把请求转发给对应的 Agent。这个模式本质上和第 6 节的实战案例一致但你可以把分发节点变成一个真正的大模型调用让模型基于对话上下文决定派发哪个 Agent。在多 Agent 系统中“谁来负责最终回答”是一个值得思考的问题。常见的做法是把所有子 Agent 的结果放入 State再由一个汇总节点统一组织成用户友好的回复。这样既保留了每个子 Agent 的专业性也保证了整体体验的一致性。8. 常见问题与排查清单8.1 LangGraph 常见报错对照表在编写和运行 LangGraph 的过程中下面这些问题是高频出现的我整理成了对照表。问题现象常见原因解决思路ModuleNotFoundError: No module named langgraph未安装或安装到了别的环境执行pip install langgraph确认激活正确的虚拟环境无法从 langgraph.graph 导入 START、END版本较旧常量命名不同新版本使用START、END旧版本尝试from langgraph.graph import START_NODE, END_NODE或直接使用空字符串节点返回后 State 没有更新字段可能被默认覆盖逻辑合并或节点返回了 None检查字段是否使用了 Annotated Reducer确认节点函数 return 的是 dict条件路由总是走到兜底分支路由函数返回值不在映射字典中在路由函数中打印返回值与 add_conditional_edges 的映射 key 对比compile() 报错提示存在未连通节点某个节点没有从 START 出发的可达路径检查 add_edge 与 add_conditional_edges 是否完整连接所有节点并确保最终有路径通向 END图执行陷入死循环循环边缺少终止条件在 State 中增加轮次字段在路由函数里判断达到上限后返回 END并行节点执行结果互相覆盖多个节点写同一个 State 字段且该字段使用默认覆盖 reducer为共享聚合字段指定 reducer或让多个并行节点写入不同字段多次 invoke 之间状态残留没有使用 Checkpointer 时每次 invoke 应是独立状态如果希望保持历史状态需要传入 checkpoint 配置或手动维护历史8.2 排查步骤建议当你遇到一个诡异的问题时不妨按下面的顺序排查先把图结构渲染出来确认拓扑调用app.get_graph().print_ascii()可以直观看到节点之间的连接情况。在路由函数和节点函数中增加打印语句观察每次执行的入参和出参。把大型节点拆小逐个 validate 节点函数是否正确返回。检查返回字段名称是否与 TypedDict 中的 key 完全一致少一个字母都会出问题。确认需要跨节点保留的数据是否写入了 State而不是只存在局部变量里。如果依赖了外部模型或网络调用先 mock 掉外部依赖排除环境因素。8.3 关于安装和版本的一些提醒LangGraph 迭代速度很快你在搜索资料时可能会看到不同版本的代码比如StateGraph构造方式、START常量导入、add_conditional_edges参数顺序等都有过变化。遇到 API 对不上的情况优先查看当前安装版本的官方文档不要盲改代码。另外如果项目里同时使用 LangChain注意保持两个框架版本兼容某些情况下新版本 LangChain 可能依赖更新的 LangGraph。如果你只需要图编排能力可以只安装 langgraph不安装 langchain减少依赖冲突。9. 最佳实践与工程建议9.1 State 设计要小而清晰State 是图的数据中枢但它不是数据库也不是日志系统。不要在 State 里塞入大量与当前流程无关的字段。字段越多节点之间隐性耦合就越强调试时越难定位问题。建议把 State 拆成两部分流程状态和业务数据。流程状态如task_type、round、ticket_id等业务数据如user_input、agent_result、final_report等。9.2 节点函数要保持纯函数风格尽量让节点函数不修改外部全局变量不写文件不做隐藏副作用。输入 State输出状态变更这是最安全、最好测试的形态。如果某个节点必须调用外部 API建议把网络请求封装成独立函数便于做单元测试时 mock。9.3 条件路由一定要有兜底无论路由函数判断逻辑多完善都会出现模型输出格式异常、文本匹配失败、外部结果缺失等情况。因此每个条件路由都应该提供 fallback 分支比如返回unknown从路由映射到兜底节点或者直接回到人工处理。不要让路由函数因为 key 不匹配而抛异常。9.4 使用 Checkpointer 实现状态持久化LangGraph 提供了 Checkpointer 机制可以把每一步的 State 保存下来实现断点续跑、人工确认、时间旅行等能力。如果你的业务需要多轮对话长期记忆或者需要人工审批流程强烈建议深入了解这一块。它和普通的多轮对话 memory 不同Checkpointer 保存的是图的完整执行状态可以随时恢复。from langgraph.checkpoint.memory import MemorySaver checkpointer MemorySaver() app graph.compile(checkpointercheckpointer) config {configurable: {thread_id: conversation-1}} result app.invoke(input_data, configconfig)MemorySaver 适合本地测试和教学演示。生产环境建议使用 PostgresSaver 等持久化实现并配置独立的存储服务。这一块涉及具体业务建议阅读官方文档后再落地。9.5 重视日志与可观测性LangGraph 的价值之一是“流程可追踪”。在生产环境中建议给每个节点增加结构化日志记录时间、节点名称、输入关键字段、输出关键字段。结合 LangSmith 等可观测平台可以看到每一次 Agent 执行的完整轨迹。遇到线上问题能很快定位到具体是哪个节点、哪一步导致的。9.6 从小图开始逐步构建复杂系统不要一上来就设计几十个节点的大图。正确的做法是从最小可用链路开始例如“入口节点 - 一个处理节点 - 汇总节点 - END”跑通后再逐步加入条件路由、并行分支、子图。每增加一个节点就做一次完整验证这样定位问题会容易很多。9.7 关于安全与权限如果 LangGraph 工作流涉及数据库访问、文件删除、生产环境变更等操作一定要在节点内部做严格授权校验和安全边界控制。不要把生产环境密钥直接写入代码或 State避免放进日志。Agent 调用外部工具时建议维护工具白名单对工具的入参、出参做校验防止恶意输入造成越权操作。LangGraph 真正吸引人的地方不是因为它是某一个大模型产品而是它把 Agent 开发从“靠 Prompt 碰运气”提升到了“用代码控制流程”的工程化高度。你不用再担心模型擅自跳步也不用为了实现一次条件分支而写一堆 if-else 串行调用。先用一个小例子把 State、Node、Edge 三个概念跑通再逐步往图里添加条件路由、并行分支和子图你会发现自己对 Agent 应用的控制能力有了明显提升。下一篇可以继续深入 Checkpointer、持久化记忆、以及更复杂的监督者多 Agent 协作模式。 LangGraph核心概念与实战：用State、Node、Edge构建可控Agent工作流
做 Agent 应用时最让人头疼的问题往往不是“模型不够聪明”而是“流程难以控制”。如果你之前用 LangChain 写过 Agent大概率遇到过这类情况Agent 的内部决策是一个黑盒想插入人工确认环节、想根据用户意图走不同分支、想持久化多轮对话状态都很难下手。LangGraph 正是为解决这类问题而诞生的图编排框架。它把一次 Agent 的执行过程建模成一张有向图每个业务环节是一个节点节点之间的流转关系就是边。这篇文章不会照着官方文档逐页翻译而是从零开始拆解 LangGraph 的三个核心组件State、Node、Edge最后通过一个可运行的完整案例演示如何用条件路由和多 Agent 协作搭建一个真实的业务工作流。本文适合有 Python 基础、想深入控制 Agent 流程的开发者。读完之后你能理解 LangGraph 的状态管理机制能独立编写节点函数和条件路由也能在项目里把多个 Agent 串成一条可维护的工作流。代码示例我会尽量完整方便你直接复制到本地运行。1. 背景与核心概念为什么需要 LangGraph1.1 LangChain 写 Agent 的痛点在 LangGraph 出现之前大部分开发者使用 LangChain 的 Agent 模块来构建大模型应用。LangChain Agent 的模式是“模型自主决定调用哪个工具”看起来很方便但在实际业务中会遇到几个很难绕开的问题。第一是流程不可控。模型可能跳过必须执行的校验步骤直接去调用工具也可能在同一个工具上反复循环浪费大量 token。第二是状态不透明。多轮对话中用户信息、中间结果、工具返回内容都堆在 memory 里缺少结构化的维护方式。第三是分支逻辑难以实现。业务上经常需要“根据用户问题类型走不同处理链”使用 LangChain 原生的 AgentExecutor 实现这种分支很别扭。你当然可以在 Prompt 里严格要求模型“先分类再执行”但 Prompt 约束属于软约束模型一旦出现幻觉整个流程就乱了。LangGraph 用图结构把流程变成了硬约束节点执行什么、节点之间怎么跳转都由代码决定模型只负责节点内部的具体生成任务。1.2 LangGraph 是什么LangGraph 是 LangChain 团队推出的低层编排框架专门用于构建有状态、可编排、可循环的 Agent 应用。官方定位是“LangChain 的编排层”它把 Agent 的工作流建模为一张图图中的节点可以是普通函数、工具函数、LLM 调用也可以是一个子图节点之间的边则定义了执行的先后与条件。LangGraph 和 LangChain 的关系可以这样理解LangChain 提供了模型封装、Prompt 模板、工具加载、向量存储等基础组件LangGraph 则在这些组件之上提供工作流控制能力。你完全可以脱离 LangChain只用 LangGraph 搭配原生 OpenAI SDK 写流程但更常见的做法是两者配合使用。LangGraph 有以下几个关键特性基于状态图StateGraph建模任何一次执行都可被追踪。支持条件路由可以根据节点返回值动态选择下一个节点。原生支持循环Agent 可以在循环中不断调用工具直到满足退出条件。支持并行节点执行多个 Agent 可以同时运行并汇总结果。支持子图嵌套适合把复杂流程拆成多个可复用小图。可插拔 Checkpointer方便实现持久化、断点续跑、时间旅行。1.3 核心三件套State、Node、EdgeLangGraph 的最小认知模型可以浓缩成三个词State、Node、Edge。State 是贯穿整个执行流程的共享状态对象。所有节点读取同一个 State节点执行完毕后把需要更新的字段返回LangGraph 负责合并和传递。Node 是图中的一个执行单元本质是一个 Python 函数输入是当前 State输出是更新后的 State 部分字段。Edge 是节点之间的连接线分为普通边和条件边普通边用于固定顺序流转条件边根据某个函数的结果动态决定下一步走向哪个节点。用一张简单的比喻来理解State 是流水线上的共享工单Node 是每个工位上的工人Edge 是工位之间的传送带。普通边是“做完 A 一定去 B”条件边是“做完 A 后根据工单上的标记决定去 B、C 还是 D”。2. 环境准备与版本说明2.1 安装 LangGraphLangGraph 的安装非常简单直接通过 pip 安装即可。建议在一个全新的虚拟环境中进行避免版本冲突。pip install langgraph如果需要配合 LangChain 使用可以一并安装pip install langchain langchain-openai如果你的项目里需要调用 OpenAI、通义千问等模型还需要安装对应的 Provider 包。以 OpenAI 为例pip install langchain-openai安装完成后可以通过下面这条命令检查版本pip show langgraph由于 LangGraph 迭代速度非常快API 会持续调整本文代码基于撰写时较新的 API 写法。如果你使用的版本较旧可能会遇到个别导入路径不一致的情况比如START、END的导入方式我会在代码注释和常见问题里解释。2.2 运行环境建议本文示例在以下环境验证通过操作系统Windows / macOS / Linux 均可Python 版本建议 3.10 及以上LangGraph 版本以较新版本为准包管理工具pip 或 poetry需要强调一点LangGraph 并不强制依赖 LangChain。如果你只是想体验图编排能力安装langgraph就够了。但在真实项目中我们往往需要使用现成的模型封装和工具包装所以两者通常会一起出现。2.3 本文示例代码结构为了让后面的内容更好理解我们先约定演示项目的文件结构。langgraph-demo/ ├── requirements.txt ├── 01_minimal_graph.py # 最小可运行示例 ├── 02_state_reducer.py # State 与 Reducer 示例 ├── 03_conditional_edge.py # 条件路由示例 ├── 04_multi_agent.py # 多 Agent 协作示例 └── README.md建议大家一步步手动创建这些文件不要直接复制整个项目。手写一遍能明显加深对图结构、状态传递的理解。3. 快速上手写一个最小可运行的 LangGraph 程序3.1 创建一张最简单的图先来看一个最基础的 LangGraph 程序。这个例子不涉及大模型只演示 State、Node、Edge 三个核心概念的最小组合。# 文件路径langgraph-demo/01_minimal_graph.py from typing import TypedDict from langgraph.graph import StateGraph, START, END class MyState(TypedDict): message: str def first_node(state: MyState): print(执行 first_node) return {message: state[message] 第一次处理} def second_node(state: MyState): print(执行 second_node) return {message: state[message] 第二次处理} graph StateGraph(MyState) graph.add_node(first, first_node) graph.add_node(second, second_node) graph.add_edge(START, first) graph.add_edge(first, second) graph.add_edge(second, END) app graph.compile() result app.invoke({message: Hello LangGraph}) print(result)运行效果如下执行 first_node 执行 second_node {message: Hello LangGraph第一次处理第二次处理}这里有几个关键点需要解释。TypedDict用于声明 State 的结构它告诉 LangGraph 这个图的状态包含哪些字段。每个节点函数接收一个字典类型的 state 参数返回一个字典返回内容只需要包含你希望更新的字段即可不需要返回全部 State。LangGraph 会把返回值自动合并到 State 中并传给下一个节点。StateGraph(MyState)创建了一张基于 MyState 类型的状态图。graph.add_node注册节点第一个参数是节点名称第二个参数是对应的处理函数。graph.add_edge定义边START是入口标记END是出口标记。3.2 invoke 执行流程内部到底做了什么当我们调用app.invoke({message: Hello LangGraph})时LangGraph 内部做了以下几件事。第一步接收初始字典作为 State 的初始值。第二步从START出发找到START指向的节点把当前 State 传给该节点函数。第三步节点函数执行完毕后拿到返回值与当前 State 做合并更新 State。第四步根据当前节点的出边找到下一个节点重复执行直到走到END。第五步把最终 State 返回给调用方。注意节点函数的返回值是“部分状态更新”不是完整覆盖。比如 first_node 返回{message: ...}LangGraph 会把 message 字段更新而其他字段如果存在也会保留。这个合并机制在下一步会展开讲。3.3 一个容易理解的执行顺序比喻你可以把这张图想象成一条消息处理流水线。初始消息是Hello LangGraphfirst 工位给消息追加了第一次处理second 工位又追了一句第二次处理最终从流水线出口拿到处理完成的最终消息。这个比喻能帮你理解 LangGraph 最简单的执行机制State 在节点之间按边流动每个节点只关心自己需要读取和修改的字段。4. State 状态管理深度拆解4.1 使用 TypedDict 定义状态结构在真实项目中State 往往不止一个字段。比如一个客服 Agent 可能需要以下状态用户消息列表、当前意图、检索到的知识片段、生成的最终回复、是否需要进行人工介入等。# 文件路径langgraph-demo/02_state_reducer.py from typing import TypedDict, Annotated from langgraph.graph.message import add_messages class AgentState(TypedDict): # 消息列表使用 add_messages reducer 做追加合并 messages: Annotated[list, add_messages] # 用户当前输入 user_input: str # 意图识别结果 intent: str # 中间标记 need_human_review: bool # 最终答案 final_answer: str这里最特殊的是messages字段的类型标注Annotated[list, add_messages]。Annotated是 Python 标准库typing提供的语法它允许你在类型之外附加额外元数据。LangGraph 会读取这个元数据来决定字段的合并方式。4.2 默认的更新规则覆盖合并如果字段没有使用Annotated标注 ReducerLangGraph 默认的更新规则是“每次节点返回新值就直接覆盖旧值”。这在大部分简单场景下是合理的。例如class SimpleState(TypedDict): count: int def increase_node(state: SimpleState): # 注意这里拿到的是旧值 new_count state[count] 1 return {count: new_count}当increase_node执行结束后count字段会被替换成new_count。4.3 Reducer自定义状态合并规则但有些场景下覆盖并非你想要的行为。比如多轮对话中messages 需要不断追加新的历史消息如果用覆盖逻辑每轮对话都会把旧消息冲掉这显然不对。这时就需要 Reducer。add_messages就是 LangGraph 内置的一个 Reducer它的作用是“合并新旧消息列表”。也就是说节点返回的新消息会被追加到 State 中已有消息列表的末尾而不是直接替换整个列表。from typing import TypedDict, Annotated from langgraph.graph.message import add_messages class ChatState(TypedDict): messages: Annotated[list, add_messages] def user_message_node(state: ChatState): return {messages: [{role: user, content: 你好}]} def ai_message_node(state: ChatState): return {messages: [{role: assistant, content: 有什么可以帮你}]}连续执行这两个节点后messages 会同时包含用户消息和助手回复。如果不用add_messages后一个节点返回的消息会把前一个节点的消息覆盖掉。4.4 在节点函数中修改 State 的状态值这是很多初学者会踩坑的地方节点函数到底怎样修改 State准确地说节点函数不应该直接修改传入的 state 参数。正确写法是返回一个字典字典包含你希望更新的字段LangGraph 之后会按照字段的 Reducer 规则把返回值合并进 State。def my_node(state: MyState): # 错误做法直接修改传入对象属性很多情况下不会生效且容易引入隐式 bug # state[final_answer] xxx # 正确做法返回需要更新的字段 return {final_answer: xxx}这样设计的好处是节点函数变成纯函数风格输入 state输出状态变更方便测试、追踪和并行执行。如果所有节点都直接改 state执行顺序一旦调整状态就会变得不可预测。5. Node 与 Edge 开发实战5.1 节点函数的基本要求节点函数是 LangGraph 的最小执行单元。函数签名必须满足第一个参数是当前 State 对象类型为你在StateGraph中声明的 TypedDict。返回值是一个字典包含需要更新的字段也可以返回None表示不更新任何状态。节点的定义方式有两种直接传函数或者传一个可调用对象。def call_llm(state: MyState): return {llm_response: 模拟回复} graph.add_node(llm, call_llm)节点内部通常可以做很多事情调用大模型、调用工具函数、查询数据库、调用外部 API、进行文件读写等。节点本身不关心图结构它只负责处理自己负责的事务然后把处理结果写回 State。5.2 普通边固定顺序流转普通边用于固定流程。add_edge方法接收两个参数分别是源节点和目标节点。graph.add_edge(START, node_a) graph.add_edge(node_a, node_b) graph.add_edge(node_b, END)上面这段代码意味着执行顺序固定为node_a 执行然后 node_b 执行最后结束。多个节点可以连接到同一个目标节点这叫做“汇合”。例如graph.add_edge(node_a, merge_node) graph.add_edge(node_b, merge_node) graph.add_edge(merge_node, END)执行完 node_a 和 node_b 后都会进入 merge_node。但要注意图是并行还是串行取决于add_edge是否处于并行分支中在默认串行情况下LangGraph 会按照拓扑顺序执行。5.3 条件边根据状态动态选择流程条件边是 LangGraph 最强大的能力之一。它的核心思路是通过一个路由函数根据当前 State 计算出下一个节点的名字然后跳转到对应节点。# 文件路径langgraph-demo/03_conditional_edge.py from typing import TypedDict from langgraph.graph import StateGraph, START, END import random class RouteState(TypedDict): user_input: str intent: str result: str def classify_node(state: RouteState): # 模拟意图识别 text state[user_input] if 天气 in text: intent weather elif 时间 in text: intent time else: intent chat print(识别意图, intent) return {intent: intent} def weather_node(state: RouteState): return {result: f【天气】你查询的内容是{state[user_input]}。模拟结果晴20℃。} def time_node(state: RouteState): return {result: f【时间】你查询的内容是{state[user_input]}。模拟结果2025-01-15 14:30。} def chat_node(state: RouteState): return {result: f【通用对话】你说{state[user_input]}。我收到了。} def route_by_intent(state: RouteState) - str: # 路由函数返回目标节点名称 if state[intent] weather: return weather elif state[intent] time: return time return chat graph StateGraph(RouteState) graph.add_node(classify, classify_node) graph.add_node(weather, weather_node) graph.add_node(time, time_node) graph.add_node(chat, chat_node) graph.add_edge(START, classify) graph.add_conditional_edges( classify, route_by_intent, { weather: weather, time: time, chat: chat, }, ) graph.add_edge(weather, END) graph.add_edge(time, END) graph.add_edge(chat, END) app graph.compile() for text in [今天北京天气怎么样, 现在几点了, 你好呀]: result app.invoke({user_input: text}) print(result[result])运行结果识别意图 weather 【天气】你查询的内容是今天北京天气怎么样。模拟结果晴20℃。 识别意图 time 【时间】你查询的内容是现在几点了。模拟结果2025-01-15 14:30。 识别意图 chat 【通用对话】你说你好呀。我收到了。add_conditional_edges接收三个参数第一个参数是源节点名称。第二个参数是路由函数这个函数接收当前 State返回一个字符串。第三个参数是映射字典key 是路由函数可能返回的字符串value 是目标节点名称。如果你的路由函数返回的字符串不在映射字典中LangGraph 会抛出一个运行时错误。因此保险的做法是路由函数里最后写一个兜底返回值作为默认分支。5.4 循环与回环Agent 的自主决策基础Agent 和普通链路的本质区别在于Agent 可以根据中间结果决定是否继续调用工具这需要图具有回环能力。LangGraph 天然支持循环你只需要让条件边从一个节点指回前面的节点。一个常见的“工具调用循环”模型如下开始节点接收用户问题。决策节点判断需要调用哪个工具或者是否直接给出最终答案。工具节点执行具体工具调用把结果写回状态。条件边从工具节点回到决策节点形成循环。问题的关键是要设置终止条件否则会无限循环。比如在 State 中维护一个max_rounds字段每轮循环加一当超过最大值时强制返回END。def should_continue(state: MyState) - str: if state[round] state[max_rounds]: return end return tool_node graph.add_conditional_edges( agent_node, should_continue, { tool_node: tool_node, end: END, }, )这里必须提到 LangGraph 的一个默认行为循环不会自动终止。如果你的代码里有循环边务必在路由函数中提供终止条件。6. 完整实战构建一个带条件路由的多 Agent 协作系统这一节我们把前面所有知识点串起来构建一个更接近真实业务的案例。该案例模拟一个“研发助手”系统里包含三个 Agent代码审查 Agent、单元测试 Agent、安全扫描 Agent。入口节点先判断用户请求属于哪一类然后派发给对应的 Agent最后把结果汇总输出。6.1 需求分析假设你有以下三个独立的人工智能任务代码审查 Agent接收代码片段返回代码规范问题和潜在 bug 建议。单元测试 Agent接收一个函数返回生成的测试用例思路。安全扫描 Agent接收依赖或配置文件返回安全风险提示。这三个 Agent 内部调用大模型但在本文中我们用普通函数模拟重点演示图编排而不是 Agent 内部实现。用户输入一段文本系统先通过“路由节点”判断输入类型然后决定让哪个 Agent 处理。最后无论哪个 Agent 先执行都要进入“汇总节点”生成一份统一格式的报告。6.2 定义状态# 文件路径langgraph-demo/04_multi_agent.py from typing import TypedDict class DevAssistantState(TypedDict): user_question: str task_type: str # code_review | test_generation | security_scan | unknown agent_result: str # 单个 Agent 的处理结果 final_report: str # 汇总报告我们把agent_result设计成公共字段无论哪个 Agent 执行都会把结果写进同一个字段后续的汇总节点只关心这个字段。6.3 编写节点函数def route_node(state: DevAssistantState): text state[user_question] if 审查 in text or review in text.lower(): task_type code_review elif 测试 in text or test in text.lower(): task_type test_generation elif 安全 in text or scan in text.lower(): task_type security_scan else: task_type unknown return {task_type: task_type} def code_review_agent(state: DevAssistantState): code state[user_question] result ( 【代码审查 Agent】\n f你提交的代码片段长度为 {len(code)} 个字符。\n 建议增加 try-except 异常处理避免空指针问题。\n 规范建议补充函数注释统一命名风格。 ) return {agent_result: result} def test_agent(state: DevAssistantState): content state[user_question] result ( 【单元测试 Agent】\n f针对你提供的函数{content[:30]}...\n 建议覆盖以下用例正常输入、空输入、超长输入、异常参数。 ) return {agent_result: result} def security_agent(state: DevAssistantState): content state[user_question] result ( 【安全扫描 Agent】\n f对配置或依赖内容{content[:30]}...\n 风险提示未检测到硬编码密钥但建议启用最小权限策略。 ) return {agent_result: result} def fallback_agent(state: DevAssistantState): result 【兜底 Agent】当前请求未匹配到任何专业 Agent建议提供更明确的需求描述。 return {agent_result: result} def report_node(state: DevAssistantState): report ( f 研发助手汇总报告 \n f需求类型{state[task_type]}\n f处理结果\n{state[agent_result]}\n f 报告结束 \n ) return {final_report: report}这里每个 Agent 节点都只负责更新agent_result字段整个设计是解耦的。如果以后要接入真实大模型只需要替换 Agent 节点内部的模拟实现即可图结构完全不需要变动。6.4 建立图与条件路由from langgraph.graph import StateGraph, START, END def route_by_task_type(state: DevAssistantState) - str: if state[task_type] code_review: return code_review elif state[task_type] test_generation: return test elif state[task_type] security_scan: return security return fallback graph StateGraph(DevAssistantState) graph.add_node(route, route_node) graph.add_node(code_review, code_review_agent) graph.add_node(test, test_agent) graph.add_node(security, security_agent) graph.add_node(fallback, fallback_agent) graph.add_node(report, report_node) graph.add_edge(START, route) graph.add_conditional_edges( route, route_by_task_type, { code_review: code_review, test: test, security: security, fallback: fallback, }, ) graph.add_edge(code_review, report) graph.add_edge(test, report) graph.add_edge(security, report) graph.add_edge(fallback, report) graph.add_edge(report, END) app graph.compile()6.5 运行与验证if __name__ __main__: inputs [ 请帮我审查这段代码\ndef foo(a, b):\n return a/b, 请为下面函数生成测试用例def add(x, y): return x y, 请扫描这个依赖配置requests2.28.0, 你好, ] for q in inputs: result app.invoke({user_question: q}) print(result[final_report]) print(----------------)运行效果类似 研发助手汇总报告 需求类型code_review 处理结果 【代码审查 Agent】 你提交的代码片段长度为 58 个字符。 建议增加 try-except 异常处理避免空指针问题。 规范建议补充函数注释统一命名风格。 报告结束 ----------------到这里你已经实现了一个最基本的“多 Agent 协作”系统。这里的“多 Agent 协作”体现在多个 Agent 节点被条件路由分发多个节点汇聚到同一个汇总节点最终输出统一格式的结果。7. 多 Agent 协作模式扩展7.1 并行执行模式上面的例子是“多选一”的派发模式。另一类常见的多 Agent 协作是“多个 Agent 同时执行然后汇总结果”。比如你可以同时让代码审查 Agent、测试 Agent、安全扫描 Agent 处理同一个任务最后把三份报告合并。LangGraph 中实现并行非常简单只需要从一个节点引出多条边指向不同节点然后再让这些节点都连向同一个汇总节点即可。graph.add_edge(START, split_node) graph.add_edge(split_node, code_review) graph.add_edge(split_node, test) graph.add_edge(split_node, security) graph.add_edge(code_review, merge_node) graph.add_edge(test, merge_node) graph.add_edge(security, merge_node) graph.add_edge(merge_node, END)此时merge_node会等待所有指向它的上游节点执行完毕后再运行从而读取到所有 Agent 的返回结果。并行模式适合子任务之间没有依赖关系、可以独立执行的场景。7.2 子图模式当业务越来越复杂一张图中的节点可能会超过几十个。这时候把图拆成多个子图会更清晰。LangGraph 支持把一张编译好的图作为另一个图的节点。# 子图专门处理代码审查流程 subgraph StateGraph(SubState) subgraph.add_node(...) subgraph_app subgraph.compile() # 主图把子图当作一个节点 main_graph StateGraph(MainState) main_graph.add_node(review_subgraph, subgraph_app)子图内部可以有自己的 State、Node、Edge主图只需要把子图当成一个黑盒即可。这样做的好处是子图可以独立测试、独立复用主图的逻辑也变得更清晰。7.3 监督者模式当存在多个独立 Agent 时另一种常见的组织方式是“监督者模式”。一个主控 Agent 负责监听用户请求并调度其他子 Agent 执行最后汇总结果。在 LangGraph 中监督者模式可以通过一个“分发节点”加条件路由来实现分发节点判断任务类型然后通过条件边把请求转发给对应的 Agent。这个模式本质上和第 6 节的实战案例一致但你可以把分发节点变成一个真正的大模型调用让模型基于对话上下文决定派发哪个 Agent。在多 Agent 系统中“谁来负责最终回答”是一个值得思考的问题。常见的做法是把所有子 Agent 的结果放入 State再由一个汇总节点统一组织成用户友好的回复。这样既保留了每个子 Agent 的专业性也保证了整体体验的一致性。8. 常见问题与排查清单8.1 LangGraph 常见报错对照表在编写和运行 LangGraph 的过程中下面这些问题是高频出现的我整理成了对照表。问题现象常见原因解决思路ModuleNotFoundError: No module named langgraph未安装或安装到了别的环境执行pip install langgraph确认激活正确的虚拟环境无法从 langgraph.graph 导入 START、END版本较旧常量命名不同新版本使用START、END旧版本尝试from langgraph.graph import START_NODE, END_NODE或直接使用空字符串节点返回后 State 没有更新字段可能被默认覆盖逻辑合并或节点返回了 None检查字段是否使用了 Annotated Reducer确认节点函数 return 的是 dict条件路由总是走到兜底分支路由函数返回值不在映射字典中在路由函数中打印返回值与 add_conditional_edges 的映射 key 对比compile() 报错提示存在未连通节点某个节点没有从 START 出发的可达路径检查 add_edge 与 add_conditional_edges 是否完整连接所有节点并确保最终有路径通向 END图执行陷入死循环循环边缺少终止条件在 State 中增加轮次字段在路由函数里判断达到上限后返回 END并行节点执行结果互相覆盖多个节点写同一个 State 字段且该字段使用默认覆盖 reducer为共享聚合字段指定 reducer或让多个并行节点写入不同字段多次 invoke 之间状态残留没有使用 Checkpointer 时每次 invoke 应是独立状态如果希望保持历史状态需要传入 checkpoint 配置或手动维护历史8.2 排查步骤建议当你遇到一个诡异的问题时不妨按下面的顺序排查先把图结构渲染出来确认拓扑调用app.get_graph().print_ascii()可以直观看到节点之间的连接情况。在路由函数和节点函数中增加打印语句观察每次执行的入参和出参。把大型节点拆小逐个 validate 节点函数是否正确返回。检查返回字段名称是否与 TypedDict 中的 key 完全一致少一个字母都会出问题。确认需要跨节点保留的数据是否写入了 State而不是只存在局部变量里。如果依赖了外部模型或网络调用先 mock 掉外部依赖排除环境因素。8.3 关于安装和版本的一些提醒LangGraph 迭代速度很快你在搜索资料时可能会看到不同版本的代码比如StateGraph构造方式、START常量导入、add_conditional_edges参数顺序等都有过变化。遇到 API 对不上的情况优先查看当前安装版本的官方文档不要盲改代码。另外如果项目里同时使用 LangChain注意保持两个框架版本兼容某些情况下新版本 LangChain 可能依赖更新的 LangGraph。如果你只需要图编排能力可以只安装 langgraph不安装 langchain减少依赖冲突。9. 最佳实践与工程建议9.1 State 设计要小而清晰State 是图的数据中枢但它不是数据库也不是日志系统。不要在 State 里塞入大量与当前流程无关的字段。字段越多节点之间隐性耦合就越强调试时越难定位问题。建议把 State 拆成两部分流程状态和业务数据。流程状态如task_type、round、ticket_id等业务数据如user_input、agent_result、final_report等。9.2 节点函数要保持纯函数风格尽量让节点函数不修改外部全局变量不写文件不做隐藏副作用。输入 State输出状态变更这是最安全、最好测试的形态。如果某个节点必须调用外部 API建议把网络请求封装成独立函数便于做单元测试时 mock。9.3 条件路由一定要有兜底无论路由函数判断逻辑多完善都会出现模型输出格式异常、文本匹配失败、外部结果缺失等情况。因此每个条件路由都应该提供 fallback 分支比如返回unknown从路由映射到兜底节点或者直接回到人工处理。不要让路由函数因为 key 不匹配而抛异常。9.4 使用 Checkpointer 实现状态持久化LangGraph 提供了 Checkpointer 机制可以把每一步的 State 保存下来实现断点续跑、人工确认、时间旅行等能力。如果你的业务需要多轮对话长期记忆或者需要人工审批流程强烈建议深入了解这一块。它和普通的多轮对话 memory 不同Checkpointer 保存的是图的完整执行状态可以随时恢复。from langgraph.checkpoint.memory import MemorySaver checkpointer MemorySaver() app graph.compile(checkpointercheckpointer) config {configurable: {thread_id: conversation-1}} result app.invoke(input_data, configconfig)MemorySaver 适合本地测试和教学演示。生产环境建议使用 PostgresSaver 等持久化实现并配置独立的存储服务。这一块涉及具体业务建议阅读官方文档后再落地。9.5 重视日志与可观测性LangGraph 的价值之一是“流程可追踪”。在生产环境中建议给每个节点增加结构化日志记录时间、节点名称、输入关键字段、输出关键字段。结合 LangSmith 等可观测平台可以看到每一次 Agent 执行的完整轨迹。遇到线上问题能很快定位到具体是哪个节点、哪一步导致的。9.6 从小图开始逐步构建复杂系统不要一上来就设计几十个节点的大图。正确的做法是从最小可用链路开始例如“入口节点 - 一个处理节点 - 汇总节点 - END”跑通后再逐步加入条件路由、并行分支、子图。每增加一个节点就做一次完整验证这样定位问题会容易很多。9.7 关于安全与权限如果 LangGraph 工作流涉及数据库访问、文件删除、生产环境变更等操作一定要在节点内部做严格授权校验和安全边界控制。不要把生产环境密钥直接写入代码或 State避免放进日志。Agent 调用外部工具时建议维护工具白名单对工具的入参、出参做校验防止恶意输入造成越权操作。LangGraph 真正吸引人的地方不是因为它是某一个大模型产品而是它把 Agent 开发从“靠 Prompt 碰运气”提升到了“用代码控制流程”的工程化高度。你不用再担心模型擅自跳步也不用为了实现一次条件分支而写一堆 if-else 串行调用。先用一个小例子把 State、Node、Edge 三个概念跑通再逐步往图里添加条件路由、并行分支和子图你会发现自己对 Agent 应用的控制能力有了明显提升。下一篇可以继续深入 Checkpointer、持久化记忆、以及更复杂的监督者多 Agent 协作模式。