多模型提供商支持

概述

Provider 抽象层与 API 适配器模式的统一接口设计

多模型支持是现代 AI 应用的刚需：不同的任务需要不同能力的模型，不同的成本结构要求灵活的模型切换，而供应商锁定则是企业架构师极力避免的陷阱。Agents SDK 通过 Provider 抽象层为这一问题提供了解决方案。

Provider 模式的核心思想是面向接口编程：Agent 的配置只关心"需要什么能力"（如文本生成、工具调用、结构化输出），而不关心"由谁提供"。具体的模型差异（如 API 格式、认证方式、参数命名）由 Provider 适配器封装和转换。

一个典型的 Provider 适配器需要处理以下差异：

1. API 格式转换：OpenAI 使用 /chat/completions，Anthropic 使用 /messages，Google 使用 /generateContent。适配器需要将统一的内部请求转换为各供应商的原生格式。
2. 参数映射：不同供应商对温度、top_p、max_tokens 等参数的支持程度和默认值不同。适配器需要进行参数校验和默认值填充。
3. 响应解析：各供应商的响应结构不同（如 OpenAI 的 choices[0].message.content vs Anthropic 的 content[0].text）。适配器需要将异构响应统一为内部标准格式。
4. 错误处理：不同供应商的错误码和重试策略不同。适配器需要将异构错误映射为统一的异常类型。
5. 流式协议：SSE 的事件格式在不同供应商间也有差异，适配器需要统一解析逻辑。

从架构设计来看，Provider 抽象层使得模型切换对上层业务完全透明。运营团队可以在不修改任何业务代码的情况下，通过配置中心切换底层模型，实现 A/B 测试、成本优化和供应商灾备。

然而，完全的抽象也带来了代价：特性妥协。某些供应商特有的高级功能（如 OpenAI 的 JSON Mode、Anthropic 的 Computer Use）可能无法被统一抽象层完美支持。在这种情况下，需要在通用接口和供应商特有能力之间做出权衡。

OpenAIResponsesModel、OpenAIChatCompletionsModel、LitellmAdapter：适配 100+ LLM。

正文

相关阅读

• OpenAI Agents SDK 快速上手
• SDK 核心概念一览

参考文档

• OpenAI Agents SDK 官方文档
• GitHub 仓库

完整实战示例：多模型 Provider 路由与适配器工厂

以下示例展示了如何在生产环境中构建一个支持多模型 Provider 的动态路由系统：

import asyncio
import os
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Any, AsyncIterator
from agents import Agent, Runner


@dataclass
class UnifiedRequest:
    messages: list[dict]
    model: str
    temperature: float = 0.7
    max_tokens: int | None = None
    tools: list[dict] | None = None
    stream: bool = False


@dataclass
class UnifiedResponse:
    text: str
    tool_calls: list[dict] | None = None
    usage: dict | None = None
    raw_response: Any = None


class ModelProvider(ABC):
    """模型 Provider 抽象基类。"""
    @abstractmethod
    async def generate(self, request: UnifiedRequest) -> UnifiedResponse:
        pass

    @abstractmethod
    async def generate_stream(self, request: UnifiedRequest) -> AsyncIterator[str]:
        pass


class OpenAIProvider(ModelProvider):
    """OpenAI Provider 适配器。"""
    def __init__(self, api_key: str | None = None):
        self.api_key = api_key or os.getenv("OPENAI_API_KEY")

    async def generate(self, request: UnifiedRequest) -> UnifiedResponse:
        # 简化：实际应调用 openai.AsyncClient
        return UnifiedResponse(
            text="[OpenAI response placeholder]",
            usage={"prompt_tokens": 100, "completion_tokens": 50})

多模型提供商的统一适配架构

下图展示了统一适配层如何屏蔽不同提供商的 API 差异：
mermaid
flowchart TD
A[Agent 请求] --> U[统一适配层]
U --> O[OpenAI 驱动]
U --> AN[Anthropic 驱动]
U --> G[Google 驱动]
U --> L[本地模型驱动]
O --> OAPI[OpenAI API]
AN --> AAPI[Anthropic API]
G --> GAPI[Google API]
L --> LLM[Local LLM Server]
U --> F[功能降级策略]
F --> O
F --> AN
style U fill:#e8d5b5,stroke:#5a4a3a,stroke-width:2px
style F fill:#f4b183,stroke:#5a4a3a

统一适配层的核心职责是：将 Agents SDK 的内部请求格式转换为各提供商的特定格式，并将响应统一转换回 SDK 内部格式。

## 多提供商路由的智能策略

拥有多个模型提供商后，如何智能地路由请求成为关键问题。

**成本优先路由**：

在输出质量可接受的前提下，选择最便宜的提供商。这需要建立各提供商在不同任务上的质量评估数据集。

```python
COST_PER_1K_TOKENS = {
    "openai:gpt-5-nano": 0.05,
    "anthropic:claude-3-haiku": 0.025,
    "google:gemini-1.5-flash": 0.018,
}

def cheapest_provider(task_complexity: str) -> str:
    if task_complexity == "simple":
        return "google:gemini-1.5-flash"
    elif task_complexity == "medium":
        return "anthropic:claude-3-haiku"
    return "openai:gpt-5"

容错路由：

当首选提供商不可用时，自动切换到备用提供商。切换策略应基于故障类型：

• 速率限制（429）：短暂等待后重试，或立即切换到同等级备用提供商
• 服务不可用（503）：立即切换，避免浪费等待时间
• 超时：切换的同时记录该提供商的可用性评分，连续超时则降低其优先级

class ResilientRouter:
    def __init__(self):
        self.providers = {}
        self.health_scores = defaultdict(lambda: 100)

    async def call_with_fallback(self, request, preferred: str):
        providers = sorted(
            self.providers.keys(),
            key=lambda p: self.health_scores[p],
            reverse=True
        )
        for provider in providers:
            try:
                return await self.providers[provider].call(request)
            except Exception as e:
                self.health_scores[provider] -= 10
                logger.warning(f"{provider} 失败: {e}")
        raise AllProvidersFailed()

数据主权路由：

某些数据因合规要求不能离开特定地理区域（如欧盟用户的数据必须留在欧盟）。路由层应根据数据敏感度选择合规的提供商：欧盟数据走欧盟部署的模型，中国数据走中国部署的模型。

多提供商架构的另一个好处是议价能力。当系统支持无缝切换提供商时，你就不被单一供应商锁定，可以在续费时获得更好的价格。建议每季度评估各提供商的价格和性能，动态调整路由权重。统一接口与厂商特性适配方面，不同模型提供商的 API 存在差异，统一封装层需要处理这些差异。适配层的另一个挑战是功能降级。当某个提供商不支持特定功能（如工具调用）时，需要优雅降级到纯文本模式，或切换支持该功能的备用提供商。

常见问题与调试

问题一：不同 Provider 的工具调用格式不兼容

OpenAI 和 Anthropic 对 Function Calling 的格式定义不同，统一适配时可能出现参数解析错误。解决方案：

1. 在 Provider 适配器内部实现双向格式转换，确保无论上层使用哪种格式，底层都能正确解析。
2. 对工具 Schema 进行兼容性检查，发现不支持的特性（如某个 Provider 不支持 anyOf）时提前报错。
3. 在 CI 流水线中增加跨 Provider 的工具调用测试矩阵。

问题二：流式响应的事件格式差异

不同 Provider 的 SSE 事件结构和结束标志不同，客户端可能无法正确解析。应对措施：

1. 在 Provider 适配器层将所有流式响应统一为内部标准格式，客户端只消费标准格式。
2. 实现健壮的事件解析器，能够处理各种边界情况（如缺少 data: 前缀、空行、异常结束）。
3. 为每个 Provider 维护独立的流式响应测试用例。

问题三：Provider 切换后输出质量不一致

同样的提示词在不同模型上可能产生截然不同的输出，导致用户体验不一致。缓解策略：

1. 为每个 Provider 维护独立的提示词模板，针对该模型的特性进行优化。
2. 在切换 Provider 前进行充分的 A/B 测试，评估输出质量的差异。
3. 在后置处理层增加输出标准化逻辑，将不同模型的输出映射到统一的格式。

与其他方案对比

维度	Agents SDK Provider	LiteLLM	LangChain ChatModel
抽象层级	Agent 运行层	API 请求层	组件层
支持供应商数	依赖 SDK 实现	100+	数十种
功能覆盖	Agent 特性优化	通用 API 代理	框架生态集成
部署方式	库内集成	独立代理服务	库内集成

LiteLLM 是目前多模型支持领域最成熟的开源项目，它以一个轻量级代理服务的形式提供了对 100+ 种模型的统一接入，支持负载均衡、Fallback、成本追踪等高级功能。如果你的系统需要对接大量不同的模型供应商，LiteLLM 几乎是不二之选。Agents SDK 的 Provider 抽象更适合已经在 SDK 生态内、只需要在少数几个供应商间切换的团队。LangChain 的 ChatModel 抽象在易用性上做得很好，但在性能和功能完整性上不如 LiteLLM。

适配器工厂与 Fallback 熔断机制的多层架构

Provider 抽象层解决了"统一接口"的问题，但生产环境还需要回答"如何选择 Provider"和"Provider 故障时怎么办"。这就需要在适配器模式之上引入工厂模式（Factory Pattern）和熔断器模式（Circuit Breaker Pattern），构建一个多层 resilient 架构。

flowchart LR
    A[Agent 请求] --> B[Provider 路由器]
    B --> C{健康检查}
    C -->|健康| D[主 Provider]
    C -->|降级| E[备用 Provider 1]
    C -->|熔断| F[备用 Provider 2]
    E --> G{成功率监控}
    F --> G
    D --> G
    G -->|低于阈值| H[触发熔断]
    H --> I[自动恢复探测]
    I --> C
    subgraph 适配器层
        D
        E
        F
    end
    subgraph 监控层
        G
        H
        I
    end

工厂模式负责根据配置和上下文动态创建 Provider 实例。与直接实例化相比，工厂模式将创建逻辑集中管理，支持基于规则的路由（如"中文任务走模型 A，代码任务走模型 B"）和基于负载的调度（如将请求分发到延迟最低的 Provider）。

熔断器模式则解决了级联故障问题。当某个 Provider 的连续错误率超过阈值（如 50% 或连续 5 次失败），熔断器打开，后续请求自动路由到备用 Provider；同时启动定时探测，当 Provider 恢复后自动关闭熔断。这种设计避免了因单个供应商故障导致整个系统瘫痪：

import asyncio
import time
from dataclasses import dataclass
from enum import Enum
from typing import Optional, Callable


class CircuitState(Enum):
    CLOSED = "closed"       # 正常
    OPEN = "open"           # 熔断
    HALF_OPEN = "half_open" # 探测中


@dataclass
class CircuitBreakerConfig:
    failure_threshold: int = 5
    recovery_timeout: float = 30.0
    half_open_max_calls: int = 3


class CircuitBreaker:
    """熔断器：保护 Provider 免于级联故障。"""

    def __init__(self, name: str, config: CircuitBreakerConfig):
        self.name = name
        self.config = config
        self.state = CircuitState.CLOSED
        self.failures = 0
        self.last_failure_time: Optional[float] = None
        self.half_open_calls = 0
        self._lock = asyncio.Lock()

    async def call(self, fn: Callable, *args, **kwargs):
        async with self._lock:
            if self.state == CircuitState.OPEN:
                if time.time() - self.last_failure_time > self.config.recovery_timeout:
                    self.state = CircuitState.HALF_OPEN
                    self.half_open_calls = 0
                else:
                    raise Exception(f"Circuit OPEN for {self.name}")

            if self.state == CircuitState.HALF_OPEN:
                if self.half_open_calls >= self.config.half_open_max_calls:
                    raise Exception(f"Circuit HALF_OPEN limit for {self.name}")
                self.half_open_calls += 1

        try:
            result = await fn(*args, **kwargs)
            async with self._lock:
                self._on_success()
            return result
        except Exception as e:
            async with self._lock:
                self._on_failure()
            raise e

    def _on_success(self):
        self.failures = 0
        if self.state == CircuitState.HALF_OPEN:
            self.state = CircuitState.CLOSED
            self.half_open_calls = 0

    def _on_failure(self):
        self.failures += 1
        self.last_failure_time = time.time()
        if self.failures >= self.config.failure_threshold:
            self.state = CircuitState.OPEN


class ProviderRouter:
    """Provider 路由器：工厂 + 熔断器 + 负载感知。"""

    def __init__(self):
        self.providers = {}  # name -> (provider_instance, circuit_breaker)
        self.latency_window = {}  # name -> deque of recent latencies

    def register(self, name: str, provider, breaker: CircuitBreaker):
        self.providers[name] = (provider, breaker)
        self.latency_window[name] = []

    async def route(self, request, strategy: str = "latency") -> str:
        """根据策略选择最优 Provider。"""
        healthy = []
        for name, (prov, breaker) in self.providers.items():
            if breaker.state != CircuitState.OPEN:
                avg_latency = sum(self.latency_window.get(name, [100])) / max(1, len(self.latency_window.get(name, [100])))
                healthy.append((name, prov, breaker, avg_latency))

        if not healthy:
            raise Exception("No healthy provider available")

        if strategy == "latency":
            healthy.sort(key=lambda x: x[3])
        elif strategy == "round_robin":
            # 简化：实际应维护轮询计数器
            pass

        return healthy[0][0], healthy[0][1], healthy[0][2]

    async def generate_with_fallback(self, request, primary: str):
        """带熔断和 Fallback 的生成调用。"""
        names = list(self.providers.keys())
        # 将主 Provider 放第一位
        order = [primary] + [n for n in names if n != primary]
        last_error = None
        for name in order:
            if name not in self.providers:
                continue
            prov, breaker = self.providers[name]
            start = time.time()
            try:
                result = await breaker.call(prov.generate, request)
                # 记录延迟
                self.latency_window.setdefault(name, []).append(time.time() - start)
                if len(self.latency_window[name]) > 10:
                    self.latency_window[name].pop(0)
                return result
            except Exception as e:
                last_error = e
                continue
        raise last_error or Exception("All providers failed")

上述架构体现了**桥接模式（Bridge Pattern）**的核心思想：抽象（路由逻辑）与实现（具体 Provider）完全分离，两者可以独立演化和扩展。当新增一个模型供应商时，只需实现 Provider 接口并注册到路由器，无需修改任何路由逻辑；当优化路由策略时，也无需改动任何 Provider 适配器代码。这种解耦使得团队可以并行推进基础设施优化和模型接入工作。

除了延迟和熔断，成本也是 Provider 路由的重要考量维度。不同模型的定价差异巨大（如 GPT-5 与 GPT-5-nano 的价差可达 30 倍以上），对于非关键路径的请求（如内部日志分析、测试环境），路由到廉价模型可以显著降低运营成本。建议为每个 Provider 维护"能力-成本"评分矩阵，在请求级别标注所需的能力等级（如"需要推理"、"仅生成"），路由器据此在成本和质量之间做出权衡。

在实际部署中，建议将熔断器的状态、延迟和成本指标暴露为 Prometheus 指标，接入 Grafana 大盘观察。同时，设置合理的降级策略：当所有 Provider 都不可用时，返回预设的友好提示（如"服务暂时繁忙，请稍后再试"），而非直接抛出异常导致前端崩溃。对于关键业务场景，还应配置兜底模型（如本地轻量模型），确保极端情况下仍能提供基础服务。定期演练故障切换流程，验证 Fallback 链路的可靠性。

生产环境部署与性能优化

Provider 灰度发布的实践要点

将本章节的技术应用到生产环境时，首要考虑的是稳定性与可观测性。建议采用渐进式 rollout 策略：先在开发环境验证核心逻辑，再迁移到预发布环境进行压力测试，最后才全量上线。部署过程中应配置完善的日志收集和指标监控，确保任何问题都能被快速发现和定位。

具体来说，需要在基础设施层面做好以下准备：容器资源限制（CPU/内存）、网络策略配置（防火墙规则、服务网格）、持久化存储选型（SSD vs 标准盘）以及备份恢复方案。对于高可用要求严格的场景，建议部署多实例并配置负载均衡，避免单点故障导致服务中断。

模型降级率的关键指标

监控是生产系统的生命线。针对本章节涉及的功能，建议重点跟踪以下指标：请求延迟（P50/P95/P99）、错误率（4xx/5xx/超时）、吞吐量（QPS/TPS）以及资源利用率（CPU/内存/磁盘/网络）。这些指标应接入统一的监控大盘，并设置合理的告警阈值。

除了基础指标，还应关注业务层面的指标。例如功能成功率、用户满意度、成本消耗趋势等。通过将技术指标与业务指标关联分析，可以更准确地评估系统改进的实际价值，避免陷入"为了优化而优化"的陷阱。

统一计费系统的架构考量

随着业务规模增长，单实例部署很快会成为瓶颈。扩展性设计应在项目初期就纳入考量，而非事后补救。水平扩展通常比垂直扩展更具成本效益，但也引入了分布式系统的复杂性（数据一致性、服务发现、负载均衡等）。

在扩展过程中，建议遵循"无状态优先"原则：将状态外置到独立的存储层（如 Redis、PostgreSQL），使计算层可以随时水平扩容。对于无法避免的状态（如会话、缓存），采用分布式一致性协议或最终一致性模型来管理。定期进行容量规划和压力测试，确保系统在流量峰值时仍能稳定运行。

运维团队的协作建议

技术方案的落地离不开高效的团队协作。建议建立清晰的运维手册（Runbook），涵盖常见故障的诊断步骤、应急处理流程和升级路径。同时，通过定期的复盘会议，将线上事故转化为团队的学习素材，持续完善系统的健壮性。

在工具链方面，推荐将本章节的配置和脚本纳入版本控制（Git），并使用 Infrastructure as Code（IaC）工具（如 Terraform、Ansible）管理基础设施变更。这不仅能提高部署效率，还能确保环境一致性，减少"在我机器上能跑"的问题。

多模型切换时的输出一致性是常见痛点。不同模型对同一提示词的理解可能存在偏差。建议在切换模型后，使用固定的回归测试集验证输出质量，确保用户体验不下降。