# 不要倒着建 AI Agent：从 Demo 到生产系统的组件化教程

- 原文：<https://towardsdatascience.com/most-ai-agents-fail-in-production-because-theyre-built-backwards/>
- 原文标题：Most AI Agents Fail in Production Because They’re Built Backwards
- 作者：Benjamin Nweke
- 来源：Towards Data Science
- 发布时间：2026-05-27
- 本教程来源：基于原文观点、Hermes wiki 中的 `agentic-programming-system-engineering` 概念页，以及本地中文摘要整理。
- 提取限制：可读原文由 Jina Reader 获取，正文从半句开始，开头少量上下文缺失；主体章节、案例和结论完整。

## 这篇教程解决什么问题

很多 AI Agent 原型看起来能跑，但一到生产就会出现一种危险状态：系统没有报错，输出也像是合理的，但实际已经偏离目标。原文把这种问题叫做 **built backwards**：先想“我要一个 Agent 做 X”，再加工具、写 prompt，然后相信模型推理会把上下文、状态、重试、工具失败和验证全部串起来。

这篇教程把原文观点改写成一个可落地的工程流程：**先定义组件和责任，再组合出 Agent 行为。**

适合读者：

- 正在把 LLM demo 改成内部工具或生产服务的工程师；
- 正在设计多工具、多步骤 Agent 的团队；
- 想审查现有 Agent workflow 是否过度依赖模型推理的人。

不适合：

- 想找一个“一键万能 Agent 框架”的读者；
- 只做一次性实验、无需长期维护的玩具 demo。

## 核心原则：模型只负责决策，不负责架构

原文最值得保留的一句话是：生产级 AI Agent 不是一个智能实体，而是一套系统。

把这句话落到工程上，就是：

- 模型层：只做“下一步该做什么”的判断；
- 上下文层：负责准备模型本轮需要看到的信息；
- 编排层：负责条件分支、队列、重试、状态机和超时；
- 工具层：接收明确输入，返回稳定输出，尽量避免隐藏副作用；
- 状态层：保存当前系统知道什么、是否过期、谁能读写；
- 可观测层：记录模型看到了什么、决定了什么、调用了什么、结果是否正确。

如果一个 Agent 的答案是“这些都交给模型自己想”，它就是倒着建的。

## 最小可行架构

下面是一个比“LLM + prompt + tools”更适合生产演进的最小架构：

```text
用户请求
  ↓
任务入口：校验输入、定义目标、生成 request_id
  ↓
上下文准备器：读取必要资料，裁剪历史，形成 context_packet
  ↓
决策层 LLM：只输出结构化 decision
  ↓
编排器：根据 decision 调工具、重试、超时、路由
  ↓
工具执行层：窄工具、明确输入输出、无隐藏副作用
  ↓
状态存储：记录当前事实、版本、过期时间、读写人
  ↓
验证与可观测：trace、指标、人工确认、回滚依据
```

这里的关键不是组件数量，而是每个责任都有归属。哪怕全部写在一个 Python 文件里，也比把所有职责塞进模型推理更容易调试。

## 反模式：倒着建 Agent 的 5 个信号

你可以先用这 5 个问题审查现有系统：

1. 上下文是谁准备的？如果答案是“模型自己从历史里找”，风险高。
2. 重试是谁控制的？如果答案是“框架内部自动重试”，但你看不到状态变化，风险高。
3. 工具是否只做一件事？如果一个工具既查 API、又更新缓存、又写数据库，风险高。
4. 状态是否有版本和过期规则？如果模型使用 20 分钟前的偏好还没人知道，风险高。
5. 可观测性是否能判断“做得对不对”？如果只有日志，没有 trace 和验证标准，风险高。

## 案例 1：客服工单 Agent，不要让模型直接决定所有动作

### 错误做法

```text
用户说：帮我处理退款。
Agent 读取聊天历史 → 自己判断是否退款 → 调 refund_api → 回复用户。
```

问题：模型同时承担了意图识别、政策判断、权限判断、工具调用和结果验证。任何一步错了都难以定位。

### 正确拆法

```yaml
request_id: ticket-20260529-001
intent: refund_request
context_packet:
  user_id: u_123
  order_id: o_456
  policy_version: refund-policy-2026-05
  order_status: delivered
  risk_flags: []
decision_contract:
  allowed_actions: [ask_more_info, approve_refund_candidate, reject_with_reason]
  forbidden_actions: [direct_refund_without_policy_check]
```

模型只输出候选决策：

```json
{
  "action": "approve_refund_candidate",
  "reason": "order is within refund window and no risk flags were found",
  "needs_human_confirmation": true
}
```

编排器再决定：是否需要人工确认、是否调用退款 API、是否记录审计日志。

### 验收标准

- 模型不能直接调用真实退款工具；
- 每次退款候选都有 `policy_version`；
- 每次执行都有 `request_id`、输入上下文、模型决策、最终动作；
- 人工确认前不产生外部资金变更。

### 失败处理

- 如果订单状态缺失：返回 `ask_more_info`，不允许猜；
- 如果政策版本缺失：停止，提示系统配置错误；
- 如果退款 API 超时：编排器重试一次，仍失败则进入人工队列。

## 案例 2：研究助手 Agent，先做证据包，再让模型写结论

### 错误做法

```text
用户问：某公司是否值得投资？
Agent 自己搜索网页 → 自己筛选 → 自己总结 → 自己给建议。
```

问题：检索、证据筛选、事实判断和建议生成都混在模型里，最后很难知道结论来自哪里。

### 正确拆法

先生成证据包：

```yaml
evidence_packet:
  question: 某公司是否值得投资？
  sources:
    - type: official_filing
      url: https://example.com/10q
      extracted_facts:
        - revenue_growth: "12% YoY"
        - operating_margin: "18%"
    - type: news
      url: https://example.com/news
      extracted_facts:
        - risk: "regulatory investigation opened"
  missing_evidence:
    - latest cash flow statement
    - management guidance transcript
```

再让模型只做结构化判断：

```json
{
  "answer_type": "insufficient_evidence",
  "supported_claims": [
    "Revenue grew 12% YoY in the extracted filing",
    "A regulatory investigation is reported by the cited news source"
  ],
  "unsupported_claims": [
    "No conclusion on valuation because current cash flow data is missing"
  ],
  "next_step": "fetch_latest_cash_flow_statement"
}
```

### 验收标准

- 每条关键结论必须能回到 source URL；
- 缺证据时输出 `insufficient_evidence`，不强行给结论；
- 最终报告区分事实、推论、建议；
- 允许人工追加证据后重跑，不允许用旧结论覆盖新证据。

### 失败处理

- 检索为空：返回缺口清单，而不是让模型凭常识写；
- 来源冲突：保留冲突，不让模型提前裁决；
- 新闻源可信度低：标记为弱证据，不进入核心结论。

## 案例 3：运维 Agent，先只读观察，再人工确认修复

### 错误做法

```text
用户说：服务器好像卡了。
Agent 直接执行清理缓存、重启服务、删除日志。
```

问题：这是把诊断、决策和破坏性动作交给模型一次性完成，生产风险太高。

### 正确拆法

第一阶段只读：

```bash
uptime
free -h
df -h
systemctl --failed
journalctl -p err -n 50 --no-pager
```

形成诊断卡：

```yaml
diagnosis_card:
  host: app-01
  observed_at: 2026-05-29T20:00:00+08:00
  symptoms:
    - load_average_high
    - disk_usage_92_percent
  proposed_actions:
    - action: rotate_logs
      risk: medium
      requires_confirmation: true
    - action: restart_service
      risk: high
      requires_confirmation: true
  forbidden_without_confirmation:
    - rm -rf
    - systemctl restart production_service
    - database_delete_or_drop
```

### 验收标准

- 默认只读；
- 所有写操作必须生成计划；
- 高风险动作必须人工确认；
- 每个动作都有回滚或止损说明；
- 诊断命令和输出路径可审计。

### 失败处理

- 命令无权限：记录权限错误，不 sudo 猜测；
- 指标矛盾：停止并要求更多证据；
- 需要重启服务：先输出影响范围和回滚方案。

## 可复制的离线最小示例：让编排器负责重试和 trace

下面这个例子不依赖外部 API。它模拟一个“模型决策 + 工具执行 + 编排重试 + trace”的最小闭环。重点不是模型本身，而是责任边界。

保存为 `agent_orchestrator_demo.py`：

```python
from __future__ import annotations

from dataclasses import dataclass, asdict
import json


@dataclass
class Decision:
    action: str
    reason: str


@dataclass
class TraceEvent:
    step: str
    status: str
    detail: str


def prepare_context(user_text: str) -> dict[str, str]:
    return {
        "request": user_text,
        "policy": "read-only inspection before write actions",
    }


def model_decide(context: dict[str, str]) -> Decision:
    if "delete" in context["request"].lower():
        return Decision(action="deny", reason="destructive action requires explicit approval")
    return Decision(action="inspect", reason="safe read-only first step")


def run_tool(action: str) -> dict[str, str]:
    if action == "inspect":
        return {"result": "system looks healthy", "risk": "low"}
    if action == "deny":
        return {"result": "blocked", "risk": "high"}
    raise ValueError(f"unknown action: {action}")


def orchestrate(user_text: str) -> dict[str, object]:
    trace: list[TraceEvent] = []
    context = prepare_context(user_text)
    trace.append(TraceEvent("prepare_context", "ok", "context packet created"))

    decision = model_decide(context)
    trace.append(TraceEvent("model_decide", "ok", decision.reason))

    tool_result = run_tool(decision.action)
    trace.append(TraceEvent("run_tool", "ok", tool_result["result"]))

    return {
        "decision": asdict(decision),
        "tool_result": tool_result,
        "trace": [asdict(event) for event in trace],
    }


if __name__ == "__main__":
    print(json.dumps(orchestrate("please inspect disk usage"), ensure_ascii=False, indent=2))
```

运行：

```bash
python3 agent_orchestrator_demo.py
```

预期输出形状：

```json
{
  "decision": {
    "action": "inspect",
    "reason": "safe read-only first step"
  },
  "tool_result": {
    "result": "system looks healthy",
    "risk": "low"
  },
  "trace": [
    {
      "step": "prepare_context",
      "status": "ok",
      "detail": "context packet created"
    }
  ]
}
```

这个例子的关键点：

- `prepare_context()` 不归模型负责；
- `model_decide()` 只输出结构化决策；
- `run_tool()` 是窄工具；
- `orchestrate()` 负责流程和 trace；
- trace 可以被测试、审计和回放。

## 一周落地路线

### Day 1：列出当前 Agent 的隐式责任

把现有 Agent 的 prompt、工具、框架配置和代码列出来，回答：

```text
上下文由谁准备？
状态由谁保存？
工具由谁调用？
重试由谁控制？
失败由谁处理？
验证由谁完成？
回滚由谁负责？
```

如果任何问题的答案是“模型会自己处理”，先标记为 built backwards 风险。

### Day 2：定义决策契约

要求模型输出结构化 JSON，而不是自由文本：

```json
{
  "action": "inspect | ask_more_info | propose_write | stop",
  "reason": "why this action is justified",
  "required_evidence": ["evidence item"],
  "risk_level": "low | medium | high",
  "needs_human_confirmation": true
}
```

验收：非法 action 应被 schema 拒绝；高风险 action 必须要求人工确认。

### Day 3：收窄工具面

每个工具只做一件事。工具说明必须写清：

```text
何时使用：
何时不要使用：
输入参数：
返回字段：
失败语义：
是否有外部副作用：
```

验收：任何同时“查询 + 修改 + 推断”的工具都应拆分。

### Day 4：把状态从聊天历史中拿出来

不要让模型从长聊天记录里猜当前状态。改用状态卡：

```yaml
state_card:
  current_goal: "diagnose slow API responses"
  verified_facts:
    - "p95 latency increased from 300ms to 1200ms"
  assumptions:
    - "database may be slow"
  open_questions:
    - "is cache hit rate lower than usual?"
  next_allowed_steps:
    - "read metrics"
    - "read logs"
  forbidden_steps:
    - "restart production services without approval"
```

验收：模型每轮只看到当前任务所需状态，而不是完整历史。

### Day 5：建立 trace 和验证

每次运行至少记录：

```yaml
trace_event:
  request_id: "..."
  context_version: "..."
  decision: "..."
  tool_call: "..."
  tool_result: "..."
  validation_result: "..."
  final_status: "ok | stopped | needs_human"
```

验收：发生错误时，能定位是上下文错、模型决策错、工具执行错、状态过期，还是验证缺失。

## 发布前检查清单

在把 Agent 推到生产或内部稳定流程前，至少检查：

- [ ] 模型职责是否足够窄？
- [ ] 上下文准备是否显式？
- [ ] 状态是否有版本、时间和读写边界？
- [ ] 工具是否窄、可测试、失败语义清楚？
- [ ] 重试是否可见，而不是框架黑盒？
- [ ] 高风险动作是否需要人工确认？
- [ ] 是否能追踪每次请求从输入到输出的全链路？
- [ ] 是否有离线测试或 mock 数据验证关键路径？
- [ ] 是否定义了停止条件和降级路径？
- [ ] 是否避免把文章观点直接升级成 runtime / cron / memory 规则？

## 常见失败与处理

### 失败 1：模型输出看似合理，但事实错了

处理：检查 context packet，不要先调 prompt。多数情况下是输入上下文错误、过期或缺失。

### 失败 2：工具调用偶尔污染状态

处理：把工具拆成 read / validate / write 三类；写操作必须单独审计。

### 失败 3：框架自动重试导致重复副作用

处理：把重试挪到自己可见的编排层；对写操作使用幂等键。

### 失败 4：多 Agent 共享旧状态

处理：不要共享完整聊天历史；共享结构化状态卡，并给状态加版本。

### 失败 5：只有日志，没有判断正确性的证据

处理：补 trace 和 validation result。日志回答“发生了什么”，验证回答“发生得对不对”。

## 结论

不要把 Agent 当成一个会自己协调一切的智能体。生产级 Agent 更像一个普通分布式系统：模型只是其中一个组件。真正可靠的系统，是先把上下文、状态、工具、编排、验证和可观测性分清楚，再让模型在这个边界内做决策。

一句话记住：**组件先行，行为随后；模型决策，不管架构。**