走向生产：容错、部署与可观测性

《Deep Agents 实战笔记》第 13 篇（收官）| 基于 deepagents (Python) 官方文档 原文：https://docs.langchain.com/oss/python/deepagents/fault-tolerance · https://docs.langchain.com/oss/python/deepagents/going-to-production

十二篇下来，你已经能搭出会拆任务、会写文件、会派子代理的 Deep Agent。但 demo 和生产之间隔着一道墙：demo 里出错是你坐在终端前手动重跑，生产里出错是凌晨三点的告警和烧钱的 API 账单。 这一篇我们补上最后一课——容错、部署、记忆落地与可观测性，然后给整个系列画一张收尾的地图。

一、容错：不是所有错误都该一概而论

很多团队上线 Agent 的第一反应是"包一层 try/except"，这是错的。官方文档开篇就点破了一个关键认知：错误要分类，不同类别交给不同的"修复者"。

flowchart TB
    ERR["工具或模型调用出错"] --> Q{"这是什么类型的错误？"}
    Q -->|"网络超时 / 限流"| RETRY["重试中间件<br/>指数退避后自动重试"]
    Q -->|"工具入参不对等<br/>模型能自我纠正"| FEEDBACK["错误处理中间件<br/>把异常转成 ToolMessage 喂回模型"]
    Q -->|"缺订单号 / 指令不清"| HUMAN["人机协同 HITL<br/>interrupt 暂停等用户"]
    Q -->|"提供商整体宕机"| FALLBACK["回退中间件<br/>切换备用模型"]
    Q -->|"同一调用反复循环"| LIMIT["调用上限中间件<br/>熔断本次运行"]
    Q -->|"未知异常"| PROP["不捕获<br/>让它冒泡进日志"]
    style RETRY fill:#f0f7f0,color:#1d4a1d,stroke:#4a9e4a
    style FEEDBACK fill:#f0f7f0,color:#1d4a1d,stroke:#4a9e4a
    style HUMAN fill:#fbf3e6,color:#6b4a12,stroke:#c99a3d
    style FALLBACK fill:#eef4ff,color:#1a3a8b,stroke:#4a72d9
    style LIMIT fill:#fdeef0,color:#701d2a,stroke:#c94a5e
    style PROP fill:#f7f7f7,color:#333333,stroke:#999999

这张图值得贴在工位上。逐个看代码。

重试：模型调用和工具调用各有各的重试中间件，都带指数退避：

from deepagents import create_deep_agent
from langchain.agents.middleware import ModelRetryMiddleware, ToolRetryMiddleware

agent = create_deep_agent(
    model="google_genai:gemini-3.6-flash",
    middleware=[
        # 对限流、超时、5xx 自动重试模型调用
        ModelRetryMiddleware(max_retries=3, backoff_factor=2.0, initial_delay=1.0),
        # 只重试命中外部 API 的特定工具，而不是所有工具
        ToolRetryMiddleware(
            max_retries=2,
            tools=["search", "fetch_url"],
            retry_on=(TimeoutError, ConnectionError),
        ),
    ],
)

注意 tools=["search", "fetch_url"] 这个限定——官方特意强调不要无差别重试一切。失败的 read_file 重试一百次也不会有结果，但超时的网络搜索很可能值得再试一次。另外主流集成包抛出的异常（如 ModelRateLimitError、ModelTimeoutError）自带 is_retryable 标志，重试中间件默认会遵循它。

错误反馈给模型：ToolErrorMiddleware 把工具异常转成模型能看到的错误 ToolMessage，让它自己调整再试（需要 langchain>=1.3.14）：

from langchain.agents.middleware import ToolErrorMiddleware

def on_error(exc: Exception, request: ToolCallRequest) -> str | None:
    if isinstance(exc, ValueError):
        return f"`{request.tool_call['name']}` failed with {type(exc).__name__}."
    # 其余异常不返回内容，原样向上传播

agent = create_deep_agent(
    model="google_genai:gemini-3.6-flash",
    middleware=[ToolErrorMiddleware(on_error)],
)

这里藏着一条很好的工程哲学：只捕获你明确知道怎么处理的异常，其余全部放行。 on_error 返回 None 就等于"我不认识这个错误，让它崩"。全场兜底的 try/except 只会把问题埋进日志深处。

模型回退：主提供商整体宕机时切备用模型：

from langchain.agents.middleware import ModelFallbackMiddleware

agent = create_deep_agent(
    model="google_genai:gemini-3.6-flash",
    middleware=[
        ModelFallbackMiddleware("gpt-5.5"),
    ],
)

限流与调用上限——这两件事解决的是"烧钱"而不是"报错"。官方原话很直白：不加限制的话，一个陷入混乱的代理可能在几分钟内烧光你的 LLM API 预算。两层控制互补使用：

from langchain.rate_limiters import InMemoryRateLimiter
from langchain.chat_models import init_chat_model
from langchain.agents.middleware import ModelCallLimitMiddleware, ToolCallLimitMiddleware

# 第一层：控制发给提供商的请求速率
rate_limiter = InMemoryRateLimiter(
    requests_per_second=0.1,    # 每 10 秒 1 个请求
    check_every_n_seconds=0.1,  # 每 100ms 检查一次是否放行
    max_bucket_size=10,         # 允许的最大突发量
)
model = init_chat_model(
    model="google_genai:gemini-3.6-flash",
    rate_limiter=rate_limiter,
)

# 第二层：限制单次运行的调用次数
agent = create_deep_agent(
    model=model,
    middleware=[
        ModelCallLimitMiddleware(run_limit=50),
        ToolCallLimitMiddleware(run_limit=200),
    ],
)

run_limit 限制单次调用（每轮重置），thread_limit 限制整个对话累计（需要 checkpointer）。我的建议是把调用上限当成生产环境的标配保险丝——它平时毫无存在感，出事时帮你保住账单和口碑。

二、部署形态：托管还是自托管

代码写稳了，接下来是"跑在哪"。先把工程上的关键配置说清：项目根目录需要一个 langgraph.json，告诉平台如何构建和运行你的应用：

{
  "dependencies": ["."],
  "graphs": {
    "agent": "./agent.py:agent"
  },
  "env": ".env"
}

graphs 把 graph ID 映射到代码位置（"<id>": "./<file>:<variable>"），env 指向存放 API 密钥的 .env 文件。本地 langgraph dev 和生产部署用的都是这份配置。

部署路径有两条：

• Managed Deep Agents（推荐路径）：CLI 优先的托管运行时，在 LangSmith Cloud 上目前处于公开 beta。你以一组文件的形式声明代理，托管运行时替你编译和运营；
• LangSmith Deployment：需要自定义应用代码、自定义路由和高级认证的团队走这条路。传统部署开箱即用地提供认证、webhook、cron 任务、可观测性，还能通过 MCP 或 A2A 暴露你的代理。

两条路都会替你备好线程、运行、store 和 checkpointer，不用自己搭。

生产调用的两个必传参数值得单独强调。本地玩 demo 时 invoke 只传消息，生产里每次调用都应带上：

• thread_id（通过 config={"configurable": {"thread_id": ...}}）：对话的稳定标识符，checkpointer 用它持久化和恢复消息历史；
• context：每次运行的数据（如 user_id、功能开关），用 context_schema 定义形状，工具和中间件通过 runtime.context 读取。

from dataclasses import dataclass
from deepagents import create_deep_agent
from langchain_core.utils.uuid import uuid7

@dataclass
class Context:
    user_id: str

agent = create_deep_agent(
    model="google_genai:gemini-3.6-flash",
    context_schema=Context,
)

# 开启一段对话
config = {"configurable": {"thread_id": str(uuid7())}}
agent.invoke(
    {"messages": [{"role": "user", "content": "Plan a 3-day trip to Tokyo"}]},
    config=config,
    context=Context(user_id="user-123"),
)

# 同一段对话追问：复用同一个 thread_id
agent.invoke(
    {"messages": [{"role": "user", "content": "Make it 5 days instead"}]},
    config=config,
    context=Context(user_id="user-123"),
)

thread_id 限定对话，context 携带每次运行的数据，两者相互独立。

持久化执行是自托管时最容易被低估的红利。 Deep Agents 跑在 LangGraph 上，每一步都对状态做 checkpoint：运行中途因故障、超时或 HITL 审批暂停，恢复时从最后记录的状态继续，不重放之前的步骤。对一个派了十几个子代理、跑了半小时的研究任务来说，这意味着第 25 分钟崩溃不用从头再来。Checkpoint 还附赠两样东西：HITL 中断可以暂停几天再精确恢复；每个步骤都是可回退的快照（时间旅行），支付这类不可逆操作出问题时可以审计导致该动作的确切状态。

三、持久化记忆落地：作用域决定一切

第 8 篇讲过记忆就是文件，但那是单机视角。生产环境里真正的难题不是"存什么"，而是**"谁能看到谁的记忆"**。官方给出三个作用域选项：

落地方式就是第 4 篇讲过的 Backend 组合拳：默认 StateBackend 提供线程内的草稿空间，/memories/ 路径路由到 StoreBackend，跨线程持久化：

from deepagents import create_deep_agent
from deepagents.backends import CompositeBackend, StateBackend, StoreBackend

agent = create_deep_agent(
    model="google_genai:gemini-3.6-flash",
    backend=CompositeBackend(
        default=StateBackend(),
        routes={
            "/memories/": StoreBackend(
                namespace=lambda rt: (
                    rt.server_info.assistant_id,
                    rt.server_info.user.identity,
                ),
            ),
        },
    ),
    system_prompt="""You have persistent memory at /memories/.

    Read /memories/instructions.txt at the start of each conversation for
    accumulated knowledge and preferences. When you learn something that
    should persist, update that file.""",
)

（rt.server_info 命名空间模式需要 deepagents>=0.5.0。）

flowchart LR
    REQ["生产请求<br/>thread_id + context"] --> AG["Deep Agent"]
    subgraph FS["CompositeBackend 按路径路由"]
        SB["StateBackend<br/>线程内草稿空间<br/>随 checkpoint 持久化"]
        ST["StoreBackend<br/>跨线程长期记忆"]
    end
    AG -->|"工作文件、中间产物"| SB
    AG -->|"/memories/ 路径"| ST
    ST -->|"namespace: assistant_id + user_id"| SCOPE["按助手和用户隔离<br/>互不可见"]
    style AG fill:#eef4ff,color:#1a3a8b,stroke:#4a72d9
    style SB fill:#f0f7f0,color:#1d4a1d,stroke:#4a9e4a
    style ST fill:#fbf3e6,color:#6b4a12,stroke:#c99a3d
    style SCOPE fill:#f7f7f7,color:#333333,stroke:#999999

这里有一个官方用"警告"级别标出的坑：共享记忆是提示注入的传播渠道。 如果一个用户能写入另一个用户的对话会读取的记忆，恶意用户就能往共享状态里注入指令。对策：共享路径强制只读——组织级策略只应由应用代码写入，用权限规则声明式拒绝对共享路径的写操作。

另外两条生产红线，来自官方原文：FilesystemBackend 和 LocalShellBackend 直接访问宿主机，不要在已部署的代理中使用它们；代理需要跑代码就用沙箱（隔离容器崩了也只影响沙箱本身），而 API 密钥通过沙箱认证代理注入出站请求，原始密钥永远不进沙箱。还可以用 PIIMiddleware 在用户输入到达模型前做脱敏（redact / mask / hash / block 四种策略）。

四、可观测性与评估

好消息是这一块几乎免费：LangSmith Cloud 部署会自动把追踪发送到以你的部署命名的项目，打开 smith.langchain.com 就能调试每次运行、监控用量。自托管部署则需要按文档单独配置追踪。官方还建议同时设置 LangSmith Engine——它会监控你的追踪记录、检测问题并提出修复建议。

第 2 篇说过"调试 Deep Agent 的第一原则：先看 trace，再猜问题"，在生产语境下这句话要升格：trace 不只是调试工具，更是评估数据源。 每一次生产运行都是一条带完整输入输出的样本，积累起来之后就可以做系统化的评估——比如用 LLM-as-a-judge 的思路，拿强模型按评分标准（rubric，第 12 篇实战里已经练过）批量审查生产 trace 里的最终报告质量，把"感觉变好了"变成"分数变高了"。评估不是收官之后的作业，而是从上线第一天就该开始攒的数据资产。

五、全系列总结：一张地图收尾

十三篇走完，回头画一张知识地图。整个系列的脉络是五级台阶：先建立认知，再拆核心机制，然后装上安全与体验的护栏，接着进真实项目，最后推向生产。

flowchart TB
    subgraph STAGE1["认知：为什么这样做"]
        B01["01 文件系统与子代理<br/>两大设计的动机"]
        B02["02 跑通第一个研究 Agent<br/>40 行代码上手"]
    end
    subgraph STAGE2["核心机制：引擎拆解"]
        B03["03 五大内置能力"]
        B04["04 Backend 外接大脑"]
        B05["05 子代理上：委派"]
        B06["06 子代理下：异步编排"]
        B07["07 上下文工程"]
    end
    subgraph STAGE3["安全与体验：装上护栏"]
        B08["08 记忆与技能"]
        B09["09 沙箱与解释器"]
        B10["10 权限与 HITL"]
    end
    subgraph STAGE4["实战：真刀真枪"]
        B11["11 复刻 Deep Research"]
        B12["12 Agentic RAG"]
    end
    subgraph STAGE5["生产：最后一公里"]
        B13["13 容错、部署、可观测性<br/>（本篇）"]
    end
    B01 --> B02 --> B03
    B03 --> B04 --> B05 --> B06 --> B07
    B07 --> B08 --> B09 --> B10
    B10 --> B11 --> B12 --> B13
    style STAGE1 fill:#f4effb,color:#4a2170,stroke:#8b52c9
    style STAGE2 fill:#eef4ff,color:#1a3a8b,stroke:#4a72d9
    style STAGE3 fill:#fbf3e6,color:#6b4a12,stroke:#c99a3d
    style STAGE4 fill:#f0f7f0,color:#1d4a1d,stroke:#4a9e4a
    style STAGE5 fill:#fdeef0,color:#701d2a,stroke:#c94a5e

如果用一句话浓缩十三篇：Deep Agents 的全部设计都在回答同一个问题——怎么让一个上下文有限、会犯错的 LLM，稳定地干完又长又杂的活。 文件系统和 Backend 解决"装得下"，子代理解决"想得清"，上下文工程解决"跑得久"，记忆和技能解决"越用越好"，权限和 HITL 解决"不出事"，容错和可观测解决"上了线还能活着"。

如果你只想读三篇，我的推荐组合是：

1. 第 02 篇——40 行代码跑通，建立手感，判断这个框架值不值得投入；
2. 第 07 篇——上下文工程是 Deep Agents 最独门的能力，也是长任务成败的真正分水岭，读懂它等于读懂了框架的设计哲学；
3. 第 13 篇——本篇。容错中间件 + 持久化 + 记忆作用域，是从 demo 工程师到生产工程师的分界线。

如果你做的是多代理协作类应用，把第 2 项换成第 05 篇（子代理委派）。

写完这篇，系列正式收官。工具会过时，版本号会过期，但"管理上下文、隔离风险、给错误留退路"这三件事，会比你用的任何一个框架都活得久。祝你的 Agent 顺利上线。

参考资料：

• 容错（官方文档，本文主要素材）：https://docs.langchain.com/oss/python/deepagents/fault-tolerance
• 走向生产（官方文档，本文主要素材）：https://docs.langchain.com/oss/python/deepagents/going-to-production
• 完整示例仓库：https://github.com/langchain-ai/deepagents/tree/main/examples