2026工程首选：用LangGraph构建高可靠性企业Agent的实战指南

摘要
当大模型从玩具走向生产，开发者面临的不是“能不能调通”，而是“挂了怎么恢复”“人工如何介入”“成本怎么算”这类工程级问题。本文将基于 LangGraph 的状态图、检查点持久化与人机协同机制，结合 CrewAI 的多 Agent 协作模式，以企业客服工单系统为例，从零构建一个可自动重试、状态可恢复、成本可审计的高可靠 Agent。读完你会理解为什么这类框架正在成为 Agentic AI 落地的工程底座。

问题背景：从 Demo 到生产，还差一个“工程底座”

2025 年，绝大多数团队都能用 LangChain、LlamaIndex 快速搭一个智能问答或自动化脚本，但一谈到部署到生产环境，问题就层出不穷：

不可靠：LLM 调用超时或返回格式错误，整个流程崩塌，没有自动重试和降级策略。
无状态：进程重启后，Agent 的执行进度全部丢失，无法从中断点继续。
无法人工干预：客服回答错误、高风险操作需要确认时，缺乏人机协作的标准协议。
成本黑盒：多次 retry、长上下文消耗了大量 token，但没人说得清钱花在哪。

这些问题本质上不是模型能力不足，而是缺少一套工程框架来统筹流程控制、持久化、协同和审计。进入 2026 年，LangGraph、CrewAI 等库通过有向图、检查点持久化、中断回调等设计，正在填补这一空白，让 Agent 具备了生产级骨架。

技术方案：用状态图刻画 Agent 生命周期

LangGraph 的核心思想是把 Agent 的执行过程建模为一个状态图（StateGraph）：每个节点代表一个处理步骤（如分类、回答、审查），边代表转移条件，整个流程的状态（包括对话历史、中间结果、审计日志）都被序列化到检查点中。

关键能力包括：

持久化检查点（Checkpointer）：每一步执行后自动保存状态，中断恢复、回溯、分支执行成为可能。
人机协作（Interrupt）：可以在任意节点悬挂流程，等待人工输入后再继续，人工决策也作为状态的一部分持久化。
条件路由与循环：支持根据状态动态决定下一步（如是否需要升级人工、是否重试）。
成本追踪：通过节点包裹记录每次 LLM 调用的 token 消耗，写入审计日志。

CrewAI 则更多从多 Agent 协作的角度切入，通过角色定义、任务委托和工具共享来实现复杂工作流，但它内部也可以与 LangGraph 结合使用。本文聚焦 LangGraph，因为其“状态图+持久化”模型对单 Agent 流程控制最为直接。

核心实现解析：构建一个企业客服工单 Agent

我们以一个典型的客服场景为例：用户提交工单 → Agent 自动分类 → 尝试自动回答 → 若置信度低则升级人工 → 人工审查后可修改回答并确认 → 发送回答、总结归档。

整个 Agent 使用 LangGraph 定义，代码在 Python 3.12、langgraph==0.2.x（模拟 2026 年版本）下运行。

定义状态结构

from typing import TypedDict, List, Optional
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver

class TicketState(TypedDict):
    messages: List[dict]          # 对话历史
    category: Optional[str]       # 分类结果
    auto_answer: Optional[str]    # 自动回答
    confidence: float             # 置信度
    needs_human: bool             # 是否需要人工
    human_decision: Optional[str] # 人工决策（批准/修改/拒绝）
    final_answer: Optional[str]   # 最终回答
    audit_log: List[dict]         # 成本审计日志

构建节点

每个节点负责一个明确子任务，并在状态上操作。为演示，核心逻辑用伪 LLM 调用替代，实际可接入 OpenAI、Claude 等。

def classify(state: TicketState) -> TicketState:
    # 模拟 LLM 分类
    last_msg = state["messages"][-1]["content"]
    # 简化的分类逻辑
    if "退款" in last_msg:
        state["category"] = "退款"
        state["confidence"] = 0.95
    elif "咨询" in last_msg or "帮助" in last_msg:
        state["category"] = "咨询"
        state["confidence"] = 0.7
    else:
        state["category"] = "其他"
        state["confidence"] = 0.4
    state["audit_log"].append({"node": "classify", "tokens": 120})
    return state

def auto_reply(state: TicketState) -> TicketState:
    # 根据分类尝试生成回复
    if state["category"] == "退款":
        state["auto_answer"] = "我们已收到您的退款请求，将在3个工作日内处理。"
    else:
        state["auto_answer"] = "您好，关于您的问题，请稍等，我们正在查看。"
    state["audit_log"].append({"node": "auto_reply", "tokens": 80})
    return state

def decide_intervention(state: TicketState) -> str:
    if state["confidence"] < 0.8 or state["category"] == "其他":
        return "human_review"
    else:
        return "finalize"

def human_review(state: TicketState) -> TicketState:
    # 标记需要中断，并等待人工输入
    state["needs_human"] = True
    return state

def finalize(state: TicketState) -> TicketState:
    # 合成最终回答
    if state.get("human_decision"):
        state["final_answer"] = state["human_decision"]
    else:
        state["final_answer"] = state["auto_answer"]
    state["audit_log"].append({"node": "finalize", "tokens": 30})
    return state

搭建图与条件边

builder = StateGraph(TicketState)

builder.add_node("classify", classify)
builder.add_node("auto_reply", auto_reply)
builder.add_node("human_review", human_review)
builder.add_node("finalize", finalize)

builder.set_entry_point("classify")
builder.add_edge("classify", "auto_reply")
builder.add_conditional_edges(
    "auto_reply",
    decide_intervention,
    {
        "human_review": "human_review",
        "finalize": "finalize"
    }
)
builder.add_edge("human_review", "finalize")
builder.add_edge("finalize", END)

# 使用内存检查点持久化
checkpointer = MemorySaver()
graph = builder.compile(checkpointer=checkpointer, interrupt_before=["human_review"])

关键点：

**interrupt_before=[“human_review”]**：图在执行到 human_review 节点前自动中断，此时状态已被检查点保存，等待外部恢复。这实现了人机协作的标准协议。
条件边：根据 confidence 动态决定是否需要人工介入。
持久化：MemorySaver 保存每次状态变更，进程重启后可通过 thread_id 恢复。

运行与人工恢复

import uuid

thread_id = str(uuid.uuid4())
initial_state = {
    "messages": [{"role": "user", "content": "我刚买的耳机有杂音，要退款"}],
    "audit_log": [],
}

# 第一次运行，会停在 human_review 之前
config = {"configurable": {"thread_id": thread_id}}
for event in graph.stream(initial_state, config):
    print(event)

# 此时状态持久化，可以查看
current_state = graph.get_state(config)
print("Pending human review:", current_state.values["needs_human"])
print("Auto answer:", current_state.values["auto_answer"])

# 人工审查后，决定修改回答
graph.update_state(config, {"human_decision": "已为您升级处理，将在4小时内联系您。", "needs_human": False})

# 从中断点继续执行（finalize 节点）
for event in graph.stream(None, config):
    print(event)

final_state = graph.get_state(config)
print("Final answer:", final_state.values["final_answer"])
print("Audit log:", final_state.values["audit_log"])

自动重试与错误处理

LangGraph 支持在节点函数内抛出 NodeInterrupt 或自定义异常，配合 retry 策略实现自动重试。也可以设置最大步数防止无限循环。在生产中，我们通常为 LLM 调用节点包裹重试装饰器：

from tenacity import retry, stop_after_attempt, wait_exponential
from langgraph.errors import GraphRecursionError

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
def robust_llm_call(prompt):
    # 模拟可能失败的调用
    raise ConnectionError("API 超时")

将该调用放入节点即可实现自动化重试，同时每次重试的 token 消耗也会被审计。

运行效果：可靠、可控、可审计的闭环

以“退款”工单为例，运行流程如下：

用户消息进入，分类为“退款”，置信度 0.95，自动生成回复草稿。
条件分支判断置信度大于 0.8，直接进入 finalize，发送回复。
若用户消息为“这个产品怎么用？”，分类置信度 0.7，流程会被路由到 human_review 并中断。
运维平台检测到该线程处于暂停状态，展示自动回答草稿，支持人员可直接修改或批准。
提交决策后，图恢复执行，最终回答被更新并发送。

在中断与恢复期间，因为检查点持久化，服务重启不影响流程进度。审计日志清晰记录了每个节点的 token 开销，便于成本归因。这一闭环将 Agent 从“一次性调用”提升为可干预、可恢复、可复核的持续过程。

总结与展望

LangGraph 和 CrewAI 等框架在 2026 年不再是实验性工具，而是构建企业级 Agent 的工程标准。它们解决了状态持久化、人机协作、自动重试和成本审计这类原本需要大量自研基础设施的问题，让团队能够专注于业务逻辑。

未来，随着多 Agent 协作和工具调用的深度集成，我们会看到更多“Agent 即服务”的架构模式。但内核仍然是清晰的状态管理和可靠的恢复机制——这正是每个有 2-3 年后端经验的开发者此时值得掌握的核心能力。从今天起，别再写无状态的 prompt 脚本，用一个状态图，把你的 Agent 变成可靠的生产服务。