LangGraph实战：用图编排搞定复杂智能体流程-北京尧图网络

做应用层的朋友应该都有同感模型本身往往不是最难的真正难的是把模型塞进一条会转弯、会分叉、还会重复执行的业务流水线里。我帮某电商业务搭智能客服时需求其实很朴素——用户进来先判断意图查订单的走查询工具要退货的走退货流程客服搞不定的转人工。结果用传统方式写if-else 越套越深一个工具返回异常整套流程就乱了想中途插一个人工审批节点更是无从下手。后来把整套逻辑迁移到 LangGraph用图来组织“复杂智能体编排”才真正把流程控制、状态管理和分支决策拆开代码和业务都理顺了。这篇保姆级教程不打算只贴一段能跑的代码而是完整讲一遍我对这套框架的理解、踩过的坑以及真实项目里真正实用的写法。适合正在写智能体应用的工程师也适合刚接触 LangGraph 但对“图编排”这个概念还没形成直觉的同学。1. 为什么复杂智能体编排绕不开 LangGraph——先想明白用它解决什么1.1 从 ReAct 循环到真实业务流AgentExecutor 到底卡在哪早期做智能体大多数人都是从 ReAct 模板起步模型先思考再看要不要调用工具脚本解析出动作后执行把结果塞回提示词然后继续循环。这个模式在 Demo 里非常顺畅因为只处理一个问题、只调一个工具循环几轮就结束。但真实业务不是这样。我曾在一个知识库问答项目里硬用 AgentExecutor 顶着业务方加了一个需求当用户问“帮我查一下订单顺便把电子发票发我邮箱”系统要先调订单查询、再判断是否需要调发票接口发票接口如果失败必须通知人工介入过程中还要把每一步的原因记录下来给用户看。这个流程至少有四个状态要维护当前步骤、工具结果、失败标记、用户可见日志。用 AgentExecutor 的循环结构做出来代码里全是标志位和嵌套分支跑不了几轮就不知道状态丢到哪儿了。AgentExecutor 本质是一个封装好的单智能体循环它帮你处理“模型输出 - 工具调用 - 带回结果”的闭环。问题在于它的控制流是写死的你没法轻松插入“走到一半暂停去问人”也没法在同一个循环里同时维护两路互相独立的任务。当业务复杂度上去这个循环就成了约束。1.2 LangGraph 把智能体变成一张图改动的不只是框架LangGraph 的切入角度很巧妙它不管你内部是 ReAct 还是别的什么它把整个智能体流程当成一张有向图。图的节点是你要执行的函数边是节点之间的流转关系条件边负责分支和循环而整张图上流动的唯一载体是一个 State 对象。这个设计直接解决了我前面说的三类问题。状态不再散落在局部变量和闭包里而是整张图共享、随时可读、每一步都能落盘流程控制从写死的循环变成显式配置想加一个节点就在图里加一个节点想让它循环无数次就在两个节点之间画一条回头边分支逻辑全部收敛到条件边函数里测试的时候可以单独拎出来跑。可以用一个现实类比来理解普通代码流程像流水线上的传送带零件只能往前送想绕回去就得人工搬运LangGraph 像工厂里的调度系统传送带、升降台、质检返工通道都画在一张图纸上零件走到哪个节点、要不要返工、哪批并行处理全由调度规则决定。1.3 什么场景必须上 LangGraph什么场景不必上先泼一盆冷水如果你只是做一个单轮问答、或者一个简单的 RAG 检索链路LangGraph 是过度设计。模型调用一次就能返回答案的场景用普通代码串起来反而更轻快。但我建议你遇到下面四种情况之一时认真考虑 LangGraph流程里有多个业务节点需要根据状态反复跳转或循环而不是一插到底需要在运行中暂停等人工确认后从同一位置继续执行需要多个智能体协作一个智能体处理一半发现要让另一个专业智能体接手需要审计和回放每一步状态变更都要保留出错时能定位到具体节点。2. 核心概念扫盲状态、节点、边、条件边2.1 状态整条流程里唯一可读写的“共享流转单”LangGraph 里的 State 通常是一个 TypedDict用来定义整张图的全局数据结构。你可以把它想象成仓库里的流转单从货物进仓开始每经过一个环节工作人员就在流转单上贴一张记录条下一个环节的人能看到前面所有环节的记录。from typing import Annotated, TypedDict from langgraph.graph.message import add_messages from langchain_core.messages import BaseMessage class State(TypedDict): messages: Annotated[list[BaseMessage], add_messages] intent: str order_id: str上面这个 State 里messages用了Annotated和add_messages意思是新节点返回的消息会追加进列表而不是把旧列表整个替换掉intent和order_id则是普通字段谁最后写了就以谁为准。这个“合并规则”叫 reducer是理解 LangGraph 状态更新最关键的一步。需要注意节点函数不是把整个 State 传给你随便改而是接收一个只读视图靠返回值提交“更新”。返回的字典会按 reducer 规则合并进总状态其他字段保持原样。这个设计让每个节点都像函数式编程里的纯函数逻辑独立、可单独测试。2.2 节点只干一件事的处理单元节点是图里最小的执行单位通常是一个接收 State 参数、返回部分更新的普通函数。节点里可以调大模型、调工具、查数据库什么都能干但它最好只负责一件事。def classify_node(state: State) - State: # 示例从最后一条消息判断意图 last_message state[messages][-1].content if 退货 in last_message or 退款 in last_message: intent refund elif 订单 in last_message: intent order_query else: intent chat return {intent: intent}节点函数不依赖图对象本身所以单元测试非常简单直接丢进去一个 State检查返回值是否符合预期。这也是 LangGraph 项目比传统 if-else 代码更好维护的核心原因之一。2.3 边和条件边流程的“公路”与“立交匝道”普通边就是流程线图里从 A 节点到 B 节点拉一条线执行完 A 自动走 B。START 是图的入口END 是出口这两个是内置常量。条件边是 LangGraph 最有价值的地方。它的本质是在一条边上挂一个路由函数函数读取当前 State返回一个字符串这个字符串决定下一步进入哪个节点。def route_by_intent(state: State) - str: if state[intent] refund: return refund_node if state[intent] order_query: return order_query_node return chat_node你可以把条件边理解为高速公路立交桥上的匝道控制一辆车开到这里系统读取车牌信息决定它左转还是右转而不是所有车都走同一条路。条件边让“一个节点执行完按结果分流”变成了图的显式能力。3. 保姆级实操5 分钟搭出一个可运行的 LangGraph 状态图3.1 装好依赖注意 Python 版本实操前先把环境准备好。LangGraph 目前对 Python 3.9 以上的支持都比较成熟我自己常用 3.10 或 3.11建议避开太老的版本。安装直接用 pippip install langgraph langchain-openai如果你只是先跑通流程不急着接付费大模型也可以先用langgraph自带的模拟函数做节点后面再接真实模型。关键是把图的结构、执行流程跑熟模型接入反而是最简单的一步。接模型时记得配置好 API Key 环境变量。下面示例里我用 OpenAI 接口写法只是演示你完全可以用自己项目里接好的大模型客户端只要节点函数能拿到模型返回结果就行。3.2 先写状态和节点函数不要碰图很多人一开始就急着搭图结果图还没搭明白节点逻辑一团糟。我的建议是反过来先把节点函数当作普通函数写好再连线。from langchain_core.messages import HumanMessage, AIMessage def greet_node(state: State) - State: # 模拟一个问候节点 return {messages: [AIMessage(content你好我是智能助手请问有什么可以帮你)]} def assistant_node(state: State) - State: # 模拟一个模型回复节点 last_human state[messages][-1].content reply f收到你的问题{last_human}。这是一个模拟回复。 return {messages: [AIMessage(contentreply)]}这两个函数都没有碰图对象输入输出都是 State 相关类型。你可以直接跑单测验证test_state {messages: [HumanMessage(content你好)], intent: chat, order_id: } result greet_node(test_state) print(result[messages][-1].content)测完再往下走这样排查问题时边界非常清晰。3.3 组装状态图并跑通第一次调用节点函数就绪后开始连线。用StateGraph创建图add_node注册节点add_edge表达顺序最后调用compile得到可执行对象。from langgraph.graph import StateGraph, START, END workflow StateGraph(State) workflow.add_node(greet, greet_node) workflow.add_node(assistant, assistant_node) workflow.add_edge(START, greet) workflow.add_edge(greet, assistant) workflow.add_edge(assistant, END) app workflow.compile()调用时传入初始状态返回最终状态init {messages: [HumanMessage(content我想退货)], intent: , order_id: } final_state app.invoke(init) print(final_state[messages][-1].content)如果你想看每一步执行时状态怎么变化换成app.stream(init)按步骤输出即可。这里注意compile()是必须的它会把图结构编译成可调用的执行对象很多新手漏掉这一步导致一直报错找不到入口。3.4 第一次运行后必须理解的三个观察点跑通之后有三个现象值得你停下来想一想。第一节点的执行顺序跟注册顺序无关只跟边有关。你把assistant_node先注册再把greet_node注册执行仍然先从 START 到 greet 再到 assistant。因为图结构已经通过边定义好了。第二messages字段因为add_messagesreducer 的作用会自动把所有节点的消息追加在一起。节点无需手动合并原来的历史消息这避免了一个非常常见的 bug每个节点都拿完整历史拼 prompt结果代码越写越长。第三普通字段是覆盖式更新。如果两个节点都写order_id后执行的节点会覆盖先执行的节点。所以如果你要保留多步结果一定用列表或字典这类可合并的结构或者自己写 reducer而不是塞进一个普通字符串字段。4. 条件路由与工具调用复杂编排的真正主战场4.1 场景拆解工单分类助手为什么需要分支接着拿开头的工单系统举例。假设公司内部有一个工单入口员工提问题后系统需要自动判断这是“IT 故障”“人事咨询”还是“财务报销”再分别路由到不同的处理节点。三个节点的处理逻辑差异很大没必要塞进同一个大模型调用里更不适合用 if-else 硬串。用 LangGraph 的思路是先有一个分类节点统一做意图识别返回一个category字段然后通过条件边把流程分流到三个分支节点三个分支完成后汇聚到公共出口。4.2 用条件边做三通道路由代码比 if-else 干净在哪先定义带category字段的状态class TicketState(TypedDict): messages: Annotated[list[BaseMessage], add_messages] category: str分类节点用大模型识别意图def classify_ticket(state: TicketState) - TicketState: # 这里简化成关键词匹配真实项目里换成 LLM 调用 text state[messages][-1].content if 电脑 in text or 网络 in text: return {category: it} if 工资 in text or 社保 in text: return {category: hr} if 发票 in text or 报销 in text: return {category: finance} return {category: other}三个处理节点各写各的def it_handler(state: TicketState) - TicketState: return {messages: [AIMessage(contentIT 工单已受理请描述具体故障现象。)]} def hr_handler(state: TicketState) - TicketState: return {messages: [AIMessage(content人事工单已受理请补充员工编号。)]} def finance_handler(state: TicketState) - TicketState: return {messages: [AIMessage(content财务工单已受理请上传相关票据。)]} def other_handler(state: TicketState) - TicketState: return {messages: [AIMessage(content工单已记录将由值班人员处理。)]}组装图核心是add_conditional_edgesworkflow StateGraph(TicketState) workflow.add_node(classify, classify_ticket) workflow.add_node(it, it_handler) workflow.add_node(hr, hr_handler) workflow.add_node(finance, finance_handler) workflow.add_node(other, other_handler) workflow.add_edge(START, classify) workflow.add_conditional_edges( classify, route_by_category, { it: it, hr: hr, finance: finance, other: other, }, ) workflow.add_edge(it, END) workflow.add_edge(hr, END) workflow.add_edge(finance, END) workflow.add_edge(other, END) app workflow.compile()路由函数def route_by_category(state: TicketState) - str: return state[category]这段代码的好处是路由逻辑从业务代码里抽了出来以后加一个“法务工单”分支只要加一个节点、加一条路由映射主流程几乎不用动。这在 if-else 版本里会改动一大片调用链。4.3 Agent ToolNode 工具调用回路的标准写法条件边最典型的应用还不是普通分支而是工具调用循环。模型决定调工具工具执行后把结果喂回模型模型再决定下一步如此循环直到模型认为不需要调用工具、给出最终答案。LangGraph 官方预置了ToolNode它帮你封装了“解析工具调用 - 执行工具 - 构造 ToolMessage - 写回消息列表”的标准逻辑。先定义工具def get_order_status(order_id: str) - str: # 模拟查询订单接口 if order_id A123: return 已发货预计 3 天内送达 return 订单不存在或已取消 def refund_request(order_id: str) - str: return f已为订单 {order_id} 提交退款申请1 个工作日内处理定义模型节点和工具节点from langgraph.prebuilt import ToolNode def agent_node(state: TicketState) - TicketState: response llm_with_tools.invoke(state[messages]) return {messages: [response]} tools [get_order_status, refund_request] tool_node ToolNode(tools)这里假设你已经把工具绑定到了模型对象上也就是llm_with_tools llm.bind_tools(tools)。真实项目里模型是否发起工具调用由模型判断我们需要做的是在图上配置循环条件。def should_continue(state: TicketState) - str: last_message state[messages][-1] if getattr(last_message, tool_calls, None): return continue return end workflow StateGraph(TicketState) workflow.add_node(agent, agent_node) workflow.add_node(tools, tool_node) workflow.add_edge(START, agent) workflow.add_conditional_edges( agent, should_continue, {continue: tools, end: END}, ) workflow.add_edge(tools, agent) app workflow.compile()执行流程是这样的agent 节点调用模型模型如果决定查询订单会在 AIMessage 里带上tool_calls条件边函数检查到tool_calls把流程引向 tools 节点ToolNode 执行工具后把结果追加进messages边把流程又引回 agent 节点模型看到工具结果如果认为信息足够就不再发起工具调用条件边把流程引向 END。这个循环特别适合“问订单状态 - 查库 - 再回复”这类多轮交互。模型第一次只能说出“让我查一下”第二次拿到工具结果后真正回答用户中间的上下文全部通过messages保留。4.4 这里最容易被忽略的细节第一个细节是tool_calls的判断。模型在决定调用工具时返回的 AIMessage 会带tool_calls属性不调用时通常是 None 或空列表。条件函数里用getattr兜底避免某些情况下属性不存在直接抛 AttributeError。第二个细节是 ToolNode 返回的 ToolMessage 必须能被模型看到而它进入状态列表的路径就是add_messagesreducer。如果你在状态定义里把messages写成了普通字段工具结果会被后面的节点覆盖模型永远看不到工具返回循环就会卡死。第三个细节是工具函数本身要保持简单和独立。每个工具只做一件事错误处理尽量在工具内部完成或者用自定义异常包装这样 ToolNode 报错时你能很快定位到具体工具而不是在一个很长的函数里翻找。5. 多智能体与高级编排从单节点到团队协作5.1 拆多个智能体的理由不只是为了“更专业”把一个大 prompt 塞给一个模型虽然简单但有两个问题一是上下文越来越长费用和延迟都上去了二是职责混在一起后prompt 里的指令互相打架。比如同一个模型既要扮演客服又要扮演质检员系统提示稍微模糊一点行为就会漂移。拆成多个智能体节点后每个节点只负责一件事prompt 各管各的状态字段各自独立。测试时也可以按节点分别验证而不是整个流程拉起来跑一遍才知道哪里不对。5.2 Supervisor 模式一个调度节点管一群执行节点多智能体协作里最常用的模式是 Supervisor有一个调度节点负责看当前局势、决定下一步派给哪个执行节点执行节点干完活把结果回报调度节点再决定下一步。在 LangGraph 里实现这个模式不复杂。调度节点读状态里当前任务类型条件边决定路由执行节点干完活后从各自出口回到调度节点调度节点再根据新状态判断是结束还是继续派单。这个“回到上级节点”的边就是多智能体协作的骨架。具体到代码结构大概是workflow.add_node(supervisor, supervisor_node) workflow.add_node(researcher, researcher_node) workflow.add_node(writer, writer_node) workflow.add_edge(START, supervisor) workflow.add_conditional_edges( supervisor, supervisor_router, {research: researcher, write: writer, finish: END}, ) workflow.add_edge(researcher, supervisor) workflow.add_edge(writer, supervisor)supervisor_router每次读到最新的执行结果判断下一步。如果 researcher 刚返回了一批资料就转 writerwriter 返回了成稿就转 finish。这个模式的好处是你可以在不改变其他节点的情况下替换调度策略逻辑非常集中。5.3 扇出扇入并行用 Send 一次性派发多个任务很多业务场景需要并行处理多个子任务比如客服工单里有多条待分析消息每条消息都要单独做一遍意图识别或者一个报告任务要同时检索三个资料库。LangGraph 提供了Send机制允许在条件边里动态创建多个并行分支。from langgraph.types import Send def distribute_tasks(state: State) - list[Send]: return [ Send(analysis_worker, {task: task}) for task in state[pending_tasks] ] workflow.add_conditional_edges( START, distribute_tasks, [analysis_worker], )Send的每个元素会启动一个独立的 worker 实例每个实例拿到属于自己的子状态互不干扰。当所有 worker 都返回后再通过汇合节点把结果合并。这个 map-reduce 式结构特别适合批量任务能显著降低总耗时。需要注意的是并行分支的 worker 如果依赖同一个字段做结果收集要设计好 reducer不然多个 worker 同时写一个字段会互相覆盖。常见做法是每个 worker 把结果写入一个按任务 ID 索引的字典汇合节点再把字典汇总成列表。5.4 断点恢复、状态持久化与人工介入LangGraph 的持久化能力是它区别于普通代码框架的一大优势。只要在编译时传入一个 checkpointer每次执行完一个节点状态就会被保存下来。from langgraph.checkpoint.memory import MemorySaver memory MemorySaver() app workflow.compile(checkpointermemory)编译后调用invoke时需要传入config里面带一个线程 ID。同一个线程 ID 的执行可以在任何节点暂停之后从指定节点继续跑。这对“人工审批”场景非常重要流程跑到“提交订单”节点前先暂停等管理员确认后再继续。config {configurable: {thread_id: ticket_123}} app.invoke(init_state, configconfig)个人体会是持久化的价值不只是“防止程序挂了重跑”更关键的是它给了你审计和回放能力。每次状态变更都有记录出问题就能定位到具体节点和具体输入这在生产环境里就是救命稻草。6. 避坑清单真实项目里最容易翻车的 6 个问题6.1 节点函数没返回状态被吃掉了新手最容易犯的错在节点函数里把状态改了却没 return。比如下面这段代码修改只发生在函数内部图层面根本感知不到。def bad_node(state: State) - State: state[intent] query # 错误示范直接改入参 # 注意这里没有 return 任何内容正确写法是返回一个部分更新的字典或者返回修改后的整个字典def good_node(state: State) - State: updated dict(state) updated[intent] query return updated需要明确的是State 对象在节点里是只读视图直接修改不保证生效。只有返回值才会被 reducer 处理并合并。6.2 消息列表被整体覆盖为什么必须用 add_messages如果你把 State 里的messages定义成普通list字段节点每次返回{messages: new_messages}都会把原来的消息历史整个替换掉。模型聊了几轮前面的话全没了智能体立刻失忆。在messages字段上挂Annotated[list, add_messages]才能保证新消息追加、历史消息保留。这个 reducer 也会自动处理 AIMessage、HumanMessage、ToolMessage 之间的关系所以强烈建议所有消息相关状态都走这个模式。6.3 循环停不下来max_recursion 与退出条件工具调用循环如果退出条件没写对图会一直在 agent 和 tools 之间打转。模型每次都说要调用工具工具也正常返回但模型就是不肯给出最终答案。要么是 prompt 里的指令不够明确要么是工具返回的信息格式有问题模型没法从中得出结论。LangGraph 默认有递归上限超过会抛RecursionError算是一层保险。但更建议在条件边里写得足够明确只有最后一条消息存在tool_calls才继续否则直接 END。有条件的情况下限流工具调用次数比如最多执行 5 次避免模型在长任务里反复试探。6.4 ToolNode 内部抛错整个流程直接炸工具是会出错的网络超时、接口限流、参数格式不对。如果工具函数内部不做异常处理错误会直接抛穿整个图导致流程中断状态也停留在半路。我的做法是给工具包一层统一异常捕获返回带有错误标记的结构化结果让模型能看见错误信息并决定下一步。比如查询接口超时返回“查询超时请稍后重试”模型会根据这个信息决定是重试还是转人工而不是图直接崩掉。6.5 Checkpointer 序列化问题类型要能 pickle启用持久化以后State 里所有字段都会被 checkpointer 序列化保存。如果 State 里塞了一个自定义对象而这个对象没有实现可序列化接口保存状态时就会报错。踩过这个坑后我的原则是所有塞进 State 的字段尽量用基础类型字符串、数字、列表、字典非要放对象就先转成字典或者只存一个标识 ID。这样不仅 checkpointer 稳定后续做状态回放和审计也更方便。6.6 上下文越滚越大长任务里的消息裁剪工具调用循环跑十几轮后messages会越来越长模型输入可能被截断费用也水涨船高。这不是 LangGraph 的 bug图本来就是保留全量状态但实际跑长任务时你要自己设计裁剪策略。简单做法是在 agent 节点调用模型前对messages做一次裁剪保留系统指令、保留最近几轮对话把中间过于久远的工具结果压缩成摘要。更精细的做法是用单独字段维护一个“关键信息摘要”让模型主要看摘要而不是每次把完整历史从头拼一遍。最后再分享一个我的使用习惯一开始接触 LangGraph 时我也被它的概念绕得头晕。后来我发现一个很管用的方法不要一上来就做多智能体先把一个最普通的工具调用流程用图跑通确认自己理解了状态怎么流、条件边怎么跳再逐步加分支和并行。等图的结构熟悉了再回头看 AgentExecutor 那种写死的循环会有一种“原来之前被限制住了”的感觉。另外建议把路由函数和业务节点分开维护。路由函数职责单一输入是 State输出是一个字符串这种纯函数非常好测试。我通常会给每个条件边单独建一个测试文件放十几组模拟状态进去断言路由结果保证每次改图都不会破坏原有分支逻辑。LangGraph 能做的事还有很多比如带记忆的多轮会话、定时批量任务、复杂审批流甚至把整张图打包成独立服务。先把这篇里的状态、节点、边、条件边吃透再往高级模式走你会发现复杂智能体编排并没有想象中那么玄关键是把每一步流转都变得看得见、控得住、可恢复。

LangGraph实战：用图编排搞定复杂智能体流程

相关推荐

学术速递信息筛选与论文解读实战：从信息源到日更体系

自动驾驶基座模型：从多模态感知到决策能力升级

Spring AI Alibaba停更后，Java AI应用开发的工程化选型与落地实践

更多资讯

Rust正成为AI基础设施的底层新标准

Rust为何成为AI基础设施的确定性选择

FreeRTOS任务通知实战详解：内核原理、API与避坑指南

数据服务监控运维实战：从指标体系到告警治理与故障排查

Verilog实现FPGA密码锁：矩阵键盘与状态机设计实战

AI大模型开发转行指南：从学习路线到项目实战的完整路径

今日更新

Verilog实现FPGA密码锁：矩阵键盘与状态机设计实战

数据服务监控运维实战：从指标体系到告警治理与故障排查

FreeRTOS任务通知实战详解：内核原理、API与避坑指南

本周精选

金融AI落地指南：从调研报告到技术选型与工程实践

大模型上下文成本与注意力优化技术深度解析——为何O(n²)平方复杂度无法被常规优化消除｜TaoToken 统一 Key 通道实测

基于GMM的语音识别系统原理与MATLAB实现详解

本月热门

我发现了一个新思路：用 Remotion + Claude Code 像写代码一样自动化生成短视频

Windows下 Codex 中 Chrome 和 Computer Use 插件不可用问题排查及解决参考方式：TaoToken 统一 Key 配置与验证

2026 大模型集体涨价：用 Python 做企业 Token 成本测算与选型避坑（附配置）

读完文章，想聊聊您的网站？