在接触 LangGraph 的时候很多人的第一反应是这不就是 LangChain 出的一个状态机工具吗等我真正用 LangGraph 写完几个多智能体应用之后才意识到这个判断只对了一半。LangGraph 确实做状态编排但它真正改变的是 Agent 应用从“调 API”到“控流程”的思维转变。如果你之前只把 LangChain 当成一个 LLM 封装库来用会觉得它什么都帮你干了但当你开始做复杂的 Agent 项目比如多步推理、人工审批、条件分支、循环修正你会发现 LangChain 的链式调用非常别扭。而 LangGraph 提供了一套完全不同的抽象把智能体应用看成一张图节点是具体动作边是流转规则状态是全局共享的数据。这个思路跟传统后端开发中的工作流引擎非常像所以对于有后端经验的开发者来说上手 LangGraph 反而比硬啃 LangChain 的链式抽象要更自然。这篇文章不会去复读官方文档而是从一个真实的开发视角出发把 LangGraph 中最重要的概念、最容易踩坑的地方以及实际项目里真正用得上的写法一次讲透。如果你正打算往多智能体方向深入或者已经在 LangChain 里写了不少代码但总觉得“链式调用不够灵活”这篇文章值得你读完并收藏。1. 这篇文章真正要解决的问题LangGraph 最近在开发者社区的热度上升得很快但很多人看资料的时候会有一种感觉看官方文档觉得概念很清晰真到自己写一个项目的时候还是不知道从哪下手。这个问题的根源在于LangGraph 的知识点非常分散而且相互依赖不理解 State 的设计后面看条件路由会懵不理解条件路由子图的写法就会非常绕不搞清楚并行分支的底层机制就没法理解为什么它可以优化响应延迟。所以这篇文章要解决的不是“LangGraph 有哪些 API”而是“LangGraph 到底怎么用来组织一个完整的 Agent 应用”。我会从下面几个角度展开LangGraph 和 LangChain 的核心区别以及为什么说 LangGraph 不是 LangChain 的简单升级。官方教程里反复讲但很多人没理解的五个核心概念State、Node、Edge、Conditional Edge、Subgraph。一个从零开始的多分支 Agent 项目示例覆盖条件路由、状态更新、并行分支、子图嵌套和循环检测。实际开发中最常遇到的报错、坑点以及排查思路。在生产环境中使用 LangGraph 的工程建议包括持久化、记忆管理、可观测性和测试策略。读完这篇文章你至少应该能做到不看文档、只凭思路就能设计出一个包含条件分支和子图的多智能体工作流。2. LangGraph 基础概念与核心原理2.1 LangGraph 是什么LangGraph 是一个基于图结构的大语言模型应用编排框架。这里的“图”不是指图表可视化而是计算机科学里的有向图由节点Node和边Edge组成。每个节点负责一个具体的任务比如调用模型、执行代码、查询数据库每一条边定义了节点之间的流转关系。在 LangGraph 中一次完整的 Agent 运行可以看作一个状态State从初始节点出发沿着图的边在节点之间流转最终到达结束节点的过程。2.2 LangGraph 和 LangChain 的区别网上很多资料把 LangGraph 说成 LangChain 的下一代框架其实这种说法并不准确。LangChain 的核心抽象是 Chain也就是链式调用一个环节接一个环节地执行LangGraph 的核心抽象是 Graph它比链式调用多出了几个能力能力维度LangChain ChainLangGraph Graph流程结构线性前后拼接有向图支持分支、合并、循环状态管理通过 Chain 参数手动传递全局 State 对象节点共享条件控制需要外部代码 if/elseConditional Edge 原生支持循环控制不擅长原生支持并带递归限制适合场景简单问答、文档检索链复杂 Agent、多步骤工作流LangGraph 不是要取代 LangChain而是补足了 LangChain 在复杂流程控制上的短板。在实际项目中你可以把两者结合起来使用用 LangChain 的组件比如 ChatPromptTemplate、LCEL组织节点内部的调用逻辑用 LangGraph 控制节点之间的流转。2.3 为什么要用图结构来编排 Agent这里我们可以做一个类比。传统后端开发中处理复杂业务流程时核心思路是状态机定义状态、定义状态之间的转移条件、定义每个状态下要执行的动作。这种思路的好处是流程清晰、容易排错、方便扩展。Agent 应用本质上也是一种业务流程只是这个流程里的“动作”变成了模型调用、工具调用而且流程往往不是线性的——Agent 需要根据模型的输出决定下一步做什么。比如模型可能认为应该先查资料再回答也可能认为资料不足需要让用户补充信息还可能出现工具调用失败需要重试的情况。如果只用链式调用这些分支逻辑会散落在代码里流程被 if/else 切得支离破碎。用图结构后分支逻辑被显式地建模为图中的边每个分支要做的事情就是图中的一条路径这样整个 Agent 的行为变得可以预测、可以追踪、可以测试。3. LangGraph 核心概念详解3.1 State全局状态对象State 是 LangGraph 中最基础也最重要的概念。它本质上是一个数据结构在图的整个运行过程中被所有节点共享。每个节点执行时都能读取当前 State执行完之后可以返回一个字典这个字典里的键值会被更新到 State 中。from typing import Annotated, TypedDict from langgraph.graph import StateGraph class AgentState(TypedDict): messages: list task: str steps: int这里有一个非常容易踩坑的点如果你在节点里直接返回一个列表字段LangGraph 的行为是覆盖而不是追加。如果你想让多个节点往同一个字段里追加内容必须使用 Annotated 加上 reducer 函数from typing import Annotated from langgraph.graph.message import add_messages class AgentState(TypedDict): messages: Annotated[list, add_messages] task: str steps: intadd_messages 是 LangGraph 内置的 reducer它会把新返回的消息追加到已有的消息列表而不是覆盖。3.2 Node执行单元Node 是图中的一个执行单元本质上就是一个 Python 函数。函数接收 State 作为参数返回一个字典用于更新 State。def process_task(state: AgentState) - dict: print(f当前任务{state[task]}) return {steps: state[steps] 1}这个设计非常朴素但恰恰是它的优势你不需要学习任何新语法只要保持“接收 State返回字典”这个约定函数内部可以用任何你熟悉的方式处理逻辑。3.3 Edge 与 Conditional Edge流程控制Edge 定义节点之间的固定流转关系。比如节点 A 执行完后必然进入节点 B用 Edge 就够了。Conditional Edge 则是在节点执行完之后根据 State 的内容决定下一步进入哪个节点。它的核心是一个路由函数def route_after_query(state: AgentState) - str: if error in state.get(status, ): return retry_node elif state[need_manual_approval]: return human_approval_node else: return generate_answer_node这里容易忽略的一点是conditional edge 的路由函数返回的是字符串这个字符串必须和目标节点的名称完全一致否则运行时会报 KeyError。在实际项目中建议把节点名称定义成常量避免手写字符串出错。3.4 Subgraph子图Subgraph 就是把一张图作为另一个图的节点。它的价值在于模块化组合。比如一个完整的 Agent 应用可能需要“数据预处理”“多轮推理”“结果格式化”三个子流程你可以分别构建三张子图然后在主图里把它们作为三个节点连接起来。子图可以有自己的 State也可以和主图共享 State。子图之间可以互相嵌套LangGraph 对嵌套层数没有硬性限制但从工程角度建议不要超过三层否则排查问题会非常痛苦。3.5 循环与递归限制LangGraph 支持循环边也就是说节点 A 可以流转回节点 C。这在很多场景下非常有用比如模型输出格式不正确需要重新生成、工具调用失败需要重试。但是循环必须设置递归限制recursion_limit否则可能出现无限循环。默认的递归限制是 25 次也就是图中节点的总执行次数超过 25 次就会抛异常。实际项目中建议根据业务需要显式设置而不是依赖默认值。4. LangGraph 环境准备与基础配置在动手写代码之前我们需要先把环境准备好。下面是 LangGraph 开发的基础环境要求Python 3.9 及以上版本建议 3.10 或 3.11对类型注解支持更好。一个可用的 LLM API比如 OpenAI、通义千问、智谱或本地部署的模型。LangGraph 本身不强绑定某个模型厂商它只负责编排模型调用是你节点函数内部的事情。LangChain 相关的包用于调用模型和处理消息格式。版本方面需要特别注意LangGraph 的 API 更新比较频繁特别是在 0.2 到 0.4 这段期间部分接口有调整。本文提供的示例代码以稳定可运行为目标不绑定某一个具体版本。你在安装时建议先创建独立虚拟环境避免和已有项目冲突。推荐安装命令pip install langgraph langchain langchain-openai如果你需要使用 LangGraph 提供的持久化和检查点功能还需要额外安装pip install langgraph-checkpoint-sqlite下面是项目目录结构本篇文章后面的代码会按这个目录组织langgraph_demo/ ├── agent.py # Agent 核心逻辑 ├── state.py # State 定义 ├── nodes.py # 节点函数 ├── routes.py # 路由函数 ├── main.py # 主入口构建图并运行 └── requirements.txt # 依赖列表5. LangGraph 完整流程示例多分支 Agent 实战接下来我们通过一个完整的例子把前面讲的概念全部串起来。这个例子模拟的是一个“智能客服工单处理 Agent”需求是用户提交一个工单包含问题描述。Agent 先判断工单类型故障咨询、投诉、建议。根据类型走不同的处理分支。每种分支都可能需要调用外部工具这里用简单的打印函数模拟。如果处理失败进行重试。全部完成后输出最终结果。5.1 定义 State# 文件路径state.py from typing import Annotated, TypedDict from langgraph.graph.message import add_messages class AgentState(TypedDict): messages: Annotated[list, add_messages] ticket_type: str retry_count: int result: str这里给 messages 字段加了 reducer目的是让所有节点产生的新消息都能追加到同一个历史列表中。ticket_type 用于记录工单类型retry_count 用于循环重试result 保存最终结果。5.2 定义节点函数# 文件路径nodes.py from state import AgentState def classify_ticket(state: AgentState) - dict: 模拟工单分类实际项目中这里通常调用 LLM content state[messages][-1].content if 投诉 in content: ticket_type complaint elif 建议 in content: ticket_type suggestion else: ticket_type fault print(f[分类节点] 工单类型{ticket_type}) return {ticket_type: ticket_type} def handle_fault(state: AgentState) - dict: 故障咨询处理 print([故障节点] 执行故障排查流程) # 模拟工具调用 print([故障节点] 查询设备状态...) return {result: 已完成故障排查建议重新插拔设备后观察。} def handle_complaint(state: AgentState) - dict: 投诉处理 print([投诉节点] 转人工客服处理) return {result: 已转接人工客服请保持电话畅通。} def handle_suggestion(state: AgentState) - dict: 建议处理 print([建议节点] 记录建议并提交产品组) return {result: 感谢您的建议已提交产品团队评审。} def check_retry(state: AgentState) - dict: 模拟处理结果检查如果失败则增加重试次数 # 这里故意模拟失败一次第二次成功后退出 current state.get(retry_count, 0) if current 1: print([检查节点] 处理失败准备重试) return {retry_count: current 1, result: FAILED} print([检查节点] 处理成功) return {retry_count: current 1, result: SUCCESS}5.3 定义路由函数# 文件路径routes.py def route_after_classify(state) - str: 根据工单类型路由到不同的处理节点 ticket_type state.get(ticket_type, fault) if ticket_type complaint: return handle_complaint elif ticket_type suggestion: return handle_suggestion else: return handle_fault def route_after_check(state) - str: 根据检查结果决定是否重试 result state.get(result, ) if result FAILED: return handle_fault return end5.4 构建主图# 文件路径main.py from langgraph.graph import StateGraph, START, END from state import AgentState from nodes import classify_ticket, handle_fault, handle_complaint, handle_suggestion, check_retry from routes import route_after_classify, route_after_check def build_graph(): graph StateGraph(AgentState) # 添加节点 graph.add_node(classify, classify_ticket) graph.add_node(handle_fault, handle_fault) graph.add_node(handle_complaint, handle_complaint) graph.add_node(handle_suggestion, handle_suggestion) graph.add_node(check, check_retry) # 添加边 graph.add_edge(START, classify) graph.add_conditional_edges( classify, route_after_classify, { handle_fault: handle_fault, handle_complaint: handle_complaint, handle_suggestion: handle_suggestion, }, ) graph.add_edge(handle_fault, check) graph.add_edge(handle_complaint, END) graph.add_edge(handle_suggestion, END) # 条件循环检查不通过则重新进入故障处理 graph.add_conditional_edges( check, route_after_check, { handle_fault: handle_fault, end: END, }, ) return graph.compile() if __name__ __main__: app build_graph() # 这里先用一个简化的消息对象演示 from langchain_core.messages import HumanMessage result app.invoke({ messages: [HumanMessage(content我的设备坏了请帮我排查故障)] }) print(最终状态, result)5.5 代码逻辑说明这段代码完整地展示了 LangGraph 的核心用法使用StateGraph(AgentState)创建图对象传入状态类型。使用add_node注册所有节点。使用add_edge建立固定流转关系。使用add_conditional_edges建立条件分支和循环。使用graph.compile()将图编译成可执行对象。使用app.invoke(state)启动整个工作流。你可能会注意到一个细节check_retry节点后面的条件边既连接了handle_fault循环也连接了END结束。这就是 LangGraph 中实现循环的方式——通过条件边让流程回到之前的节点。这个循环不是无限循环因为retry_count会累加一旦超过阈值就走向结束。6. LangGraph 子图Subgraph实战扩展上面这个例子已经覆盖了条件路由和循环但在真实的项目中我们经常需要把一部分逻辑抽成子图让主图的逻辑更清晰。下面扩展一下把“故障处理流程”抽成一个子图。6.1 构建故障处理子图# 文件路径subgraph.py from langgraph.graph import StateGraph, START, END from state import AgentState def retrieve_device_info(state: AgentState) - dict: print([子图] 正在获取设备历史工单...) return {messages: [{role: assistant, content: 设备历史工单获取完成}]} def diagnose_fault(state: AgentState) - dict: print([子图] 正在诊断故障原因...) return {result: 设备系统版本过低导致兼容性问题。} fault_subgraph StateGraph(AgentState) fault_subgraph.add_node(retrieve, retrieve_device_info) fault_subgraph.add_node(diagnose, diagnose_fault) fault_subgraph.add_edge(START, retrieve) fault_subgraph.add_edge(retrieve, diagnose) fault_subgraph.add_edge(diagnose, END) fault_subgraph fault_subgraph.compile()6.2 在主图中接入子图# 文件路径main_with_subgraph.py from langgraph.graph import StateGraph, START, END from state import AgentState from subgraph import fault_subgraph graph StateGraph(AgentState) graph.add_node(classify, classify_ticket) # 将子图作为节点加入 graph.add_node(fault_process, fault_subgraph) graph.add_node(check, check_retry) graph.add_edge(START, classify) graph.add_conditional_edges( classify, route_after_classify, {handle_fault: fault_process, handle_complaint: handle_complaint, handle_suggestion: handle_suggestion}, ) graph.add_conditional_edges( fault_process, route_after_check, {handle_fault: fault_process, end: END}, )子图在 LangGraph 中会被当作一个“黑盒节点”主图只关心子图的输入和输出。如果子图内部升级了处理逻辑只要保证输入输出接口不变主图完全不需要改动。这是子图最重要的工程价值。6.3 子图调试注意事项调试子图时有一个常见问题子图内部产生的中间状态不会自动暴露到主图的最终状态中。如果你在调试时发现某些字段丢失先用graph.get_graph().draw_mermaid()仅本地调试用不在文章中渲染查看图结构确认边连接是否正确再确认子图的输入字段是否在主图中存在。这个问题在实际项目中几乎是必踩的。7. LangGraph 并行分支与长期记忆7.1 并行分支提升响应速度的关键多智能体应用里有一个常见的需求多个任务彼此独立可以同时执行。比如客服工单处理中既需要查订单状态又需要查物流信息这两个操作互不依赖。LangGraph 的fanout机制允许我们从同一个节点出发同时连接多个节点这些节点会并行执行。要想做到并行需要做一个额外的配置在创建StateGraph时传入一个自定义的State其中并行分支涉及到的字段需要设置 reducer 来处理并发写入。# 文件路径parallel.py from typing import Annotated from langgraph.graph import StateGraph, START, END from concurrent.futures import ThreadPoolExecutor def query_orders(state) - dict: print([并行节点1] 查询订单) return {order_info: 订单已发货} def query_logistics(state) - dict: print([并行节点2] 查询物流) return {logistics_info: 物流运输中} def collect_results(state) - dict: print([汇总节点] 汇总并行结果) return {result: f订单{state.get(order_info)}物流{state.get(logistics_info)}} parallel_graph StateGraph(AgentState) parallel_graph.add_node(orders, query_orders) parallel_graph.add_node(logistics, query_logistics) parallel_graph.add_node(collect, collect_results) parallel_graph.add_edge(START, orders) parallel_graph.add_edge(START, logistics) parallel_graph.add_edge(orders, collect) parallel_graph.add_edge(logistics, collect) parallel_graph.add_edge(collect, END)这里有一个明显的好处orders 和 logistics 两个节点是并行执行的总耗时不等于两个节点耗时的叠加而是等于较慢节点的耗时。如果你的业务中确实有多个相互独立的耗时操作用并行分支可以显著降低整体延迟。需要注意的是并行分支只有不同节点没有数据依赖时才可以使用。如果节点 B 需要节点 A 的计算结果那只能串行。7.2 长期记忆跨会话状态保存默认情况下LangGraph 的 State 在一次 invoke 结束后就消失了。但真实业务中Agent 往往需要记住用户的偏好、历史对话、之前的处理结果。LangGraph 通过 Checkpointer检查点机制来实现持久化。from langgraph.checkpoint.sqlite import SqliteSaver with SqliteSaver.from_conn_string(checkpoints.db) as checkpointer: graph build_graph() app graph.compile(checkpointercheckpointer) config {configurable: {thread_id: user_123}} result app.invoke({messages: [HumanMessage(content我的设备坏了)]}, config)第二次用同一个thread_id调用时LangGraph 会自动恢复该会话的历史状态Agent 就能“记住”上下文。这个机制在长期记忆场景中很关键比如客服机器人需要记住用户之前的工单编号或者个人助理需要记住用户的偏好设置。除了 CheckpointerLangGraph 从 0.2 版本开始还引入了更完整的持久化 API用于区分短期会话记忆和长期业务记忆。实际项目中建议把“对话上下文”放在 LangGraph 状态里把“业务数据”放到外部数据库或者对象存储中这样查询和统计更加灵活。8. LangGraph 常见问题与排查方法LangGraph 的报错信息虽然不算特别复杂但很多新手第一次接触时会觉得无从下手。下面总结了实际开发中最常见的几类问题问题现象可能原因排查方式解决方案报错 KeyError: node条件路由函数返回了不存在的节点名称打印路由函数返回值与add_conditional_edges映射表对比使用节点名称常量避免手写字符串节点返回的字段没有更新到 State返回字典中的键名和 State 定义的键名不一致检查节点返回字典的键名保持键名一致注意大小写列表字段被覆盖而不是追加没有给列表字段配置 reducer检查字段是否使用了Annotated[list, add_messages]配置正确的 reducer循环执行次数过多异常递归限制默认值较小查看错误日志中的递归计数设置recursion_limit或在路由函数里增加终止条件子图修改的状态主图看不到子图有自己的内部状态作用域打印主图最终状态检查有没有对应字段确认子图的输入和输出字段在主图中定义并行分支执行报错并行节点操作了同一个共享资源查看具体异常确认是否线程安全在节点中避免直接操作全局变量使用线程安全的队列或锁使用invoke时状态丢失没有配置检查点或 thread_id 不一致确认编译时是否传入了 checkpointer使用compile(checkpointer...)并固定 thread_id如果你遇到难以定位的问题可以先做一个最小化复现只保留两个节点和一个边跑通之后再逐步加回其他节点。这虽然是笨办法但在图编排框架中往往是最有效的排查路径。另外一个排查技巧是用graph.get_graph().draw_mermaid()输出图结构来看节点和边连接是否符合预期。在本地开发环境可以配合 markdown 预览查看如果发现节点的边连接和你设计的思路不一致基本上问题就出在add_edge或add_conditional_edges的配置上。9. LangGraph 最佳实践与工程建议9.1 节点设计单一职责节点的设计一定要保持单一职责。一个节点只做一件事比如“调用模型生成回复”“调用工具查询天气”“判断用户意图”。不要做一个“万能节点”什么都往里面塞否则后期排查问题时你会不知道该看哪个逻辑片段。如果节点内部逻辑确实很复杂应该把这个逻辑拆成多个函数然后作为多个节点放进图里。这样每个节点都可以单独调试和测试图的演化历史也更清晰。9.2 状态设计最小化原则State 中只放真正需要全局共享的数据。临时变量、中间结果、大段文本尽量不要全部塞进 State否则每次传递都会消耗大量内存并且日志打印时会非常冗长。如果你的节点之间需要传递一个比较大的数据比如一个完整的文档尽量在 State 中只保存文档的引用比如数据库 ID而不是把文档内容直接放进 State。9.3 错误处理不要让图裸奔在节点函数内部你应该用 try/except 捕获异常并把异常信息写入 State。这样即使某个环节失败流程也可以走到一个统一的异常处理节点而不是整个图直接中断。def safe_node(state: AgentState) - dict: try: # 业务逻辑 result do_something() return {result: result} except Exception as e: return {error: str(e)}配合条件路由你可以根据error字段是否存在来决定是否进行重试或人工介入。9.4 持久化与回溯生产环境请务必配置 Checkpointer。这不仅是为了长期记忆更是为了可观测性。有了 Checkpointer你可以随时回溯某一次对话的完整状态演变过程这对定位 Agent 的异常行为非常有帮助。9.5 测试策略图编排应用的测试应该分两层单元测试独立测试每个节点函数用构造好的 State 字典调用节点函数断言返回的字典是否符合预期。集成测试编译完整的图用不同输入触发不同的路径比如覆盖分类为“投诉”“建议”“故障”三条分支路径。实际项目中建议把每个节点路由的映射表单独抽出来这样测试时可以直接针对路由函数做断言而不需要完整跑一遍图。9.6 可观测性LangGraph 提供了langgraph-cli和云端监控能力但在本地或者公司内部最简单的方案是在每个节点打印关键日志。日志格式建议包含节点名称、当前 State 的关键字段、耗时等信息。这看起来土但排查问题非常有效。10. 总结与后续学习方向这一篇文章从 LangGraph 的核心概念讲起通过一个完整的多分支 Agent 示例把 State、Node、Edge、条件路由、子图、并行分支、持久化这些关键知识点全部串了起来。如果你能照着示例代码跑通一遍相信对 LangGraph 的认知会比之前清晰很多。关于为什么 LangGraph 值得认真学在我看来最核心的一点是它把 LLM 应用从“调用模型”提升到了“编排系统”的层面。现在单纯调用一个聊天模型已经没什么门槛了真正有难度的是如何设计一套稳定可控的 Agent 流程让它能在复杂的业务场景中可靠地工作。LangGraph 提供的正是这套流程控制的基本框架。接下来你可以按这个顺序继续深入先把本文的代码自己动手敲一遍每一行都要理解它为什么存在。然后尝试改造成自己的业务场景把节点函数替换成真实业务逻辑。再深入了解 LangGraph 的持久化 API把 Checkpointer 用熟。最后研究多智能体协作模式比如 Supervisor 模式、Hierarchical Agent 模式。文章中如果有什么地方不清晰或者你在实际运行中遇到了本文没有覆盖到的报错欢迎在评论区留言我会尽量回复。也建议把这篇文章收藏备用等你真正开始写 LangGraph 项目的时候一定能用得上。