ARTICLE · INTELLIGENCE

战地情报 · 详情页

来自尧图项目组的一线实战观察与深度解析

LangGraph实战:构建企业级Agent状态机与工作流编排

LangGraph实战:构建企业级Agent状态机与工作流编排 如果你正在做 AI 应用开发尤其是 Agent 方向的开发最近一定绕不开一个名字LangGraph。从 LangChain 到 LangGraph再到各类 Agent 框架的涌现这一波技术演进的速度非常快很多开发者已经明显感觉到过去“调模型、拼 Prompt、串 API”就能交付 AI 应用的阶段正在过去真正决定应用上限的已经从模型能力转移到了流程控制、状态管理和工程化落地上。这篇文章不是简单的 “LangGraph 入门文档翻译”而是一份面向 CSDN 技术读者的实战型解析。我会从实际开发中最关心的问题出发LangGraph 到底解决了 LangChain 的什么痛点Agent、工作流、图执行这些概念的本质是什么如何从零构建一个带条件分支、循环、持久化和人工审批的真实 Agent在进入企业级生产环境时有哪些容易踩的坑和必须做的设计如果你正在准备 AI 大模型应用开发、Agent 开发相关的技术面试或者想在自己的项目中引入更可控的 Agent 编排能力这篇文章建议收藏后用。1. 先搞清楚LangGraph 到底解决了什么问题先说一个核心判断LangGraph 不是 LangChain 的简单升级版而是一次设计范式上的补充。LangChain 核心提供的是一套标准化的组件抽象模型封装、Prompt 模板、检索器、工具调用。你在 LangChain 里写 AI 应用本质上是把一段段逻辑串联成链Chain。链的问题在于它更像一条直线虽然可以做分支判断但一旦涉及循环、回退、分支合并、人工介入、状态回滚这类真实的 Agent 场景代码会迅速变得混乱且不可维护。很多人第一次接触 LangGraph 时会问我不是用 LangChain 也能写 Agent 吗确实是。你可以用代码手写一个while循环反复调用模型判断要不要继续调工具。这在 Demo 阶段没有问题但到了企业级场景你需要考虑几个实际问题会话状态如何持久化进程重启后Agent 的上下文还在吗某一步执行失败如何精确回滚或重试需要人工审核的环节如何暂停流程等待用户输入再继续执行多个分支并行执行如何合并结果并维护清晰的数据流如何记录每一步的耗时、 Token 消耗和中间输出做到可观测和审计LangGraph 给出的答案是把 Agent 的执行过程建模为一张图。每个节点是一个计算步骤每条边定义了步骤之间的流转关系再加上一个全局的 State 对象承载数据、一个 Checkpointer 负责持久化和断点恢复。这套抽象借鉴了数据流编程和图计算的思想放到 Agent 编排场景里天然地解决了循环、分支、并行和人工介入这些核心问题。如果说 LangChain 解决的是“模型和工具怎么封装”LangGraph 解决的是“整个 Agent 过程的执行逻辑怎么编排才可控”。从当下 AI 应用开发的大趋势看2026 年最值得投入的不再是简单的 Prompt 拼接而是 Agent 的工程化控制能力。LangGraph 在这一点上提供了近乎工业级的参考实现。2. 基础概念State、Node、Edge 与 CheckpointerLangGraph 的核心概念并不复杂但初次接触时容易被术语绕晕尤其是 State 和 LangChain 中的 Message 搞混。下面逐个拆解。2.1 StateAgent 的全局中央状态State 是 LangGraph 中的灵魂对象。你可以把它理解成一个随着图执行不断更新的字典里面保存了所有节点之间需要共享的数据。不同于普通函数参数State 的设计有几个关键点所有节点共享同一个 State 对象。每个节点执行完后可以返回一份更新LangGraph 会把更新合并回 State。更新方式可以自定义是覆盖字段还是往列表字段里追加元素。举个例子简单定义一个 Statefrom typing_extensions import TypedDict class AgentState(TypedDict): messages: list current_step: str retry_count: int这里messages保存对话历史current_step记录当前流程阶段retry_count用于控制循环次数。在实际执行中不同节点会修改不同字段共同维护 Agent 的完整上下文。2.2 Node每个节点就是一个执行单元Node 就是图中的节点可以是任意一个 Python 函数。函数签名是固定的接收一个 State返回 State 的部分更新或新值。例如一个简单节点def call_model(state: AgentState): messages state[messages] response llm.invoke(messages) return {messages: messages [response]}注意你没有必要把整个 State 都返回。只返回需要更新的字段即可LangGraph 会自动合并。2.3 Edge定义流转路径Edge 分为普通边和条件边。普通边表示“执行完 A 后固定执行 B”条件边则根据当前 State 的值动态决定下一步走向哪个节点。条件边是 Agent 具备决策能力的关键。graph.add_conditional_edges( analyze, route_by_intent, { answer: generate_answer, use_tool: call_tool, human: ask_human } )上面这段代码的意思是执行完analyze节点后调用route_by_intent函数根据返回的字符串选择下一步进入哪个节点。2.4 Checkpointer让 Agent 拥有记忆和执行可恢复能力Checkpointer 是 LangGraph 比较独特的机制。它会定期把 State 的完整快照保存下来。基于这个机制你可以实现时间旅行回到历史某个步骤重新执行。断点恢复Agent 执行到一半进程崩溃可以从最近检查点恢复。人工介入Agent 遇到需要人工审批的节点时暂停等待人工输入后再继续。from langgraph.checkpoint.memory import MemorySaver checkpointer MemorySaver() graph workflow.compile(checkpointercheckpointer)开发阶段用MemorySaver就足够生产环境建议使用 Postgres 或 Redis 等外部存储实现持久化检查点。2.5 和 LangChain 的关系可以这样理解LangChain 是面向模型、Prompt、工具的组件库LangGraph 是面向流程控制的编排引擎。LangGraph 本身不依赖 LangChain但二者配合效果最好用 LangChain 封装模型和工具用 LangGraph 控制流程。两者定位并不冲突有大量项目是 LangChain 负责 RAG 和工具接入LangGraph 负责整体 Agent 状态机。3. 环境准备与依赖安装本文的实操部分使用 Python建议版本为 Python 3.9 及以上。LangGraph 从 0.4 版本开始 API 已经相对稳定安装时建议直接安装较新版本确保和当前生态兼容。创建虚拟环境并安装依赖mkdir langgraph-demo cd langgraph-demo python -m venv .venv # Linux / macOS source .venv/bin/activate # Windows # .venv\Scripts\activate安装核心依赖pip install langgraph langchain langchain-openai这里我使用 OpenAI 接口作为示例你可以根据实际情况替换为 DeepSeek、通义千问、智谱等国内模型的 OpenAI 兼容接口。配置方式是通过环境变量设置 API Key 和 Base URLexport OPENAI_API_KEYyour-api-key export OPENAI_BASE_URLhttps://api.openai.com/v1使用国内兼容接口时将OPENAI_BASE_URL换成服务商对应的地址即可。所有对话模型都走 OpenAI 兼容协议LangChain 接入后不需要改业务代码。验证安装是否成功python -c from langgraph.graph import StateGraph; print(LangGraph OK)如果这一行没有报错说明环境已经就绪。4. 从零实现一个企业级客服工单 Agent理论知识容易理解真正有价值的是把各种机制组合到一个完整的场景中。下面我们用 LangGraph 实现一个接近真实业务场景的客服工单处理 Agent。这个 Agent 需要做的事接收用户提交的工单内容。分析工单并判断意图。能回答的智能问答直接生成答复。需要查询知识库或外部系统的走工具查询。识别到用户情绪强烈或问题复杂时自动转人工并等待审批。对 AI 生成的答复进行质量评估不满意时自动重写最多重写两轮。整个流程状态可持久化业务方可以随时查看进度甚至中断回滚。4.1 定义 Statefrom typing_extensions import TypedDict class TicketState(TypedDict): ticket_id: str user_message: str intent: str ai_response: str response_rating: int rewrite_count: int need_human: bool history: list4.2 定义节点函数先定义 5 个节点意图识别、知识库检索、AI 生成、质量评估、转人工。每个节点本质上是一个接收 State、返回部分更新的 Python 函数。# 文件路径demo/nodes.py def analyze_intent(state: TicketState): 识别工单意图 prompt f 用户提交了新的客服工单请你分析其意图。 工单内容{state[user_message]} 只返回以下类型之一一般咨询、退款申请、产品故障、投诉。 resp llm.invoke(prompt) intent resp.content.strip() return { intent: intent, history: state[history] [{step: analyze_intent, result: intent}] } def retrieve_knowledge(state: TicketState): 根据意图检索知识库这里用关键字模拟实际项目可替换为向量检索 knowledge_db { 退款申请: 退款流程用户可在订单页面发起退款申请3 个工作日内审核完成。, 产品故障: 请用户先尝试重启应用若问题仍存在提供日志文件以便进一步排查。, 投诉: 需要优先安抚用户情绪并记录问题细节转交高级客服处理。 } answer knowledge_db.get(state[intent], 暂时没有检索到匹配的解决方案。) return {ai_response: answer, history: state[history] [{step: retrieve, result: answer}]} def generate_answer(state: TicketState): 生成最终回复文案 prompt f 你是客服助手请根据已知信息和工单内容生成一段友好、专业、简洁的回复。 工单内容{state[user_message]} 已知解决方案{state[ai_response]} 注意控制回复字数在 100 字以内。 resp llm.invoke(prompt) return {ai_response: resp.content.strip(), history: state[history] [{step: generate, result: resp.content.strip()}]} def evaluate_response(state: TicketState): 给回复质量打分分数低于 6 分视为不合格 prompt f 请评价下面这条客服回复的完整度、语气和专业性只输出 0 到 10 的数字。 用户问题{state[user_message]} 客服回复{state[ai_response]} resp llm.invoke(prompt) try: score int(resp.content.strip()) except ValueError: score 5 return {response_rating: score, history: state[history] [{step: evaluate, result: score}]} def ask_human(state: TicketState): 转人工审批流程在这里会暂停等待人工处理 return {need_human: True, history: state[history] [{step: human_handoff, result: waiting}]}4.3 构建图和条件边上面完成了节点定义接下来把这些节点串联成一张有分支、有循环的图。# 文件路径demo/graph.py from langgraph.graph import StateGraph, START, END from langgraph.checkpoint.memory import MemorySaver # 1. 创建图实例 builder StateGraph(TicketState) # 2. 添加节点 builder.add_node(analyze, analyze_intent) builder.add_node(retrieve, retrieve_knowledge) builder.add_node(generate, generate_answer) builder.add_node(evaluate, evaluate_response) builder.add_node(human, ask_human) # 3. 设置入口 builder.add_edge(START, analyze) # 4. 根据意图决定下一步 def route_by_intent(state: TicketState): if state[intent] 投诉: return human return retrieve builder.add_conditional_edges( analyze, route_by_intent, {human: human, retrieve: retrieve} ) # 5. 工具知识库检索后生成回答 builder.add_edge(retrieve, generate) # 6. 生成回答后进入质量评估 builder.add_edge(generate, evaluate) # 7. 如果质量不合格且未超过重写次数限制重新生成 def route_after_evaluate(state: TicketState): if state[response_rating] 6 and state[rewrite_count] 2: return rewrite return end builder.add_conditional_edges( evaluate, route_after_evaluate, { rewrite: generate, end: END } ) # 8. 人工审批结束回到生成节点继续完成回复 builder.add_edge(human, generate) # 9. 编译图传入 checkpointer 以支持断点与持久化 checkpointer MemorySaver() graph builder.compile(checkpointercheckpointer)注意这里第 7 步的条件路由有个细节rewrite_count需要在某个节点递增否则会因为重写次数一直为 0 而无限循环。可以在generate_answer中增加计数def generate_answer(state: TicketState): # 原有逻辑... current_count state.get(rewrite_count, 0) return { ai_response: resp.content.strip(), rewrite_count: current_count 1, history: state[history] [{step: generate, result: resp.content.strip()}] }每次重新生成都会让rewrite_count加 1当超过 2 时route_after_evaluate会返回end流程自然终止。这里真正容易踩坑的地方是如果不修改rewrite_count条件边会一直认为“次数没有达到上限”从而反复调用生成节点造成 Token 费用失控。在真实项目中凡是有条件边的地方都要确保对应的 State 字段会在某个节点被正确更新。4.4 条件边与循环机制的原理上面的代码展示了两种典型的条件路由意图分流analyze之后根据意图去不同节点。质量驱动的循环evaluate之后决定是重新生成还是结束。LangGraph 的循环实现并没有特殊的语法它就是靠条件边指向前置节点实现的。图软件层面允许有环状态机天然支持这种回退逻辑。这也是 LangGraph 比普通 Chain 灵活很多的核心原因。你可以把 LangGraph 想象成一个带反馈回路的流程图而 LangChain 的 Chain 是一根从输入到输出的直线。前者可以处理“一次结果不满意再重来一次”这类真实业务需求后者只能在链路上做有限的前置处理。4.5 人工审批interrupt 让流程暂停等待用户输入企业级场景中Agent 全自动处理并不是常态更多是“AI 完成大部分工作关键节点交给人确认”。LangGraph 通过interrupt机制实现这种人工介入。在ask_human节点中如果希望流程真正暂停、等待人工输入而不是立即返回可以使用interruptfrom langgraph.types import interrupt def ask_human(state: TicketState): 暂停流程等待人工审核结果 human_review interrupt({ ticket_id: state[ticket_id], user_message: state[user_message], reason: 工单被判定为投诉需要人工介入处理 }) return { need_human: False, human_feedback: human_review, history: state[history] [{step: human_approved, result: human_review}] }使用interrupt后图执行到这个节点会暂停并返回一个__interrupt__对象。外部系统收到这个对象后可以在人工审核通过后通过Command恢复执行。这种模式非常适合工单审批、交易确认、知识库更新审核等场景。完整的恢复调用方式需要在图中引入一个显式状态字段from langgraph.types import Command from typing_extensions import Annotated class TicketState(TypedDict): # ...原有字段... human_feedback: str后续通过保存的 checkpoint 标识即线程 ID和 Command 恢复执行graph.invoke( Command(resume人工已确认继续生成回复), config{configurable: {thread_id: ticket-10001}} )这个步骤在企业 Agent 开发中是关键能力。没有它人工介入只能是“流程外介入”一旦人审通过Agent 无法从断点恢复整个状态机就要重建。5. 运行测试与效果验证构建完整的测试脚本验证你刚才设计的图是否能跑通。# 文件路径demo/main.py config {configurable: {thread_id: ticket-10086}} # 第一轮一般咨询工单 initial_state { ticket_id: ticket-10086, user_message: 我想咨询一下订单的发货时间为什么延迟了, intent: , ai_response: , response_rating: 0, rewrite_count: 0, need_human: False, history: [] } result graph.invoke(initial_state, config) print(最终处理结果) print(意图识别, result[intent]) print(AI 回复, result[ai_response]) print(回复质量分, result[response_rating]) print(是否转人工, result[need_human]) print(执行历史节点, [h[step] for h in result[history]])运行脚本python demo/main.py预期输出效果类似意图识别 一般咨询 AI 回复 您好非常抱歉给您带来不便。您的订单发货延迟可能与物流高峰期相关查询到最新进展后我们会第一时间联系您。 回复质量分 9 是否转人工 False 执行历史节点 [analyze_intent, retrieve, generate, evaluate]如果流程运行成功可以看到执行历史节点顺序先分析意图再检索知识库生成回复最后评估质量。评估分数合格流程正常结束。想测试循环重写功能可以在测试数据中把模型刻意调低输出质量或者在 Prompt 中要求“故意生成不完整的回复”观察rewrite_count是否递增执行历史中是否出现两次generate。想测试人工转交把工单内容改成“我要投诉你们的服务质量”预期执行历史中会包含human_handoff。6. 企业级 Agent 工程化持久化、监控与安全Demo 跑通只是第一步。从企业级角度看需要关注几个关键问题6.1 持久化检查点的实现选型前面用的是MemorySaver它把检查点保存在内存中进程重启就丢失。生产环境需要把检查点持久化到外部存储。from langgraph.checkpoint.postgres import PostgresSaver DB_URI postgresql://user:passwordlocalhost:5432/langgraph with PostgresSaver.from_conn_string(DB_URI) as checkpointer: graph builder.compile(checkpointercheckpointer)使用 PostgresSaver 后每个线程即一次会话流程的执行状态都会保存在数据库中。即使应用重启也可以根据thread_id恢复到原来的执行位置。6.2 使用 LangSmith 做全链路可观测Agent 环境和传统 Web 后端不同同一个流程中可能发生多次 LLM 调用且每次调用的输入输出都受前一步影响。排错时必须能看清每一步的输入、输出、耗时、Token 消耗。LangSmith 是目前 LangChain 生态中主流的可观测平台。在代码中简单配置export LANGCHAIN_TRACING_V2true export LANGCHAIN_API_KEYyour-langsmith-api-key export LANGCHAIN_PROJECTticket-agent配置完成后LangGraph 的每一步执行都会自动上报到 LangSmith后台可以看到完整的 Trace 链路。在一个生产级 Agent 项目中可观测能力是上线前必须解决的问题否则根本无法分析“为什么 Agent 在某个特殊输入上表现异常”。6.3 版本管理与回滚LangGraph 的图不是一个线上不可变组件。随着业务发展你会不断调整节点逻辑、修改 Prompt、替换模型。每一次调整都可能影响整个执行链路。推荐的做法是引入版本化部署机制图逻辑与配置分离Prompt 模板放在配置中心不写死在代码里。保留旧版本图实例发布新版本时旧版本保留一段时间用于快速回滚。使用线程 ID 隔离不同版本的 Agent 服务使用不同的线程 ID 前缀避免状态串扰。6.4 安全边界与敏感信息处理Agent 的核心机制是“把工具交还给模型决定使用”这本身就是安全风险。你必须明确画出一条安全边界所有涉及资金、数据删除、权限变更的操作必须经过人工审批节点Agent 没有最终执行权。对 Tool 的输入参数做白名单校验禁止传入任意代码或任意路径。State 中不要存放敏感明文信息。对外暴露 API 时必须做用户身份校验和鉴权不能让人随意传入thread_id读取他人的执行状态。在客服工单这个场景中转人工审批必须走interrupt机制确保最敏感的一步始终有人参与决策。7. 常见报错与排查方法LangGraph 使用过程中报错最多的地方集中在几个方向状态更新格式错误、条件边返回值不匹配、检查点持久化配置出错、循环次数控制失效。下面用一个常见问题排查表整理。问题现象可能原因排查方式解决方案执行时报错 “Invalid update for field”State 节点返回值与字段类型不一致或返回了未定义的字段查看错误堆栈中字段名检查 State 定义确保节点返回的键存在于 State 中且类型一致条件边报错 “No node found for value”条件函数返回了图上不存在的节点名打印条件函数的返回值将条件函数返回值与 add_conditional_edges 的映射字典对齐使用 checkpointer 后流程从旧状态恢复多个请求复用了同一个 thread_id查看每次请求的 thread_id每次新的业务请求使用新的 thread_id手动恢复中断时一直报错调用的 thread_id 不存在或该流程没有执行到 interrupt查询检查点状态确认上一次 invoke 返回了interrupt对象后再调用 Command(resume...)循环节点无限执行State 中没有维护循环计数器或计数器没有递增在历史记录中观察 generate 节点出现次数在循环路径上的节点内对计数器字段递增并设置上限Token 消耗远超预期每次模型调用都输出了超长内容循环重写逻辑触发多次查看每次调用的 Token 用量的 Trace合理设置 max_tokens优化 Prompt 输出约束在条件边上限制重写次数PostgresSaver 连接失败数据库连接串配置错误或依赖库版本不匹配打印数据库连接异常检查连接串、驱动安装和网络策略8. LangGraph 学习路径与进阶建议很多开发者问学 LangGraph 应该从哪里开始如何避免走弯路。如果从头规划一条学习路径我会建议分四个阶段推进。第一阶段理解核心抽象。用最小示例跑通 State、Node、Edge、Checkpointer 这四件事。不要急着写复杂业务逻辑先写一个只有两个节点的图手工触发几次观察 State 的变化和 Checkpoint 的持久化效果。第二阶段掌握条件边与循环。找三个典型的流程模式练手意图分流、工具调用循环、质量评估重写。这三个模式基本覆盖了日常 Agent 开发中的大部分控制流需求。第三阶段熟悉中断与人工审批。实现一个带人工确认的订单审批流程。理解interrupt的行为它如何暂停执行、如何携带数据给外部系统、如何通过Command(resume...)恢复执行。第四阶段企业级工程化。把 PostgresSaver 集成进来接入 LangSmith 追踪设计安全的 Tool 边界做版本管理和灰度发布。在这个阶段你可以去研究langgraph-platform等更上层的部署方案。这里也想给出一个判断LangGraph 的学习重点从来不是记住 Api 的调用格式而是建模能力。拿到一个业务场景能不能拆出节点、状态和条件边能不能合理设计循环和中断才是拉开初级开发者与资深 Agent 工程师差距的关键。9. 总结与最后几点提醒LangGraph 之所以在 Agent 开发领域快速流行是因为它真正解决了流程控制不确定性的问题。它给 Agent 开发带来了状态图这种工程化建模方式把持久化、断点恢复、人工介入这些企业级需求落到了标准机制中而不是靠开发者自己手写一整套状态机。如果你当前正在做一个 Agent 项目我的建议是先用 LangGraph 把完整链路跑通重点验证状态流转和分支逻辑是否正确再逐步补上持久化和可观测能力最后才考虑复杂工具链和多智能体协作。直接上多 Agent 架构很容易失败因为问题往往出在基础控制流不扎实。最后提醒一点数据安全和权限管控在任何 Agent 项目中都是最高优先级。一条不变的原则是涉及资金变动、数据删除、权限修改的操作永远把决策权保留给人类。LangGraph 的interrupt机制已经提供了足够的工具剩下的取决于你如何设计整个流程的安全边界。
RELATED READING

延伸阅读

更多一线实战笔记与深度复盘,助您持续精进