之前在业务迭代中使用 LangGraph 时反复卡在 State 数据流转与条件分支的逻辑设计上网上的资料要么只讲单个概念要么直接甩一个复杂 Agent 项目对零基础同学不太友好。本文会从 LangGraph 最核心的 State、Node、Edge 三个组件讲起完整拆解一个可运行的实战项目手把手带你实现状态管理、条件分支与多 Agent 协作。如果你是刚接触 LangGraph 的初学者或者已经简单了解过 LangChain 但不太清楚 LangGraph 的实际用法这篇文章都适用。学完后你能理解 LangGraph 的图结构设计思路能写出自己的状态管理节点也能基于条件边实现分支跳转最终搭建一个简单的多 Agent 协作流程。1. LangGraph 是什么Agent 应用的工作流引擎1.1 从 LangChain 到 LangGraph很多同学接触 LangChain 的时间比较早LangChain 的核心能力是封装大模型调用、构建 Prompt 模板、管理外部工具和文档检索。它强调的是“组件复用”比如你写一个带知识库的问答机器人LangChain 可以把加载文档、切分文本、向量化检索、拼接 Prompt 这几件事串成一条链。但实际生产环境里Agent 应用并不是一条直线走到底的。一个完整的 Agent 可能要做这些事接收用户问题。判断需要调用哪个工具。调用工具得到结果。根据结果再次判断是否需要继续调用工具。最终整理答案返回用户。这是一个有状态、有分支、甚至可能有循环的过程。如果用 LangChain 的 Chain 去表达偏死板因为 Chain 本质上是一个“顺序调用”思维。LangGraph 的出现就是为了解决这个问题。它把 Agent 的工作流程抽象成一张图Graph图中最核心的三个元素叫 State、Node、Edge。你可以理解成State 是全局状态所有节点共享。Node 是处理逻辑的节点每个节点读取 State、处理数据、更新 State。Edge 是连接节点的线决定下一步执行哪个节点。这种设计和我们写程序时的状态机非常像只是它的执行节点可以是大模型调用、工具调用、普通 Python 函数。1.2 LangGraph 和 LangChain 的区别这里需要注意一个常见的认知误区。LangGraph 并不是 LangChain 的“升级版”也不是一个需要完全替换 LangChain 的新框架。两者的关系更接近“互补”。LangChain 提供的是组件库和模型抽象LangGraph 提供的是编排能力。很多 LangGraph 项目里依然会用到 LangChain 的 ChatPromptTemplate、OutputParser、Tool 等模块。所以更准确的理解是如果你只需要简单的一次 Prompt 调用用 LangChain 就够了。如果你需要管理多轮状态、工具调用循环、条件分支、多 Agent 协作LangGraph 是更合适的编排层。从项目结构上也能看出来LangChain 是链式封装LangGraph 是图式编排。链式封装的缺点在于“中间判断”和“循环”很难优雅表达而图可以轻松实现这些逻辑。1.3 LangGraph 的典型应用场景LangGraph 常见的落地场景包括客服机器人根据用户意图进入不同对话流程。数据分析 Agent判断用户意图后选择查询数据库、调用 API 或生成图表。多 Agent 协作一个主管 Agent 负责任务拆解多个子 Agent 分别执行不同专业任务。人工审核流程Agent 生成内容后节点判断是否需要人工介入。自动化报告生成先规划报告结构再逐段生成并校验。本文的实战案例会覆盖其中两个核心场景带条件分支的状态管理流程以及一个简化的多 Agent 协作流程。2. 环境准备与版本说明2.1 运行环境本文示例以 Python 3.10 环境为主操作系统使用 Windows / macOS / Linux 均可。LangGraph 的核心运行和操作系统无关但如果你在 Windows 下遇到某些依赖安装失败建议优先检查 Python 版本和 pip 版本。版本需要根据你的项目实际情况调整本文示例以当前常见环境为例重点演示配置思路。建议你创建独立的虚拟环境避免项目之间互相污染依赖。python -m venv langgraph-demo # Windows langgraph-demo\Scripts\activate # macOS / Linux source langgraph-demo/bin/activate2.2 安装依赖安装 LangGraph 和 LangChain 核心依赖pip install langgraph langchain langchain-openai如果你的模型来自 OpenAI需要额外配置 API Key。如果使用国内模型或本地模型只需要把模型实例换成对应接口即可LangGraph 的图结构代码本身不依赖具体模型。为了方便演示这里使用 dotenv 管理环境变量pip install python-dotenv在项目根目录创建.env文件OPENAI_API_KEY你的API_KEY OPENAI_API_BASEhttps://api.openai.com/v12.3 项目结构我们计划实现一个“智能客服 数据分析”的流程演示项目结构如下langgraph-demo/ ├── .env ├── main.py ├── state.py ├── nodes.py └── graph.pystate.py定义 State 数据结构。nodes.py定义每个节点函数。graph.py组装图和边。main.py入口文件编译图并执行流程。这个结构不算复杂但它能清晰地把“数据定义”和“业务逻辑”分离开。实际项目中我还喜欢把每个节点单独拆文件但示例项目保持这个粒度即可方便理解。3. 核心组件拆解State、Node、Edge3.1 State图里的全局状态State 是 LangGraph 里最容易理解也最容易踩坑的地方。你可以把它理解为一张“共享表单”图里的每一个节点都能读取它、修改它修改后的结果会继续传给下一个节点。在 LangGraph 中State 通常用 TypedDict 或 Pydantic BaseModel 定义。官方默认推荐 TypedDict简单直观。以下是一个示例# state.py from typing import TypedDict, Annotated, List class AgentState(TypedDict): messages: Annotated[List[str], append] # 消息列表mermaid中会与节点拼接 current_step: str user_intent: str final_answer: str这里大部分字段是普通类型直接覆盖更新即可。但messages字段是一个列表如果多个节点都往这个字段里追加内容默认行为是“覆盖”不是“追加”。为了让多个节点都能往messages里追加内容我们需要给messages字段加上 reducer 操作符。LangGraph 内置了一个简单方式使用operator.add。# state.py from typing import TypedDict, Annotated, List import operator class AgentState(TypedDict): messages: Annotated[List[str], operator.add] current_step: str user_intent: str final_answer: str加上operator.add后每个节点返回的messages片段会自动拼接到全局 messages 列表里而不会覆盖原有内容。这是 LangGraph 新手最容易忽略的细节。在 LangGraph 的术语里这种机制叫 Reducer。Reducer 可以理解成一个“自定义合并函数”它决定了当多个节点想更新同一个字段时新旧值如何合并。3.2 Node执行逻辑的最小单元Node 是 LangGraph 图中的执行单元。它可以是普通 Python 函数也可以是一个对象。节点函数的输入是当前的 State输出是一个字典字典里是更新后的 State 字段。节点函数的基本写法def my_node(state: AgentState) - dict: # 读取当前 State current_step state[current_step] # 业务处理 result f处理完成{current_step} # 返回要更新的字段 return {final_answer: result}节点函数不需要手动修改传入的 state 对象只需要返回一个包含新增字段或修改字段的字典。LangGraph 会自动把返回值和当前 State 合并。下面模拟一个简单的处理节点# nodes.py from state import AgentState def intent_node(state: AgentState) - dict: user_message state[messages][-1] if state[messages] else # 这里简化了意图识别真实场景可以调用 LLM if 数据 in user_message or 分析 in user_message: intent data_analysis else: intent general_chat return {user_intent: intent} def data_node(state: AgentState) - dict: return { messages: [【数据分析节点】正在查询数据库...], final_answer: 这是你的数据查询结果本月销售额同比上涨 15%。 } def chat_node(state: AgentState) - dict: return { messages: [【普通聊天节点】正在处理日常问题...], final_answer: 你好有什么可以帮你的 }注意messages字段返回时必须是一个列表因为它配置了operator.addreducerLangGraph 会把返回的列表追加到原来的 messages 后面。还有一个常见的疑问节点函数里如何改变 State 值答案就是通过返回值。不要把节点函数设计成“直接修改传入字典”那样可能不会生效还容易造成逻辑混乱。3.3 Edge连接节点的路径Edge 定义了节点与节点之间的执行顺序。LangGraph 有几种边普通边从 A 节点执行完后直接进入 B 节点。条件边从 A 节点执行完后根据条件动态决定进入哪个节点。入口边指定 START 节点到哪个节点开始。出口边指定哪些节点执行后进入 END 结束。普通边和入口边的写法如下from langgraph.graph import StateGraph, START, END graph StateGraph(AgentState) # 添加节点 graph.add_node(intent, intent_node) graph.add_node(data, data_node) graph.add_node(chat, chat_node) # 入口边 graph.add_edge(START, intent) # 普通边 graph.add_edge(data, END) graph.add_edge(chat, END)现在整张图的逻辑是START - intent - data/chat - END但这里还没接入意图判断所以不管用户输入什么都会先走 data 节点然后直接结束。接下来需要加入条件边。4. 条件分支与路由控制4.1 使用 conditional_edges 实现分支条件边的作用是根据 State 当前的值动态决定下一步走向。语法上需要定义一个路由函数路由函数输入 State返回目标节点名称。def route_after_intent(state: AgentState) - str: if state[user_intent] data_analysis: return data return chat然后在 StateGraph 上注册条件边graph.add_conditional_edges( intent, route_after_intent, { data: data, chat: chat } )这段代码的含义是路由从intent节点出发后调用route_after_intent函数判断下一步走向。函数返回值如果是data进入data节点如果是chat进入chat节点。完整的 graph.py 如下# graph.py from langgraph.graph import StateGraph, START, END from state import AgentState from nodes import intent_node, data_node, chat_node def route_after_intent(state: AgentState) - str: if state[user_intent] data_analysis: return data return chat def build_graph(): graph StateGraph(AgentState) graph.add_node(intent, intent_node) graph.add_node(data, data_node) graph.add_node(chat, chat_node) graph.add_edge(START, intent) graph.add_conditional_edges( intent, route_after_intent, { data: data, chat: chat } ) graph.add_edge(data, END) graph.add_edge(chat, END) return graph.compile()这里要解释一下add_conditional_edges的第三个参数。它是一个映射字典key 是路由函数的返回值value 是实际要进入的节点名称。也可以不写这个字典直接让路由函数返回节点名称本身LangGraph 会默认使用返回值作为节点名。不过为了可读性建议显式写出映射。4.2 循环控制让图“回头”而不是一直线性走条件边不仅能实现分支还能实现循环。比如一个 Agent 需要多次调用工具第一次调用后可能结果不充分需要重新调用一次。假设有一个should_continue判断函数def should_continue(state: AgentState) - str: if state[current_step] done: return end return processing如果返回值是end进入END节点如果是processing回到processing节点继续执行。这样图就形成了循环结构。graph.add_conditional_edges( processing_node, should_continue, { end: END, processing: processing_node } )注意 LangGraph 本身支持循环这一点是它和 LangChain Chain 最大的不同。在 LangChain 里如果要表达循环你需要自己在外部用 Python 循环控制在 LangGraph 里循环是图结构的一部分状态由 State 统一管理更容易跟踪。4.3 并行分支一个节点触发多个下游节点LangGraph 也支持从一个节点出发同时进入多个节点。你可以这样写graph.add_edge(start_node, node_a) graph.add_edge(start_node, node_b)当start_node执行完后node_a和node_b会并行执行。LangGraph 的默认执行策略是并行执行所有可执行节点最后把所有更新结果合并到 State。并行分支适合这样的场景分析 Agent 需要同时调用多个工具比如同时查数据库、查文档、查外部 API最后汇总结果。多 Agent 协作里也经常用到并行分支一个任务拆成两路一路做摘要一路做翻译最后汇总。5. 实战案例一带状态管理的客服机器人5.1 需求分析现在我们把前几个章节的内容串起来做一个带状态管理的客服机器人。需求如下用户输入一句话。系统判断用户意图。如果意图是“数据分析”进入数据分析节点模拟查询数据库并返回结果。如果意图是“普通聊天”进入普通聊天节点返回问候语。整个过程中的 messages 被完整记录在 State 里方便追溯。5.2 定义 State# state.py from typing import TypedDict, Annotated, List import operator class AgentState(TypedDict): messages: Annotated[List[str], operator.add] current_step: str user_intent: str final_answer: str5.3 定义节点# nodes.py from state import AgentState def intent_node(state: AgentState) - dict: user_message state[messages][-1] if state[messages] else if 数据 in user_message or 分析 in user_message or 报表 in user_message: intent data_analysis else: intent general_chat return { current_step: intent_done, user_intent: intent, messages: [f【意图识别】判断结果为{intent}] } def data_node(state: AgentState) - dict: return { messages: [【数据分析】正在查询数据库...], final_answer: 本月销售额为 1200 万同比增长 15%环比增长 3%。 } def chat_node(state: AgentState) - dict: return { messages: [【普通聊天】正在生成回复...], final_answer: 你好我是智能客服你可以问我数据分析相关的问题。 }5.4 组装图# graph.py from langgraph.graph import StateGraph, START, END from state import AgentState from nodes import intent_node, data_node, chat_node def route_after_intent(state: AgentState) - str: return state[user_intent] def build_graph(): graph StateGraph(AgentState) graph.add_node(intent, intent_node) graph.add_node(data, data_node) graph.add_node(chat, chat_node) graph.add_edge(START, intent) graph.add_conditional_edges( intent, route_after_intent, { data_analysis: data, general_chat: chat } ) graph.add_edge(data, END) graph.add_edge(chat, END) return graph.compile()这里路由函数直接返回state[user_intent]因为我们的意图字段值刚好就是data_analysis或general_chat映射表里的 key 和节点名对上了。5.5 运行验证# main.py from graph import build_graph app build_graph() def run(user_input: str): result app.invoke({ messages: [user_input], current_step: , user_intent: , final_answer: }) print(当前状态字段, result) if __name__ __main__: run(我想看一下本月的数据分析) print( * 50) run(你好呀)运行后第一句输入会走data节点第二句走chat节点。结果里能看到 messages 被追加了多条内容这正是 reducer 生效的结果。6. 实战案例二简化版多 Agent 协作6.1 多 Agent 协作的核心思路多 Agent 协作是当前比较热的话题。LangGraph 实现多 Agent 协作的方式不是“多个模型同时思考”而是用图结构把多个 Agent 节点连接起来让它们分工协作。一种常见模式是“主管 工人模式”一个负责拆解任务的主管 Agent。多个负责执行具体任务的工人 Agent。主管根据任务内容把任务分配给不同工人。工人完成后把结果返回主管再判断是否可以结束。6.2 定义多 Agent State# state.py from typing import TypedDict, Annotated, List import operator class MultiAgentState(TypedDict): messages: Annotated[List[str], operator.add] task: str assigned_agent: str report: str6.3 定义主管 Agent这里简化处理用关键词判断任务类型。真实项目中可以使用 LLM 做意图判断。# nodes.py from state import MultiAgentState def supervisor_node(state: MultiAgentState) - dict: task state[task] assigned_agent writer if 代码 in task or 编程 in task: assigned_agent coder elif 翻译 in task: assigned_agent translator return { assigned_agent: assigned_agent, messages: [f【主管】任务拆解完成分配给{assigned_agent}] }6.4 定义工人 Agentdef coder_node(state: MultiAgentState) - dict: return { messages: [【程序员Agent】正在编写代码...], report: 代码任务完成已生成 Python 脚本。 } def translator_node(state: MultiAgentState) - dict: return { messages: [【翻译Agent】正在翻译文本...], report: 翻译任务完成已输出英文版本。 } def writer_node(state: MultiAgentState) - dict: return { messages: [【写作Agent】正在撰写文章...], report: 写作任务完成已输出初稿。 }6.5 组装多 Agent 图# graph.py from langgraph.graph import StateGraph, START, END from state import MultiAgentState from nodes import supervisor_node, coder_node, translator_node, writer_node def route_after_supervisor(state: MultiAgentState) - str: return state[assigned_agent] def build_multi_agent_graph(): graph StateGraph(MultiAgentState) graph.add_node(supervisor, supervisor_node) graph.add_node(coder, coder_node) graph.add_node(translator, translator_node) graph.add_node(writer, writer_node) graph.add_edge(START, supervisor) graph.add_conditional_edges( supervisor, route_after_supervisor, { coder: coder, translator: translator, writer: writer } ) for agent in [coder, translator, writer]: graph.add_edge(agent, END) return graph.compile()6.6 运行多 Agent 流程# main.py from graph import build_multi_agent_graph def run_multi_agent(): app build_multi_agent_graph() result app.invoke({ messages: [请帮我写一段 Python 代码], task: 写一段 Python 代码读取 CSV 文件并统计行数, assigned_agent: , report: }) print(执行节点消息) for msg in result[messages]: print( -, msg) print(最终报告, result[report]) if __name__ __main__: run_multi_agent()这个例子虽然简单但已经把多 Agent 协作的“主管拆任务、按条件分发、子 Agent 执行、汇总结果”走通了。真实项目里每个子 Agent 可以拥有自己的 State 子图可以调用不同的模型和工具从而实现更复杂的协作。7. 进阶概念子图与长期记忆7.1 子图Subgraph当图变得复杂时全部节点都放在同一张图里会很难维护。LangGraph 允许把一部分节点封装成子图作为整体嵌入父图。子图的用法定义子图时和普通图一样编译后得到一个 CompiledStateGraph。在父图中直接用add_node(agent_node, subgraph)把子图作为节点加入。父图调用子图时会传递 State子图的返回值会更新父图的 State。子图适合做“可复用模块”比如上面的多 Agent 协作流程可以封装成一个coding_team子图然后被更大的业务图调用。7.2 长期记忆MemoryLangGraph 默认的 State 是短期记忆也就是只在一次调用内有效。如果应用需要跨对话保存状态比如记住用户偏好就需要引入外部存储。LangGraph 官方提供了长期记忆的解决方案核心思路是使用一个持久化存储层保存聊天历史或用户档案。实际项目中你可以用 Redis、SQLite 或者向量数据库保存这些信息。具体实现方式取决于你的部署环境。需要注意较新的 LangGraph 版本中记忆功能模块有过结构调整。如果你使用的版本较新建议优先参考你安装版本对应的官方文档。7.3 如何使用 LangGraph 实现大模型节点前面示例里的节点都是纯 Python 函数。真实项目里节点内大概率要调用大模型。这里给一个结合 LangChain 的节点示例# llm_node.py from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate llm ChatOpenAI(modelgpt-4o-mini, temperature0) def llm_agent_node(state): prompt ChatPromptTemplate.from_messages([ (system, 你是一个智能助理请根据用户问题给出回答。), (human, {question}) ]) chain prompt | llm question state[messages][-1] response chain.invoke({question: question}) return { final_answer: response.content }这里的关键点是LangGraph 的节点函数和普通 Python 函数没有本质区别所以你可以把任何 LangChain 组件、其他第三方库封装进节点里。8. 常见问题与排查思路8.1 节点返回的 messages 没有追加而是覆盖这是 LangGraph 新手最常见的问题。原因通常是 State 定义里没有配置 reducer。解决方案messages: Annotated[List[str], operator.add]加上operator.add后节点 return 的列表会自动追加到原来的 messages 后面。8.2 条件边路由函数返回值与映射表不匹配如果路由函数返回的字符串没有出现在映射字典里LangGraph 会报错。排查时优先检查路由函数和映射表 key 是否一致。建议路由返回值使用英文小写加下划线例如data_analysis、general_chat。避免中文或带空格的值减少匹配出错的概率。8.3 图编译报错start node not found 或 missing edge检查以下两点是否使用了START作为入口节点。是否所有非 END 节点都有出边。LangGraph 要求从 START 出发能到达所有节点并且所有节点最终都能到达 END否则编译时会报错。8.4 LangGraph 版本升级后 API 变化LangGraph 迭代速度较快不同小版本之间的 API 可能有细微差异。如果你看到网上代码和你本地环境不兼容优先查看你项目中 LangGraph 的实际版本pip show langgraph然后根据版本调整 API 写法。本文示例用的是常见较新版本的写法如果你的版本不同请以实际环境为准。8.5 与 Node.js 相关报错混淆有些同学会搜索 LangGraph 时看到 Node.js 相关的报错比如node:util模块导出问题、nvm切换版本、npm安装等。这里要提醒一下python -m pip install langgraph安装的是 Python 的 LangGraph 库和 Node.js 运行环境没有直接关系。如果你只是在做 Python 项目不需要安装 Node.js。9. 最佳实践与工程建议9.1 State 设计建议State 是 LangGraph 的灵魂。我建议在设计 State 时注意以下几点字段粒度不要过粗。一个字段只放一类数据。messages 这类追加型字段统一使用Annotated reducer。不要把所有临时变量都塞进 State只保存需要跨节点传递的数据。如果字段较多可以使用嵌套 TypedDict保持结构清晰。9.2 节点职责单一每个节点只做一件事。比如意图识别节点只负责识别意图工具调用节点只负责调用工具。如果一个节点里塞了太多逻辑后续调试会非常困难。9.3 条件路由建议显式写出映射虽然 LangGraph 允许路由函数直接返回节点名但我建议显式写出映射字典。因为显式映射能让图结构一目了然也方便你在路由函数里做更复杂的逻辑。9.4 善用调试输出开发阶段可以在节点里加上日志输出def my_node(state): print(进入 my_node当前 state:, state) return {...}生产环境记得移除或改成标准日志模块。9.5 错误处理与容错节点内部建议做好异常捕获尤其是在调用外部 API、数据库、工具时。不要让底层异常直接抛出导致整个图崩溃。可以在节点内部 catch 异常后返回一个错误信息字段让上层 Agent 根据错误信息决定下一步。9.6 生产环境注意事项使用独立的 API Key 管理不要把 Key 硬编码在代码里。对模型调用增加超时和重试机制。如果图中有长时间运行的任务考虑异步处理。涉及用户隐私数据时注意脱敏和权限控制。重要变更尽量先在测试环境跑完再到生产执行。10. 总结与下一步学习路线本文从 LangGraph 的核心组件开始讲解了 State、Node、Edge 的概念和用法然后通过两个实战案例分别演示了带条件分支的状态管理流程和多 Agent 协作流程。核心掌握点有三个State 利用 reducer 控制字段的合并方式特别是messages这类需要追加的字段。Node 是普通的 Python 函数通过返回值更新 State。Edge 控制执行顺序条件边实现分支和循环。下一步你可以继续学习这几个方向如何把 LangChain 的 Tool 接入 LangGraph 节点。如何用子图拆分大型 Agent 项目。如何引入外部存储实现长期记忆。如何把图流程部署成 API 服务。在实际项目中优先关注状态字段的设计和条件路由的健壮性这两处是坑最多的地方。动手把示例跑通后再往里面加自己的业务逻辑会比直接看复杂文档轻松很多。如果本文对你有帮助可以收藏备用也欢迎在评论区聊聊你踩过的坑。