Agent 工程体系 · 第 93/98 篇。内容以 2026 年 9 月可验证的公开规范和稳定接口为基线；框架版本敏感能力会明确标注，不把实验行为写成通用保证。

Agent 网关：认证、模型路由、配额、策略、审计和协议适配

Agent 网关位于客户端、业务系统与模型提供商之间。它不是“把请求转发到某个大模型”的反向代理，而是一个面向 Agent 运行时的控制平面与数据平面边界：

  • 控制平面负责身份、租户、模型、工具、配额、策略、密钥和审计规则；
  • 数据平面负责接收一次 Agent 请求，解析会话与消息，执行鉴权、路由、限流、策略检查和协议转换，然后调用模型或 Agent 运行时；
  • 观测平面负责记录请求、模型调用、工具调用、策略决策、成本和最终结果。

Agent 本身通常需要规划、调用工具、协作多个专长 Agent，并保存足够的状态来完成多步任务。OpenAI 将 Agent 描述为能够规划、调用工具、协作并保留状态的应用；Anthropic 则区分了由固定代码路径编排的 workflow 与由模型动态决定过程和工具使用的 agent。(developers.openai.com)

因此，网关要处理的对象不只是一次 HTTP 请求，而是一个可能持续数分钟、产生多次模型调用和工具调用的 Agent Run。

一、先区分三种边界：API 网关、模型网关和 Agent 网关

1. API 网关解决“谁能访问哪个服务”

传统 API 网关主要关心：

客户端
  ↓
认证、限流、路由、日志
  ↓
订单服务、用户服务、支付服务

它通常以一次请求为基本计量单位：

request → response

2. 模型网关解决“这次模型调用发给谁”

模型网关位于应用与多个模型供应商之间：

Agent 应用
  ↓
模型网关
  ├── Provider A
  ├── Provider B
  └── 私有部署模型

它主要处理：

  • 模型名称映射；
  • 供应商 API 差异；
  • 重试与故障转移；
  • 令牌统计；
  • 供应商密钥管理；
  • 价格和延迟选择。

3. Agent 网关解决“一个多步运行是否允许继续”

Agent 请求的实际结构可能是：

用户请求
  ↓
第 1 次模型调用：判断意图
  ↓
工具调用：查询订单
  ↓
第 2 次模型调用：解释结果
  ↓
工具调用：申请退款
  ↓
人工审批
  ↓
第 3 次模型调用：生成确认消息

如果网关只在第一次请求时检查配额和权限，就无法控制后续的模型调用与工具动作。

Agent 网关的基本计量对象至少包括：

tenant
principal
session
run
model_call
tool_call
policy_decision
audit_event

其中：

  • tenant：租户，表示数据、模型、工具、记忆、配额和密钥的隔离边界；
  • principal：主体，可以是用户、服务账号、Agent 或工作负载；
  • session：会话，保存跨轮次上下文；
  • run：一次 Agent 执行，从接收任务开始，到完成、失败、取消或等待审批结束；
  • model_call：Run 内的一次模型调用；
  • tool_call：Run 内的一次工具调用；
  • policy_decision：一次策略判断及其结果；
  • audit_event：不可抵赖地记录关键动作的审计事件。

二、总体架构：网关不应吞掉 Agent 运行时

一个可维护的部署关系如下：

flowchart LR
    C[客户端或业务服务]
    G[Agent 网关]
    R[Agent Runtime]
    M[模型适配层]
    P1[模型供应商 A]
    P2[模型供应商 B]
    T[工具网关]
    D[业务工具与数据系统]
    S[(会话/状态存储)]
    Q[(配额账本)]
    A[(审计日志)]
    O[观测系统]
    K[密钥管理系统]

    C --> G
    G --> R
    R --> M
    M --> P1
    M --> P2
    R --> T
    T --> D
    G --> S
    G --> Q
    G --> A
    G --> O
    G --> K

这里有一个重要的职责边界：

  • Agent Runtime 决定下一步做什么；
  • 模型路由器 决定使用哪个模型或供应商；
  • 工具网关 决定工具调用是否具备权限、是否需要审批；
  • Agent 网关 决定这个租户、主体、会话和 Run 是否允许继续。

如果把规划、工具执行、配额、身份和审计全部塞进一个网关进程，网关将变成不可测试的“超级服务”。更合理的方式是让网关拥有强控制能力，但不拥有业务 Agent 的全部执行逻辑。

OpenAI 当前文档也区分了两类运行方式：Responses API 由应用自行控制模型交互、工具、状态和编排；Agents SDK 则由 SDK 管理 Agent loop、工具调用、handoff、guardrail 和可恢复审批。(developers.openai.com)

请求主流程

sequenceDiagram
    participant U as 调用方
    participant G as Agent 网关
    participant I as 身份服务
    participant P as 策略引擎
    participant Q as 配额服务
    participant R as Agent Runtime
    participant M as 模型供应商
    participant T as 工具服务
    participant A as 审计系统

    U->>G: POST /v1/runs + Idempotency-Key
    G->>I: 验证令牌、主体、租户
    I-->>G: principal + tenant + scopes
    G->>P: 输入策略与能力策略检查
    P-->>G: allow / deny / require_approval
    G->>Q: 预留 Run、请求数和预算
    Q-->>G: reservation_id
    G->>A: 记录 request.accepted
    G->>R: 启动或恢复 Run

    loop Agent loop
        R->>G: 请求模型能力
        G->>Q: 预留本次模型调用预算
        G->>M: 转换后的模型请求
        M-->>G: 模型响应或工具调用
        G->>A: 记录 model.call
        G-->>R: 统一模型结果

        alt 需要工具
            R->>G: tool.call
            G->>P: 工具策略检查
            P-->>G: allow / deny / approval
            G->>T: 执行工具
            T-->>G: 工具结果
            G->>A: 记录 tool.call
        end
    end

    G->>Q: 结算实际用量
    G->>A: 记录 run.completed
    G-->>U: 统一响应或事件流

三、认证不是授权：先确定“谁”，再确定“能做什么”

1. 认证的定义

认证 Authentication 是验证调用方身份的过程。网关需要得到一个可信的主体：

{
  "principal_id": "user_123",
  "principal_type": "user",
  "tenant_id": "tenant_acme",
  "scopes": [
    "agent:run",
    "model:invoke"
  ],
  "token_id": "tok_456"
}

常见凭证包括：

  • OAuth 2.0 Bearer Token；
  • OIDC 身份令牌；
  • 服务账号令牌；
  • mTLS 证书；
  • 工作负载身份；
  • 内部签发的短期 Agent Token。

网关不应直接相信客户端提交的 tenant_id、user_id 或 role。这些字段只能作为业务输入，不能作为身份来源。

错误做法：

POST /v1/runs
Authorization: Bearer user-token
X-Tenant-ID: tenant_other

如果网关直接使用 X-Tenant-ID 查询资源，就可能造成跨租户访问。正确做法是：

token → subject → identity mapping → tenant membership → effective permissions

2. 授权的定义

授权 Authorization 是判断已认证主体能否执行某个动作。

一次授权判断至少包含：

主体 + 租户 + 资源 + 动作 + 运行上下文

可以形式化为：

Allow=Authenticated(principal)∧Member(principal,tenant)∧Scope(principal,action)∧Policy(context,resource,action)Allow = Authenticated(principal) \land Member(principal, tenant) \land Scope(principal, action) \land Policy(context, resource, action)

例如：

user_123
在 tenant_acme
对 run_789
执行 tool.refund

与普通 API 授权不同，Agent 授权还要考虑：

  • 该工具是否允许此 Agent 使用；
  • 工具参数中是否包含敏感数据；
  • 是否超过金额阈值；
  • 是否必须人工审批；
  • 当前 Run 是否已经进入终态；
  • 当前模型是否有权生成该类动作；
  • 工具结果是否可以回流到当前租户的上下文。

3. 凭证传递

模型供应商密钥属于网关或模型适配层，不应下发给终端用户，也不应放进 Agent prompt。

工具凭证则应采用“按调用注入”：

用户令牌
  ↓ 认证
网关内部主体
  ↓ 授权
短期工具访问令牌
  ↓
工具服务

不要把长期数据库密码、云厂商密钥或第三方 OAuth refresh token 放入：

  • 消息正文；
  • 工具参数；
  • tracing attributes；
  • 普通应用日志；
  • Agent 记忆。

四、模型路由：不是按模型名转发，而是约束下的选择问题

1. 模型路由的定义

模型路由 Model Routing 是根据请求属性、租户策略、模型能力、供应商状态和成本约束，选择具体模型部署的过程。

请求中的逻辑模型名不应等同于供应商模型名：

{
  "model": "support.reasoning"
}

网关内部可以映射为：

{
  "logical_model": "support.reasoning",
  "candidates": [
    {
      "provider": "provider_a",
      "model": "model-x",
      "capabilities": ["tool_calling", "json_schema"],
      "regions": ["cn-east", "us-west"],
      "priority": 10
    },
    {
      "provider": "provider_b",
      "model": "model-y",
      "capabilities": ["tool_calling"],
      "regions": ["cn-east"],
      "priority": 20
    }
  ]
}

这样做的原因是：业务代码依赖逻辑能力，而不是依赖某个供应商的字符串。

2. 路由约束

设候选模型集合为 MM，请求约束为 CC，合法模型集合为：

M′={m∈M∣Capability(m)⊇Ccapability∧Region(m)∈Cregion∧Policy(m,tenant)=allow}M' = \{m \in M \mid Capability(m) \supseteq C_{capability} \land Region(m) \in C_{region} \land Policy(m, tenant) = allow\}

如果 M′=∅M' = \varnothing，网关必须返回明确错误，而不是悄悄降级到不满足约束的模型。

例如请求要求：

{
  "requires": {
    "tool_calling": true,
    "structured_output": true,
    "data_region": "cn"
  }
}

某模型支持工具调用，但不支持结构化输出，则不能因为“看起来差不多”而路由过去。

3. 路由评分

过滤出合法模型后，可以使用评分函数：

Score(m)=wqQ(m)−wlLatency(m)−wcCost(m)+waAvailability(m)Score(m) = w_q Q(m) - w_l Latency(m) - w_c Cost(m) + w_a Availability(m)

其中：

  • Q(m)Q(m)：质量或评测分；
  • Latency(m)Latency(m)：历史延迟；
  • Cost(m)Cost(m)：估算成本；
  • Availability(m)Availability(m)：当前健康度；
  • wq,wl,wc,waw_q,w_l,w_c,w_a：租户或产品配置的权重。

这不是说“质量最高的模型永远最好”。一个延迟高、成本高、且不支持目标区域的模型，评分前就应被过滤掉。

4. 失败转移必须考虑副作用

模型调用失败时可以重试，但不能把“所有错误都重试”当作可用性策略。

适合自动重试的情况：

  • 连接超时，且请求尚未被供应商接受；
  • 供应商明确返回临时过载；
  • 读取流中断，且供应商支持安全续接；
  • 请求使用了相同幂等键，供应商保证幂等。

不应盲目重试的情况：

  • 已经产生工具调用；
  • 模型输出要求执行外部副作用；
  • 请求超过上下文限制；
  • 参数校验失败；
  • 内容策略拒绝；
  • 供应商已经返回了完整结果但客户端断线。

尤其要避免下面的失败路径：

模型返回：请执行退款
网关超时
网关重试
模型再次返回：请执行退款
工具执行两次

模型重试与工具执行必须分层。对外部副作用工具，应使用独立的工具幂等键：

tool_idempotency_key =
  hash(tenant_id, run_id, tool_call_id)

五、配额：限制的不只是请求数

1. 配额的定义

配额 Quota 是租户、用户、Agent 或项目在时间窗口、并发度、令牌量、金额和动作次数上的资源上限。

至少应区分：

配额维度 典型含义
请求数 每分钟或每天允许的 Run 数
并发数 同时运行的 Run 数
输入令牌 发送给模型的 token
输出令牌 模型生成的 token
总 token 输入与输出之和
金额预算 账期内最大费用
工具次数 单个 Run 或租户的工具调用数
运行时长 单个 Run 的最长持续时间
记忆容量 可写入租户记忆的大小

仅使用 QPS 限流是不够的。一个请求可能触发几十次模型调用和工具调用。

2. 预留、使用和结算

配额应该采用三阶段模型：

reserve → consume → settle
  • reserve：请求开始时预留可能需要的资源；
  • consume：每次模型调用或工具调用实际消耗资源；
  • settle：Run 结束后用实际值结算，释放未使用预留。

设租户余额为 BB，本次预估成本为 c^\hat{c}，安全余量为 ss：

Reserve  ⟺  B≥c^+sReserve \iff B \ge \hat{c} + s

模型调用结束后，实际成本为 cc：

Bnew=B−cB_{new} = B - c

如果只在结束时扣费，会出现并发超卖：

余额：100
请求 A 读取余额 100，可使用 80
请求 B 同时读取余额 100，也可使用 80
A 成功扣 80
B 成功扣 80
最终余额变成 -60

因此预留必须是原子操作：

UPDATE quota_bucket
SET reserved = reserved + :estimate,
    version = version + 1
WHERE tenant_id = :tenant_id
  AND reserved + consumed + :estimate <= hard_limit;

如果更新行数为 0，说明配额不足。生产实现还应使用数据库事务、乐观锁或原子脚本，不能依赖应用层“先读后写”。

3. 并发配额与令牌配额不同

并发配额限制的是正在运行的工作数量：

active_runs < max_concurrent_runs

令牌配额限制的是模型使用量：

used_input_tokens + used_output_tokens <= token_budget

两者不能互相替代：

  • 一个长时间等待人工审批的 Run 会占用并发，但不消耗很多 token；
  • 一个快速循环调用模型的 Run 会消耗大量 token，但可能只占用一个并发槽。

4. 配额不足时的行为

配额不足不是统一返回 429：

  • 速率超限：429 rate_limited；
  • 账期预算耗尽：429 quota_exhausted；
  • 单 Run 预算不足：409 run_budget_exhausted 或业务定义的配额错误；
  • 模型供应商临时限流：可以返回 503 upstream_rate_limited；
  • 没有满足最低模型能力的降级候选：503 no_eligible_model。

错误响应必须告诉客户端是否可以重试：

{
  "error": {
    "type": "quota_exhausted",
    "code": "tenant_daily_budget_exhausted",
    "message": "租户今日模型预算已用尽",
    "retryable": false,
    "request_id": "req_01"
  }
}

六、策略：把“允许继续”变成可解释的决策

1. 策略的定义

策略 Policy 是对主体、资源、动作和上下文施加约束的规则集合。

策略至少分为五类：

  1. 访问策略：谁能调用哪个 Agent；
  2. 模型策略：允许哪些模型、区域和供应商；
  3. 工具策略：允许调用哪些工具；
  4. 数据策略：哪些数据可以发送给模型或写入记忆；
  5. 运行策略：最大迭代数、最长时间、审批节点和输出限制。

2. 策略检查点

策略不能只在入口检查一次。应在以下节点重复检查：

入口
  → 创建 Run
  → 每次模型调用
  → 每次工具调用
  → 工具结果回流
  → 写记忆
  → 输出给用户

原因是上下文会变化。初始请求可能只是“查询订单”，但模型在后续步骤中可能生成：

tool.refund(amount=9999)

此时需要重新进行工具授权、金额检查和审批判断。

3. Allow、Deny 与 Require Approval

策略结果最好不是二值，而是三值：

allow
deny
require_approval
  • allow：可以继续；
  • deny：立即阻止；
  • require_approval：暂停 Run，等待指定主体批准。

人工审批不是普通错误。它是一个合法的中间状态：

RUNNING
  ↓
WAITING_FOR_APPROVAL
  ↓ approve
RUNNING
  ↓ reject
REJECTED

OpenAI Agents SDK 的当前文档也将 guardrails、人工审批和可恢复的 approval flow 作为 Agent 运行生命周期的一部分。(developers.openai.com)

审批请求必须绑定不可变上下文：

{
  "approval_id": "apr_123",
  "run_id": "run_456",
  "tool_call_id": "tc_789",
  "tool": "refund",
  "arguments_hash": "sha256:...",
  "policy_version": "refund-v3",
  "expires_at": "2026-09-01T12:00:00Z"
}

审批人批准的是这一次具体的工具调用，而不是模糊地批准“这个 Agent 可以退款”。

4. Prompt 不是安全策略

下面的做法不能代替网关策略：

系统提示词：不要给用户退款超过 100 元。

提示词属于模型可见上下文，模型可能误解、遗漏或受到不可信内容影响。金额上限必须在工具执行前由确定性代码检查：

def authorize_refund(ctx, args):
    if ctx.tenant_id != args["tenant_id"]:
        return "deny"
    if args["amount"] > ctx.policy.max_refund_amount:
        return "require_approval"
    return "allow"

模型可以提出动作，但不能成为最终权限裁决者。

Anthropic 对 Agent 的描述强调了工具结果和环境反馈的重要性，并建议设置最大迭代次数、沙箱和适当的 guardrails，以控制自主运行的成本和错误累积。(anthropic.com)

七、审计：记录“发生了什么”，而不是只记录最终答案

1. 审计的定义

审计 Audit 是以可检索、可关联、尽量不可篡改的方式记录关键决策和动作，使系统能够回答：

  • 谁发起了这个 Run？
  • 使用了哪个租户和身份？
  • 最终选择了哪个模型？
  • 为什么选择它？
  • 发生了哪些工具调用？
  • 哪条策略允许或拒绝了动作？
  • 谁批准了高风险操作？
  • 实际消耗了多少 token 和金额？
  • 失败发生在哪一步？
  • 是否进行了重试或故障转移？

2. 事件模型

审计事件应带有统一关联字段：

{
  "event_id": "evt_001",
  "event_type": "tool.call",
  "occurred_at": "2026-09-01T10:00:01.123Z",
  "tenant_id": "tenant_acme",
  "principal_id": "user_123",
  "session_id": "sess_001",
  "run_id": "run_001",
  "model_call_id": "mc_001",
  "tool_call_id": "tc_001",
  "policy_version": "policy-2026-08-20",
  "request_id": "req_001",
  "idempotency_key": "idem_001",
  "decision": "allow",
  "input_hash": "sha256:...",
  "output_hash": "sha256:..."
}

建议将原文和索引分离：

  • 索引保存事件类型、租户、时间、Run ID、状态和哈希；
  • 原文保存到有访问控制和保留期限的对象存储；
  • 敏感字段默认脱敏或只保存哈希；
  • 审计日志本身不能回流给模型。

3. 审计与普通日志的区别

普通日志用于调试：

model request failed

审计事件用于追责和重建：

model.call.failed
tenant=tenant_acme
run=run_001
provider=provider_a
logical_model=support.reasoning
attempt=2
error=upstream_timeout
retryable=true
fallback=provider_b

不要只记录最终响应。最终响应无法解释：

  • 为什么没有调用某个工具；
  • 为什么切换了模型；
  • 为什么发生了两次工具调用；
  • 为什么某次请求被配额拒绝；
  • 为什么 Run 等待审批。

OpenAI Agents SDK 当前文档将 tracing 覆盖到模型调用、工具、Agent、guardrail 和 handoff 等层级，这体现了 Agent 运行必须具备步骤级可观测性，而不是只有 HTTP access log。(developers.openai.com)

八、协议适配：统一语义，不要强行统一所有能力

1. 协议适配的定义

协议适配 Protocol Adaptation 是将不同客户端、Agent Runtime、模型供应商和工具系统之间的请求与响应，转换为网关内部统一语义的过程。

适配分为两层：

语法适配：字段、路径、流格式、错误格式
语义适配：消息角色、工具调用、状态、幂等、取消、审批

只做字段重命名是不够的。

例如，供应商 A 可能返回：

{
  "tool_call": {
    "name": "search_order",
    "arguments": "{\"id\":\"o-1\"}"
  }
}

供应商 B 可能返回：

{
  "content": [
    {
      "type": "tool_use",
      "name": "search_order",
      "input": {
        "id": "o-1"
      }
    }
  ]
}

网关内部应统一成：

{
  "type": "tool_call",
  "id": "tc_001",
  "name": "search_order",
  "arguments": {
    "id": "o-1"
  }
}

2. 内部规范化对象

围绕关联主题，网关内部至少需要以下对象：

{
  "run_id": "run_001",
  "session_id": "sess_001",
  "messages": [
    {
      "id": "msg_001",
      "role": "user",
      "content": [
        {
          "type": "text",
          "text": "查询订单 o-1"
        }
      ],
      "attachments": []
    }
  ],
  "events": [],
  "idempotency_key": "idem_001"
}

注意以下区别：

  • message 是语义内容；
  • event 是运行过程中的状态变化；
  • attachment 是外部文件或媒体引用；
  • error 是机器可处理的失败对象；
  • idempotency_key 是客户端对同一意图的稳定标识。

3. 流式协议不能只转发文本

Agent 流中可能同时存在：

run.started
message.delta
model.call.started
tool.call.created
approval.required
tool.call.completed
message.completed
run.completed

如果网关把所有事件都拼成字符串，客户端将无法知道：

  • 哪段内容是最终回答；
  • 哪段内容是工具调用参数；
  • Run 是否暂停；
  • 是否可以安全重连；
  • 当前事件是否已经被消费。

一个合理的 SSE 事件格式可以是：

event: tool.call.created
data: {"run_id":"run_001","tool_call_id":"tc_001","name":"search_order"}

event: approval.required
data: {"run_id":"run_001","approval_id":"apr_001"}

event: run.completed
data: {"run_id":"run_001","status":"completed"}

每个事件应有单调递增的 sequence：

{
  "event_id": "evt_12",
  "sequence": 12,
  "type": "run.completed"
}

客户端断线后可以携带：

Last-Event-ID: evt_11

网关根据事件日志重放 evt_12 之后的事件，而不是重新执行 Agent。

4. 能力差异必须显式暴露

不同模型协议的能力并不相同。网关不能将缺失能力伪装成已支持。

例如：

{
  "capabilities": {
    "streaming": true,
    "tool_calling": true,
    "structured_output": false,
    "vision": true
  }
}

如果客户端要求：

{
  "requires": ["structured_output"]
}

网关应在路由阶段拒绝不满足能力的候选，而不是调用后再尝试解析。

九、幂等、重试和状态机：Agent 网关的可靠性核心

1. 幂等键的语义

幂等键 Idempotency Key 表示客户端认为“这些重复提交属于同一个意图”。

它必须绑定请求摘要：

fingerprint =
  hash(
    tenant_id,
    principal_id,
    endpoint,
    normalized_request_body
  )

同一个幂等键再次出现时：

  • 请求摘要相同：返回原结果或当前状态；
  • 请求摘要不同：返回冲突；
  • 原请求仍运行：返回 running；
  • 原请求等待审批：返回 waiting_for_approval；
  • 原请求已完成：返回已缓存结果。

错误做法是只用：

idempotency_key = "abc"

而不绑定租户和请求内容。这样可能造成不同租户或不同操作之间的错误复用。

2. Run 状态机

stateDiagram-v2
    [*] --> ACCEPTED
    ACCEPTED --> RUNNING
    ACCEPTED --> REJECTED

    RUNNING --> WAITING_FOR_APPROVAL
    RUNNING --> COMPLETED
    RUNNING --> FAILED
    RUNNING --> CANCELLED
    RUNNING --> TIMED_OUT

    WAITING_FOR_APPROVAL --> RUNNING
    WAITING_FOR_APPROVAL --> REJECTED
    WAITING_FOR_APPROVAL --> TIMED_OUT

    COMPLETED --> [*]
    FAILED --> [*]
    REJECTED --> [*]
    CANCELLED --> [*]
    TIMED_OUT --> [*]

每个状态转换都应该有前置条件：

RUNNING → COMPLETED
条件：Agent Runtime 已产生最终输出，所有必需工具调用已结束

RUNNING → WAITING_FOR_APPROVAL
条件：策略引擎返回 require_approval，且 approval 记录持久化成功

WAITING_FOR_APPROVAL → RUNNING
条件：审批主体有权限，审批内容哈希与原工具调用一致，审批未过期

如果审批后重新构造工具参数，而不是校验参数哈希，就可能发生“审批 A，执行 B”。

3. 故障路径

网关在调用供应商前崩溃

如果还未写入 model.call.started，可以安全重试。

网关在供应商已接受请求后崩溃

此时不能根据本地超时判断供应商没有执行。应使用：

  • 供应商请求 ID 查询；
  • 上游幂等键；
  • 本地状态恢复；
  • 或将 Run 标记为 unknown，进入人工或后台核查。

工具执行成功，但返回网关失败

工具必须使用幂等键。恢复时先查询工具执行记录：

if tool_call_id exists and status=completed:
    reuse stored result
else:
    execute with same idempotency key

审计写入失败

对高风险动作，审计写入失败应阻止工具执行；对低风险模型文本调用，可以采用异步审计，但必须明确“审计缺失窗口”。

十、一个最小可运行的策略与路由核心

下面的代码不调用真实模型，而是展示网关中最关键的确定性部分：租户授权、模型能力过滤、策略决策和幂等冲突。它可以直接用 Python 运行。

from dataclasses import dataclass
from hashlib import sha256
import json


@dataclass(frozen=True)
class Principal:
    principal_id: str
    tenant_id: str
    scopes: frozenset[str]


@dataclass(frozen=True)
class Model:
    logical_name: str
    provider: str
    model_name: str
    capabilities: frozenset[str]
    enabled: bool = True


@dataclass
class IdempotencyRecord:
    fingerprint: str
    status: str
    result: dict | None = None


def fingerprint(tenant_id: str, endpoint: str, body: dict) -> str:
    normalized = json.dumps(body, sort_keys=True, separators=(",", ":"))
    raw = f"{tenant_id}\n{endpoint}\n{normalized}".encode()
    return sha256(raw).hexdigest()


def authorize_run(principal: Principal, requested_tenant: str) -> None:
    if principal.tenant_id != requested_tenant:
        raise PermissionError("cross-tenant access denied")

    if "agent:run" not in principal.scopes:
        raise PermissionError("missing scope: agent:run")


def choose_model(
    candidates: list[Model],
    required_capabilities: set[str],
) -> Model:
    eligible = [
        m for m in candidates
        if m.enabled and required_capabilities <= m.capabilities
    ]

    if not eligible:
        raise RuntimeError("no eligible model")

    # 示例规则：先按供应商优先级，实际系统应从配置中心读取
    priority = {"provider_a": 10, "provider_b": 20}
    return min(eligible, key=lambda m: priority.get(m.provider, 100))


def decide_tool(tool_name: str, arguments: dict, max_refund: int) -> str:
    if tool_name == "refund":
        amount = arguments.get("amount")
        if not isinstance(amount, int) or amount < 0:
            return "deny"
        if amount > max_refund:
            return "require_approval"
    return "allow"


def accept_idempotent_request(
    store: dict[str, IdempotencyRecord],
    key: str,
    request_fingerprint: str,
) -> IdempotencyRecord:
    old = store.get(key)

    if old is not None:
        if old.fingerprint != request_fingerprint:
            raise ValueError("idempotency key reused with different request")
        return old

    record = IdempotencyRecord(
        fingerprint=request_fingerprint,
        status="accepted",
    )
    store[key] = record
    return record


if __name__ == "__main__":
    principal = Principal(
        principal_id="user_123",
        tenant_id="tenant_acme",
        scopes=frozenset({"agent:run"}),
    )

    authorize_run(principal, "tenant_acme")

    candidates = [
        Model(
            logical_name="support.reasoning",
            provider="provider_a",
            model_name="model-x",
            capabilities=frozenset({"tool_calling"}),
        ),
        Model(
            logical_name="support.reasoning",
            provider="provider_b",
            model_name="model-y",
            capabilities=frozenset({"tool_calling", "structured_output"}),
        ),
    ]

    selected = choose_model(
        candidates,
        required_capabilities={"tool_calling", "structured_output"},
    )

    decision = decide_tool(
        "refund",
        {"amount": 500},
        max_refund=100,
    )

    store = {}
    body = {
        "agent": "support",
        "input": "查询订单 o-1",
        "requires": ["structured_output"],
    }
    fp = fingerprint("tenant_acme", "/v1/runs", body)
    record = accept_idempotent_request(store, "idem-001", fp)

    print(selected.provider, selected.model_name)
    print(decision)
    print(record.status)

预期输出：

provider_b model-y
require_approval
accepted

这个示例有三个重要边界：

  1. choose_model 只负责能力约束，不负责真实健康度和成本；
  2. decide_tool 只返回策略结果，不执行工具；
  3. accept_idempotent_request 使用内存字典，只适合演示，生产中必须使用持久化存储和原子写入。

十一、生产诊断：从错误表象定位控制点

1. 请求频繁返回 401

检查顺序：

令牌是否存在
→ 签名是否有效
→ 是否过期
→ issuer/audience 是否匹配
→ token kid 是否能找到
→ 时钟是否漂移

401 表示身份凭证无效或缺失；如果身份有效但没有调用权限，通常应返回 403。

2. 请求返回 403，但用户认为“有权限”

检查：

主体属于哪个 tenant
→ scope 是否包含 agent:run
→ Agent 是否对该租户开放
→ 模型和工具是否被租户策略禁用
→ 是否命中了数据区域或敏感数据策略

不要只检查用户角色名称。最终权限应来自有效策略评估结果。

3. 请求返回 503，但供应商其实正常

可能不是供应商故障，而是：

  • 所有候选模型都不满足能力；
  • 租户区域策略排除了所有候选；
  • 当前模型预算不能覆盖最小请求；
  • 网关熔断器尚未恢复；
  • 模型配置版本没有发布到当前实例。

错误中应区分：

no_eligible_model
upstream_unavailable
upstream_rate_limited
policy_blocked
quota_exhausted

4. 成本突然升高

需要沿着：

tenant → run → model_call → tool_call → retry

聚合，而不是只看供应商账单。

典型原因包括：

  • Agent 迭代上限未设置；
  • 工具错误导致模型循环重试；
  • 上下文每轮完整复制；
  • 路由降级到了高成本模型；
  • 流断线后重复执行；
  • 并行化任务数量未限制；
  • 审批等待期间重复启动 Run。

Anthropic 指出，Agent 相比固定 workflow 往往以更高延迟和成本换取灵活性，并存在错误累积风险；因此 Agent 网关必须同时控制迭代、并发、预算和工具次数。(anthropic.com)

5. 审计日志无法重建一次运行

检查是否缺少：

run_id
model_call_id
tool_call_id
approval_id
policy_version
request_id
attempt
sequence

如果只有：

user asked something
assistant answered something

就无法解释模型路由、工具动作、审批和失败转移。审计字段应在请求进入时生成，并贯穿所有下游调用。

十二、常见错误设计及其反例

错误一：把网关做成“万能 Agent”

网关内部直接实现规划、记忆、工具执行、重试和业务逻辑，会导致：

  • 权限边界模糊；
  • 业务变更需要修改基础设施；
  • 运行状态无法独立恢复；
  • 测试需要启动整个系统。

改进方式是：

网关：身份、租户、策略、配额、审计、协议
Runtime：规划、上下文编排、Agent loop
工具服务：业务动作与数据访问

错误二：只限制入口请求

反例：

POST /runs 时检查一次 token 和预算
之后 Runtime 可以无限调用模型和工具

这相当于把一个长事务当作一个短请求处理。正确做法是每次模型调用、工具调用和敏感输出都重新经过相关策略与配额检查。

错误三：允许模型直接携带供应商密钥

模型输出：

{
  "tool": "http_request",
  "headers": {
    "Authorization": "Bearer long-lived-secret"
  }
}

这会把密钥暴露给上下文、日志、追踪和可能的外部模型。正确做法是工具服务根据主体和租户上下文，从密钥管理系统短期获取凭证。

错误四：用“降级模型”掩盖能力不兼容

如果业务要求结构化输出，降级模型不支持该能力，就不能仅因为它响应更快而使用。降级只能在满足最小能力集合的候选中进行。

错误五：把审批当作同步 HTTP 阻塞

人工审批可能持续数小时。保持 HTTP 连接会占用连接、线程、内存和网关实例。正确方式是：

返回 run.status = waiting_for_approval
客户端订阅事件或轮询
批准后恢复同一个 run

错误六：把原始 prompt 全量写入日志

原始消息可能包含个人信息、商业秘密、访问令牌或文件内容。审计需要可追溯性，但可追溯不等于无条件保存原文。应根据数据分类决定：

原文保存、字段脱敏、摘要保存、哈希保存或完全丢弃

十三、落地顺序：先建立不可绕过的控制点

一个可行的建设顺序是：

第一阶段：统一身份和租户上下文

所有请求在网关内生成不可伪造的：

principal_id
tenant_id
request_id
run_id

同时拒绝客户端直接覆盖这些字段。

第二阶段：建立统一 Run 和事件契约

先把会话、消息、附件、事件、错误、幂等键和 Run 状态定义清楚，再接入多个模型供应商。否则每个供应商都会把状态模型带进业务代码。

第三阶段：加入逻辑模型和能力路由

业务只请求：

support.fast
support.reasoning
embedding.search

网关负责映射供应商模型，并显式校验能力、区域和租户策略。

第四阶段：加入预留式配额

至少实现：

并发预留
模型调用预留
实际用量结算
Run 级预算
租户级预算

没有预留和结算，配额在并发下通常是不正确的。

第五阶段：将工具调用纳入策略和审计

工具调用是 Agent 从“生成文本”变成“执行动作”的边界。必须有：

tool_call_id
arguments_hash
policy_decision
approval_id
execution_status
tool_idempotency_key

第六阶段：实现可恢复事件流和故障处理

客户端断线、网关重启、上游超时和人工审批都不能导致重复副作用。事件重放、Run 恢复和工具幂等要在生产前验证。

Agent 网关的核心价值不是隐藏模型供应商，而是把 Agent 的不确定性限制在可管理边界内：

模型可以提出计划，
Runtime 可以推进步骤，
但网关决定身份、资源、权限、能力、数据和审计边界。

当认证、模型路由、配额、策略、审计和协议适配分别拥有清晰语义，并通过 tenant_id、run_id、model_call_id、tool_call_id 和 policy_version 贯通时，Agent 才能从一次可运行的演示，变成可以计量、诊断、恢复和追责的生产系统。

系列导航与关联阅读

官方资料

本文依据 Agent、模型、协议与框架官方资料重新梳理；正文、示例与生产清单由 WR BLOG 编写。