丹尼拉（Dayfing）
返回文章列表
4,005 字16 分钟

生产环境 AI 智能体架构：工具、状态、审批与评估

生产环境中的 AI 智能体不是套着更大上下文窗口的 prompt。它是一项服务，由模型选择动作、观察动作结果，并持续执行，直到得到受约束的结果或请求人工帮助。模型只是这项服务的一个组件。周围的控制平面决定模型可以看到什么、可以调用哪些工具、哪些动作需要审批、状态如何跨越重启，以及发布新版本前需要什么证据。

本文给出一个拥有少量工具的单智能体参考架构。同样的边界也可以支撑固定步骤的 workflow，或日后扩展为多智能体系统。先实现能够完成任务的最小循环。Anthropic 关于有效智能体的指南也区分了可预测的 workflow，以及由模型动态决定工具使用方式的系统。自主性是产品决策，不是默认的架构选项。

参考架构

将请求路径、决策循环和改变外部世界的副作用分开。Gateway 负责认证调用者、创建请求 ID 与 trace ID、应用配额，并移除或标记敏感字段。Orchestrator 负责循环。Model adapter 把中立状态转换成供应商请求，再把供应商响应转换成小型决策类型。Tool gateway 校验参数、授权执行主体、应用策略，并调用隔离的 connector。状态与事件存储应位于进程之外，这样 worker 在崩溃后可以恢复。

[用户或 API 客户端]
          |
[Gateway：身份、限制、request ID]
          |
[Orchestrator：policy -> model -> tool loop]
       /       |              \
[State]   [Approval service]  [Tool gateway]
   |             |              |
[Memory/RAG] [人工决定] [API、文件、队列]
                         |
                  [事件与追踪]

不要让模型接触原始 credentials、无限制的网络访问或直接写数据库。Tool gateway 应暴露 "search_orders"、"draft_refund"、"send_message" 这类狭窄操作，而不是通用 HTTP 客户端或 SQL 控制台。每个操作都要有负责人、输入 schema、permission、risk class、timeout 和记录在案的结果格式。

使用明确的请求封装：

字段 用途
tenantId 与 actorId 将每次读取和副作用绑定到经过认证的主体。
requestId 与 traceId 关联重试、审批、工具调用和日志。
goal 将用户目标与模型消息分开保存。
policyVersion 与 modelVersion 让一次运行足够可复现，便于调查。
deadline 与 stepBudget 限制墙钟时间以及模型到工具的回合数。

NIST AI RMF 生成式 AI 配置文件可帮助团队按生命周期组织风险工作。将其中的问题转化为封装和 tool gateway 中的具体控制，不要把框架本身当成某个部署安全的证明。

工具循环

循环应由一个组件负责，并在每个回合产生清晰的状态转移：

  1. 加载 run、policy、对话摘要和允许使用的工具目录。
  2. 构造有界的 model input。把用户文本、检索文本、工具输出和系统指令标记为不同的信任类别。
  3. 要求模型返回最终回答或类型化的 tool call。在执行前拒绝格式错误的名称和参数。
  4. 在模型之外完成 authorization 和 risk policy 判断。模型声称动作安全，不等于授权决定。
  5. 如果动作跨越 approval gate，保存 pending action 并停止。获得明确决定后，从保存的动作恢复。
  6. 使用 deadline 和 idempotency key 执行 connector。记录经过脱敏的结果，并将其追加到 state。
  7. 检查步数、token、成本和时间预算。只有 policy 允许继续时才开始下一回合。
  8. 返回说明已发生什么、未发生什么，以及仍待处理的审批或不确定性的回答。

不要把模型调用放进没有上限的 "while" 循环。Step budget 防止工具或 prompt 把一次请求变成长而昂贵的链。Deadline 保护调用者和 worker 池。重复调用检测器可以发现模型持续请求同一个失败动作。Circuit breaker 可以暂时停用异常 connector，同时保留只读工具。

Model Context Protocol 规范统一了 resources、prompts 和 tools，但不会替代宿主应用的 consent、authorization 和 isolation 控制。把 MCP server 视为外部依赖。固定服务器身份，检查它声明的工具，限制发送给它的数据，并对它应用与内部 connector 相同的 gateway policy。

状态、记忆与上下文

State 是一次 run 的持久记录。保存用户目标、消息或消息引用、工具调用与结果、审批、policy 决策、模型和工具版本、时间戳、状态以及错误分类。先追加事件，再从事件派生当前 run view。这样恢复和审计比修改一个不透明的 JSON blob 更容易。加密敏感字段，隔离 tenants，定义 retention，并让删除操作覆盖 snapshots、indexes、traces 和 caches。

Conversation history 不等于 memory。为当前 run 保留短期 working context。只有在有明确用途、retention 规则、来源，以及用户或管理员可以纠正的机制时，才保存用户或业务 memory。事实应带有 provenance 和 confidence 或 freshness 字段。不要因为模型猜测听起来合理，就把它写入持久 memory。

Retrieval 是具有 trust boundary 的工具。排名前先应用 tenant 和 authorization filters。将 source identifiers 和 timestamps 带进 context。明确告诉模型，retrieved text 是数据，不是指令。限制 chunks，移除不需要的 secrets，默认记录 query 和文档标识而不记录私有内容。在 PostgreSQL 设计中，可以比较 lexical 和 vector candidates，并且只在授权后执行 rerank。内部的 pgvector 混合 RAG 指南进一步说明这条边界。

Context compaction 应足够确定，便于调试。用固定 schema 总结旧回合，保留 decisions、未解决的问题、工具副作用、citations 和用户约束。原始 event log 要保留在 prompt 之外。如果 summary 改变了 pending action 的含义，应停止并请求 review，而不是静默继续。

工具与威胁边界

Tool schema 是安全契约，不只是模型 metadata。Gateway 要校验类型、长度、enum 值、资源所有权和字段之间的关系。完成授权后，再将名称解析成内部标识。分开 read tools 和 write tools。给 connector 的 credentials 应是短期的，并限制到一个 tenant 和一个操作。高风险代码或文件任务应在隔离 worker 中运行，并使用文件系统和网络 allowlist。

把所有外部内容都视为可能有害。Email、网页、issue、文档、工具描述或 MCP resource 可能包含间接 prompt injection。在 model input 中清楚划定这些内容，必要时删除可执行标记，并让应用决定 permissions。Output validation 必须独立于模型。例如，在 payment connector 看到退款前，根据订单、货币、执行主体权限和 policy maximum 检查 refund amount。

威胁模型至少要覆盖：

边界 需要阻止的失败
用户到 gateway 账户接管、超大请求以及其他 tenant 的标识符。
模型到 tool gateway 参数注入、权限混淆和过度自主。
Retrieval 或 MCP 到模型 间接 prompt injection 以及数据中的恶意指令。
Connector 到外部服务 凭据泄漏、SSRF、重放和数据外泄。
Worker 到 state store 事件被篡改、策略过期以及审计记录不完整。

OWASP GenAI LLM Top 10将 prompt injection、excessive agency、insecure output handling 和 unbounded consumption 列为需要应用层缓解的风险。相关的 Dayfing prompt injection 与 MCP 安全指南提供了威胁检查清单。System prompt 是有用的指导，但不是 sandbox、authorization layer 或 secret store。

审批门与人工控制

把 approval 放在副作用之前，而不是放在无害的 reasoning 周围。读取用户自己的日历可以自动完成。发送外部消息、修改记录、发放资金、删除数据或部署代码，通常要根据 actor、target、amount、reversibility 和 confidence 做 policy decision。审批门必须在 application code 中，不能让 prompt 绕过。

持久化 approval request，其中包含 proposed tool、normalized arguments、受影响资源、原因、policy version、过期时间，以及相关 state 的 hash。审核者应看到同样的数据。将审核决定绑定到 action hash 和 actor。批准后，在执行前重新检查 authorization、freshness 和 budget。拒绝或过期后，记录决定并告诉模型动作没有发生。旧 approval 不能授权已修改的 payload。

Human-in-the-loop 可以有几种模式：

模式 适合的用途
Observe 记录或抽样低风险动作，同时保持 agent 自动运行。
Confirm 在不可逆副作用前立即请求批准。
Review 让人检查完整 draft 和选定的证据。
Take over 将 run 连同当前 state 和 lock 转交给 operator。

把暂停设计成正常状态，而不是异常。队列可以交付待处理审批，通知可能过期，worker 可以在另一台主机上恢复 run。用户应能看到 agent 是在思考、等待数据、等待审批、重试，还是已经完成。

重试、幂等与恢复

重试前先分类错误。Validation error 需要修正调用或向用户提问。Authentication 和 authorization errors 应停止。Rate limit 应遵守供应商返回的 retry signal。对于写操作，timeout 和连接重置是不明确的，因为远程服务可能已经应用了副作用。只有当操作语义或 idempotency key 能保证重复安全时才重试。RFC 9110 第 9.2.2 节解释了为什么没有办法确认副作用安全时，不应自动重试非幂等方法。

从 run 和逻辑动作生成稳定 key，不要从 attempt number 生成。Connector 在 retry window 内保存 key 和最终结果。如果同一个 key 携带不同 arguments，应拒绝。使用带 jitter 的 exponential backoff，并设置较小的最大 attempts。Retry 不是新的 model decision。持久化 original call、attempt number、response class 和 connector request ID。

Recovery 属于 state machine。使用 "running"、"waiting_for_approval"、"retrying"、"failed"、"completed" 和 "cancelled" 等状态。Lease 防止两个 worker 同时执行同一个 run。失去 lease 后，在下一个副作用前停止。Worker 崩溃后，reconciler 可以将 pending actions 与 connector records 对照。若供应商支持，cancellation 应传播到 model requests、tool calls、queues 和 approval requests。

可观测性

每个用户请求创建一个 trace，并为 model calls、retrieval、policy checks、approvals 和 tools 创建 spans。记录 duration、status、retry count、可用时的 input 和 output token counts、model 和 prompt versions、tool name、risk class 以及 cost estimate。导出前先 redact secrets 和敏感内容。使用稳定的 run ID 将审批或支持工单与 trace 关联，但不要把个人数据放入 baggage。OpenTelemetry context propagation说明了跨服务关联 trace context 的方式，也提醒开发者注意不可信 headers 和敏感 baggage。

应测量有明确 rubric 的任务完成率、成功工具调用率、validation failures、approval rate、retry 与 timeout rate、p50 与 p95 latency、每个完成任务的 token 和工具成本、cancellation rate 以及 policy blocks。按 model version、tool、tenant class 和 release 分组。少量错误可能掩盖大量无声的错误回答，因此要把 traces 与抽样 transcripts 和 evaluator results 关联。不要记录 chain-of-thought。仅保存 policy 允许的简短 decision metadata，以及用户可见的 reasoning 或 citations。

内部的 AI 智能体可观测性指南说明如何利用这些信号，同时不把日志变成另一套 secrets 数据库。

发布前后的评估

Agent eval 是一个 scenario，具有确定的 starting state、允许的 tools、预期 invariants 和 scoring rule。仅检查最终回答不够。要断言 agent 使用了授权工具、保持 tenant scope、在需要时请求 approval、没有重复 write、引用了正确 source，并在 budget 到达时停止。加入 adversarial scenarios：恶意 retrieved text、不可用 connector、write 后 timeout、过期 approval、含义不明的 request，以及格式错误的 tool data。

采用分层 suite：

  1. 对 schemas、authorization、redaction、idempotency、state transitions 和 budget enforcement 做 deterministic unit tests。
  2. 使用记录的 tool responses 和固定的 model decisions 做 replay tests，验证 recovery。
  3. 使用 model 和 rubric 做 scenario tests，评价任务结果、安全性和沟通。
  4. 做 red-team tests，覆盖 direct 与 indirect injection、数据泄漏、过度自主和 denial of service。
  5. 在有隐私控制和人工 review 的情况下做 production sampling，并将 failures 转为 regression cases。

为 scenario、tools、policy、prompts、model 和 evaluator 做版本管理。将 failures 与 trace 以及最小可复现 input 一起保存。把 candidate release 与 baseline 比较，即使平均质量提升，也不允许 hard safety invariants 回归。Anthropic 的智能体评估指南解释了为什么多回合工具使用需要 trajectory-level evaluation。内部的 AI 智能体评估指南提供实践测试矩阵。

成本与延迟取舍

每个 model turn、retrieved token、tool call、approval pause 和 retry 都会增加时间或费用。按任务类别设定 budgets，而不是只设一个全局数字。如果测得的准确率足够，可用较小模型完成 routing、extraction 和 policy prechecks。把更强模型留给含糊的 planning 或 final synthesis。缓存稳定的 tool catalogs 和 retrieval embeddings。在 context 变大前总结它，但测量 summary 是否带来额外回合或事实丢失。

并行执行相互独立的只读 calls，再用明确 provenance 合并结果。除非 connector 提供 transaction 或设计好的 compensation，否则保持 writes 顺序执行。在不暴露 secrets 的情况下流式显示进度。长任务应持久化为 job，由 worker 在 HTTP request 结束后继续。更快的模型不一定更便宜，因为错误可能触发人工审核、补偿写入或重复运行。要衡量每个正确且符合 policy 的结果的总成本。

最小可运行循环

下面的 protocol-independent TypeScript 示例使用 fake model 和本地 tools，因此无需 provider SDK 或网络即可运行。它展示类型化决策循环、写操作审批、有界重试、稳定的幂等 key 和防止重复副作用的保护。实际 adapter 可以替换 model，同时保留 policy 和 tool 边界。

type ToolCall = { id: string; name: string; input: unknown };
type Message =
  | { role: "user"; content: string }
  | { role: "assistant"; content: string; toolCall?: ToolCall }
  | { role: "tool"; callId: string; content: string };
type Decision =
  | { kind: "answer"; text: string }
  | { kind: "call"; call: ToolCall };
type Tool = {
  sideEffect: "read" | "write";
  run(input: unknown, idempotencyKey: string): Promise<string>;
};

const issued = new Set<string>();
const tools: Record<string, Tool> = {
  getBalance: {
    sideEffect: "read",
    async run(): Promise<string> {
      return JSON.stringify({ account: "demo", cents: 4200 });
    },
  },
  sendInvoice: {
    sideEffect: "write",
    async run(input: unknown, idempotencyKey: string): Promise<string> {
      if (issued.has(idempotencyKey)) return "already-sent";
      if (typeof input !== "object" || input === null) throw new Error("invalid input");
      issued.add(idempotencyKey);
      return "invoice-sent";
    },
  },
};

const model = {
  async decide(messages: readonly Message[]): Promise<Decision> {
    const toolCount = messages.filter((message) => message.role === "tool").length;
    if (toolCount === 0) {
      return { kind: "call", call: { id: "balance-1", name: "getBalance", input: {} } };
    }
    if (toolCount === 1) {
      return { kind: "call", call: { id: "invoice-1", name: "sendInvoice", input: { cents: 1200 } } };
    }
    return { kind: "answer", text: "The balance was checked and the invoice was sent." };
  },
};

const wait = (milliseconds: number) => new Promise((resolve) => setTimeout(resolve, milliseconds));

async function execute(call: ToolCall, tool: Tool, key: string): Promise<string> {
  for (let attempt = 0; attempt < 3; attempt += 1) {
    try {
      return await tool.run(call.input, key);
    } catch (error) {
      if (attempt === 2) throw error;
      await wait(10 * 2 ** attempt);
    }
  }
  throw new Error("unreachable");
}

async function run(): Promise<string> {
  const state: { messages: Message[]; completed: Set<string> } = {
    messages: [{ role: "user", content: "Check the balance and send the invoice." }],
    completed: new Set<string>(),
  };
  const sessionId = "session-demo";
  for (let step = 0; step < 6; step += 1) {
    const decision = await model.decide(state.messages);
    if (decision.kind === "answer") return decision.text;
    const tool = tools[decision.call.name];
    if (!tool) throw new Error("unknown tool");
    const key = sessionId + ":" + decision.call.id;
    if (tool.sideEffect === "write" && !process.argv.includes("--approve")) {
      return "Paused for approval: " + decision.call.name;
    }
    if (state.completed.has(key)) continue;
    const result = await execute(decision.call, tool, key);
    state.completed.add(key);
    state.messages.push(
      { role: "assistant", content: "", toolCall: decision.call },
      { role: "tool", callId: decision.call.id, content: result },
    );
  }
  throw new Error("step budget exceeded");
}

run().then(console.log).catch((error: unknown) => {
  console.error(error);
  process.exitCode = 1;
});

使用项目的 TypeScript toolchain 编译代码，然后运行生成的 JavaScript，不带 flag 时会看到 approval pause，带 --approve 时会允许写操作。示例故意把 state 保存在内存中。生产环境的 state 必须持久、限定 tenant、在需要时加密，并由遵守 lease 的 worker 恢复。

构建顺序

在选择模型前先定义任务结果和禁止的副作用。先在 schema 和 authorization 后面实现一个 read tool 和一个可逆的 write tool。再添加 durable events、budgets、approval states 和 idempotency，之后才增加更多工具。尽早建立第一条端到端 trace。根据真实 failure modes 创建 eval scenarios，并在每次 prompt、policy、tool 或 model 变更时运行。只有有测量依据证明需要时，才加入 MCP 或多智能体 delegation。

更多文章

实践中的 Agentic coding：用 Codex、Claude Code 和 Copilot CLI 完成 PR 与测试

10 分钟

没有一千行指令的 AGENTS.md：给编码代理的上下文

12 分钟

2026 年 AI Act 开发者指南：Article 50、披露、日志与人工监督

11 分钟