AI Agent 生产环境崩溃自救指南：从 5 个真实事故中学习

在 Hacker News 上，一个帖子引起了我的注意：”What are your worst war stories bringing agentic applications into prod“。帖子作者描述了一个熟悉的场景：他正在构建一个由多个 AI Agent 组成的报告生成系统，Agent 们需要扇出（fan-out）到大量子 Agent 处理转录数据。当某个步骤失败时——比如 API 调用返回错误或机器内存不足——级联错误会破坏整个生成流程，而且几乎没有任何可见性。

这个问题绝非个例。评论区里其他开发者分享了类似经历：有人花了整整一个月用 DBOS 重写作业为持久化执行；有人感叹”构建那些无聊的基础设施才是真正的耗时黑洞”；还有人分享了文化层面的惨痛教训——AI 突然”罢工”引发管理层恐慌，最终发现只是 API 额度用完了。

从这些真实故事中，我总结了 5 条让 AI Agent 在生产环境中稳定运行的实战经验。

1. 拥抱 Durable Execution（持久化执行）

帖子作者花了整整一个月将子作业从简单的函数调用迁移到 DBOS（Durable Execution）。为什么这么折腾？

传统的 Agent 调用链是这样的：

Agent A → Agent B → Agent C → Agent D → 完成

如果 Agent C 在第 9 步失败了，前面所有的工作都白费了。你用 try/catch 包裹每个步骤？但 Agent 的失败往往是半路状态丢失——变量没了、上下文丢了、重试时状态不一致。

Durable Execution 的核心思想是：让每一步的执行状态持久化，即使进程崩溃，也能从中断的地方继续。目前主流的方案包括：

• Temporal：工业级的持久化工作流引擎，已经有很多 AI Agent 项目用 @durable 装饰器集成它
• DBOS：数据库原生的持久化执行，和 PostgreSQL 深度绑定
• 分布式状态管理：用 Redis 或 PostgreSQL 记录每一步的输入、输出和中间状态

async def run_agent_pipeline(data):
    step_state = await load_state("pipeline_001")
    
    if not step_state.get("step_1_done"):
        result_1 = await agent_step_1(data)
        await save_state("pipeline_001", {"step_1_done": True, "step_1_result": result_1})
    
    if not step_state.get("step_2_done"):
        result_2 = await agent_step_2(step_state["step_1_result"])
        await save_state("pipeline_001", {**step_state, "step_2_done": True})
    
    # 即使在这里崩溃，下次执行也会从 step_2 继续
    return await finalize(step_state)

一句话总结：不要相信 Agent 进程的生存期，把你的状态交给可靠的存储。

2. 用沙箱隔离每个 Agent 运行时

评论区有人提到用 VM 隔离 + --yolo-mode 运行无人值守的 Agent。这背后的关键是：每个 Agent 都应该在独立的沙箱中运行。

为什么？因为一个 Agent 的内存泄漏不应该影响另一个 Agent。一个 Agent 安装了不兼容的依赖不应该污染全局环境。

实践中，三种隔离级别都很常见：

docker run --rm \
  --memory="2g" \
  --cpus="1" \
  --network none \
  -v /workspace/agent_001:/work \
  agent-runner:latest \
  python agent.py --task build_feature

docker rm $(docker ps -aq --filter "label=agent-session")

3. 构建 Agent 专属的可观测性栈

普通应用的监控（CPU、内存、QPS）对 AI Agent 远远不够。你需要知道的不是”响应时间 500ms”，而是：

• Agent 当前在哪个步骤？
• 它为什么会卡住超过 5 分钟？
• 它调用了哪个工具，传了什么参数？
• 它的 token 消耗速率是否异常？

关键实践：

• 结构化日志：每个 Agent 步骤都输出 {step_id, agent_name, tool_call, duration_ms, token_count} 格式的日志
• 状态面板：实时显示所有 Agent 的活跃状态、当前步骤、已用 tokens
• 失败回溯：能一键查看失败的 Agent 的完整对话历史

from uuid import uuid4
import structlog

logger = structlog.get_logger()

async def agent_step(input_data):
    step_id = uuid4()
    logger.info("agent.step.start", step_id=step_id, input_size=len(str(input_data)))
    
    try:
        result = await llm.call(tools=[...])
        logger.info("agent.step.complete", step_id=step_id, 
                     tokens_used=result.usage.total_tokens,
                     tool_calls=len(result.tool_calls))
        return result
    except Exception as e:
        logger.error("agent.step.failed", step_id=step_id, 
                      error=str(e), error_type=type(e).__name__)
        raise

4. 区分 Agent 逻辑和 Agent 基础设施

帖子作者问了一个扎心的问题：“你们在 Agent 基础设施（持久化、监控、人工审批、实时 UI）上花了多少工程师周，跟实际 Agent 逻辑相比？”

答案是：非常多，而且比例失衡是正常的。

这不是问题，而是正常状态。Agent 应用本质上是一个分布式系统，而分布式系统的基础设施天然复杂。关键是要意识到这个比例，给基础设施留出足够的预算，而不是把所有时间花在优化 prompt 上。

实用建议：先用现成的工具（Temporal、LangSmith、Braintrust）搭建基础设施骨架，等系统稳定后再考虑替换为自研方案。

5. 预算和配额管理不是可选而是必需

评论区里最让人哭笑不得的故事：整个管理团队因为”AI 停止工作”开了紧急全员会议，最后发现只是 API 额度用完了。更讽刺的是，团队三天前就预警过额度不足，但没人当回事，直到生产故障发生。

这暴露了两个问题：

1. 技术层面：没有在 Agent 运行时集成配额检查
2. 管理层面：没有把 AI 成本纳入运营预算的常规流程

技术方案：

class TokenBudget:
    def __init__(self, daily_limit: int, weekly_limit: int):
        self.daily_limit = daily_limit
        self.weekly_limit = weekly_limit
    
    async def check_budget(self) -> bool:
        daily_used = await get_daily_token_usage()
        weekly_used = await get_weekly_token_usage()
        
        if daily_used > self.daily_limit * 0.9:
            logger.warning("approaching daily token limit", 
                          used=daily_used, limit=self.daily_limit)
        
        if daily_used > self.daily_limit:
            return False  # 优雅降级，而不是直接崩溃
        
        return True
    
    async def get_agent_budget(self) -> int:
        """返回当前 Agent 还可以使用的 token 数"""
        remaining = self.daily_limit - await get_daily_token_usage()
        return max(0, remaining)

在 Agent 的入口处调用 check_budget()，如果配额不足则优雅降级（切换到更小的模型、减少上下文长度、或直接返回错误提示），而不是让 Agent 在半路上突然挂掉。

总结

AI Agent 的生产环境部署还是一个相对年轻的领域。从这些真实的生产事故中，我们可以看到一些清晰的模式：持久化、隔离、可观测、预算管理——这些对于传统分布式系统来说已经是常识，但在 AI Agent 领域还经常被忽视。

如果你也在构建生产级的 Agent 应用，建议从这五点开始排查。不要等”第 9 步中的第 12 步”失败时才后悔——在你写第一个 Agent 逻辑之前，先把基础设施搭建好。

值得关注的开源工具：

• Temporal — 工业级工作流引擎
• DBOS — 数据库原生持久化执行
• Lite-Harness — 自托管 Agent 沙箱
• Braintrust — Agent 可观测性
• LangSmith — LLM 应用监控