这篇我按“先跑起来、再讲取舍”的方式写《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(f"Analyzing 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大模型里的哪类内容。