1. 项目概述：为什么我们需要可移植的API代码？

在AI应用开发的第一线待了十几年，我亲眼见证了模型生态从一家独大到百花齐放的变迁。几年前，你可能只需要对接OpenAI的API，写几行代码就能让应用跑起来。但现在呢？客户可能要求同时支持Claude、Gemini，甚至是一些开源的本地模型。更常见的情况是，为了成本、性能或功能冗余，一个应用内部需要根据不同的任务，动态切换调用不同的模型。这时候，如果你当初的代码是硬编码了某个特定厂商的SDK调用方式，那改起来简直就是一场灾难——每个接口的入参格式、错误处理、流式响应解析都不同，牵一发而动全身。

这个项目要解决的，就是这个日益普遍的痛点： 构建一套可移植的API调用代码 。它的核心目标不是简单地封装几个SDK，而是设计一个抽象层，让你写的业务逻辑（比如“生成一段营销文案”）与底层具体是哪个模型（OpenAI的GPT-4、Anthropic的Claude 3、Google的Gemini）彻底解耦。今天用GPT-4，明天因为预算或政策原因要换到Claude，你只需要改一行配置，或者甚至由系统根据策略自动选择，业务代码完全不用动。

这听起来像是又一个“设计模式”教学，但实操中的坑远比理论多。比如，不同模型的上下文长度（context window）单位可能不同（有的按token，有的按字符），计费方式各异，流式响应（streaming）的数据格式更是千差万别。一个健壮的可移植层，必须优雅地处理这些差异，而不是让调用方去操心。接下来，我会拆解如何从零开始构建这样一个系统，分享我趟过的坑和总结出的最佳实践。

2. 核心架构设计：抽象与适配的艺术

构建可移植API代码的核心思想，与计算机科学中经典的“适配器模式”（Adapter Pattern）和“依赖倒置原则”（Dependency Inversion Principle）一脉相承。我们不应该让高层模块（业务逻辑）依赖低层模块（具体的模型SDK），而是让两者都依赖于一个抽象的接口。

2.1 定义统一的抽象接口（Interface）

这是整个架构的基石。你需要定义一组所有模型提供商都必须实现的方法。这个接口应该足够通用，以覆盖主流模型的核心功能，但又不能过于宽泛而失去实际意义。通常，一个聊天补全（Chat Completion）接口是起点。

from abc import ABC, abstractmethod
from typing import AsyncIterator, List, Optional, Dict, Any

class LLMProvider(ABC):
    """大语言模型提供商的抽象基类。"""
    
    @abstractmethod
    async def chat_completion(
        self,
        messages: List[Dict[str, str]],
        model: str,
        temperature: float = 0.7,
        max_tokens: Optional[int] = None,
        stream: bool = False,
        **kwargs
    ) -> Any:
        """
        统一的聊天补全接口。
        
        Args:
            messages: 消息列表，格式通常为 [{"role": "user", "content": "Hello"}]
            model: 模型标识符（如 'gpt-4', 'claude-3-opus-20240229'）
            temperature: 温度参数，控制随机性
            max_tokens: 生成的最大token数
            stream: 是否使用流式响应
            **kwargs: 其他模型特定的参数
        
        Returns:
            如果 stream=False，返回一个包含完整响应的对象。
            如果 stream=True，返回一个异步迭代器，逐块产生响应。
        """
        pass
    
    @abstractmethod
    def get_model_list(self) -> List[str]:
        """获取该提供商支持的模型列表。"""
        pass
    
    @abstractmethod
    def calculate_cost(self, input_tokens: int, output_tokens: int, model: str) -> float:
        """根据输入输出token数计算本次调用的成本（美元）。"""
        pass

为什么这么设计？

• 异步优先 ( async ) ：现代网络IO密集型应用普遍采用异步，避免阻塞，提升并发能力。即使你当前用同步，预留异步接口也为未来扩展留有余地。
• messages 参数标准化 ：尽管底层API格式可能不同（如OpenAI用 messages ，Anthropic用 messages 但结构略有差异），我们在抽象层统一成最常见的列表字典格式。适配器的职责就是做转换。
• **kwargs 的妙用 ：这是处理不同模型独有参数的关键。比如，OpenAI有 top_p ， Anthropic有 top_k 。调用方可以通过 kwargs 传入，由具体适配器决定是否使用及如何映射。
• 成本计算 ：将成本计算封装在提供商内部，是因为不同模型的定价表（每百万token输入/输出价格）不同，甚至同一提供商不同模型价格也不同。统一接口方便做预算监控和成本优化。

2.2 实现具体适配器（Adapter）

有了接口，接下来就是为每个支持的模型实现一个适配器类。每个适配器内部封装了对该模型原生SDK的调用和参数转换。

import openai
from .base import LLMProvider

class OpenAIProvider(LLMProvider):
    def __init__(self, api_key: str, base_url: Optional[str] = None):
        self.client = openai.AsyncOpenAI(api_key=api_key, base_url=base_url)
        self._model_list = ["gpt-4-turbo", "gpt-4", "gpt-3.5-turbo"] # 可以动态获取
        
    async def chat_completion(self, messages, model, temperature=0.7, max_tokens=None, stream=False, **kwargs):
        # 将通用参数映射到OpenAI特定参数
        params = {
            "model": model,
            "messages": messages, # 格式一致，无需转换
            "temperature": temperature,
            "max_tokens": max_tokens,
            "stream": stream,
        }
        # 处理可能传入的OpenAI特有参数，如 top_p, presence_penalty
        if 'top_p' in kwargs:
            params['top_p'] = kwargs['top_p']
        # 注意：如果传入了Anthropic的top_k，这里可以选择忽略或记录警告
        
        try:
            if stream:
                response = await self.client.chat.completions.create(**params)
                # 返回一个处理过的异步生成器，统一输出格式
                async def _stream_generator():
                    async for chunk in response:
                        if chunk.choices[0].delta.content is not None:
                            yield chunk.choices[0].delta.content
                return _stream_generator()
            else:
                response = await self.client.chat.completions.create(**params)
                # 返回一个统一结构的对象，例如我们自定义的 SimpleCompletionResponse
                return SimpleCompletionResponse(
                    content=response.choices[0].message.content,
                    model=response.model,
                    usage=response.usage
                )
        except openai.APIError as e:
            # 将提供商特定的异常转换为通用异常
            raise LLMProviderError(f"OpenAI API调用失败: {e}") from e
    
    def calculate_cost(self, input_tokens: int, output_tokens: int, model: str) -> float:
        # 简化示例，实际应从配置或常量中读取价格表
        price_map = {
            "gpt-4-turbo": {"input": 0.01, "output": 0.03}, # 美元/千token
            "gpt-3.5-turbo": {"input": 0.001, "output": 0.002},
        }
        if model not in price_map:
            return 0.0
        cost = (input_tokens / 1000) * price_map[model]["input"] + (output_tokens / 1000) * price_map[model]["output"]
        return cost

适配器实现要点：

1. 异常转换 ：必须捕获原生SDK的异常（如 openai.APIError , anthropic.APIError ），并转换为你自己定义的通用异常（如 LLMProviderError ）。这样上层业务逻辑只需要处理一种异常类型。
2. 流式响应统一 ：流式处理是最容易出问题的地方。不同API返回的chunk对象结构天差地别。适配器必须将它们解析并统一成简单的字符串（或你定义的增量数据对象） yield 出去。确保调用方拿到的是一个纯净的内容流。
3. 参数过滤与映射 ：小心处理 kwargs 。对于本提供商不支持的参数，应该记录警告或直接忽略，而不是报错，以保证接口的灵活性。

2.3 设计工厂与路由层（Factory & Router）

当有多个提供商时，我们需要一个中心化的地方来创建和管理它们。这就是工厂模式的应用场景。更进一步，可以引入路由层，实现基于规则或策略的自动模型选择。

class LLMProviderFactory:
    """LLM提供商工厂，负责创建和缓存提供商实例。"""
    
    _providers: Dict[str, LLMProvider] = {}
    
    @classmethod
    def get_provider(cls, provider_name: str, **config) -> LLMProvider:
        if provider_name not in cls._providers:
            if provider_name == "openai":
                cls._providers[provider_name] = OpenAIProvider(api_key=config["api_key"])
            elif provider_name == "anthropic":
                cls._providers[provider_name] = AnthropicProvider(api_key=config["api_key"])
            elif provider_name == "gemini":
                cls._providers[provider_name] = GeminiProvider(api_key=config["api_key"])
            # 可以轻松扩展新的提供商
            elif provider_name == "azure_openai":
                cls._providers[provider_name] = AzureOpenAIProvider(
                    api_key=config["api_key"],
                    endpoint=config["endpoint"],
                    api_version=config["api_version"]
                )
            else:
                raise ValueError(f"不支持的提供商: {provider_name}")
        return cls._providers[provider_name]

class LLMRouter:
    """智能路由层。根据策略自动选择最合适的提供商和模型。"""
    
    def __init__(self, strategy: str = "fallback"):
        self.strategy = strategy
        self.factory = LLMProviderFactory
        
    async def chat_completion(self, messages, **kwargs):
        # 策略1: 故障转移（Fallback）
        if self.strategy == "fallback":
            providers = kwargs.pop("provider_priority", ["openai", "anthropic"])
            last_error = None
            for provider_name in providers:
                try:
                    provider = self.factory.get_provider(provider_name, **kwargs)
                    # 可以从kwargs中提取模型，或根据策略选择默认模型
                    model = kwargs.get('model', provider.get_model_list()[0])
                    return await provider.chat_completion(messages=messages, model=model, **kwargs)
                except (LLMProviderError, ConnectionError) as e:
                    last_error = e
                    continue # 尝试下一个提供商
            raise LLMProviderError(f"所有提供商均失败: {last_error}")
        
        # 策略2: 基于负载或成本的路由
        # 策略3: 基于任务类型的路由（如创意写作用Claude，代码生成用GPT）
        # ... 可以根据业务需求扩展

路由策略的价值 ：

• 故障转移 ：当主提供商（如OpenAI）宕机或达到速率限制时，自动无缝切换到备用提供商（如Anthropic），极大提升系统可用性。
• 成本优化 ：根据任务的复杂度和对模型能力的要求，路由到性价比更高的模型。例如，简单的分类任务用便宜的GPT-3.5，复杂的分析任务再用GPT-4。
• 性能优化 ：根据地理位置路由到延迟最低的API端点（如果提供商支持多区域）。

3. 关键实现细节与魔鬼陷阱

架构搭好了，但魔鬼藏在细节里。下面这些点，是决定你的抽象层是否真正“可用”和“好用”的关键。

3.1 输入输出的标准化与验证

不同模型对输入消息的历史长度、角色定义（ system , user , assistant ）的容忍度不同。Anthropic对 system 提示词有单独字段，而OpenAI是放在 messages 列表里的一个角色为 system 的消息。你的抽象层需要做一层转换。

def normalize_messages_for_provider(messages: List[Dict], provider: str) -> Any:
    """将统一的消息格式转换为特定提供商所需的格式。"""
    if provider == "anthropic":
        # Anthropic 需要将 system 消息分离出来
        system_messages = [msg["content"] for msg in messages if msg["role"] == "system"]
        other_messages = [msg for msg in messages if msg["role"] != "system"]
        system_text = "\n".join(system_messages) if system_messages else None
        # 转换角色名称：OpenAI的'assistant'对应Anthropic的'assistant'
        # 注意：Anthropic的消息结构是列表，每个元素是包含'role'和'content'的字典，与OpenAI类似但API参数不同
        # 这里简化处理，实际调用时，system参数单独传，messages传其他消息
        return system_text, other_messages
    elif provider in ["openai", "azure_openai", "gemini"]:
        # OpenAI、Gemini等格式基本兼容
        return messages
    else:
        # 对于未知提供商，原样返回，由适配器处理
        return messages

注意 ：这里有一个大坑。Anthropic Claude 3系列模型虽然也接受类似OpenAI的 messages 数组，但其内部的 system 字段是顶级参数。而像Claude 2等旧版本可能格式差异更大。 务必查阅你所用模型版本的最新官方API文档 ，适配器代码需要紧跟API更新。

3.2 流式响应（Streaming）的异构处理

流式响应对于提升用户体验（尤其是生成长文本时）至关重要，但各家的实现五花八门。

• OpenAI ：返回的每个chunk是一个 ChatCompletionChunk 对象，内容在 choices[0].delta.content 。
• Anthropic ：返回的SSE（Server-Sent Events）流，每个事件有不同的类型（如 message_start , content_block_delta ），文本内容在 delta.text 。
• Gemini ：Google的流式响应又是另一套逻辑。

你的适配器必须将这些全部归一化为一个简单的 字符串迭代器 。更高级的做法是定义一个统一的增量响应事件对象，包含类型（如“content”、“function_call”、“finish”）和数据。

# 在OpenAI适配器中的流式处理部分细化
async def _stream_generator():
    async for chunk in response:
        # 检查是否有内容增量
        if chunk.choices and chunk.choices[0].delta.content is not None:
            yield chunk.choices[0].delta.content
        # 可以在这里检查其他增量，如工具调用（function call）
        # if chunk.choices[0].delta.tool_calls:
        #     yield 处理工具调用的统一事件...
        # 检查是否结束
        if chunk.choices and chunk.choices[0].finish_reason:
            yield {"type": "finish", "reason": chunk.choices[0].finish_reason}

实操心得 ：处理流式响应时，一定要考虑 网络中断 和 缓冲区 。客户端可能在任意时刻断开连接，你的服务器端代码需要妥善关闭底层的HTTP连接，避免资源泄漏。此外，对于快速消费的流，可以考虑加入一个小的异步缓冲区来平滑数据推送。

3.3 上下文窗口与Token计算的迷雾

这是成本控制和功能正确性的核心。不同模型上下文窗口大小不同（从4K到200K不等），且计量单位可能是token（如OpenAI）或字符（某些开源模型）。更复杂的是，不同模型的分词器（Tokenizer）不同，同样的文本算出来的token数差异很大。

解决方案 ：

1. 在抽象层使用“字符数”或“估计token数”作为通用度量 ：提供一个 estimate_tokens 的通用方法，内部可以调用各提供商SDK自带的分词器（如OpenAI的 tiktoken ， Anthropic的官方库）进行相对准确的估算。对于没有官方分词器的，使用一个近似算法（如按空格和常见标点分割）。
2. 在配置中明确每个模型的上下文上限 ：在路由或调用前，根据估算的token数判断是否会超限，如果超限，可以自动触发“总结前文”、“滑动窗口”或直接拒绝请求并给出友好提示。
3. 成本计算依赖准确的Usage数据 ：最准确的token数来自API响应中的 usage 字段（如OpenAI返回的 prompt_tokens , completion_tokens ）。你的 calculate_cost 方法应该优先使用这个数据。如果没有（如某些开源API），再回退到估算。

3.4 配置管理与密钥安全

多个提供商意味着多套API密钥和配置（端点URL、超时时间、重试策略等）。硬编码在代码里是绝对不可取的。

• 使用配置文件或环境变量 ：推荐使用 pydantic-settings 或 python-dotenv 来管理配置。为每个提供商定义一个配置类。
• 密钥安全 ：永远不要将密钥提交到版本控制系统。使用密钥管理服务（如AWS Secrets Manager, HashiCorp Vault）或在部署时通过环境变量注入。
• 超时与重试 ：网络请求必须设置合理的超时（如30秒）和重试逻辑（针对可重试的错误，如网络抖动、速率限制429）。可以使用 tenacity 或 backoff 库优雅地实现指数退避重试。

from pydantic_settings import BaseSettings
from typing import Dict, Any

class LLMSettings(BaseSettings):
    openai_api_key: str = ""
    openai_base_url: Optional[str] = None
    openai_timeout: int = 30
    anthropic_api_key: str = ""
    anthropic_timeout: int = 30
    # 重试配置
    max_retries: int = 3
    retry_delay: float = 1.0
    
    class Config:
        env_file = ".env"

4. 完整工作流与代码集成示例

让我们看一个从配置到调用的完整示例，模拟一个简单的问答服务。

步骤1：定义配置与工厂

# config.py
import os
from typing import Dict, Any
from dataclasses import dataclass

@dataclass
class ProviderConfig:
    name: str
    api_key: str
    base_url: Optional[str] = None
    default_model: Optional[str] = None
    timeout: int = 30

class LLMConfigManager:
    def __init__(self):
        self.providers: Dict[str, ProviderConfig] = self._load_from_env()
    
    def _load_from_env(self) -> Dict[str, ProviderConfig]:
        configs = {}
        # OpenAI
        if key := os.getenv("OPENAI_API_KEY"):
            configs["openai"] = ProviderConfig(
                name="openai",
                api_key=key,
                base_url=os.getenv("OPENAI_BASE_URL"),
                default_model=os.getenv("OPENAI_DEFAULT_MODEL", "gpt-4-turbo")
            )
        # Anthropic
        if key := os.getenv("ANTHROPIC_API_KEY"):
            configs["anthropic"] = ProviderConfig(...)
        # ... 加载其他提供商
        return configs
    
    def get_provider_config(self, name: str) -> ProviderConfig:
        if name not in self.providers:
            raise ValueError(f"未配置的提供商: {name}")
        return self.providers[name]

# factory.py
from .providers import OpenAIProvider, AnthropicProvider, GeminiProvider
from .config import LLMConfigManager

class LLMProviderFactory:
    _instances: Dict[str, LLMProvider] = {}
    config_manager = LLMConfigManager()
    
    @classmethod
    def get_provider(cls, provider_name: str) -> LLMProvider:
        if provider_name not in cls._instances:
            config = cls.config_manager.get_provider_config(provider_name)
            if provider_name == "openai":
                cls._instances[provider_name] = OpenAIProvider(
                    api_key=config.api_key,
                    base_url=config.base_url,
                    timeout=config.timeout
                )
            elif provider_name == "anthropic":
                cls._instances[provider_name] = AnthropicProvider(...)
            # ... 其他提供商
            else:
                raise ValueError(f"不支持的提供商: {provider_name}")
        return cls._instances[provider_name]

步骤2：实现业务服务层

# service.py
from .factory import LLMProviderFactory
from .router import LLMRouter
import logging

logger = logging.getLogger(__name__)

class AIChatService:
    def __init__(self, routing_strategy: str = "fallback"):
        self.router = LLMRouter(strategy=routing_strategy)
        # 或者直接使用特定提供商
        # self.provider = LLMProviderFactory.get_provider("openai")
    
    async def ask_question(self, question: str, context: Optional[str] = None, **kwargs) -> str:
        """一个简单的问答方法。"""
        messages = []
        if context:
            messages.append({"role": "system", "content": f"基于以下背景信息回答问题：\n{context}"})
        messages.append({"role": "user", "content": question})
        
        try:
            # 使用路由器，自动选择提供商
            response = await self.router.chat_completion(
                messages=messages,
                temperature=0.5,
                max_tokens=500,
                provider_priority=["openai", "anthropic"], # 故障转移顺序
                **kwargs
            )
            # 假设router返回的是我们自定义的SimpleCompletionResponse
            return response.content
        except LLMProviderError as e:
            logger.error(f"AI服务调用失败: {e}")
            # 可以返回一个友好的默认回复，或者抛出业务异常
            return "抱歉，AI服务暂时不可用，请稍后再试。"
    
    async def ask_question_stream(self, question: str, **kwargs) -> AsyncIterator[str]:
        """流式问答。"""
        messages = [{"role": "user", "content": question}]
        try:
            stream = await self.router.chat_completion(
                messages=messages,
                stream=True,
                **kwargs
            )
            async for chunk in stream:
                # chunk 已经是适配器统一后的字符串
                yield chunk
        except LLMProviderError as e:
            logger.error(f"AI流式服务调用失败: {e}")
            yield "【服务中断】"

步骤3：在Web框架（如FastAPI）中集成

# main.py (FastAPI示例)
from fastapi import FastAPI, HTTPException
from fastapi.responses import StreamingResponse
import asyncio
from .service import AIChatService

app = FastAPI()
chat_service = AIChatService(routing_strategy="fallback")

@app.post("/v1/chat/completions")
async def chat_completion(request: ChatRequest):
    """兼容OpenAI格式的聊天接口，但内部可路由到任何模型。"""
    try:
        if request.stream:
            async def event_generator():
                async for chunk in chat_service.ask_question_stream(
                    question=request.messages[-1].content,
                    model=request.model,
                    temperature=request.temperature
                ):
                    # 将统一的内容块，封装成OpenAI兼容的SSE格式
                    data = json.dumps({
                        "choices": [{
                            "delta": {"content": chunk},
                            "index": 0,
                            "finish_reason": None
                        }]
                    })
                    yield f"data: {data}\n\n"
                yield "data: [DONE]\n\n"
            return StreamingResponse(event_generator(), media_type="text/event-stream")
        else:
            answer = await chat_service.ask_question(
                question=request.messages[-1].content,
                model=request.model,
                temperature=request.temperature
            )
            return {
                "choices": [{
                    "message": {"role": "assistant", "content": answer},
                    "index": 0,
                    "finish_reason": "stop"
                }],
                "model": request.model
            }
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

这个工作流展示了从配置管理、提供商抽象、智能路由到最终集成的完整链条。你的业务代码（如 AIChatService ）完全不知道底层调用的是GPT还是Claude，它只与统一的接口交互。

5. 常见问题、调试与优化实录

在实际部署和运行中，你会遇到各种各样的问题。下面是我总结的一些典型场景和解决方案。

5.1 速率限制（Rate Limiting）与优雅降级

所有云API都有速率限制。当你的请求量变大时，必然会碰到 429 Too Many Requests 错误。

应对策略：

1. 指数退避重试 ：在适配器或HTTP客户端层实现重试逻辑。对于429错误，等待一段时间后重试。使用 tenacity 库可以很优雅地实现。

   from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
   import openai
   
   @retry(
       stop=stop_after_attempt(3),
       wait=wait_exponential(multiplier=1, min=4, max=10),
       retry=retry_if_exception_type(openai.RateLimitError)
   )
   async def call_openai_with_retry(client, **params):
       return await client.chat.completions.create(**params)
   

2. 请求队列与限流 ：在应用层面，使用像 asyncio.Semaphore 或更高级的库（如 aiometer ）来控制并发请求数，避免瞬间爆发请求冲垮API。
3. 多密钥轮询 ：如果一个提供商有多个API密钥（如团队不同成员的账号），可以实现一个简单的轮询或随机选择机制，将负载分散到不同密钥上。
4. 快速失败与降级 ：当重试多次仍失败，或所有备用提供商都不可用时，应有明确的降级策略。比如，返回一个缓存的通用答案，或者将任务放入队列稍后重试，并立即给用户一个“正在处理”的反馈。

5.2 响应格式不一致与解析错误

即使抽象层做了归一化，不同模型生成的内容风格和结构也可能差异很大。比如，你要求模型以JSON格式回答，GPT-4可能严格遵守，而其他模型可能在外面包裹了Markdown代码块或附加了解释文字。

解决方案：

• 后处理层 ：在拿到模型响应后，增加一个后处理步骤。例如，使用正则表达式尝试从响应文本中提取JSON部分。

  import json
  import re
  
  def extract_json_from_response(text: str):
      # 尝试匹配 ```json ... ``` 代码块
      match = re.search(r'```json\n(.*?)\n```', text, re.DOTALL)
      if match:
          text = match.group(1)
      # 尝试直接解析
      try:
          return json.loads(text.strip())
      except json.JSONDecodeError:
          # 如果失败，记录日志并返回原始文本或错误结构
          logger.warning(f"无法从模型响应中解析JSON: {text[:200]}...")
          return {"error": "invalid_format", "raw_text": text}
  

• 在System Prompt中强化指令 ：在请求的system消息中，非常明确地指定输出格式，并举例说明。可以要求模型“必须只输出纯JSON，不要有任何额外的解释、标记或代码块”。

5.3 超时与长文本生成的不确定性

生成长文本（如一篇千字文章）时，即使设置了 max_tokens ，实际生成时间也可能远超预期，导致HTTP超时。

应对策略：

1. 分而治之 ：对于超长文本生成，不要依赖模型一次性完成。设计“大纲生成 -> 段落展开 -> 合并润色”的多步流水线，每一步都在可控的token数和时间内完成。
2. 设置合理的超时时间 ：根据任务类型动态设置超时。简单QA可以设5-10秒，长文本生成可以设60-120秒，并在UI上给用户“正在生成，请耐心等待”的提示。
3. 使用服务器发送事件（SSE） ：对于流式响应，超时问题不突出，因为数据是分块到达的。即使整体生成时间长，用户也能持续看到进度，体验更好。确保你的前端和后端都正确支持SSE。

5.4 监控、日志与可观测性

当系统依赖多个外部API时，监控变得至关重要。你需要知道：每个模型的调用成功率、延迟分布、token消耗和成本。

必须记录的关键指标：

• 调用日志 ：提供商、模型、请求token数（估算）、响应token数、耗时、是否成功、错误信息。
• 性能指标 ：使用像Prometheus这样的工具暴露指标，如 llm_api_duration_seconds （直方图）、 llm_api_calls_total （计数器，按提供商、模型、状态打标签）。
• 成本跟踪 ：每次成功调用后，根据 usage 数据累加成本。可以按项目、用户或API密钥进行聚合，定期报告。

# 在适配器的chat_completion方法中集成日志和监控
async def chat_completion(self, ...):
    start_time = time.time()
    request_id = str(uuid.uuid4())
    logger.info(f"[{request_id}] 开始调用 {self.provider_name}.{model}，估算输入token: {estimated_tokens}")
    
    try:
        response = await self._make_api_call(...)
        end_time = time.time()
        duration = end_time - start_time
        
        # 记录成功日志
        logger.info(f"[{request_id}] 调用成功，耗时{duration:.2f}s， 使用token: {response.usage}")
        # 记录指标
        metrics.observe_api_call(
            provider=self.provider_name,
            model=model,
            duration=duration,
            success=True,
            input_tokens=response.usage.prompt_tokens,
            output_tokens=response.usage.completion_tokens
        )
        # 计算并记录成本
        cost = self.calculate_cost(response.usage.prompt_tokens, response.usage.completion_tokens, model)
        metrics.record_cost(cost)
        
        return response
    except Exception as e:
        end_time = time.time()
        logger.error(f"[{request_id}] 调用失败，耗时{end_time-start_time:.2f}s， 错误: {e}")
        metrics.observe_api_call(..., success=False)
        raise

5.5 版本管理与向后兼容

AI模型的API迭代很快。今天用的 gpt-4-turbo-preview 明天可能就变成了 gpt-4-turbo ，参数也可能有细微调整。

最佳实践：

1. 将模型标识符作为配置 ：不要在代码里硬编码模型字符串（如 "gpt-4" ）。应该从配置文件中读取，或者有一个模型别名的映射表。这样当模型升级时，只需更新配置。

   MODEL_ALIAS = {
       "smart": "gpt-4-turbo", # 业务别名 -> 实际模型名
       "fast": "gpt-3.5-turbo",
       "long": "claude-3-sonnet-20240229",
   }
   

2. 适配器版本化 ：如果某个提供商的API发生重大不兼容更新（比如v1到v2），可以考虑创建新的适配器类（如 OpenAIProviderV2 ），并通过配置开关逐步迁移业务流量，而不是直接修改原有适配器，避免线上服务中断。
3. 定期测试与更新 ：将模型API的调用集成到CI/CD的测试套件中，定期运行，确保所有适配器仍然工作正常。订阅各厂商的更新公告，提前规划适配工作。

构建一套可移植的AI API调用代码，初期会花费一些设计和解耦的功夫，但从中长期来看，它带来的灵活性、可维护性和系统韧性是巨大的。当你的产品需要快速接入一个新的明星模型，或者因为某个主流服务不稳定而需要切换时，你会庆幸当初做了这个决定。这套架构不仅适用于商业API，同样可以扩展到集成本地部署的Llama、Qwen等开源模型，为你的AI应用打造一个真正模型无关的坚实底座。