LangGraph 工作流：从概念到可交付结果

这篇我按“先跑起来、再讲取舍”的方式写《LangGraph 工作流从概念到可交付结果》。概念会讲但重点放在代码怎么组织、哪里容易踩坑。摘要本文概述文章目标、核心观点和实践价值。上周生产环境出了个典型的 Agent 事故一个自动处理客户投诉的 LLM Agent 在周五晚上发了两条完全矛盾的邮件给同一个用户。一条是道歉并退款另一条是拒绝退款并要求提供额外证据。事后复盘代码逻辑本身没大问题问题出在状态管理的混乱上。之前的实现里我直接用Chain串接 LLM 调用虽然简单但无法追踪中间状态更别提“后悔药”了。这次重构我决定引入 LangGraph不是为了炫技而是为了解决两个核心痛点确定性和可观测性。如果你也在做生产级 Agent别只盯着 Prompt 调优。工作流的结构设计才是决定系统能不能扛住高并发和复杂业务逻辑的关键。目录为什么需要图工作流State 与 Node定义你的业务契约Edge 与条件分支控制流的灵魂人工审批节点生产环境的救命稻草工程化落地监控与回滚总结为什么需要图工作流传统的 LangChain Chain 模式是线性的像一根绳子。一旦中间某个环节出错比如 LLM 幻觉输出了非法 JSON整条链就断了且很难回溯到具体哪一步出了问题。而在实际业务中尤其是涉及多步推理或外部 API 调用的场景我们需要的是“图”结构。图的好处在于1.显式状态管理每一步输入输出都写入 State调试时可以随时查看中间变量。2.循环与分支支持循环调用如 ReAct 模式和动态路由根据意图跳转不同节点。3.检查点机制这是 LangGraph 最强的功能。它可以保存每一步的状态快照允许我们在出错时直接回滚到上一个稳定状态甚至支持人工介入修改状态后继续执行。State 与 Node定义你的业务契约在 LangGraph 里一切皆 State。你需要定义一个 TypedDict 来描述整个工作流经过的所有数据。这不仅仅是数据结构更是你和团队成员或者未来的你自己之间的契约。比如在一个“订单异常处理”场景中我的 State 设计如下from typing import TypedDict, Annotated, List import operator class OrderState(TypedDict): # 核心业务数据 order_id: str status: str # pending, processing, completed, failed # 交互历史用于多轮对话或日志记录 messages: Annotated[List[str], operator.add] # 临时标记位用于控制流程走向 needs_human_review: bool retry_count: int max_retries: int def analyze_state(state: OrderState) - dict: 节点函数分析当前订单状态 print(fAnalyzing order {state[order_id]}...) # 模拟 LLM 判断逻辑 if state[status] failed and state[retry_count] 3: return {messages: [Order failed, retrying...], retry_count: state[retry_count] 1} elif state[status] failed: return {needs_human_review: True} return {}注意Annotated[List[str], operator.add]这个细节。它告诉 LangGraph当多个消息进入这个字段时要追加而不是覆盖。这在构建聊天机器人或多步思考过程时非常关键。很多初学者容易忽略类型的严格定义。我建议State 定义要尽量细粒度。不要把复杂的字典塞进一个 key 里拆分成独立的字段这样你在后续添加节点或修改逻辑时才不会牵一发而动全身。Edge 与条件分支控制流的灵魂Node 负责执行逻辑Edge 负责决定下一步去哪。在 LangGraph 中边可以分为普通边和条件边。普通边就是固定的 A - B。但大多数时候我们需要条件边。比如如果 LLM 判断需要人工审核就走一条路否则走自动处理的路。这里有一个坑循环依赖。如果你在条件函数里直接引用尚未更新的 State 字段可能会导致死循环或无限重试。我的做法是在条件边中使用StateGraph.add_conditional_edges传入一个路由函数。这个函数接收当前 State返回下一个节点的名称。from langgraph.graph import StateGraph, END workflow StateGraph(OrderState) # 添加节点 workflow.add_node(analyze, analyze_state) workflow.add_node(retry, retry_action) workflow.add_node(human_review, human_review_action) workflow.add_node(finish, finish_action) # 添加边 workflow.set_entry_point(analyze) # 条件边根据状态决定去向 def route_next(state: OrderState): if state.get(needs_human_review): return human_review elif state.get(retry_count, 0) 0: return retry else: return finish workflow.add_conditional_edges( analyze, route_next, { human_review: human_review, retry: retry, finish: finish } )这段代码看起来很简洁但背后是复杂的有向无环图DAG构建过程。调试的时候如果流程图不对首先检查的就是条件函数的返回值是否与边定义的 Key 完全匹配。大小写敏感千万别马虎。人工审批节点生产环境的救命稻草这是我重构中最重要的一环。纯自动化的 Agent 在金融、医疗或高价值交易场景中是不可接受的。我们需要“人机协作”的能力。LangGraph 提供了Interrupt机制或者说我们可以利用 State 中的特定字段配合外部事件来实现人工审批。在实际项目中我通常这样做1. 当流程到达human_review节点时不直接执行而是将任务挂起。2. 将当前的 State 快照保存到数据库或 Redis并生成一个唯一的thread_id。3. 向前端或管理员发送通知附带thread_id和待处理信息。4. 管理员批准后通过 API 调用graph.update_state修改 State 中的标记位如needs_human_review: False并触发图继续执行。这种做法的优势在于你可以随时“暂停”正在运行的 Agent查看它刚才做了什么思考过程是什么然后决定是否放行。这不仅是风险控制更是建立用户信任的关键。记得加上超时机制。如果人工审批超过 24 小时没响应图应该自动进入“超时降级”流程而不是无限期挂起占用资源。工程化落地监控与回滚有了图结构剩下的就是工程化细节。1. 持久化与快照不要相信内存。在GraphCompiler编译前配置checkpointer。推荐使用SqliteSaver进行本地开发生产环境换PostgresSaver。每一个 State 的变更都会被记录你可以随时查询历史轨迹。2. 监控指标在 Node 执行前后埋点。记录每个节点的耗时、Token 消耗、以及是否触发了条件分支。如果某个“重试”节点被频繁触发说明上游 LLM 的输出质量有问题需要优化 Prompt 或引入校验层。3. 版本控制LangGraph 的图定义代码应该像微服务一样进行版本管理。当你的业务逻辑发生变化时更新 State Schema 和节点逻辑发布新版本。对于正在运行的旧实例尽量保持兼容或者优雅地终止它们。总结从 Chain 到 Graph不仅仅是技术栈的升级更是思维方式的转变。Chain适合简单的、线性的、一次性任务。Graph适合复杂的、有状态的、需要容错和人工干预的业务场景。这次重构让我深刻体会到构建可靠的 Agent 系统70% 的精力应该花在状态设计、错误处理和监控告警上只有 30% 在调优 Prompt。如果你正准备搭建生产级 Agent建议先从定义清晰的 State 开始画出手绘的工作流图再动手写代码。别急着上 LLM先把“骨架”搭结实了。毕竟再聪明的头脑也跑不出失控的车辙。资料展示下面是我整理的AI大模型学习资料和工具包预览适合收藏后按主题逐步学习。如果你想看完整资料目录可以在评论区留言「资料」也欢迎告诉我你更关注AI大模型里的哪类内容。