只要用 LangChain 写过几个 Demo就一定体会过那种“自由过了火”的感觉链也好、代理也好跑通简单一上生产就抓瞎。状态乱成一团、分支逻辑写死在代码里、中间步骤想断点续跑完全没门。我大概在半年前开始把 LangGraph 用到正式项目里一开始也觉得“不就是图编排嘛有什么了不起”直到真正用它重构了一个积重难返的工作流系统才发现这玩意儿的设计哲学确实不一样。这篇东西就是我个人从 0 到 1 搭建生产级 LangGraph 智能体工作流的完整记录。会讲清楚我为什么弃用老方案、核心概念到底怎么理解才算通、代码层面怎么从零开始搭一个带人工审批的工单处理智能体再聊聊生产环境里那些躲不开的持久化、可观测性、测试和部署问题。适合已经跑通过 LangChain 基础、但想更进一步把智能体做成工程的开发者也适合正在选型时纠结“到底用框架还是自己写状态机”的团队技术负责人。1. 为什么最终选了 LangGraph先解决“状态管理”这个要命的问题先说结论LangGraph 本质上是一个面向 LLM 应用的可编程状态机。它把你脑子里的流程图变成了机器能跑、能暂停、能恢复、能观测的代码。很多人在第一印象里会觉得 LangGraph 只是 LangChain 出的又一个编排工具跟 LangChain 自己的 Agent 或者外面的 AutoGPT 没本质区别。这个理解偏差挺大的。我做过一个对比测试用传统 LangChain Agent 和 LangGraph 分别实现同一个“数据分析邮件发送”任务。前者的代码确实更短但一旦涉及条件回退、人工确认、结果不一致时重跑某个子流程代码就开始变成一团糨糊。原因在于传统 Agent 把所有逻辑都放在“循环调用工具直到认为完成”这个隐式循环里开发者对中间状态基本失控。LangGraph 换个思路把整个流程显式建模成一个图。图里有节点要做的事和有向边下一步去哪。节点之间传递的是一个全局状态对象这个对象被 schema 严格定义。任何时刻这个图的状态都是可序列化、可检查、可恢复的。这就让智能体从“一个谁都没法完全解释的黑盒”变成了“一个条理清晰的业务系统”。用生活化的比喻传统 Agent 像一个能力很强但想到哪干到哪的自由职业者LangGraph 则像一套成熟的作业流水线每道工序有明确的输入输出质检不合格要回退到上一站整个车间的监控室能看到每个工位的实时状态。还有一个让我下定决心替换旧方案的原因是“人类介入”——这是生产级 AI 应用几乎必然遇到的需求。不管你的模型有多强涉及资金操作、对外发送内容、删除数据这种动作业务方都会要求“机器先出方案人审批后执行”。LangGraph 对这种模式的实现自然得可怕只是在图上加一个 interrupt 节点运行到那里就自动停下来等人类指令。这一点吸引了我因为它把 AI 从“只能做全自动玩具”升级到了“能进正规业务流程干活”的层次。2. 核心概念逐一说透图、状态、节点、边和检查点这个章节是给想真正用起来、而不是停留在“看教程觉得很酷”的读者写的。LangGraph 的官方文档把概念写得比较分散我按自己踩坑之后的认知顺序重新捋一遍。2.1 状态State是整个图的灵魂在 LangGraph 里节点之间共享的唯一数据载体就是 State。它通常是一个 TypedDict 或 Pydantic 模型例如from typing_extensions import TypedDict class AgentState(TypedDict): messages: list # 对话/思考历史 current_step: str # 当前执行到哪一步 report: str # 最终输出 requires_review: bool # 是否需要人工审批节点函数接收的是这个 State 的“当前快照”返回的是一个“部分更新”字典。这个设计初看简单实际影响深远。因为所有节点只依赖入参、只产出更新你就把整个工作流的每一步变成了可单测的纯函数。我们团队现在写新流程的第一件事不是画流程图而是先定义最完整的 State 字段。字段定义清楚了整个系统的一半边界也就清楚了。更要留意的是 State 的更新机制有两个模式覆盖默认和归约。比如 messages 这种要不断累积的字段你应该用add_messages这类 reducer 去合成否则新消息会直接把历史顶掉。我见过不少初学者在这里懵了“我明明往 messages 里加了东西怎么每次节点进来都是空的”答案就在 reducer 没配。2.2 节点Node和边Edge用代码把业务画出来节点就是普通的 Python 函数签名长得一样def generate_report(state: AgentState) - dict: ... return {report: report_text, requires_review: True}只要返回的 dict 里的 key 符合 State 定义LangGraph 就会自动用它们更新全局状态。写节点函数时有一条我特别想强调的军规不要在里面偷偷维护任何函数外的模块级变量。因为图是可暂停可恢复的内部节点可能在某个时间点被重复执行、被序列化任何“藏在节点外的记忆”都会成为事故源。状态都要走 State 对象进出。边分两种普通边就是“走完 A 一定去 B”条件边则是一个注册函数根据当前 State 判断下一步到底去哪个节点。逻辑很像写if/else但它把这个决策变成了图中一等公民非常适合那些“先判断门槛条件再决定走哪条处理链”的流程。2.3 检查点器CheckpointerLangGraph 真正的杀手锏如果说图结构是 LangGraph 的骨架那检查点器就是它的神经系统。它让图在每一步执行后把完整状态快照保存到存储器里从而获得两项核心能力时间旅行回到任意历史节点重新跑和跨线程恢复下次带着同一线程 id 来可以从上次暂停的位置接着走。这个能力在开发期极具幸福感。以前用传统 Agent 调试时一旦某一步 LLM 输出异常基本只能加日志重跑整个流程不仅慢而且不便宜。有了检查点器我可以直接graph.get_state(config)查看出问题那一刻的完整状态甚至可以graph.update_state(config, {...})手动改字段后再从那个节点继续执行。我在开发阶段几乎天天用这一招来复现线上问题效率比传统方式高出不止一截。生产环境里检查点器也天然实现了“长时对话记忆断点续跑”。比如用户做多轮资料填写填到一半关掉页面下次打开带着同一个 thread_id系统能准确从上次中断的节点恢复这体验已经是“带持久化的业务系统”级别了。3. 从 0 到 1 实战搭建带人工审批的工单处理智能体光讲概念没意义直接拿我最近做的“智能工单处理系统”作为例子完整展示从零开始怎么在 LangGraph 里落地一套生产级流程。3.1 业务需求拆解业务背景大概是这样的客服团队每天收到大量工单希望 AI 先自动完成分类、情绪识别、知识库检索和答复草稿生成但所有要发给客户的答复必须经过人工审批才能发送。如果 AI 信心不足或者客户情绪激烈则直接升级给高级客服专员。整个流程还需要全程留痕要能查“当时 AI 为什么给了这个结论”。拆解下来核心节点有五个classify_ticket判断工单类型产生结构化标签escalation_check根据情绪强度、是否 VIP、是否投诉升级等条件决定是否走人工特批search_articles在知识库检索相关内容draft_response基于检索结果生成回复草稿human_review挂起等待人工审批后决定发送、修改还是重新生成3.2 搭图与条件分支实现定义一个完整的状态模型是第一步from typing_extensions import TypedDict class TicketState(TypedDict): ticket_text: str category: str sentiment: float confidence: float matched_articles: list draft: str review_status: str # pending / approved / rejected接着实现节点函数。这里为了压缩篇幅展示核心结构节点间逻辑做了一定合并。真实项目中建议一个文件只放一个节点的实现和测试别把节点函数全部堆在同一个文件里后面维护会很痛苦。def classify_ticket(state: TicketState) - dict: # 这里通常要走 LLM 调用或者你自己训练的文本分类模型 ... return {category: billing, sentiment: 0.2, confidence: 0.94} def escalation_check(state: TicketState) - dict: # 判定是否升级低置信度或负面情绪过高 ... return {needs_escalation: True} def search_articles(state: TicketState) - dict: ... return {matched_articles: [...]} def draft_response(state: TicketState) - dict: ... return {draft: 您好关于账单扣费问题...}核心的图编排如下。这里用了两个条件边一个是是否升级一个是人工审批评判结果from langgraph.graph import StateGraph, END from langgraph.checkpoint.memory import MemorySaver builder StateGraph(TicketState) builder.add_node(classify, classify_ticket) builder.add_node(escalate, escalation_check) builder.add_node(search, search_articles) builder.add_node(draft, draft_response) builder.set_entry_point(classify) builder.add_edge(classify, escalate) # 条件边 1根据是否升级决定去向 builder.add_conditional_edges( escalate, lambda state: human_specialist if state[needs_escalation] else search, {human_specialist: human_specialist, search: search} ) builder.add_edge(search, draft) builder.add_edge(draft, human_review)这里只是部分代码完整图还应该包含人工审批节点human_review以及后续的发送节点send_response。审批结果如果是不通过可以用条件边让图回到draft重新生成。这种“重新生成”的场景在普通流程框架里非常痛苦但在这里无非是多加一条边。3.3 人工审批中断的实现细节人工审批我单独拿出来讲因为这是理解 LangGraph 能力的绝佳样本。核心用到了interrupt函数from langgraph.types import interrupt def human_review(state: TicketState) - dict: action interrupt({ draft: state[draft], question: 是否批准发送该答复, }) # action 是外部传入的 {decision: approve} 或 {decision: reject} ... return {review_status: action[decision]}图执行到interrupt()时会抛出特殊异常并向上传播LangGraph 自动把运行状态封存进检查点。外部系统想审阅时从 API 拉取当前状态审批人点击按钮后再通过graph.invoke(None, config, config{thread_id: ...})的方式把决策喂回图图会从上次中断的地方继续执行。这里有一个必须注意的 API 用法细节你传给invoke的第一个位置参数带了值是作为新的输入从头开始跑传None时LangGraph 才理解为“从上次中断处恢复”。我第一次用的时候没看文档直接又传了一遍工单文本结果图从头重跑了一遍工单被重复处理。这个坑说小不小生产环境里会造成严重的重复操作事故。同时在写这个节点时要预设一个 timeout 机制如果人工超过 24 小时未审批由外部定时任务自动强制回退或转交。可以给图配置不同的recursion_limit或者在节点内用外部存储记录超时时间两条路都可行根据自己的运维复杂度和团队习惯来选。4. 生产级改造持久化、可观测性与测试Demo 能跑和系统能上线中间隔着一条大江。这章全是我在生产环境里踩出来的硬经验。4.1 选对检查点器后端别换了个系统官方默认的MemorySaver只配本地调试进程一重启所有状态灰飞烟灭。生产环境我直接换成基于数据库的检查点器。官方提供 Postgres 和 Redis 两种方案我推荐优先看 Postgres 方案毕竟很多团队本来就有 Postgres 基础设施。检查点数据本质上是一张存储状态快照的表使用数据库方案后可以用 SQL 直接查状态历史排查问题时有时比日志还直观。配置方式大致是from langgraph.checkpoint.postgres import PostgresSaver checkpointer PostgresSaver.from_conn_string(postgresql://user:passhost/db) # 首次使用记得初始化表 checkpointer.setup()而且检查点信息里天然包含了图运行的每一步耗时、token 使用量、LLM 请求参数等执行元数据如果你有对应的回调插件。这意味你可以直接跑报表分析每个节点的平均耗时快速定位系统里最慢的环节。老方案里想拿到这种全链路数据需要自己在每个环节手动加埋点维护成本不可同日而语。4.2 可观测性必须从第一天就做别等线上出事LangGraph 生态里现成的观测工具是 LangSmith。它能自动追踪每次图的完整运行过程包括每一步的输入输出、模型的 prompt、token 消耗和延迟。接入方式简单到令人发指就是配两个环境变量和一个 API Keyexport LANGCHAIN_TRACING_V2true export LANGCHAIN_API_KEYyour_key我见过有人说“我们团队不用 LangSmith因为不想依赖第三方平台”这个观点我以前也有。但后来想通了你自己拿日志系统能做到按图节点和线程去全链路检索吗能做到跨时间对比同一节点的输入输出吗能。但成本至少是一两周的开发工作量。对于一个两周就能上线的项目花两周做日志系统属实不划算。如果确实对数据合规有严格要求那就要评估自己团队是否有能力在 ClickHouse 或 Elasticsearch 上实现同样的链路追踪能力。我的经验是开发阶段就开 LangSmith每周翻了翻 trace往往能在用户反馈之前发现明显异常。4.3 给 LLM 工作流做测试和传统 CRUD 完全不同给图做自动化测试时最大的变化在于输出是非确定性的。直接断言“输出必须包含某某字符串”基本必挂。我们的策略是用快照测试加语义相似度双重验证。每个节点函数先用固定 prompt 快照跑一遍看结构是否稳定再用 embedding 模型算输出与期望答案的余弦相似度设定一个阈值来判断“语义对不对”。对于条件边不测最终输出而是测条件函数本身。把预构造好的状态喂给条件函数断言返回的目标节点名是否正确。由于条件函数是普通 Python 函数这个测试极其稳定。我还强烈建议为图的每个节点设计“状态不变量”。比如“任何节点执行完state 里必须存在一条 messages 字段的最后一条 assistant 消息”。有了这种不变量结合图中天生带有的检查点你可以在每次运行后自动校验状态合法性。给 LLM 应用写测试逻辑一定要放在“状态转换是否合法”而不是“文本是否精确匹配”上。4.4 部署形态别把图塞进 Web 服务器里当普通函数调LangGraph 官方提供了一个部署工具langgraph dev和对应的运行时服务。但在容器化普及的今天我更建议直接把图封装成一个独立服务用 FastAPI 或者 Cloud Run 一类的计算服务包一层。关键在于遵循一个原则长时间运行的工作流不要放在 HTTP 请求线程里同步等结果。我们现在的做法是HTTP 接口收到请求后立刻把图执行丢进后台任务队列接口只返回一个线程 ID前端轮询或 WebSocket 推送获取执行结果。因为对人工审批这种场景一次运行可能持续好几个小时HTTP 连接早就断了。用异步任务队列比如 Celery 或者简单的 Redis 队列配合 Postgres 检查点器整个系统非常稳定。进程重启、机器崩溃都不会丢状态因为检查点都在数据库里。5. 踩坑实录与经验速查最后这部分是纯经验输出很多是我付出了真实代价才换来的教训。5.1 节点函数必须幂等别在函数里做自增LangGraph 在重试、恢复执行时节点函数可能会被执行多次。最经典的坑就是我在一个节点函数里写了“写入工单日志日志序号自增”结果由于执行超时自动重试日志被写了两遍。这很隐蔽因为不是所有重试都会触发日志重复问题排查起来极其恶心。解决方式是把压力放到下游去重或者利用检查点器自带的状态版本号做幂等控制。一句话节点函数内永远不要做有副作用的操作要做就做成可重入的。5.2 Checkpointer 对 State 序列化的要求极其严格State 里的所有值必须能被检查点库正确序列化。我这里遇到过自定义业务对象塞进 State 导致恢复时报错的情况。后来统一改成只存可序列化类型字符串、数字、列表、字典必要时把自定义对象转成 JSON 字符串存进 State。自定义的复杂对象如果必须传递就单独放到外部存储State 只存引用 ID。这个原则极大减少了生产事故。5.3 大上下文爆炸问题工作流跑长任务时messages 会越攒越长有一天你突然发现 LLM 的 context window 扛不住了。解决思路有两个一是消息裁剪给 State 配一个“保留最近 N 轮 所有摘要”的机制二是需要总结时单独加一个 summarize 节点定期把历史消息压缩成摘要。不要等线上报警再想这个事情长周期工作流里这是必然事件不是偶然事件。5.4 测试环境里记得跑一次完整的人工审批闭环我们上线前最大的教训是所有自动化测试都通过了但忘了手动测试人工审批那条链路。结果部署后发现审批系统回传决策绑错了 thread_config审批人员点“通过”后系统从错误的状态恢复生成了完全错误的答复。这种集成类问题只有靠完整的端到端演练才能发现。建议上线新流程之前建一条测试专用工单完整走一遍“提交—自动处理—人工审批—发送”的整个闭环所有审批操作也真点一遍按钮。写在最后我做这个项目前后有两个感受转变。一开始觉得 LangGraph 无非就是把 LangChain 的链换成图多此一举。真正用起来以后发现它的价值从来不在“图”这个形式本身而在于围绕图建立的一整套工程体系状态可持久化、流程可中断恢复、逻辑可测试、历史可追溯。这一套组合下来AI 工作流才真正从“模型调用的拼装玩具”变成了“能和现有业务系统对接的正式模块”。我现在的建议是如果你新项目里已经确定要用智能体处理多步骤、有条件分支、且存在人工介入的场景可以直接从 LangGraph 起步。如果你还在维护老的自由式 Agent 代码也不一定急着推翻重写可以挑一个低风险场景用 LangGraph 做一次重构让团队完整走一遍状态设计、检查点调试和生产部署的流程之后你会对这套模式有非常具象的体感。状态设计是一切的地基花时间把 State 定义想清楚后面所有节点的实现都会顺很多。