1. 为什么需要状态机来设计 Agent 工作流1.1 从一段式到三段式Agent 工作流的演进逻辑如果你用 LangChain 写过 Agent大概率经历过这样的场景一个 ReAct 循环里模型自己决定调什么工具、什么时候停代码写起来很爽但一旦上线就开始出问题——有时候工具调错了顺序有时候该停的时候不停有时候中间某一步失败了整个流程直接崩掉你连它走到哪一步了都不知道。这就是典型的“一段式”工作流所有逻辑塞在一个循环里靠模型的自由推理来驱动。它的优点是灵活缺点是不可控。在 Demo 阶段没问题但到了生产环境你需要的是可预测、可恢复、可观测的流程。“三段式”状态机的思路就来源于这个痛点。它把 Agent 的工作流拆成三个明确的阶段感知阶段输入解析与意图识别、决策阶段工具选择与参数构造、执行阶段工具调用与结果处理。每个阶段是一个独立的状态节点节点之间的转移条件由代码显式定义而不是交给模型自由发挥。LangGraph 正是为这种设计模式而生。它在 LangChain 的基础上引入了一张有向图图中的每个节点是一个函数或一个 Agent边定义了状态如何流转。你可以把它理解成给 Agent 装了一个“流程控制器”——模型仍然负责推理但流程的骨架由你来定。1.2 LangGraph 和 LangChain 到底什么关系很多人第一次接触 LangGraph 时会困惑我已经有 LangChain 了为什么还要多学一个东西简单说LangChain 解决的是“组件复用”问题——它提供了 LLM 封装、工具定义、提示词模板、输出解析器这些积木。而LangGraph 解决的是“流程编排”问题——它告诉你这些积木该怎么拼、按什么顺序拼、拼错了怎么回退。打个比方LangChain 像是一盒乐高零件LangGraph 是那张拼装图纸加上一个可以随时暂停、回退、重放的拼装台。你可以只用 LangChain 的零件用 while 循环自己拼但一旦流程复杂到需要分支、需要人工介入、需要失败重试自己写循环的维护成本就会指数级上升。LangGraph 的核心抽象只有三个State状态一个贯穿全流程的共享数据结构通常是一个 TypedDict 或 Pydantic 模型所有节点都能读写它。Node节点一个函数接收当前 State返回 State 的更新部分。Edge边定义节点之间的转移关系可以是固定的也可以是条件判断的。这三个东西组合起来就是一张状态机图。你编译这张图得到一个可执行对象每次调用就是一次状态流转。1.3 哪些场景真的需要上 LangGraph不是所有 Agent 项目都需要 LangGraph。如果你只是做一个简单的问答机器人或者一个单轮工具调用用 LangChain 的 AgentExecutor 就够了没必要引入额外的抽象层。但以下场景我强烈建议直接上 LangGraph多步骤流程比如“先检索知识库 → 再判断是否需要追问 → 再生成回答 → 再审核敏感词 → 最后输出”这种流程用状态机表达非常自然。需要人工介入比如审批流、内容审核流程走到某一步需要暂停等人确认后再继续。LangGraph 的interrupt机制原生支持这个。需要失败重试和回退某个节点执行失败了可以回到上一个状态重新走而不是整个流程重来。需要多 Agent 协作比如一个“规划 Agent”负责拆解任务一个“执行 Agent”负责调工具一个“审核 Agent”负责检查结果它们之间的协作关系用图来表达比用代码硬编码清晰得多。需要持久化和恢复流程跑到一半服务重启了能从上次的状态继续跑。LangGraph 的 Checkpointer 机制就是干这个的。反过来如果你的流程是线性的、不需要暂停、不需要回退、不需要多 Agent那用 LangChain 的 Chain 或者简单的函数调用就够了。工具选型的第一原则是匹配需求而不是追新。2. LangGraph 核心概念拆解与实操要点2.1 State 设计整个工作流的地基State 是 LangGraph 里最重要的设计决策。它决定了整个流程中哪些数据需要被传递、哪些节点需要读写哪些字段。最基础的 State 是一个 TypedDictfrom typing import TypedDict, Annotated from langgraph.graph import StateGraph, END from langgraph.graph.message import add_messages class AgentState(TypedDict): messages: Annotated[list, add_messages] current_step: str tool_result: str retry_count: int这里有几个关键点第一Annotated和 reducer 函数。默认情况下每个节点返回的 State 更新会覆盖原值。但messages字段用了add_messages这个 reducer意思是新消息会追加到列表里而不是替换。这个设计非常关键——对话历史需要累积不能覆盖。第二State 字段要精简。我见过有人在 State 里塞了十几个字段结果每个节点都要处理一堆不相关的数据。原则是只有需要跨节点传递的数据才放进 State节点内部的临时变量不要放。第三retry_count这类控制字段要显式定义。如果你需要重试逻辑重试次数必须放在 State 里否则节点之间无法共享这个信息。注意State 的字段类型要尽量明确。用str而不是Any用list而不是object。这样在调试时你能清楚地知道每个字段应该是什么类型减少运行时错误。2.2 Node 编写每个节点只做一件事Node 就是一个普通的 Python 函数签名是def node_name(state: AgentState) - dict。返回值是一个字典表示要更新 State 的哪些字段。def parse_intent(state: AgentState) - dict: last_message state[messages][-1] # 调用 LLM 做意图识别 intent llm.invoke(f判断以下用户输入的意图{last_message.content}) return {current_step: intent_parsed, intent: intent.content} def call_tool(state: AgentState) - dict: intent state.get(intent, ) # 根据意图选择工具 if 查询天气 in intent: result weather_tool.invoke(state[messages][-1].content) else: result 未匹配到工具 return {tool_result: result, current_step: tool_called}写 Node 有几个实操心得单一职责。一个节点只做一件事。如果你发现一个节点里做了意图识别又做了工具调用又做了结果格式化那应该拆成三个节点。拆开的好处是每个节点可以独立测试、独立重试、独立替换。返回值只包含需要更新的字段。不需要把整个 State 都返回只返回你修改的字段即可。LangGraph 会自动合并。异常处理要显式。节点内部如果可能抛异常要么在节点内捕获并返回错误信息到 State要么让异常抛出由 LangGraph 的重试机制处理。不要静默吞掉异常。2.3 Edge 与条件路由流程的“红绿灯”Edge 定义了节点之间的流转关系。最简单的形式是固定边graph.add_edge(parse_intent, call_tool)但真正强大的是条件边它根据 State 的内容决定下一步走哪个节点def route_after_intent(state: AgentState) - str: intent state.get(intent, ) if 查询 in intent: return call_tool elif 闲聊 in intent: return chat_response else: return fallback graph.add_conditional_edges( parse_intent, route_after_intent, { call_tool: call_tool, chat_response: chat_response, fallback: fallback } )条件路由函数返回一个字符串这个字符串对应到映射表里的节点名。这个设计的好处是路由逻辑和节点逻辑分离——你可以在不改动节点代码的情况下调整流程走向。提示条件路由函数应该是纯函数只依赖 State 的内容做判断不要在里面做副作用操作比如调 API、写数据库。这样路由逻辑才能可预测、可测试。2.4 Checkpointer让流程可以暂停和恢复Checkpointer 是 LangGraph 里最容易被忽视但最有价值的功能之一。它会在每个节点执行后自动保存 State 的快照这样你可以在流程中间暂停等人工确认后继续服务重启后从上次的状态恢复回放整个流程的执行历史from langgraph.checkpoint.memory import MemorySaver memory MemorySaver() app graph.compile(checkpointermemory) # 调用时传入 thread_id config {configurable: {thread_id: user_123}} result app.invoke({messages: [(user, 帮我查一下明天北京的天气)]}, config)thread_id是会话标识同一个 thread_id 的多次调用会共享状态历史。这意味着你可以先调一次让流程走到一半然后用同一个 thread_id 再调一次继续走。生产环境建议用SqliteSaver或PostgresSaver做持久化MemorySaver 只适合开发和测试。3. 从零搭建一个可靠 Agent 工作流的完整过程3.1 环境准备与依赖安装先明确版本。LangGraph 的 API 在 0.1 到 0.2 之间有较大变化网上很多教程还是旧版写法直接抄会报错。截至我写这篇内容时稳定版本是 0.2.x。pip install langgraph0.2.60 langchain0.3.13 langchain-openai0.2.14如果你用 conda 管理环境建议单独建一个环境避免和已有的 LangChain 项目冲突conda create -n langgraph-demo python3.11 conda activate langgraph-demo pip install langgraph langchain langchain-openaiPython 版本建议 3.10 以上因为 LangGraph 用了一些较新的类型注解特性。3.2 定义 State 与搭建图骨架我们以一个“智能客服工单处理”场景为例完整走一遍。这个场景的流程是接收用户消息判断意图咨询/投诉/其他如果是咨询走知识库检索如果是投诉走工单创建如果是其他走人工转接最后统一做回复生成先定义 Statefrom typing import TypedDict, Annotated, Literal from langgraph.graph.message import add_messages class CustomerServiceState(TypedDict): messages: Annotated[list, add_messages] intent: Literal[consult, complaint, other, ] knowledge_result: str ticket_id: str final_response: str retry_count: int然后创建图from langgraph.graph import StateGraph, END workflow StateGraph(CustomerServiceState) # 添加节点 workflow.add_node(classify_intent, classify_intent) workflow.add_node(search_knowledge, search_knowledge) workflow.add_node(create_ticket, create_ticket) workflow.add_node(transfer_human, transfer_human) workflow.add_node(generate_response, generate_response) # 设置入口 workflow.set_entry_point(classify_intent) # 添加条件边 workflow.add_conditional_edges( classify_intent, route_by_intent, { consult: search_knowledge, complaint: create_ticket, other: transfer_human } ) # 各分支汇聚到回复生成 workflow.add_edge(search_knowledge, generate_response) workflow.add_edge(create_ticket, generate_response) workflow.add_edge(transfer_human, generate_response) workflow.add_edge(generate_response, END) app workflow.compile()这个骨架清晰表达了流程逻辑分类之后走三条不同路径最后汇聚到统一回复。图的结构一目了然新人接手也能快速理解。3.3 各节点函数的具体实现意图分类节点from langchain_openai import ChatOpenAI from langchain_core.messages import SystemMessage llm ChatOpenAI(modelgpt-4o-mini, temperature0) def classify_intent(state: CustomerServiceState) - dict: user_message state[messages][-1].content prompt f判断以下用户消息的意图只返回一个词 - consult咨询类问题 - complaint投诉类问题 - other其他 用户消息{user_message} result llm.invoke([SystemMessage(contentprompt)]) intent result.content.strip().lower() if intent not in [consult, complaint, other]: intent other return {intent: intent}知识库检索节点def search_knowledge(state: CustomerServiceState) - dict: query state[messages][-1].content # 这里替换成你实际的知识库检索逻辑 # 比如用向量数据库做相似度搜索 result knowledge_base.search(query, top_k3) return {knowledge_result: result}工单创建节点import uuid def create_ticket(state: CustomerServiceState) - dict: ticket_id fTK-{uuid.uuid4().hex[:8].upper()} # 实际项目中这里会写入数据库 return {ticket_id: ticket_id}回复生成节点def generate_response(state: CustomerServiceState) - dict: intent state[intent] if intent consult: context state.get(knowledge_result, ) prompt f根据以下知识库内容回答用户问题\n{context} elif intent complaint: ticket_id state.get(ticket_id, ) prompt f用户投诉已受理工单号{ticket_id}请生成安抚回复。 else: prompt 用户需要人工服务请生成转接提示。 response llm.invoke([ SystemMessage(contentprompt), state[messages][-1] ]) return {final_response: response.content, messages: [response]}3.4 条件路由函数的实现细节def route_by_intent(state: CustomerServiceState) - str: intent state.get(intent, other) if intent consult: return consult elif intent complaint: return complaint else: return other这个函数看起来简单但有一个容易踩的坑返回值必须和add_conditional_edges里的映射键完全一致。如果返回了映射表里没有的键LangGraph 会直接报错。所以我在函数里做了兜底任何意外情况都走other分支。另一个坑是条件路由函数不能是异步的除非你用ainvoke。如果你在路由函数里调了异步 API会直接报错。路由函数应该只做同步的、轻量的判断。3.5 编译、调用与结果验证from langgraph.checkpoint.memory import MemorySaver memory MemorySaver() app workflow.compile(checkpointermemory) config {configurable: {thread_id: test_001}} result app.invoke( {messages: [(user, 你们的产品怎么退货)]}, config ) print(result[intent]) # consult print(result[final_response]) # 生成的回复内容如果你想看整个流程走了哪些节点可以用stream方法for event in app.stream( {messages: [(user, 我要投诉)]}, config ): print(event)输出会显示每个节点执行后的 State 更新非常直观。调试阶段我强烈建议用stream而不是invoke因为你能看到每一步的中间状态。4. 常见问题与排查技巧实录4.1 节点返回值类型错误这是新手最常遇到的问题。报错信息通常是InvalidUpdateError或者Expected dict, got ...。原因通常是节点函数返回了非字典类型或者返回的字典键不在 State 定义里。比如 State 里没有intent字段但节点返回了{intent: consult}LangGraph 会直接报错。排查方法检查 State 的 TypedDict 定义确保每个返回的键都在里面。如果你用的是 Pydantic 模型做 StateLangGraph 会做更严格的类型校验。4.2 条件边路由不到目标节点报错信息类似Branch route returned unknown target。原因通常是路由函数返回的字符串和映射表的键不匹配。比如路由函数返回consult但映射表里写的是consult_node。排查方法把路由函数的返回值和映射表的键打印出来对比。我习惯在路由函数里加一行日志def route_by_intent(state): intent state.get(intent, other) print(f[DEBUG] routing with intent{intent}) ...4.3 状态更新被覆盖而不是追加如果你发现messages列表每次只有最新一条而不是累积的说明 reducer 没生效。检查两点一是 State 定义里messages字段是否用了Annotated[list, add_messages]二是节点返回时是否用了{messages: [new_message]}这种列表形式而不是{messages: new_message}。4.4 流程卡住不结束如果流程走到某个节点后不再继续通常是两个原因一是该节点没有出边也没有连到END二是条件路由函数返回了一个没有对应边的值。排查方法用app.get_graph().draw_mermaid()生成图的可视化注意这里只是生成文本描述不是渲染图表检查每个节点是否都有出边。4.5 常见问题速查表问题现象可能原因解决方法InvalidUpdateError节点返回了 State 中不存在的字段检查 TypedDict 定义对齐字段名路由报 unknown target路由返回值与映射键不匹配打印路由返回值核对映射表messages 不累积reducer 未生效确认使用 Annotated add_messages流程不结束节点缺少出边或未连 END检查图的边定义确保所有路径可达 END恢复后状态丢失未配置 Checkpointer编译时传入 checkpointer 参数异步节点报错混用 sync/async统一用 async 节点 ainvoke或全用 sync提示调试 LangGraph 时养成用app.stream()而不是app.invoke()的习惯。stream 会逐步输出每个节点的执行结果能快速定位问题出在哪一步。4.6 几个我踩过的坑坑一在节点里直接修改 State。LangGraph 的 State 应该是不可变的节点只能通过返回值来更新 State。如果你在节点里写了state[intent] consult虽然 Python 不会报错但 LangGraph 不会感知到这个修改流程会出问题。正确做法是return {intent: consult}。坑二条件路由函数里做耗时操作。我见过有人在路由函数里调 LLM 做判断结果整个流程变得极慢。路由函数应该是轻量的、确定性的判断耗时的推理应该放在节点里。坑三忘记设置 recursion_limit。如果你的图里有循环比如重试逻辑一定要设置recursion_limit否则可能无限循环。默认是 25 步可以通过 config 调整config {configurable: {thread_id: x}, recursion_limit: 50}坑四Checkpointer 的 thread_id 冲突。如果你用同一个 thread_id 跑不同的流程状态会串。每个独立的会话应该用独立的 thread_id通常用用户 ID 会话 ID 组合。5. 进阶多 Agent 协作与人工介入5.1 用子图拆分复杂流程当你的流程复杂到一张图放不下时可以用子图Subgraph来拆分。子图本质上就是一个编译好的 StateGraph可以作为另一个图的节点使用。# 定义一个子图 sub_workflow StateGraph(SubState) sub_workflow.add_node(step_a, step_a) sub_workflow.add_node(step_b, step_b) sub_workflow.set_entry_point(step_a) sub_workflow.add_edge(step_a, step_b) sub_workflow.add_edge(step_b, END) sub_graph sub_workflow.compile() # 在主图中使用子图 main_workflow.add_node(sub_process, sub_graph)子图的好处是关注点分离——每个子图负责一个独立的业务逻辑主图只负责编排。这在多团队协作时特别有用每个团队维护自己的子图。5.2 人工介入的 interrupt 机制LangGraph 的interrupt允许你在某个节点执行前暂停流程等外部输入后再继续from langgraph.types import interrupt def review_node(state: CustomerServiceState) - dict: # 暂停等待人工审核 human_input interrupt(请审核以下回复是否合适...) if human_input approve: return {final_response: state[final_response]} else: return {final_response: human_input}调用时流程会在这个节点暂停返回一个中断信号。外部系统处理后用Command(resume...)继续from langgraph.types import Command # 第一次调用流程会暂停 result app.invoke(input_data, config) # 人工审核后继续 result app.invoke(Command(resumeapprove), config)这个机制在审批流、内容审核、高风险操作确认等场景非常实用。5.3 多 Agent 协作的两种模式模式一Supervisor 模式。一个“主管 Agent”负责决策把任务分发给不同的“工人 Agent”工人完成后汇报给主管主管决定下一步。模式二Handoff 模式。Agent 之间直接传递控制权比如客服 Agent 判断需要技术支持直接把对话交给技术 Agent。两种模式在 LangGraph 里都可以用条件边来实现。Supervisor 模式更可控适合流程明确的场景Handoff 模式更灵活适合对话式交互。6. 一些实际使用后的体会LangGraph 最大的价值不是让你写出更复杂的 Agent而是让你用结构化的方式管理复杂度。在没有 LangGraph 之前我写多步骤 Agent 的方式是在一个大函数里塞各种 if-else代码超过 200 行就开始失控。用了 LangGraph 之后每个节点是独立的函数流程是显式的图调试时能精确定位到哪个节点出了问题。但它也不是银弹。如果你的流程很简单引入 LangGraph 反而增加了抽象层得不偿失。我的判断标准是当你发现自己在用 while 循环 if-else 管理流程状态时就该考虑 LangGraph 了。另外LangGraph 的文档更新很快网上很多教程的 API 已经过时。遇到报错时第一件事是去官方文档确认当前版本的 API 签名而不是盲目搜索。我踩过好几次这个坑浪费了不少时间在过时的写法上。最后分享一个小技巧在开发阶段给每个节点函数加上日志输出记录输入 State 和输出 State。这样当流程出问题时你能快速定位是哪个节点的输入不对还是输出不对。上线前再把日志级别调低即可。
