1. 从“会聊天”到“会干活”LangGraph到底解决了什么问题在做 AI Agent 之前我跟你一样也觉得 LangChain 已经够用了——链式调用、工具调用、模板提示词跑个客服机器人、写个文档总结模型都挺顺利。直到开始做真正要落地的业务系统才发现纯链式方案在三个地方特别别扭。第一个是状态共享。一个复杂任务往往要拆成“理解需求 → 查数据库 → 调外部接口 → 生成回复”好几步纯 LangChain 里每一步都要手动把上一轮的输出塞给下一轮变量越传越多函数参数越来越长代码写着写着就变成一堆样板代码。第二个是控制流。业务里充满了“如果用户要退款就走审批如果只是查物流就直接回复”这种分叉逻辑链式调用只能靠硬编码 if-else 在每一步里判断逻辑稍微一多就和业务代码缠在一起你根本分不清是提示词问题还是流程问题。第三个是人工介入。很多真实场景不能全自动跑完比如转账、退款、发送重要通知这些动作必须“卡在人审环节”链式调用一旦跑起来就只能跑完很难在中间停下来等一个人确认再继续。后来我换到 LangGraph思路一下子顺了。它的核心非常朴素把一次任务执行定义成一张有向图节点是“要做的事”边是“怎么流转”而所有节点共享一份全局状态。你可以随时根据状态决定下一步去哪也可以在某个节点上主动暂停等待人工输入恢复之后接着往下跑。这篇文章我就从状态管理、条件路由、人工介入这三块讲透最后给出一个完整可跑的客服 Agent 案例希望能帮到那些正从“Demo 能跑”迈向“业务能上线”的朋友。2. 状态管理的底层逻辑State Schema、Reducer 与持久化2.1 State Schema 声明你定义的其实是一份“公共黑板”LangGraph 里的 State 是贯穿整个图的核心数据结构所有节点都能读写它。它的定义方式和 Pydantic 模型或 TypedDict 差不多但有一个关键差异不同字段的更新策略由你是否给字段配 Reducer 决定。from typing import Annotated, TypedDict from langgraph.graph import add_messages class AgentState(TypedDict): # 带 Reducer每次更新都会“追加”而不是覆盖 messages: Annotated[list, add_messages] # 不带 Reducer后写覆盖先写 order_id: str status: str # 中间过程用的临时变量 candidate_action: dict默认情况下同一个字段被多次写入时是“后写覆盖先写”这像一块黑板谁最后擦掉重写黑板上的字就是最新的。这适合 status、order_id 这种单值字段。而 messages 用了 add_messages 这个 Reducer表示“新消息进来时不是覆盖而是追加到原列表里”。这种设计在第一眼看起来有点绕但实际上它是整套状态管理的命门。如果你把状态当成普通字典每次节点 return 一个字段就整体覆盖那多轮对话的历史会被越冲越薄用 Reducer 则让字段拥有了自定义的“合并策略”事实上 LangGraph 里的状态管理本质上就是**“字段粒度”的读写策略管理**。2.2 Reducer 是怎么被触发的节点返回值与状态更新流程每个节点函数的返回值会“合并”进全局状态Reducer 就在合并这一刻生效。一个很常见的自定义需求是合并字典字段比如你要在状态里记录多个工具调用结果而不是只保留最后一个from typing import Annotated, TypedDict def merge_dict(left: dict, right: dict) - dict: 把两个字典合并并以 right 为准覆盖同名字段。 if not left: return right return {**left, **right} class WorkerState(TypedDict): tool_results: Annotated[dict, merge_dict] messages: Annotated[list, add_messages]在写这个自定义 Reducer 之前我踩过一个挺隐蔽的坑多个工具并行执行后各自向状态里写 tool_results如果不做合并后完成任务的一方会把先完成的结果整个覆盖掉最后状态里只剩一份数据。加了 Reducer 之后才意识到它本质上就是让你给每个字段定义“收发室规则”——是快递堆在门口追加还是新包裹替换旧包裹覆盖。2.3 状态持久化与 Thread ID多用户多会话必须提前想清楚LangGraph 的状态默认存在内存里图跑完就没了。但只要传入一个 checkpoint_saver状态就能被持久化到指定存储同时每一条执行线都靠 thread_id 隔离。from langgraph.checkpoint.memory import MemorySaver from langgraph.graph import StateGraph graph workflow.compile(checkpointerMemorySaver()) config {configurable: {thread_id: user-001-conversation-1}} result graph.invoke({messages: [(user, 查一下订单号A10086的状态)]}, configconfig)这个设计对我帮助很大的是thread_id 天然等于会话 ID。同一个用户传同一个 thread_id图就知道这是接着上次继续聊不同用户传不同 thread_id状态天然隔离互不串线。不过要提醒一句MemorySaver 只是存在进程内存重启就没了生产环境至少换成持久化方案。我自己的项目里开始图省事用了 MemorySaver结果 UAT 环境一重启用户的历史对话全丢了排查了半天才意识到是 memory 的问题。后来换成基于 SQLite 的 SqliteSaver 才稳定下来配置上只是 saver 的类型不同接口完全一样。3. 条件路由让流程分叉贴合真实业务语义3.1 条件边的两个关键约定LangGraph 里路由一共两种普通边Edge和条件边Conditional Edge。普通边表示“A 跑完一定去 B”条件边则表示“A 跑完要根据某个判定函数的结果选择下一个节点”。条件边的判定函数有两条约定必须记住输入是整个 State 字典返回值是下一个节点的名称字符串。def route_after_llm(state: AgentState) - str: # 根据 LLM 给出的结构化决策结果路由 action state.get(candidate_action, {}) if action.get(needs_tool): return call_tool if action.get(needs_human): return human_approval return final_reply然后构建图时这样连from langgraph.graph import StateGraph, START, END builder StateGraph(AgentState) builder.add_node(llm_reason, llm_reason_node) builder.add_node(call_tool, call_tool_node) builder.add_node(human_approval, human_approval_node) builder.add_node(final_reply, final_reply_node) builder.add_edge(START, llm_reason) builder.add_conditional_edges(llm_reason, route_after_llm) builder.add_edge(call_tool, llm_reason) # 工具返回后回 LLM 继续推理 builder.add_edge(human_approval, final_reply) builder.add_edge(final_reply, END)这里最需要注意的就是工具调用后的回流。我一开始习惯写“call_tool → END”结果是 LLM 只能调一次工具遇到需要先查订单再调退款接口的复合场景流程直接断掉。正确做法是工具执行完成后重新回到 LLM 节点让模型基于工具结果做下一轮判断形成一个“推理 → 行动 → 再推理”的循环回路。3.2 什么时候让 LLM 决策什么时候用硬规则条件路由里最常见的纠结是分叉逻辑到底让 LLM 判断还是代码判断我的建议是画一条线涉及自然语言理解的让 LLM 判断涉及系统状态的用规则判断。比如“用户这句话是不是要退款”属于语义理解应该走 LLM而“退款金额是否超过 2000 元需要人工审批”是明确数值规则应该走代码。这两者不要混在同一个路由函数里否则既难调试又难测试。我在案例中把路由函数拆成两层先判断状态里的 order_id 是否存在、金额是否超过阈值等硬性条件再根据 LLM 的结构化输出字段决定最终流向。为了让 LLM 的判断尽量稳定强烈建议给 LLM 一个结构化输出from langchain_core.pydantic_v1 import BaseModel, Field class Decision(BaseModel): needs_tool: bool Field(description是否需要调用工具获取更多信息) needs_human: bool Field(description是否需要人工介入审批) reply: str Field(description若不需要工具和人工直接生成的用户答复) llm_with_structure llm.with_structured_output(Decision)之所以这么做是因为直接让 LLM 在自由文本里说“我需要工具”然后你写正则去匹配 keywords做法极其脆弱。我第一次用 LangGraph 做交易 Agent 时就是这么干的结果用户换了种说法“帮我看看扣款有没有问题”模型写出了“maybe use tool to check”正则完全没匹配上整个流程掉进了死分支。换成结构化输出后判定字段永远在那里只是个别场合值可能为 False逻辑分支也就稳定了。3.3 一个值得复制的路由范式复杂系统中路由函数往往不止一个。一个我常用的设计模式是一个节点一个专属路由。LLM 节点后面跟“判断是否调工具/是否需要人工/是否直接回复”的路由人工确认节点后面跟“确认通过/拒绝”的路由。每个路由函数只解决一个节点的去向问题不要试图写一个“万能路由”包揽全部判断。def route_after_human(state: AgentState) - str: if state.get(human_approved): return execute_action return final_reply一个路由只处理一个决策点这样的图逻辑才能被人一眼看懂也方便后面在 LangSmith 上排查每一条链路的走向。4. 人工介入interrupt 机制与 Human-in-the-loop 的实现细节4.1 interrupt 不是“sleep”而是真正把图挂起LangGraph 的人工介入核心是一个 interrupt 函数。执行到调用它的节点时图的执行会被挂起状态被持久化保存进程不阻塞、不占用资源相当于整个执行线“冻结”。之后只要通过同一个 thread_id 传入恢复指令图会从刚才暂停的位置继续往下执行。from langgraph.types import interrupt def human_approval_node(state: AgentState) - dict: action state[candidate_action] # 把待审批的内容暴露给前端/人工 approval interrupt({ action_type: action[type], amount: action[amount], target_account: action[target_account], }) # approval 是恢复时传入的数据 return {human_approved: approval.get(approved, False)}这里有一个容易让人迷惑的地方interrupt 传入的参数会被“挂起”并作为当前执行点的快照存在 checkpoint 里。你可以在外部通过 graph.get_state配置读取这个挂起信息展示给人看也可以通过 update_state 或 Command(resume...) 把审批结果传回去。传回之后interrupt 调用位置会返回这个值继续执行后半段代码。4.2 恢复执行的两种姿势update_state 与 Command如果你需要修改状态再让图继续跑用 update_state 更顺手# 人工审核后把结果写进状态 graph.update_state( config, {human_approved: True}, ) # 从暂停处继续执行 graph.invoke(None, configconfig)如果只想直接“告诉 interrupt 返回什么”用 Command 更简洁from langgraph.types import Command # 从暂停处恢复并把值传给 interrupt graph.invoke( Command(resume{approved: True, note: 金额可接受同意退款}), configconfig, )两者的差别很微妙update_state 是在“当前暂停节点之前”把状态改了再继续Command(resume...) 则是直接给 interrupt 的返回值。这个细节我在实际开发中花了一些时间才完全理解但现在记忆方式很简单——interrupt 本身是一个等待输入的特殊“工具”resume 传进去的就是这个工具的返回值。4.3 人工介入最容易被低估的坑超时与逾期处理人工介入不是没有代价的。业务上如果人工一直不点按钮图就一直挂在那个状态。生产环境里必须有超时清理机制我实际的处理方式是在应用层加一个定时扫描任务定时查询所有“等待人工”状态的 thread超过 30 分钟未处理的自动写入超时决定并恢复执行同时通过消息队列通知相关人工审核人员。这个机制看似简单但是没有它线上跑了一阵就会出现一堆“僵尸会话”——用户以为系统坏了其实是图在等一个人而那个人出差了三天。另外人工介入期间用户如果重新发送消息需要区分处理。我的设计是如果图正处于中断状态新消息不进主流程而是走“留言队列”等人工决定完成后一起处理。5. 完整案例一个带人工审核的订单客服 Agent5.1 场景定义与整体链路下面我把上面讲的三块内容组合成一个能直接落地的例子一个订单客服 Agent用户提出订单相关问题Agent 先查订单如果用户要退款且金额超过阈值暂停请求人工审批否则自动生成回复。整条链路如下用户发起消息LLM 判断意图并决定是否需要查订单查订单工具返回订单数据LLM 再次判断是直接回复还是进入人工审批需要审批则中断等人工决定人工通过则执行退款动作并回复拒绝则直接回复拒绝文案。5.2 定义状态与节点函数from typing import Annotated, TypedDict from langgraph.graph import add_messages from langgraph.types import interrupt class OrderState(TypedDict): messages: Annotated[list, add_messages] order_id: str | None order_info: dict | None decision: dict | None human_approved: bool | None def llm_reason_node(state: OrderState) - dict: LLM 推理节点判断意图、是否查单、是否需要人工。 from langchain_openai import ChatOpenAI llm ChatOpenAI(modelgpt-4o-mini, temperature0) decision llm.with_structured_output(Decision).invoke( [{role: user, content: state[messages][-1].content}] ) return {decision: decision.model_dump(), messages: [(assistant, decision.reply)]} def lookup_order_node(state: OrderState) - dict: 模拟查订单工具。 order_id state[decision].get(order_id) # 实际项目里换成真实数据库或接口调用 order { id: order_id, amount: 3280.00, status: 已发货, can_refund: True, } return {order_info: order} def human_approval_node(state: OrderState) - dict: order state[order_info] approval interrupt({ order_id: order[id], amount: order[amount], action: 退款, }) return {human_approved: approval.get(approved, False)}5.3 条件路由与图定义def route_after_llm(state: OrderState) - str: decision state.get(decision, {}) if decision.get(needs_tool): return lookup_order if decision.get(needs_human): return human_approval return final_reply def route_after_lookup(state: OrderState) - str: # 查完订单后再让 LLM 判断一次是否需要人工介入 return llm_reason def route_after_human(state: OrderState) - str: return execute_refund if state.get(human_approved) else final_reply def execute_refund_node(state: OrderState) - dict: # 实际项目里对接支付/财务接口 return {messages: [(assistant, 退款已执行原路退回预计1-3个工作日到账。)]} def final_reply_node(state: OrderState) - dict: # 如果已经由 execute_refund_node 写过回复这里可以不再覆盖 return {}构建图的代码from langgraph.graph import StateGraph, START, END builder StateGraph(OrderState) builder.add_node(llm_reason, llm_reason_node) builder.add_node(lookup_order, lookup_order_node) builder.add_node(human_approval, human_approval_node) builder.add_node(execute_refund, execute_refund_node) builder.add_node(final_reply, final_reply_node) builder.add_edge(START, llm_reason) builder.add_conditional_edges(llm_reason, route_after_llm) builder.add_conditional_edges(lookup_order, route_after_lookup) builder.add_conditional_edges(human_approval, route_after_human) builder.add_edge(execute_refund, final_reply) builder.add_edge(final_reply, END) graph builder.compile(checkpointerMemorySaver())这个图有一个有意思的地方lookup_order 节点之后没有直接连 END而是连回 llm_reason让 LLM 看到订单数据后重新决策。这种“工具结果喂回模型”的循环是 Agent 系统稳定工作的关键只有这样才能处理“查完发现金额超限需要改走人工审批”这种后续演变。5.4 用 FastAPI 把 Agent 跑起来图定义好之后对外服务需要一个薄薄的应用层。我的做法是三个端点创建会话、发消息并运行图、查询或恢复等待中的任务。from fastapi import FastAPI from pydantic import BaseModel app FastAPI() class ChatRequest(BaseModel): thread_id: str content: str class ResumeRequest(BaseModel): thread_id: str approved: bool note: str app.post(/chat) def chat(req: ChatRequest): config {configurable: {thread_id: req.thread_id}} result graph.invoke( {messages: [(user, req.content)]}, configconfig, ) return { thread_id: req.thread_id, messages: result[messages][-1].content, state: graph.get_state(config).values, } app.post(/resume) def resume(req: ResumeRequest): config {configurable: {thread_id: req.thread_id}} # 恢复执行并把审批结果传给 interrupt result graph.invoke( Command(resume{approved: req.approved, note: req.note}), configconfig, ) return {messages: result[messages][-1].content}这样一个最小闭环就通了。用户发消息如果触发了人工审批/chat 返回的 state 里会有 interrupt 快照信息前端就可以根据这个信息渲染一个“待审批”的卡片人工点击同意或拒绝后调用 /resume 续跑。这里没有用复杂的 WebSocket业务不复杂时 REST 完全够用。5.5 前端交互与 interrupt 状态的识别人工介入在前端呈现时需要判断当前会话是否处于“等待恢复”状态。LangGraph 提供了 get_state 的 next 字段来帮忙state_snapshot graph.get_state(config) if state_snapshot.next: # 非空说明有节点待执行即等待恢复 resume_input state_snapshot.tasks[0].interrupts[0].value print(待审批内容:, resume_input)next 为空且已完成说明流程正常结束next 非空则说明执行停在某个节点前等待外部输入。这个判断逻辑我封装在 API 层的一个工具函数里前端轮询状态时直接拿到“waiting_approval: true/false”。6. 落地过程中的坑与优化建议6.1 状态字段被静默覆盖一定要先想好 Reducer我曾在一个图里同时让两个工具节点写同一个 status 字段本来想的是“以最后执行为准”但因为并行执行状态里最终 status 是随机某一个工具的。排查了半天才发现问题出在状态的“覆盖”语义上。哪个字段需要累积、哪个字段需要覆盖必须在定义 State 时就想清楚尤其是 messages、tool_results、errors 这些天然需要追加的字段没有 Reducer 一定会出问题。6.2 大状态导致 Token 浪费消息裁剪策略一次长对话如果所有消息都堆在 messages 里LLM 节点的上下文会越来越长成本也随之上升。这并非 LangGraph 特有但 LangGraph 的 add_messages 会保留全部历史放大了这个问题。我的做法是在每次 invoke 前做一次预处理把超过 N 轮的早期消息压缩成摘要作为一条系统消息注入。def trim_messages(state: OrderState) - OrderState: msgs state[messages] if len(msgs) 10: # 简单策略保留最近6条更早的合并成摘要 recent msgs[-6:] summary f早期对话摘要用户近期的前N轮内容已省略。 return {**state, messages: [(system, summary)] recent} return state加了这个节点之后的实测效果很明显GPT-4o-mini 的 token 消耗大概降了四成而且对多步工具类任务影响很小因为关键状态已经落在 order_info、decision 这些结构化字段里了。6.3 长期运行任务的超时与重试生产环境里图不是随时都能正常跑完。外部接口超时、LLM 报错、人工迟迟不响应这些都会让图“挂住”。建议在应用层定义整体的执行超时比如单轮任务超过 120 秒自动标记失败对可重试的任务用 thread_id 存一份输入快照失败后可以从头重新发起而不是在一个坏状态上反复重试。6.4 给 LLM 节点设计好“无法决策”的兜底只要是靠 LLM 路由就必然有模型抽风的时候。我在路由函数里加了一个兜底分支def route_after_llm(state: OrderState) - str: decision state.get(decision, {}) if decision.get(needs_tool): return lookup_order if decision.get(needs_human): return human_approval # 兜底如果什么都没匹配上走 final_reply 给用户一句友好提示 return final_reply这段代码看起来很简单但它救了我很多次。模型偶尔会返回一个空的 decision 对象如果路由函数抛异常整个图都会崩溃而走 final_reply 给一个“我暂时没能理解您的需求请换个方式描述”的兜底文案用户体验会好很多。6.5 可观测性优先尽早接入 LangSmith 或等价追踪这句话可能有点老生常谈但做 LangGraph 项目时尤其重要。多节点、多分支、状态叠加的情况下报错定位非常难。接入追踪之后你能看到每一步的输入、输出、状态快照以及路由函数走的是哪个分支。排查 6.1 节那个状态覆盖问题的时候如果没有追踪我可能还得再多熬半天。写在最后的实际操作体会如果你正打算用 LangGraph 重构手头的 Agent 项目我的建议是从一个最小的状态定义开始先跑通单一链路再逐步添加条件路由和人工介入。不要一上来就搭十个节点、五个分支的大图那种图的调试成本会成倍上升。我个人体会最深的是LangGraph 的最大价值不是把代码从链式改成图式而是它逼着你把“状态管理、流程控制、人工边界”这三件事分开来思考。状态定义清楚了Reducer 写对了条件路由和人工介入自然就水到渠成。文章里的完整案例可以直接拉下来跑建议你在跑通之后再试着改一改阈值条件、加一个新的工具节点感受一下“加节点、加边”对现有逻辑的影响这样才能真正理解图式工作流和链式调用之间的区别。