一、引言：API的“巴别塔”困境

做AI应用开发的人迟早会面对这个问题：今天用GPT写代码生成，明天想试试Claude处理长文本，后天又接到国产模型DeepSeek的性价比需求。每次切换都要重写一套调用逻辑，改到怀疑人生。

2024年以前，我们还在争论“哪个模型最强”。到了2026年，成熟的架构师都知道：没有最强，只有最适合。GPT适合代码生成和通用任务，Claude在长文本理解上表现稳健，Gemini在超长上下文中优势明显，DeepSeek则以性价比取胜。

问题在于，OpenAI、Anthropic、Google、DeepSeek……每家都有自己的SDK、鉴权方式和参数结构。OpenAI喜欢用messages=[{"role": "user"}]，Claude的结构完全不同，Gemini又有自己的格式。如果业务代码里充斥着针对不同厂商的if-else，维护成本会随模型数量线性增长。

本文的核心目标：设计一套统一抽象层，让切换模型只改一行配置，代码零改动。

---

二、核心设计模式

解决这个问题的标准答案是适配器模式 + 工厂模式的组合。

2.1 适配器模式：抹平接口差异

适配器模式的核心思想是：定义一个统一接口，为每个具体模型实现这个接口，内部完成协议转换。

统一接口（Application层）
        ↓
┌───────┼───────┬──────────┐
↓       ↓       ↓          ↓
OpenAI   Claude  Gemini   DeepSeek
Adapter  Adapter Adapter   Adapter
↓       ↓       ↓          ↓
OpenAI   Claude  Gemini   DeepSeek
API      API     API       API

无论底层是哪家厂商，应用层看到的都是同一个方法签名：chat(messages, model, **kwargs) -> Response。

2.2 工厂模式：运行时动态创建

工厂模式负责根据配置动态创建对应的适配器实例。业务代码只需要传入provider名称，工厂返回对应的实现。

2.3 两套配置：各管各的事

更精细的设计将配置拆分为两层：

• ProviderSpec（静态注册表）：模型本身的特性，如是否支持联网、是否支持深度搜索、默认超时——这些跟着代码版本走，更新频率低
• ProviderConf（运行时配置）：API密钥、访问地址、具体模型名——这些经常变，通过JSON/环境变量动态加载

用模型名做桥梁，把Spec和Conf串起来，创建出可用的实例。

---

三、代码实战：Python版统一抽象层

3.1 环境准备

pip install openai anthropic google-generativeai requests python-dotenv

环境变量（.env）：

# OpenAI
OPENAI_API_KEY=sk-xxxxx
OPENAI_BASE_URL=https://api.openai.com/v1

# Anthropic (Claude)
ANTHROPIC_API_KEY=sk-ant-xxxxx

# Google (Gemini)
GOOGLE_API_KEY=xxxxx

# DeepSeek (兼容OpenAI协议)
DEEPSEEK_API_KEY=sk-xxxxx
DEEPSEEK_BASE_URL=https://api.deepseek.com/v1

3.2 定义统一接口

from abc import ABC, abstractmethod
from typing import List, Dict, Optional, Any
from dataclasses import dataclass, field
from enum import Enum


@dataclass
class Message:
    """统一的消息格式"""
    role: str  # system | user | assistant
    content: str


@dataclass
class LLMResponse:
    """统一的响应格式"""
    content: str
    model: str
    usage: Optional[Dict[str, int]] = None
    raw_response: Any = None  # 保留原始响应便于调试


class LLMProvider(ABC):
    """统一接口抽象基类"""
    
    @abstractmethod
    def chat(
        self,
        messages: List[Message],
        model: Optional[str] = None,
        temperature: float = 0.7,
        max_tokens: int = 1024,
        stream: bool = False,
        **kwargs
    ) -> LLMResponse:
        """统一的对话调用接口"""
        pass
    
    @abstractmethod
    def chat_stream(
        self,
        messages: List[Message],
        model: Optional[str] = None,
        temperature: float = 0.7,
        max_tokens: int = 1024,
        **kwargs
    ):
        """统一的流式对话接口（生成器）"""
        pass

3.3 各厂商适配器实现

import os
from openai import OpenAI as OpenAIClient
import anthropic
import google.generativeai as genai


class OpenAIAdapter(LLMProvider):
    """OpenAI适配器"""
    
    def __init__(self, api_key: str = None, base_url: str = None):
        self.client = OpenAIClient(
            api_key=api_key or os.getenv("OPENAI_API_KEY"),
            base_url=base_url or os.getenv("OPENAI_BASE_URL")
        )
    
    def chat(self, messages, model=None, temperature=0.7, max_tokens=1024, 
             stream=False, **kwargs):
        # 转换为OpenAI格式
        openai_messages = [{"role": m.role, "content": m.content} for m in messages]
        
        response = self.client.chat.completions.create(
            model=model or "gpt-4o",
            messages=openai_messages,
            temperature=temperature,
            max_tokens=max_tokens,
            stream=stream,
            **kwargs
        )
        
        if stream:
            return response  # 返回流对象，由调用方处理
        
        return LLMResponse(
            content=response.choices[0].message.content,
            model=response.model,
            usage=response.usage.model_dump() if response.usage else None,
            raw_response=response
        )
    
    def chat_stream(self, messages, model=None, temperature=0.7, 
                    max_tokens=1024, **kwargs):
        stream = self.chat(
            messages, model=model, temperature=temperature,
            max_tokens=max_tokens, stream=True, **kwargs
        )
        for chunk in stream:
            if chunk.choices[0].delta.content:
                yield chunk.choices[0].delta.content


class ClaudeAdapter(LLMProvider):
    """Anthropic Claude适配器"""
    
    def __init__(self, api_key: str = None):
        self.client = anthropic.Anthropic(
            api_key=api_key or os.getenv("ANTHROPIC_API_KEY")
        )
    
    def _messages_to_claude(self, messages: List[Message]) -> tuple:
        """将统一消息格式转换为Claude格式"""
        system_prompt = None
        claude_messages = []
        
        for m in messages:
            if m.role == "system":
                system_prompt = m.content
            else:
                claude_messages.append({
                    "role": m.role,  # user or assistant
                    "content": m.content
                })
        
        return system_prompt, claude_messages
    
    def chat(self, messages, model=None, temperature=0.7, max_tokens=1024,
             stream=False, **kwargs):
        system_prompt, claude_messages = self._messages_to_claude(messages)
        
        response = self.client.messages.create(
            model=model or "claude-3-5-sonnet-20241022",
            messages=claude_messages,
            system=system_prompt,
            temperature=temperature,
            max_tokens=max_tokens,
            stream=stream,
            **kwargs
        )
        
        if stream:
            return response
        
        return LLMResponse(
            content=response.content[0].text,
            model=response.model,
            usage={"input_tokens": response.usage.input_tokens,
                   "output_tokens": response.usage.output_tokens},
            raw_response=response
        )
    
    def chat_stream(self, messages, model=None, temperature=0.7,
                    max_tokens=1024, **kwargs):
        stream = self.chat(
            messages, model=model, temperature=temperature,
            max_tokens=max_tokens, stream=True, **kwargs
        )
        for chunk in stream:
            if chunk.type == "content_block_delta":
                yield chunk.delta.text


class GeminiAdapter(LLMProvider):
    """Google Gemini适配器"""
    
    def __init__(self, api_key: str = None):
        genai.configure(api_key=api_key or os.getenv("GOOGLE_API_KEY"))
        self.client = genai.GenerativeModel
    
    def _messages_to_gemini(self, messages: List[Message]) -> tuple:
        """转换为Gemini格式"""
        system_prompt = None
        gemini_messages = []
        
        for m in messages:
            if m.role == "system":
                system_prompt = m.content
            else:
                gemini_messages.append({"role": m.role, "parts": [m.content]})
        
        return system_prompt, gemini_messages
    
    def chat(self, messages, model=None, temperature=0.7, max_tokens=1024,
             stream=False, **kwargs):
        system_prompt, gemini_messages = self._messages_to_gemini(messages)
        
        model_name = model or "gemini-1.5-pro"
        model_instance = self.client(
            model_name,
            system_instruction=system_prompt,
            generation_config={
                "temperature": temperature,
                "max_output_tokens": max_tokens,
            }
        )
        
        # Gemini的chat接口需要用History组装
        chat = model_instance.start_chat()
        for m in gemini_messages:
            chat.history.append(m)
        
        response = chat.send_message(
            gemini_messages[-1]["parts"][0] if gemini_messages else "",
            stream=stream
        )
        
        if stream:
            return response
        
        return LLMResponse(
            content=response.text,
            model=model_name,
            usage={"prompt_tokens": response.usage_metadata.prompt_token_count
                   if hasattr(response, 'usage_metadata') else None},
            raw_response=response
        )
    
    def chat_stream(self, messages, model=None, temperature=0.7,
                    max_tokens=1024, **kwargs):
        stream = self.chat(
            messages, model=model, temperature=temperature,
            max_tokens=max_tokens, stream=True, **kwargs
        )
        for chunk in stream:
            if chunk.text:
                yield chunk.text


class DeepSeekAdapter(OpenAIAdapter):
    """
    DeepSeek适配器：直接复用OpenAI协议
    因为DeepSeek API完全兼容OpenAI格式，只需修改base_url
    """
    
    def __init__(self, api_key: str = None, base_url: str = None):
        super().__init__(
            api_key=api_key or os.getenv("DEEPSEEK_API_KEY"),
            base_url=base_url or os.getenv("DEEPSEEK_BASE_URL")
        )

3.4 工厂模式：统一创建入口

from enum import Enum
from typing import Optional


class ProviderType(Enum):
    OPENAI = "openai"
    CLAUDE = "claude"
    GEMINI = "gemini"
    DEEPSEEK = "deepseek"


class LLMProviderFactory:
    """工厂：根据provider类型创建对应适配器"""
    
    _providers = {}  # 缓存已创建的实例
    
    @classmethod
    def create_provider(
        cls,
        provider_type: ProviderType,
        api_key: Optional[str] = None,
        base_url: Optional[str] = None,
        **kwargs
    ) -> LLMProvider:
        """
        创建或获取provider实例
        支持缓存，避免重复初始化
        """
        cache_key = f"{provider_type.value}:{api_key}:{base_url}"
        
        if cache_key in cls._providers:
            return cls._providers[cache_key]
        
        if provider_type == ProviderType.OPENAI:
            provider = OpenAIAdapter(api_key=api_key, base_url=base_url)
        elif provider_type == ProviderType.CLAUDE:
            provider = ClaudeAdapter(api_key=api_key)
        elif provider_type == ProviderType.GEMINI:
            provider = GeminiAdapter(api_key=api_key)
        elif provider_type == ProviderType.DEEPSEEK:
            provider = DeepSeekAdapter(api_key=api_key, base_url=base_url)
        else:
            raise ValueError(f"Unsupported provider: {provider_type}")
        
        cls._providers[cache_key] = provider
        return provider

3.5 统一调用客户端

class UnifiedLLMClient:
    """
    统一LLM客户端
    业务层只和这个类打交道，不感知底层厂商
    """
    
    def __init__(self, provider_type: ProviderType, **config):
        self.provider = LLMProviderFactory.create_provider(provider_type, **config)
        self.default_model = config.get("model")
    
    def chat(
        self,
        prompt: str,
        model: Optional[str] = None,
        system_prompt: Optional[str] = None,
        temperature: float = 0.7,
        max_tokens: int = 1024,
        **kwargs
    ) -> str:
        """同步对话"""
        messages = []
        if system_prompt:
            messages.append(Message(role="system", content=system_prompt))
        messages.append(Message(role="user", content=prompt))
        
        response = self.provider.chat(
            messages=messages,
            model=model or self.default_model,
            temperature=temperature,
            max_tokens=max_tokens,
            **kwargs
        )
        return response.content
    
    def chat_stream(
        self,
        prompt: str,
        model: Optional[str] = None,
        system_prompt: Optional[str] = None,
        temperature: float = 0.7,
        max_tokens: int = 1024,
        **kwargs
    ):
        """流式对话"""
        messages = []
        if system_prompt:
            messages.append(Message(role="system", content=system_prompt))
        messages.append(Message(role="user", content=prompt))
        
        for chunk in self.provider.chat_stream(
            messages=messages,
            model=model or self.default_model,
            temperature=temperature,
            max_tokens=max_tokens,
            **kwargs
        ):
            yield chunk

---

四、高级特性

4.1 Fallback自动降级

生产环境中，主模型可能因限流、超时、配额不足等原因不可用。需要设计主备降级机制。

import logging
from typing import List, Tuple


class FallbackLLMClient:
    """带降级能力的客户端"""
    
    def __init__(self, fallback_chain: List[Tuple[ProviderType, dict]]):
        """
        fallback_chain: [(ProviderType, config_dict), ...]
        按优先级排列
        """
        self.fallback_chain = fallback_chain
        self._clients = {}
    
    def _get_client(self, provider_type: ProviderType, config: dict):
        if provider_type not in self._clients:
            self._clients[provider_type] = UnifiedLLMClient(
                provider_type, **config
            )
        return self._clients[provider_type]
    
    def chat_with_fallback(self, prompt: str, **kwargs) -> str:
        """依次尝试，直到成功"""
        last_error = None
        
        for provider_type, config in self.fallback_chain:
            try:
                client = self._get_client(provider_type, config)
                logging.info(f"Using provider: {provider_type.value}")
                return client.chat(prompt, **kwargs)
            except Exception as e:
                logging.warning(f"Provider {provider_type.value} failed: {e}")
                last_error = e
                continue
        
        raise RuntimeError(f"All providers failed. Last error: {last_error}")


# 使用示例
fallback_client = FallbackLLMClient([
    (ProviderType.OPENAI, {"model": "gpt-4o"}),
    (ProviderType.CLAUDE, {"model": "claude-3-5-sonnet-20241022"}),
    (ProviderType.DEEPSEEK, {"model": "deepseek-chat"}),
])

response = fallback_client.chat_with_fallback("介绍一下RAG技术")

4.2 配置驱动的模型路由

更进一步的方案是将模型选择逻辑抽到配置层，支持按任务类型动态路由。

# config/models.yaml
routing_rules:
  - task_type: code_generation
    primary: openai
    model: gpt-4o
    fallback: deepseek
  - task_type: long_document
    primary: claude
    model: claude-3-5-sonnet
    fallback: gemini
  - task_type: cost_sensitive
    primary: deepseek
    model: deepseek-chat
    fallback: openai
    model: gpt-3.5-turbo

4.3 错误透传与可观测性

错误处理的一个常见误区是把所有厂商错误统一转成同一种格式。更好的做法是尽量透传原始错误，让上层自行判断如何处理。

@dataclass
class LLMError:
    kind: str  # http_error | rate_limit | auth_error | ...
    status: Optional[int] = None
    body: Optional[str] = None
    raw: Optional[Any] = None

同时建议在网关层统一记录：

• 每次调用的模型、token消耗、延迟
• 状态码和失败原因
• 费用估算
• fallback触发次数

---

五、工程化落地建议

如果你所在的团队准备统一模型调用，建议按以下步骤推进：

1. 把模型名抽到配置中心，不要硬编码在业务代码里
2. 统一错误码和重试策略，不同供应商的错误格式不同，应用层不应感知这些差异
3. 加观测指标，至少记录模型、token、延迟、状态码、费用估算、fallback次数
4. 建立模型评测集，每次切模型之前用固定样本跑一遍，不要只凭感觉换

---

六、小结

本文从“API的巴别塔困境”出发，系统介绍了大模型统一抽象层的设计与实现：

• 适配器模式：为每个厂商实现统一接口，抹平协议差异
• 工厂模式：运行时动态创建适配器，业务代码零改动切换模型
• Fallback机制：主模型不可用时自动降级，保障服务可用性
• 可观测性：统一记录调用指标，支撑成本优化和故障排查

这套方案已在多个生产环境中落地，核心价值是让业务代码与模型厂商解耦——当DeepSeek R1发布、GPT-5.5面世、Claude 4.7上线时，团队只需新增一个适配器，已有的业务逻辑无需任何修改。

关于多智能体场景下的统一调用、细粒度的成本控制策略、或LangChain等框架的统一抽象集成，欢迎在评论区交流讨论。