从原型到生产：用LangGraph构建可持久化、可观测的AI智能体

最近这半年，我身边陆续有好几个团队在搞AI智能体，但说句实话：真正敢把智能体放到生产环境里，让真实用户天天用的，屈指可数。大部分人停在了“Demo很惊艳，一上线就翻车”的阶段——会话状态说丢就丢，多轮流程走到一半卡死，出了问题连是哪个环节错了都查不出来。这也是我为什么最终选择了LangGraph来搭智能体。它不是那种“拿提示词包一层循环”的玩具框架，而是一套把智能体当成有状态、可持久化、可回放的有向图来构建的完整方案。这篇文章想从零开始，把我从一个最小的StateGraph跑到上生产这一路上的思路、代码和坑都整理出来。适合已经写过Agent原型、现在想认真做生产就绪系统的读者，也适合刚接触LangGraph、希望直接跑通一个可用项目的人。

1. 为什么我最终选了LangGraph而不是那些“一条提示词”的方案

1.1 大多数Agent框架的死穴：状态管理缺失

如果你搭过基于“大模型+工具调用”的原型，一定很熟悉这套标准写法：系统提示词里塞一份工具清单，模型说“我要调用工具A”，代码执行工具，把结果拼回去，再让模型继续。这个循环在演示阶段非常好用，漂亮得不像话，我见过不少团队拿这种原型去做内部汇报，效果都非常好。可一旦进入生产，问题会接二连三地冒出来：用户聊到一半断线了，重新连接之后上下文在哪里？服务重启一次，所有会话状态直接清空？流程里某一步算错了，想退回到那一步重跑，它根本没有“上一步”的概念。

这些问题的根源其实是同一个：大多数框架把“智能”押在模型的推理能力上，却忽略了系统本身必须具备的状态管理能力。状态是什么？说白了就是“当前这个Agent走到了哪一步、已经收集了哪些信息、各节点之间需要共享哪些数据”。没有这层抽象，你就只能靠全局变量、回调函数、外部数据库手工去拼。Demo阶段还能撑住，一旦并发上来、多轮交互变复杂，代码就会失控成一团乱麻。

1.2 一个图走到底：State、Node、Edge、Conditional Edge

LangGraph解决这个问题的方式，是把整个智能体定义成一张有向图。核心抽象其实只有四个：State、Node、Edge、Conditional Edge。

我习惯用一个流水线类比：状态是工件，节点是工位，普通边是传送带，条件边是装有传感器的分流器。工件走到分流器前，传感器判断它的属性，决定它去哪个工位。这套模型最值钱的地方是，每一段都在明面上：工件在哪个工位、被改造成了什么、下一站去哪，全部可以被记录和回溯。生产系统需要的正是这种可解释性，而不是一个无法拆开的“黑盒循环”。

1.3 生产环境真正值得信任的不是“智能”，而是“可控”

这里我想说一个可能反直觉的观点：生产环境里的智能体，核心价值不在于“它有多聪明”，而在于“它有多可控”。幻觉要靠提示词和工具约束去压，出错之后要能快速定位，流程被打断要能恢复，用户反馈不佳要能回放当时的上下文来排查。

LangGraph能让我在项目里做到这几点，是因为图的每个节点天然就是一个可观测、可恢复的单元。即便模型某一次判断是错的，只要路由逻辑和节点代码没问题，我们就有办法从错误路径上拉回来，而不是让整个对话螺旋飞升。这也是我把生产方案从“Agent框架A”“Agent框架B”换成LangGraph的真正原因：它给的不是便利，是结构。

2. 先把最小可运行的骨架搭起来：环境和第一个StateGraph

2.1 环境要求与安装

如果你用Python，建议3.9以上。实际生产我一般用3.11，类型标注和异步支持都更顺手。安装只需要两个核心包：langgraph负责图运行时，langchain-core提供消息结构和工具装饰器。如果你要接OpenAI、通义等模型，还需要对应模型厂商的SDK包装，但这不是LangGraph本身的核心依赖。

提示：安装时不要一股脑把langchain全家桶全装进来。很多组件这年头已经拆分了，按需装反而清爽，避免依赖冲突。

我第一次搭的时候图省事直接装了完整包，结果和已有的HTTP客户端版本冲突，排查了半天。后来学乖了，按最小依赖来，用到一个装一个，环境干净很多。

2.2 第一个StateGraph：从零跑通

不急着上模型，先用一个最简单的图跑通流程。这里我们定义状态为两个字段，节点分别给状态追加内容，然后把节点连起来。

from typing import TypedDict from langgraph.graph import StateGraph, START, END class DemoState(TypedDict): messages: list[str] step: int def node_a(state: DemoState): return { "messages": state["messages"] + ["A处理完毕"], "step": state["step"] + 1, } def node_b(state: DemoState): return { "messages": state["messages"] + ["B处理完毕"], "step": state["step"] + 1, } graph = StateGraph(DemoState) graph.add_node("node_a", node_a) graph.add_node("node_b", node_b) graph.add_edge(START, "node_a") graph.add_edge("node_a", "node_b") graph.add_edge("node_b", END) app = graph.compile() result = app.invoke({"messages": [], "step": 0}) print(result)

跑完的结果是：

{'messages': ['A处理完毕', 'B处理完毕'], 'step': 2}

这里注意一个和普通模型调用完全不同的点：invoke返回的不是模型的输出，而是整个执行结束后的完整状态。这意味着你拿到的不只是“结果”，还有“过程”。哪怕中途某一步要调试，也能从返回状态里看到每一步遗留的数据。这一点在后面做生产排障时特别重要。

2.3 加入大模型与工具调用：第一个Agent雏形

有了图骨架，下一步把大模型加进来。最典型的Agent循环是：模型决定是“调用工具”还是“直接回复”，如果调用工具，就进入工具节点，工具执行完把结果写回状态，再让模型继续判断。

from langchain_core.tools import tool from langgraph.prebuilt import ToolNode, tools_condition @tool def get_order_status(order_id: str) -> str: """查询订单当前状态。""" return f"订单{order_id}状态：已发货，预计三天内送达。" # 假定你已经有 llm = ...，并且把 llm.bind_tools([get_order_status]) 的返回作为 agent 节点 def agent_node(state): response = llm_with_tools.invoke(state["messages"]) return {"messages": [response]} graph = StateGraph(AgentState) graph.add_node("agent", agent_node) graph.add_node("tools", ToolNode([get_order_status])) graph.add_edge(START, "agent") # 条件边：如果模型输出 tool_calls 就进 tools，否则直接走 END graph.add_conditional_edges("agent", tools_condition) graph.add_edge("tools", "agent") app = graph.compile()

到这步，一个“能跑”的Agent就出来了。我实测过，从零写到能跑通，熟练的话十几分钟，不熟练也不超过一小时。但这只是开始，真正麻烦的是后面那些“生产才考虑”的问题。

3. 状态、Reducer与条件路由：让Agent从“死流程”变成“活循环”

3.1 State字段的更新策略与Reducer

很多人第一次写LangGraph时会把State当成普通字典，觉得节点返回什么，状态就替换成什么。实际上，State里的每个字段都需要定义一个“更新策略”：是直接覆盖，还是追加合并。

默认情况下确实是覆盖。但诸如messages这类需要累积的字段，我们想要的是“把新消息追加到历史消息后面”，而不是“用新消息覆盖历史”。LangGraph用Reducer机制来做这件事。Annotated类型标注就是用来指定Reducer的：

from typing import Annotated from langgraph.graph.message import add_messages class AgentState(TypedDict): messages: Annotated[list, add_messages] order_id: str

这里的add_messages是LangGraph内置的Reducer，专门处理消息列表的合并。节点只要返回{"messages": [新消息]}，新消息就会被追加到历史后面。没有这个机制，你每个节点都得自己写“取历史-拼列表-返回完整列表”的逻辑，代码会变得又臭又长，而且很容易漏。

3.2 条件边：真正让Agent“活”过来的核心

先思考一个问题：如果图只能按固定顺序走，那不就是一个升级版流水线吗？它凭什么处理Agent这种需要不断决策的场景？

答案是条件边带来的“回路”。Agent的决策逻辑是这样的：模型先看当前状态，如果认为需要查数据，就输出工具调用指令；如果认为信息够了，就输出最终答案。图拿到模型输出之后，必须根据输出内容选择下一站——去工具节点，还是直接终止。

这就是tools_condition做的事：它检查模型输出里有没有工具调用指令，有就返回“tools”这个节点，没有就返回END。因为工具节点执行完后还会连回Agent节点，整张图就形成了“决策—执行—再决策”的循环，直到模型决定结束。

如果你有自己的路由需求，条件边的函数就是一个普通的普通Python函数：

def route_by_intent(state): intent = state.get("intent", "unknown") if intent == "refund": return "refund_node" if intent == "exchange": return "exchange_node" return "fallback_node" graph.add_conditional_edges("parse_node", route_by_intent)

这个函数返回值就是下一个要进入的节点名。你可以把任意复杂的路由判断塞进去，只要返回的节点名是已经在图上注册过的，并且最终能到达END，就不会出问题。

3.3 节点返回值与异步执行：生产环境的第一个分水岭

节点函数既可以是同步的，也可以是async def。生产环境里，节点要调用外部API、查数据库、调模型，这些都是IO密集操作，我优先用异步版本。LangGraph的invoke是同步入口，ainvoke是异步入口，内部节点只要返回可等待对象，运行时就能自动调度。

节点返回值不一定要覆盖整个State，它可以只返回部分字段的新值，甚至返回空字典“什么都不改”。这种“增量更新”设计对并发比较友好。不过要记住：不要在节点内部偷偷修改传入的State对象本身，而是始终返回一个新的字典。LangGraph的状态管理默认是“不可变”的，绕过这套约定去直接改对象，短期跑不出毛病，但一旦上了并发和持久化，你会付出惨痛代价。

4. 生产就绪的四个必备能力：持久化、断点恢复、并发与可观测性

4.1 Checkpointer：把状态从“内存变量”升级成“录像带”

在本地开发时，State存在内存里，一切安好。但生产环境里的用户可不会保证“一次把流程走完”。用户可能聊到一半关掉页面，第二天再回来继续；服务也可能因为发版、容量调整而重启。如果状态只躺在内存里，这些场景全部完蛋。

LangGraph解决这个问题的方式叫Checkpointer。它的思路是在每一步节点执行结束后，把当前状态快照保存到存储后端。跑一个节点存一次，流程什么样、用户进度在哪，全都能恢复。

from langgraph.checkpoint.memory import MemorySaver checkpointer = MemorySaver() app = graph.compile(checkpointer=checkpointer) # 每次调用传入 thread_id，同一会话共用一个线程 config = {"configurable": {"thread_id": "user-1001"}} app.invoke({"messages": [{"role": "user", "content": "我的订单发货了吗？"}]}, config=config)

MemorySaver只是内存实现，进程重启就没了，适合开发调试。生产环境建议换成数据库实现，把快照落到Postgres这类持久化存储里，服务重启、实例漂移都不影响会话状态。

4.2 断点恢复与时间旅行：排障时的“回放功能”

Checkpointer带来的另一个福利是时间旅行。因为每一步快照都在，我们可以把状态恢复到任意一个历史节点，然后改掉某个节点的输出，重新执行后面的流程。打个比方，普通系统只能看最终日志，有了时间旅行，你等于有了录像带，可以拖回任何一帧。

我在实际生产里遇到过这种场景：某用户在一次售后流程的第二步被模型误判了意图，结果整条流程走偏。传统做法是让用户重新开始，体验极差。用LangGraph的话，我可以读取那个会话的历史状态，定位到parse_intent节点，修正意图字段，然后从那个节点重新执行后续流程。用户甚至感知不到出过错。

实现上用get_state读历史状态，用update_state把修正后的状态写回去，之后再invoke继续跑就行。这几个API起初可能觉得陌生，但花半小时掌握，换来的排障效率远超投入。

4.3 并发隔离与流式输出

并发方面，LangGraph用thread_id做会话隔离，不同用户不同线程，状态天然不互相干扰。但请记住：同一个thread_id下，不要同时发起多个并发更新。这就像两个人同时编辑一份文档，必然产生冲突。如果业务上确实需要同一会话内的并发，你得自己加锁或者改造状态结构，这是经验之谈。

流式输出也是生产体验的重要一环。用户等模型思考几秒和看到逐字输出，心理体验天差地别。LangGraph支持stream和astream两种接口，既能拿到节点级别的完整事件，也能拿到模型输出级别的token流。长耗时任务建议一律用流式，第一个token出来之后，用户焦虑感会直线下降。

4.4 结构化日志与可观测性：给每个节点装一个仪表盘

生产环境最怕的不是报错，而是“不知道错在哪”。模型的输出是概率性的，同样的输入可能产生不同结果，如果没有足够的过程记录，排查一个线上问题可能要翻遍所有模型调用日志和用户原始消息。

LangGraph的节点天然就是结构化日志单元。我给每个节点加了一段统一回调：记录节点名、耗时、状态体积、模型输入输出长度。这样一张图跑完，就能梳理出一条完整的执行轨迹，配合外部可观测平台，可以直观看到每个环节的耗时和token消耗。记住，可观测性不是上线之后再加的，而是在设计节点函数的时候就要留好埋点位置。

5. 实战案例：一个订单售后工单智能体从设计到落地的全过程

5.1 需求与图设计

拿我们项目组最近做的一个订单售后智能体来举例。用户在订单页发起售后，可能诉求是退款、换货、退货，也可能只是查进度。Agent要做的是：理解诉求、查订单状态、判断是否符合售后条件、执行操作，如果涉及退款和人工复核，就挂起等人工审批。

这个需求映射成图，大概分五块：意图识别、订单查询、动作执行、人工审批、最终回复。状态字段除了消息列表，还需要存intent、order_id、approval_status这些业务字段。

class OrderSupportState(TypedDict): messages: Annotated[list, add_messages] intent: str order_id: str approval_status: str final_response: str

图的边不再是简单的链式结构，而是带条件路由：意图识别节点之后，根据intent字段分别进入不同处理分支；动作执行后是否需要人工审批，也要靠条件边判断。

5.2 关键节点实现

意图识别节点复用大模型，只是让模型输出结构化意图：

def parse_intent(state): prompt = f"""请判断用户售后诉求类型，只返回以下四类之一： refund / exchange / return / inquiry 用户消息：{state['messages'][-1].content}""" intent = llm.invoke(prompt).content.strip().lower() return {"intent": intent}

订单查询节点直接调用工具拉取订单数据，并写入状态：

@tool def fetch_order(order_id: str) -> str: return f"订单状态: 已签收, 签收时间: 三天前, 是否在售后期内: 是" def fetch_order_node(state): result = fetch_order.invoke({"order_id": state["order_id"]}) return {"messages": [{"role": "tool", "content": result}]}

这里我把order_id设定成用户在对话里显式提供的，意图识别节点可以顺带抽取，也可以单独用一个实体抽取小模型做。实际项目中，抽取订单号这一步比想象中坑多，用户可能说“单号是20240915”，也可能说“尾号8848的那个”，你得自己定规则。

5.3 人工审批与异常兜底

退款这类操作，我的设计是必须经过人工审批。LangGraph里可以用中断机制来暂停图的执行，等人工决策完成后，再恢复运行。业务上相当于把Agent变成一个半自动系统：模型判断、工具执行、人工确认、最后落地。

def human_review_node(state): # 这里发起中断，等待外部系统（如审批后台）反馈 decision = interrupt({"order_id": state["order_id"]}) return {"approval_status": "approved" if decision["approved"] else "rejected"}

外部审批通过后，调用Command(resume=...)把决策传回图，Agent继续往下走。这套机制比“回调+外部状态表”优雅得多，整个流程生命周期都在图里，日志完整。

异常兜底也很关键。意图识别模型给出的分类可能超出四类，动作节点可能查不到订单数据，这些场景都要落到专门的fallback_node，输出兜底文案并记录一条日志，而不是让流程死循环。兜底节点我建议永远不要让它返回“重新开始”，用户最讨厌的就是“抱歉，请重新描述”，正确的做法是保留上下文，引导用户补充关键信息。

6. 上线之前必须踩过的坑：我的教训清单

6.1 上下文无限增长：几轮就爆Token

第一个坑，也是几乎每个Agent都会踩的坑：上下文无限增长。在Agent循环里，每一轮模型输出、工具返回都会往messages里追加，几个来回之后，上下文体积就指数膨胀。我的第一次线上事故就是这个原因——某用户连续调用了十几次工具，最后一次请求直接把上下文撑爆，返回了400错误。

解决思路是“控制进模型的消息量”。不要把全部历史都丢给模型，可以做消息裁剪，只保留最近的N条，或者把工具返回的大段内容做摘要后再放回上下文。记住一个原则：状态字段可以完整保留，但送入模型的内容必须精简。状态和上下文不是一回事。

6.2 工具调用的超时与重试：外部服务永远不会“可靠”

工具调用天然依赖外部系统，而外部系统永远不会100%可靠。我见过一个线上事故：Agent调用某内部订单接口，接口超时，模型等不到反馈，直接给用户编了一个订单状态。这提醒了我：所有工具调用必须有超时保护和重试策略。超时后宁可让Agent告诉用户“系统繁忙，请稍后再试”，也不能让它凭空发挥。

给工具加超时最简单的做法是在调用外层包一层超时控制，或者用框架提供的重试策略参数。我这里额外建议：工具返回必须带一个错误标记，比如{"success": false, "error": "timeout"}，这样Agent节点能根据标记走兜底逻辑，而不是把错误文本当作正常结果。

6.3 状态字段滥用：什么东西都往State里塞，迟早出事

State字段是全局共享的，每更新一次，Checkpointer就会快照一次。如果你把大体积数据往State里塞——比如一个大文件解析结果、一个几十MB的图片Base64——那每一次快照都是一次巨大的IO开销，几轮下来性能直接崩掉。

我现在的经验是：State里只放两种东西，一种是流程控制需要的元数据（意图、状态标记、ID），另一种是模型推理需要的小体积内容。大payload一律存外部存储，State里只留一个引用ID。等你看到快照存储的磁盘占用呈线性暴涨时，就会明白这条建议有多重要。

6.4 本地同步与生产异步的执行差异

最后一个坑是关于并发执行环境的差异。本地开发时，用同步函数按顺序一条条跑，什么问题都没有。上生产换异步接口后，节点并发执行，凡是用了全局可变对象、线程不安全的客户端的代码，都会开始间歇性出问题。

我自己踩过一次：在节点里复用了某个线程不安全的内部客户端连接，生产一并发就偶发报错，本地跑一百遍都复现不了。排查了两天才定位到连接复用问题。现在我把所有外部调用的客户端都改成连接池模式，并在节点函数里坚持无状态化：不修改全局对象，不偷偷共享连接，所有依赖都通过参数传入。

结尾

从最初“模型跑通就行”的想法，到后来“每个节点都可观测、可恢复、可回放”的体系，我大概用了两个项目周期才彻底转变过来。如果你也准备把Agent从原型推向生产，我个人的建议是：先用图把你现有的Agent流程画出来，哪怕很粗糙，再逐步加入Checkpointer、条件路由和兜底节点，不要一上来就追求大而全的生产架构。等你在一次真实的线上故障里，靠状态回放定位到了问题根源，你就会明白这套设计带给你的安全感。收藏这篇文章是好事，但更建议找个小项目动手跑一遍——很多坑，亲手踩过才算真掌握。