做应用层的朋友应该都有同感：模型本身往往不是最难的，真正难的是把模型塞进一条会转弯、会分叉、还会重复执行的业务流水线里。我帮某电商业务搭智能客服时，需求其实很朴素——用户进来先判断意图，查订单的走查询工具，要退货的走退货流程，客服搞不定的转人工。结果用传统方式写，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，建议避开太老的版本。安装直接用 pip：
pip 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(content=reply)]}这两个函数都没有碰图对象，输入输出都是 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(content="IT 工单已受理，请描述具体故障现象。")]} 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_edges：
workflow = 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 刚返回了一批资料，就转 writer；writer 返回了成稿，就转 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(checkpointer=memory)编译后调用invoke时需要传入config，里面带一个线程 ID。同一个线程 ID 的执行可以在任何节点暂停，之后从指定节点继续跑。这对“人工审批”场景非常重要：流程跑到“提交订单”节点前先暂停，等管理员确认后再继续。
config = {"configurable": {"thread_id": "ticket_123"}} app.invoke(init_state, config=config)个人体会是，持久化的价值不只是“防止程序挂了重跑”，更关键的是它给了你审计和回放能力。每次状态变更都有记录，出问题就能定位到具体节点和具体输入，这在生产环境里就是救命稻草。
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 能做的事还有很多，比如带记忆的多轮会话、定时批量任务、复杂审批流，甚至把整张图打包成独立服务。先把这篇里的状态、节点、边、条件边吃透，再往高级模式走，你会发现复杂智能体编排并没有想象中那么玄，关键是把每一步流转都变得看得见、控得住、可恢复。
