AI API版本管理与向后兼容实战：从OpenAI API废弃风波中学到的血泪教训

# api_version_router.py - AI API版本路由中间件
from typing import Dict, Any, Optional
from dataclasses import dataclass, field
from enum import Enum
import httpx
import logging

logger = logging.getLogger(__name__) class Provider(Enum): OPENAI = "openai" ANTHROPIC = "anthropic" GEMINI = "gemini" @dataclass
class ModelEndpoint: """模型端点配置""" model_id: str # 平台侧的模型ID api_version: str # API版本号（可选） base_url: str # API基础URL max_tokens: int = 4096 cost_per_1k: float = 0.0  # 每1K token成本 is_active: bool = True @dataclass
class VersionConfig: """业务版本配置""" version: str # 业务版本号，如 "v1", "v2" provider: Provider endpoint: ModelEndpoint fallback_endpoint: Optional[ModelEndpoint] = None request_transform: Optional[callable] = None response_transform: Optional[callable] = None class APIVersionRouter: """ AI API版本路由器 业务代码通过业务版本号调用，路由器负责映射到具体的模型和API版本 """ def __init__(self): self._versions: Dict[str, VersionConfig] = {} self._default_version: Optional[str] = None def register(self, config: VersionConfig, is_default: bool = False): """注册一个业务版本""" self._versions[config.version] = config if is_default: self._default_version = config.version logger.info(f"设置默认版本: {config.version}") def resolve(self, version: Optional[str] = None) -> VersionConfig: """解析版本号，返回对应的配置""" v = version or self._default_version if v not in self._versions: raise ValueError(f"未知版本: {v}，可用版本: {list(self._versions.keys())}") return self._versions[v] async def call( self, messages: list, version: Optional[str] = None, **kwargs ) -> Dict[str, Any]: """ 统一调用入口 业务代码只需要传 messages 和可选的版本号 """ config = self.resolve(version) # 应用请求转换 if config.request_transform: messages, kwargs = config.request_transform(messages, kwargs) try: result = await self._call_provider(config, messages, **kwargs) except Exception as e: logger.error(f"主端点调用失败: {e}") # 尝试fallback if config.fallback_endpoint: logger.info("切换到fallback端点") result = await self._call_provider(config, messages, use_fallback=True, **kwargs) else: raise # 应用响应转换 if config.response_transform: result = config.response_transform(result) return result async def _call_provider( self, config: VersionConfig, messages: list, use_fallback: bool = False, **kwargs ) -> Dict[str, Any]: """实际调用AI API""" endpoint = config.fallback_endpoint if use_fallback else config.endpoint headers = {"Content-Type": "application/json"} payload = {"model": endpoint.model_id, "messages": messages, **kwargs} if config.provider == Provider.OPENAI: headers["Authorization"] = f"Bearer {kwargs.get('api_key')}" url = f"{endpoint.base_url}/chat/completions" elif config.provider == Provider.ANTHROPIC: headers["x-api-key"] = kwargs.get('api_key', '') headers["anthropic-version"] = endpoint.api_version url = f"{endpoint.base_url}/messages" # Anthropic的请求格式不同，需要转换 payload = self._transform_to_anthropic(payload) else: raise ValueError(f"不支持的provider: {config.provider}") async with httpx.AsyncClient(timeout=30.0) as client: resp = await client.post(url, headers=headers, json=payload) resp.raise_for_status() return resp.json() # ===== 初始化路由器 =====
router = APIVersionRouter()

# v1: 使用gpt-3.5-turbo（已废弃，保留用于向后兼容）
router.register(VersionConfig( version="v1", provider=Provider.OPENAI, endpoint=ModelEndpoint( model_id="gpt-4o-mini",  # 自动升级到替代模型 api_version="", base_url="https://api.openai.com/v1", cost_per_1k=0.15, ), fallback_endpoint=ModelEndpoint( model_id="claude-3-5-haiku-20241022", api_version="2023-06-01", base_url="https://api.anthropic.com/v1", cost_per_1k=0.25, ),
), is_default=True)

# v2: 使用gpt-4o
router.register(VersionConfig( version="v2", provider=Provider.OPENAI, endpoint=ModelEndpoint( model_id="gpt-4o", api_version="", base_url="https://api.openai.com/v1", cost_per_1k=2.5, ),
))

# v3: 使用Claude Sonnet 4
router.register(VersionConfig( version="v3", provider=Provider.ANTHROPIC, endpoint=ModelEndpoint( model_id="claude-sonnet-4-20250514", api_version="2023-06-01", base_url="https://api.anthropic.com/v1", cost_per_1k=3.0, ),
))

# ===== 业务代码调用 =====
# 旧客户端自动走v1（已静默升级到gpt-4o-mini）
result = await router.call(messages=[{"role": "user", "content": "Hello"}])

# 新客户端指定v3
result = await router.call( messages=[{"role": "user", "content": "Hello"}], version="v3", api_key="sk-xxx"
)

# 错误示范：temperature变成必填参数
async def chat_completion(messages: list, temperature: float): # 旧客户端不传temperature直接报错 ...

# 正确示范：temperature有默认值
async def chat_completion( messages: list, temperature: float = 0.7,  # 旧客户端兼容 max_tokens: int = 4096, top_p: float = 1.0, # 新参数，默认值保证向后兼容 response_format: Optional[dict] = None, tools: Optional[list] = None,
): ...

# 在响应头中标记废弃
if requested_version in DEPRECATED_VERSIONS: response.headers["Deprecation"] = "true" response.headers["Sunset"] = "2026-09-01T00:00:00Z" response.headers["Link"] = ( '<https://docs.example.com/migration-v1-to-v2>' '; rel="successor-version"' ) logger.warning(f"客户端 {client_id} 仍在使用废弃版本 {requested_version}")

from dataclasses import dataclass

@dataclass
class FeatureFlags: enable_streaming: bool = False enable_tool_calling: bool = False enable_json_mode: bool = False max_context_tokens: int = 8192

# 从配置中心读取flags
flags = load_feature_flags_from_redis("api:v2:features")

if flags.enable_tool_calling and "tools" in request: # 新功能逻辑 result = await call_with_tools(request)
else: # 旧逻辑，保证兼容 result = await call_legacy(request)

# 统一内部消息格式
@dataclass
class UnifiedMessage: role: str # "user" | "assistant" | "system" content: str metadata: dict = field(default_factory=dict)

# OpenAI适配器
class OpenAIAdapter: def to_provider(self, messages: list[UnifiedMessage]) -> list[dict]: return [{"role": m.role, "content": m.content} for m in messages] def from_provider(self, response: dict) -> str: return response["choices"][0]["message"]["content"]

# Anthropic适配器
class AnthropicAdapter: def to_provider(self, messages: list[UnifiedMessage]) -> list[dict]: # Anthropic的system消息需要单独提取 system_msg = "" chat_msgs = [] for m in messages: if m.role == "system": system_msg = m.content else: chat_msgs.append({"role": m.role, "content": m.content}) return chat_msgs, system_msg def from_provider(self, response: dict) -> str: return response["content"][0]["text"]

# 客户端请求
headers = { "Accept-Version": "v1, v2; q=0.9, v3; q=0.8", "X-API-Version": "v2"  # 优先使用v2
}

# 服务端协商逻辑
def negotiate_version(accept_header: str, preferred: str) -> str: # 1. 优先使用客户端明确指定的版本 if preferred in ACTIVE_VERSIONS: return preferred # 2. 按q值降序选择最佳匹配 versions = parse_accept_header(accept_header) for v, q in sorted(versions, key=lambda x: x[1], reverse=True): if v in ACTIVE_VERSIONS: return v # 3. 回退到最新版本 return LATEST_VERSION