用 AI Gateway 隔离模型剧变:面向智能体系统的演进式架构

2026-07-27 15 预计阅读时间: 1 分钟
来源: infoq.com AI 摘要 Original link

Disclaimer: This article is an AI-assisted summary. Read it together with the original source when precision matters. The summary may omit context, version differences, or edge cases and is not official documentation.

预计阅读时间:11 分钟

传统 API Gateway 建立在两个稳定假设之上:服务会产生相对确定的结果,接口契约也能用简单 schema 描述。智能体 AI 打破了这些假设。一次请求可能触发多轮推理、调用外部工具、修改业务数据,甚至因为模型升级而改变决策路径。企业需要的不只是一个转发模型请求的代理,而是一个能够集中执行护栏、模型路由、智能体身份、动作策略和语义审计的控制平面。

为什么传统网关不够用

传统网关擅长处理 URL、HTTP 方法、令牌、速率限制和结构化请求。面对智能体系统,风险却经常藏在协议层之上。

例如,下面两个请求可能都满足同一份 JSON Schema:

{
  "agent": "support-agent",
  "action": "refund",
  "arguments": {
    "order_id": "ORD-1042",
    "amount": 20
  }
}
{
  "agent": "support-agent",
  "action": "refund",
  "arguments": {
    "order_id": "ORD-1042",
    "amount": 20000
  }
}

从字段类型看,它们完全相同;从业务含义看,第二个动作可能需要人工批准。普通 API Gateway 可以确认调用者拥有有效令牌,却未必知道这个智能体是否可以退款、最大金额是多少,以及此次动作源自哪段用户指令。

AI Gateway 因而需要处理更丰富的上下文:

  • 模型选择:按照任务、成本、延迟和数据敏感度选择模型,而不是把模型名称写死在应用中。
  • 智能体身份:区分最终用户、智能体、委托链和目标工具,避免所有操作共用一个服务账号。
  • 动作授权:在工具真正执行之前,按动作语义、参数和环境决定放行、拒绝或转人工。
  • 输入输出护栏:检查敏感数据、提示注入、危险工具参数以及不符合业务约束的模型输出。
  • 语义审计:除状态码和耗时外,还记录“谁委托了哪个智能体、它为什么选择该工具、策略作出了什么决定”。

把变化收敛到架构接缝

这里的“演进式架构接缝”不是要求企业一次性改造所有平台。更实际的做法,是让业务应用继续依赖稳定的内部接口,再由 AI Gateway 吸收模型供应商、提示模板、策略和审计要求的变化。

用户 / 业务服务
       |
       v
  AI Gateway
  ├─ 身份与委托校验
  ├─ 输入输出护栏
  ├─ 模型与版本路由
  ├─ 工具动作策略
  └─ 语义审计
       |
       +----> 模型 A / 模型 B
       +----> CRM / 支付 / 工单工具

这个边界带来三个直接收益。模型迁移不必扩散到每个业务仓库;策略调整可以集中发布;事故调查也能从统一审计记录还原委托链和工具调用过程。

不过,网关不能替代应用自身的业务校验。支付限额、库存一致性和订单状态机仍应由领域服务强制执行。AI Gateway 提供的是额外的决策与治理层,而不是绕过核心系统约束的新入口。

可以这样实践:构建最小动作策略网关

下面是一个可运行的示例,用 FastAPI 模拟 AI Gateway 的动作控制面。它没有绑定特定模型供应商,重点演示智能体身份、动作白名单、金额阈值和结构化审计。

创建 requirements.txt

fastapi==0.115.0
uvicorn==0.30.6
pydantic==2.9.2

创建 gateway.py

import json
import logging
from typing import Any

from fastapi import FastAPI, Header, HTTPException
from pydantic import BaseModel, Field

app = FastAPI(title="Minimal AI Gateway")
logging.basicConfig(level=logging.INFO, format="%(message)s")
audit = logging.getLogger("semantic-audit")

POLICIES = {
    "support-agent": {
        "allowed_actions": {"lookup_order", "refund"},
        "refund_limit": 100.0,
    },
    "order-reader": {
        "allowed_actions": {"lookup_order"},
        "refund_limit": 0.0,
    },
}


class ActionRequest(BaseModel):
    action: str
    arguments: dict[str, Any] = Field(default_factory=dict)
    user_intent: str
    trace_id: str


@app.post("/v1/actions")
def authorize_action(
    request: ActionRequest,
    x_agent_id: str = Header(...),
    x_user_id: str = Header(...),
):
    policy = POLICIES.get(x_agent_id)
    decision = "deny"
    reason = "unknown_agent"

    if policy and request.action in policy["allowed_actions"]:
        decision = "allow"
        reason = "action_allowed"

        if request.action == "refund":
            amount = float(request.arguments.get("amount", 0))
            if amount <= 0:
                decision, reason = "deny", "invalid_refund_amount"
            elif amount > policy["refund_limit"]:
                decision, reason = "review", "refund_limit_exceeded"

    event = {
        "trace_id": request.trace_id,
        "user_id": x_user_id,
        "agent_id": x_agent_id,
        "action": request.action,
        "arguments": request.arguments,
        "user_intent": request.user_intent,
        "decision": decision,
        "reason": reason,
    }
    audit.info(json.dumps(event, ensure_ascii=False))

    if decision == "deny":
        raise HTTPException(status_code=403, detail=event)

    return event

安装依赖并启动:

python -m venv .venv
. .venv/bin/activate
pip install -r requirements.txt
uvicorn gateway:app --host 127.0.0.1 --port 8000

发送一个超出自动退款限额的动作:

curl -s http://127.0.0.1:8000/v1/actions \
  -H 'Content-Type: application/json' \
  -H 'X-Agent-Id: support-agent' \
  -H 'X-User-Id: user-42' \
  -d '{
    "action": "refund",
    "arguments": {"order_id": "ORD-1042", "amount": 250},
    "user_intent": "客户要求退回重复扣款",
    "trace_id": "trace-2025-001"
  }'

响应中的 decision 应为 review,表示网关没有直接执行退款,而是要求进入人工审批流程。实际部署时应把示例中的请求头替换为经过验证的工作负载身份,例如签名 JWT、mTLS 身份或云平台工作负载凭证;user_intent 也不能被当作可信授权依据。

路由和护栏应配置化,但不能失去版本控制

模型路由规则可以这样实践。以下 YAML 是架构示例,并不依赖某个具体网关产品:

apiVersion: ai-gateway.example/v1
kind: RoutingPolicy
metadata:
  name: enterprise-agent-routing
  version: 7
spec:
  routes:
    - when:
        task: classification
        dataSensitivity: internal
      target: fast-model
      timeoutMs: 3000
    - when:
        task: tool-planning
        dataSensitivity: restricted
      target: approved-private-model
      timeoutMs: 10000
  guardrails:
    input:
      - detectPromptInjection
      - redactSecrets
    output:
      - validateToolArguments
      - blockCredentialDisclosure
  audit:
    record:
      - userId
      - agentId
      - delegatedBy
      - model
      - promptTemplateVersion
      - toolName
      - policyDecision
    storeRawPrompt: false

这里最值得注意的是 versionpromptTemplateVersion。路由、策略和提示模板都会改变系统行为,因此应该像应用代码一样经过评审、测试、灰度发布和回滚。只把这些规则放进一个可在线编辑的后台,会让事故发生时难以回答“当时究竟运行了哪一版策略”。

审计同样需要边界。完整保存提示词和模型响应有助于排障,却可能扩大个人信息、商业秘密和凭证泄漏的范围。更稳妥的默认策略是记录身份、哈希、策略版本、决策原因和工具参数摘要,仅在明确授权的隔离环境中保存原始内容,并配置保留期限。

落地时从高风险动作开始

AI Gateway 不必在第一阶段覆盖所有模型流量。可以优先接管会产生外部副作用的路径,例如退款、转账、账号变更、代码执行和生产环境操作,再逐步纳入普通问答与内容生成。

上线前可以检查这些问题:

  • 每个智能体是否拥有独立、可撤销的身份,而不是共享万能凭证?
  • 工具授权是否检查具体动作与参数,而不只检查 API 访问权?
  • 模型、提示模板和策略版本是否写入审计事件?
  • 高风险动作是否支持拒绝、人工审批和幂等执行?
  • 网关不可用时,系统是明确失败,还是会绕过策略直接调用工具?
  • 原始提示和响应是否经过脱敏,并设置访问控制与保留期限?
  • 核心领域服务是否仍然执行最终业务约束?

AI 的变化速度不会因为企业平台追求稳定而降低。合理的架构选择,是建立一个清晰、可治理的边界:让 AI Gateway 承担频繁变化的模型接入和语义策略,让核心平台继续维护确定的数据约束与业务不变量。这样既能持续替换模型和演进智能体,也不会把每次变化都变成一次全平台改造。


相关推荐