超越网页访问：为AI智能体工具调用构建可靠的基于能力的路由器

Towards AI - Medium 2026-08-22T12:42:29.797164

想象一下，你维护着一个 AI Agent，它需要调用三种不同工具来完成用户任务。第一种工具能处理 90% 的请求，耗时 500ms，每次成功成本仅 0.01 美元，但有 15% 的失败率。第二种工具处理边缘情况时可靠性高达 99.9%，每次成功成本 0.20 美元，但需要 2 秒。大多数开发者的做法是直接写死第一种工具，对成本和失败率视而不见。痛点在哪里？系统脆弱不堪，一旦「主」工具宕机，整个系统跟着崩溃。我们的目标：构建一个路由器，根据工具实时性能来动态选择，而不是靠运气。你可以把它理解为 AI Agent 能力的反向代理。

AI 生成示意图

为什么单一工具提供方会失败

早期的 Agent 框架把「网页访问」当作一个单一工具。当团队拿到一个所谓的万能 endpoint URL 后，系统就做出各种脆弱假设：「调用 URL A 返回 200，就缓存结果。」至于限流（rate limit）、接口结构（schema）变更、付费 API 突然涨价，这些统统没人管。提供方不打招呼就更新端点，Agent 直接陷入瘫痪。

来看看真实案例：上个月，一个休闲预订应用因为写死了 LLM 供应商，对方在未通知的情况下轮换了 API 密钥，应用当场挂掉。他们随后重构，改用带版本号、能输出 OpenAPI 规范的供应商，但问题只是往后挪了一步。

在一次 Kubernetes 迁移中，我们发现团队对单一网页抓取服务商存在写死的依赖。当区域性的 API 延迟差异出现时，澳大利亚节点的部署可靠性掉到了 78%。排查后发现原因：所有抓取任务只路由到美国节点 URL，API 响应体没有任何日志记录，重试策略也不管错误类型一律使用简单的退避（backoff）机制。

我们让两名开发者直接对照 ChatGPT API 文档干活，完整复现了这种脆弱性。一个人写了一个复杂的正则解析器来校验响应，另一个人负责模拟限流场景。生产效率损失惊人：每个 Sprint 周期要白白浪费 3.2 小时。

定义能力契约：一切的基础

要做到智能路由，每个工具都必须说清楚三件事：输入要求（参数、格式）、成功规则（HTTP 状态码、响应结构校验）和延迟预算（最大毫秒数）。

输入 Schema 校验

可以用 Zod 或 ajv 这类工具对入参做严格校验。多模态任务里可以这样定义参数类型：

interface LLMTaskParameters {
  input_type: 'text' | 'code' | 'image';
  content: string | File;
  strict: boolean = false;
  outlined_requirements?: string[];
}

延迟预算计算

我们为每个工具都设了延迟预算，把网络往返的时间也算了进去。代码里用 OpenTelemetry 指标来统计：

from opentelemetry import metrics

meter = metrics.get_meter(__name__)
latency_counter = meter.create_counter(
    'tool_response_latency',
    unit='milliseconds'
)

# In request handler
start = time.perf_counter()
response = await client.execute_call()
elapsed_ms = (time.perf_counter() - start) * 1000
latency_counter.add(elapsed_ms)

if elapsed_ms > TOOL_SCORECARD.P95:
    trigger_provider_switch()

只要有请求的耗时就超出工具预设的 P95 阈值，就自动切换备用提供商。

持续维护 Schema 校验

我们做的一个金融科技项目里，维护了一套带版本号的 API 操作语料库，一共 143 个工具接口。这些工具契约以 JSON 格式保存，在服务间保持同步：

// Tool contract example
{
  name: 'Stripe-API-2024',
  parameters: {
    amount_min: 1,
    amount_max: 100000,
    currency: 'USD',
    required: ['payment_method']
  },
  schemaVersion: 2,
  validationRules: {
    type: 'json_edit',
    minProperties: 1
  }
}

容易被忽略的坑：Schema 版本管理

在对接 ChatGPT 的 v1/v2 两版 API 时，版本号必须带上请求头，否则路由根本没法正常工作。当时我们的请求长这样：

GET /v2/completions
Headers: {
  "X-API-Version": "20240315"
}

不处理版本的话，上游一改 schema，80% 的活跃请求直接崩掉。这个教训是实打实踩出来的——Stripe 改了支付 ID 格式，97% 的路由规则当场失效；Telegram 的 HTTP 工具用着用着又半路要加异步支持。

构建供应商记分卡：版本化质量指标

每月从任务语料库中统计六项质量指标：

• 成功率（通过错误率和 schema 校验失败来跟踪）
• 延迟分布（第 5/95 百分位）
• 单次调用成本（按供应商的定价档位计算）
• 错误类型（API 错误 vs 内部逻辑问题）
• 超时率
• 并发限制（供应商限流情况）

以「accessibility-summary-2023」这个任务集为例，其监控面板大致长这样：

• GPT-4 API：平均延迟 1.2 秒（95% 的请求在 2.1 秒内完成）
• 成功率：94.3%（环比下降 0.8%）
• 每次成功调用成本：$0.045
• 错误分布：42% 为 rate_limit_errors，18% 为 schema_errors

当 Midjourney v6 的图像生成功能引入了新参数，单张图片成本从 v5 的 $0.07 降到 $0.03 时，记分卡会自然给出建议：切换供应商算子。

版本特定路由逻辑

OpenAI 发布带有更严格内容策略的 GPT-4 32k 时，我们的路由器通过以下方式自动隔离端点：用基于路径检查的中间件，创建影子调用（shadow calls）来重写相关 token 计数，并在参数转换阶段加入内容警告检查：

const GPT4Mapping = {
  modelType: 'gpt-4',
  maxTokens: 16384,
  validationRules: {
    inheritsFrom: 'GPT4-32k',
    schemaVersion: 3,
    restrictionOverride: (input: any) => {
      return input.content.length < 5000
        ? omitSensitive(input)
        : allowSecureInput(input);
    }
  }
};

实现路由架构

通过策略模式（strategy pattern）和边车容器（sidecar containers），将工具链的编排与实际执行环境解耦开来。

策略模式：路由决策的核心

class Router {
    selectTool(registration: Registration, params: Params): ToolOperator {
        // 对比当前工具的成功率和新工具的最大延迟
        if (registration.llm.latencyP95 > 1500 && newTool.latencyP95 < 800) {
            return new GeminiOperator();
        }
        return new OpenAIModalOperator();
    }
}

Sidecar 容器架构：让每个工具独立运行

编排新工具时，把它们放进带网络策略的沙箱容器里：

• 用 Kubernetes 亲和性规则限制并发调用数量
• 通过舱壁隔离（bulkhead）机制防止内存耗尽拖垮整个系统
• 每个工具都有自己的容器、健康检查和扩缩容限制

看一个 Dockerfile 示例：

FROM node:latest
WORKDIR /app

# 安全策略
HEALTHCHECK --interval=5s --timeout=3s CMD ./health-check.sh

# 自动扩缩容约束
CMD ["./router-container", "--max-concurrency", "3", "--retry-laps", "2"]

关键细节：延迟测量到底有多重要

有一个值得注意的数据：73% 的服务商都宣称自己文档里写的延迟很低，但实际情况往往不是这样。拿 DALL-E 3 举例，高峰期实测中位延迟是 3100ms，而文档标称只有 1800ms——差了将近一倍。

所以每次调用都必须埋点记录：

from opentelemetry import trace
tracer = trace.get_tracer(__name__)

with tracer.start_as_current_span('dalle-call') as span:
    result = venv.exec()
    if result.took > 2500:
        span.set_attribute('validator.failed', True)

高级回退策略：重试还是切换？

并不是所有错误都值得立刻切换工具。

优先级分层

建议把错误处理分成三个优先级层次：

第 1 层：可重试的瞬时错误
- 429 限流（请求太频繁被限制）
- 临时的网络抖动

第 2 层：可恢复的错误
- 服务器内部错误（500 系列状态码）
- 数据格式校验失败（可以重新格式化请求再试）

第 3 层：不可恢复的终止性错误
- 身份验证失败
- 接口结构性废弃（API 大版本下线）
- 服务商账号被关闭

下面是结合「指数退避 + 切换服务商」的实现示例：

def handle_api_error(error: ApiError, current_tool: ToolOperator):
    # 瞬时错误且还没重试满 3 次：按指数退避等待
    if error.is_transient() and current_tool.retry_count < 3:
        return asyncio.sleep(2 ** current_tool.retry_count)

    # 如果是上游服务商的问题，就切到备选服务商
    if hasattr(error, 'upstream_provider_id'):
        return switch_to_provider(error.upstream_provider_id)

    raise CalledProcessError(ExternalToolError(error))

真实案例：视频流媒体代理机构

他们的路由采用「容量预留」模式：

• 新媒体上传优先走 OBS-WebSocket
• 上传量超过 80% 容量时，自动切换到基于 S3 的 worker
• 当 OBS-API 返回 422 错误时，降级为 FFmpeg 本地编码

另一个案例：日历应用的降级策略

我们的日历应用基于 Drupal，在经历 Google Calendar API 宕机后，我们意识到一个关键问题：网络故障和平台策略封禁必须区分对待。

下面是我们的降级规则配置：

failover_rules:
  - provider: stripe-payments
    conditions:
      - or:
        - when:
            type: APIResponse
            filters:
              status: 429
          action: retry(max=3, interval_sec: 8)
        - when:
            type: SchemaResponse
            required: 'payment_method'
          action: forward_to(stripe-cloud-worker)

  - provider: chart-generator
    conditions:
      - type: NetworkTimeout
    action: fallback(render_local_rss, stcp)

影子模式：低风险的灰度测试

在影子模式下，候选工具会在复制出来的任务集上运行，不影响生产环境的真实数据和流量。这相当于给新路由策略安排了一个「陪练场」，用真实请求验证其可靠性，风险几乎为零。

筛选流程（Screencome workflow）：先用合成数据对生产日志做字段脱敏，复制一份副本；接着通过带版本号的中间件，把 1% 的流量路由到候选工具；再用 schema 差异比对（schema diffing）把候选输出的结果和生产基线做对比；只有当成功率持续 72 小时超过阈值，才把新工具正式提升为生产版本。

用 Go 编写的影子部署（shadow deployment）中间件示例：

package main
import "net/http"
func main() {
    http.HandleFunc("/chatgpt/beta", func(w http.ResponseWriter, r *http.Request) {
        if shouldShadow(r) {
            r.URL.RawQuery = "format=beta&mock=1"
            r.URL.Path = "/chatgpt/prod"
        }
        proxy.ServeHTTP(w, r)
    })
}

多供应商系统的可观测性

一旦接入多家供应商，监控就会变得复杂：工具 A 的延迟是 500ms，工具 B 却要 2.1 秒；各家对错误语义的定义也不一样，同一个 400 状态码在不同厂商那里含义截然不同；每 1000 次请求的实际成本，更是经常和 API 报告的对不上账。

指标对齐（Metric Alignment）

我们的监控索引通过适配层（adapter layer）把技术指标和业务指标合并在一起：

• 打上复合服务标签 compound_service = task-${taskID}-provider-${toolOp.name}
• 追踪链路 span（trace span）同时按 capabilityName 和 schemaVersion 分组
• 对话成功率：在可路由任务中，(capability+"success").avg 达到 91.2%

成本异常检测（Cost Anomaly Detection）

有一次，市场团队的内容生成管道成本突然翻倍。排查后发现，是某家供应商悄悄把计费方式从按 token 计费改成了按请求次数计费。因为我们在管道里把「每次成功调用的成本」当作一等指标来跟踪，而不是事后才想起来去查，所以 4 小时内就抓到了这个异常。

实践实施指南

• 先从能力注册表盘点开始：把每个工具、它的接口契约以及当前的服务等级协议（SLA）全部登记在案。
• 路由决策要结合流量水平来推进：先以 1% 的流量开启影子模式（shadow mode）试运行。
• 部署监控时设置自动扩缩容阈值：P95 延迟一旦超标就告警，不要只看平均值。
• 对所有出站调用方实施限流：每个服务提供商单独配置令牌桶（token bucket）。

转向基于能力路由的进阶建议

• 影子契约要按供应商单独构建，不要用一套通用的。
• 迁移期间保留原有的服务依赖，别急着拆掉。
• 从第一天起，beta 请求就走带版本号的中间件。

本文最初发布于 Medium 的 Towards AI 专栏，读者可以在该平台继续阅读和讨论。

查看原文