这篇我按“先跑起来、再讲取舍”的方式写《一次LangGraph项目复盘，问题最后出在流程而不是模型》。概念会讲，但重点放在代码怎么组织、哪里容易踩坑。

摘要

上个月做了一次需求评审，团队推了一个 Agent 项目，能写文档、能查数据库、能调接口。我看了下 Demo，确实跑得挺顺。但我的第一个问题不是"模型选得对不对"，而是"出问题了，你从哪查？"

对方沉默了三秒。

这种场景我见过太多次了。大部分 Agent 项目死在 Demo 之后，不是因为模型不够强，而是因为可控性。脚本式调用可以跑通，但一上生产，权限混乱、日志缺失、异常无法兜底，问题就暴露了。

今天复盘一个真实的 LangGraph 项目，重点不在"怎么搭"，而在"怎么让它在生产里活下来"。

---

目录

• 为什么需要图工作流
• State 与 Node：把状态当成一等公民
• Edge 与条件分支：让流程可预测
• 人工审批节点：Agent 的"暂停键"
• 工程化落地：权限、日志、可观测
• 总结

为什么需要图工作流

先说结论：脚本式 Agent 适合一个人玩，图工作流适合团队协作。

我做过的项目里，有团队用简单的 while 循环 + LLM 调用实现了 Agent，效果不错。但上线两周后，问题来了：

• 工具调用失败，系统直接崩溃，没有兜底
• 不知道哪个节点出了问题，日志只有一行"LLM返回异常"
• 权限控制全靠代码里的 if-else，改一个字段要动十几处

这些问题不是模型问题，是架构问题。

图工作流的核心价值在于：把 Agent 的执行路径显式化。每个节点做什么、什么条件下跳转、状态怎么传递，全部可见可查。这不仅是工程化的需要，也是团队协作的基础。

---

State 与 Node：把状态当成一等公民

很多开发者一开始写 LangGraph，Node 函数里直接调用外部 API，状态只用来传文本。这是错误的起点。

State 应该承载完整的上下文，Node 应该是纯函数。

我复盘的一个项目里，定义了一个订单处理的 State：

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, END

class OrderState(TypedDict):
    order_id: str
    user_id: str
    product: dict
    inventory_check: Annotated[dict, operator.add]
    payment_status: str
    audit_log: list
    approval_needed: bool

关键点：

• inventory_check 用了 operator.add，因为多个 Node 可能都要往里面追加记录
• audit_log 是 list，用来记录所有操作，供后续排查
• approval_needed 是布尔值，用于条件分支

每个 Node 函数长这样：

def check_inventory(state: OrderState) -> OrderState:
    """检查库存，返回更新后的状态"""
    order_id = state["order_id"]
    product = state["product"]

    # 调用库存服务（这里应该加超时和重试）
    try:
        result = inventory_service.query(order_id, product)
        state["inventory_check"].append({
            "action": "inventory_query",
            "result": result,
            "timestamp": datetime.now().isoformat()
        })
    except Exception as e:
        # 异常也要记录，不能吞掉
        state["audit_log"].append({
            "action": "inventory_query",
            "error": str(e),
            "timestamp": datetime.now().isoformat()
        })
        raise

    return state

代码解释：

• 输入是 OrderState，输出也是 OrderState，符合纯函数原则
• 异常没有被吞掉，而是记录到 audit_log 后重新抛出，这样既能追踪问题，又不会让系统静默失败
• 所有操作都带时间戳，方便后续排查

---

Edge 与条件分支：让流程可预测

图工作流的另一个核心价值是条件分支。脚本式 Agent 的分支逻辑散落在代码里，很难看清楚整个流程。

def should_request_approval(state: OrderState) -> str:
    """判断是否需要人工审批"""
    if state["approval_needed"]:
        return "request_approval"
    return "process_payment"

graph.add_conditional_edges(
    "check_inventory",
    should_request_approval,
    {
        "request_approval": "human_approval",
        "process_payment": "process_payment_node"
    }
)

这里的关键是：条件函数应该是确定性的。不要用 LLM 来做条件判断，因为 LLM 的输出不稳定。条件判断应该基于 State 中的明确字段。

失败原因：很多项目在这里翻车，是因为用了 LLM 来做路由判断。LLM 可能今天返回"approve"，明天返回"approve!"，后天返回空字符串。这种不确定性在生产环境是灾难。

---

人工审批节点：Agent 的"暂停键"

这是我最想强调的部分。生产环境的 Agent 必须有暂停能力。

很多 Demo 里的 Agent 都是全自动的，但在实际业务中，有些操作需要人工确认。比如大额转账、敏感数据查询、权限变更。

def human_approval_node(state: OrderState) -> OrderState:
    """等待人工审批"""
    # 这里应该有一个外部系统来等待审批
    # 实际项目中，这里会调用消息队列或数据库等待
    approval_result = wait_for_approval(state["order_id"])

    if approval_result["approved"]:
        state["audit_log"].append({
            "action": "approval",
            "result": "approved",
            "approver": approval_result["approver"],
            "timestamp": datetime.now().isoformat()
        })
    else:
        state["audit_log"].append({
            "action": "approval",
            "result": "rejected",
            "approver": approval_result["approver"],
            "reason": approval_result["reason"],
            "timestamp": datetime.now().isoformat()
        })
        # 审批拒绝，流程终止
        return {"payment_status": "rejected"}

    return state

排查过程：我见过一个项目，人工审批节点没有持久化状态。审批系统在重启后，所有待审批的请求都丢失了。排查了三小时才发现，是因为 State 没有序列化到数据库，只存在内存里。

验收标准：人工审批节点必须满足：
1. 状态持久化，重启不丢失
2. 审批记录可追溯
3. 超时处理机制（比如 24 小时未审批自动拒绝）

---

工程化落地：权限、日志、可观测

回到文章开头的问题：Demo 能跑，为什么一上线就崩？

我总结的失败原因有三类：

1. 业务错误

模型调用失败、工具返回异常数据。这类错误应该有明确的错误码和重试机制。

2. 配置错误

权限配置错误、模型参数错误、节点依赖配置错误。这类错误应该在启动时就能发现，而不是运行时。

3. 环境错误

网络超时、数据库连接失败、第三方服务宕机。这类错误需要熔断和降级机制。

代码解释：一个完整的工程化配置应该包括：

# 配置管理
class Config:
    MODEL_TIMEOUT = 30  # 模型调用超时30秒
    MAX_RETRIES = 3     # 最大重试3次
    APPROVAL_TIMEOUT = 86400  # 审批超时24小时

# 工具调用封装
class ToolWrapper:
    def __init__(self, tool_name: str, timeout: int = 30):
        self.tool_name = tool_name
        self.timeout = timeout
        self.retry_count = 0

    def call(self, **kwargs):
        while self.retry_count < Config.MAX_RETRIES:
            try:
                result = self._call_with_timeout(**kwargs)
                self.retry_count = 0  # 成功后重置
                return result
            except TimeoutError:
                self.retry_count += 1
                logger.warning(f"{self.tool_name} timeout, retry {self.retry_count}")
        raise Exception(f"{self.tool_name} failed after {Config.MAX_RETRIES} retries")

适用边界：

• LangGraph 适合流程复杂、需要人工干预、有明确边界条件的场景
• 不适合简单问答、单次调用、无状态的场景
• 团队人数少于3人时，过度工程化可能得不偿失

---

总结

LangGraph 让 Agent 从脚本变成可控系统，核心不是语法，而是思维方式的转变：

1. 状态显式化：State 是 Agent 的"记忆"，必须完整、可追踪
2. 流程可视化：图结构让执行路径一目了然，便于排查
3. 边界明确化：条件分支、人工审批、异常处理，每个环节都要有兜底

验收标准：一个能上线的 Agent 工作流，应该满足：

• 所有节点有日志
• 所有异常有兜底
• 所有权限有控制
• 所有状态可恢复

Demo 能跑只是开始，权限和日志才是真护城河。

---

实战建议：如果你正在做一个 Agent 项目，不要等到上线前才考虑工程化。从第一个 Node 开始，就加上日志和异常处理。这些"麻烦"的工作，会在你排查问题时变成救命稻草。

资料展示

下面是我整理的AI大模型学习资料和工具包预览，适合收藏后按主题逐步学习。

需要这份AI大模型资料清单的话，在评论区回复「清单」即可；我会根据大家的问题继续补充对应的实战内容。