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