AI SDK 向后兼容性保障方案 #

# 用户代码：显式锁定模型版本
client = AIClient(
    model="gpt-4",
    model_version="2024-04-09",  # 锁定快照，不受后续更新影响
    api_version="2024-04-01"     # 锁定 API 行为
)

# 当需要升级时，用户主动变更
client.update_model("gpt-4o", model_version="2024-05-13")

class BaseChatClient(ABC):
    """统一接口——无论底层是 OpenAI、Anthropic 还是本地模型"""
    
    @abstractmethod
    def chat(self, messages: list[Message], **kwargs) -> ChatResponse:
        pass

class OpenAIAdapter(BaseChatClient):
    def __init__(self, api_version="2024-04-01"):
        self.api_version = api_version
        self._adapter = self._load_adapter(api_version)
    
    def chat(self, messages, **kwargs):
        # 统一接口，内部适配不同 API 版本
        if self.api_version >= "2024-06-01":
            return self._call_v2(messages, **kwargs)
        else:
            return self._call_v1(messages, **kwargs)
    
    def _call_v1(self, messages, **kwargs):
        # 旧版 API 调用
        return openai.chat.completions.create(messages=messages, **kwargs)
    
    def _call_v2(self, messages, **kwargs):
        # 新版 API 调用（可能有不同的参数结构）
        return openai.chat.completions.create(
            messages=messages,
            response_format=kwargs.pop("format", None),
            **kwargs
        )

# 契约定义（简化版 Pact 风格）
contract = Pact(
    consumer="my-sdk",
    provider="openai-api",
).upon_receiving("chat completion request").with_request(
    method="POST",
    path="/v1/chat/completions",
    body={"model": "gpt-4", "messages": [{"role": "user", "content": "hello"}]},
).will_respond_with(
    status=200,
    body={
        "id": like("chatcmpl-xxx"),
        "object": "chat.completion",
        "choices": each_like({
            "index": 0,
            "message": {"role": "assistant", "content": like("Hello!")},
            "finish_reason": term("stop|length", "stop"),
        }),
    },
)

# CI 中自动验证
def test_contract():
    contract.verify()  # 失败时阻断发布

# Golden Test：检测模型输出格式是否发生变化
GOLDEN_SAMPLES = {
    "json_extraction": {
        "input": "Extract name and age from: 'Alice is 30'",
        "expected_schema": {"name": str, "age": int},
    }
}

def test_behavior_drift():
    for name, sample in GOLDEN_SAMPLES.items():
        result = client.chat(sample["input"])
        assert validate_schema(result, sample["expected_schema"]), \
            f"Behavior drift detected in {name}"

import warnings

class DeprecatedAPI:
    def old_method(self, **kwargs):
        warnings.warn(
            "old_method() is deprecated and will be removed in v3.0. "
            "Use new_method() instead. Migration guide: https://...",
            DeprecationWarning,
            stacklevel=2,
        )
        return self.new_method(**kwargs)  # 内部转发，保持兼容

HTTP/1.1 200 OK
Deprecation: true
Sunset: Sat, 01 Jun 2025 00:00:00 GMT
Link: <https://docs.example.com/migration>; rel="deprecation"

# 用户可声明期望的输出格式版本
response = client.chat(
    messages=messages,
    response_format={"version": "v1"}  # 即使底层 API 是 v2，也返回 v1 格式
)

# SDK 内部转换逻辑
class ResponseNormalizer:
    def __init__(self, target_version="v1"):
        self.target_version = target_version
    
    def normalize(self, raw_response):
        if self.target_version == "v1":
            return self._to_v1(raw_response)
        elif self.target_version == "v2":
            return raw_response  # 当前格式
    
    def _to_v1(self, v2_response):
        """将 v2 格式转换为 v1 格式"""
        return {
            "text": v2_response.choices[0].message.content,
            "finish_reason": v2_response.choices[0].finish_reason,
            # v2 新增的 usage 字段在 v1 中不存在，忽略
        }

from pydantic import BaseModel, Field
from typing import Optional

class ChatCompletion(BaseModel):
    id: str
    object: str = Field(default="chat.completion")
    created: int
    model: str
    choices: list[dict]
    
    # 新增字段必须有默认值，保证向后兼容
    usage: Optional[dict] = None        # v2 新增，v1 用户不受影响
    system_fingerprint: Optional[str] = None  # v3 新增
    
    # 如果必须删除字段，标记为 deprecated
    @property
    def deprecated_field(self):
        warnings.warn("deprecated_field is removed in v4", DeprecationWarning)
        return None

class CapabilityProbe:
    """探测 API 支持的功能"""
    
    def __init__(self, client):
        self.client = client
        self._capabilities = None
    
    def probe(self) -> dict:
        if self._capabilities:
            return self._capabilities
        
        caps = {}
        # 探测流式输出
        try:
            self.client.chat(messages=[{"role": "user", "content": "test"}], stream=True)
            caps["streaming"] = True
        except Exception:
            caps["streaming"] = False
        
        # 探测函数调用
        try:
            self.client.chat(
                messages=[{"role": "user", "content": "test"}],
                tools=[{"type": "function", "function": {"name": "test"}}]
            )
            caps["function_calling"] = True
        except Exception:
            caps["function_calling"] = False
        
        # 探测结构化输出
        try:
            self.client.chat(
                messages=[{"role": "user", "content": "test"}],
                response_format={"type": "json_object"}
            )
            caps["json_mode"] = True
        except Exception:
            caps["json_mode"] = False
        
        self._capabilities = caps
        return caps

# 使用示例
probe = CapabilityProbe(client)
caps = probe.probe()

if caps.get("function_calling"):
    # 使用新版函数调用 API
    response = client.chat(messages, tools=tools)
else:
    # 回退到旧版 prompt-based 方案
    response = client.chat(messages + [{"role": "system", "content": "Use JSON format"}])

# 请求时指定 API 版本
POST https://api.openai.com/v1/chat/completions
OpenAI-Beta: assistants=v2

# OpenAI Python SDK 内部兼容层（简化）
class OpenAI:
    @classmethod
    def Completion(cls, **kwargs):
        # 旧 API 调用会被 SDK 内部转发到新的 chat/completions
        # 保持旧代码可用
        prompt = kwargs.pop("prompt")
        return cls.ChatCompletion.create(
            model=kwargs.get("model", "gpt-3.5-turbo-instruct"),
            messages=[{"role": "user", "content": prompt}],
            **kwargs
        )

# LangChain 的适配器层
class BaseChatModel(ABC, RunnableSerializable):
    @abstractmethod
    def _generate(self, messages: List[BaseMessage], **kwargs) -> ChatResult:
        """子类实现具体 API 调用，但对外接口统一"""
        pass
    
    def invoke(self, input: str, config: Optional[RunnableConfig] = None) -> str:
        # 统一入口：无论底层是 OpenAI / Anthropic / 本地模型
        messages = [HumanMessage(content=input)]
        result = self._generate(messages)
        return result.generations[0].message.content

# 当 OpenAI API 变更时，只需更新 ChatOpenAI 适配器
# 用户代码无需修改——这就是适配器的价值
class ChatOpenAI(BaseChatModel):
    def _generate(self, messages, **kwargs):
        if self.api_version >= "2024-06-01":
            return self._call_v2(messages, **kwargs)
        else:
            return self._call_v1(messages, **kwargs)

from transformers import pipeline, GenerationConfig

# 方案6: GenerationConfig 用 dataclass 约束生成参数
config = GenerationConfig(
    max_length=200,
    temperature=0.7,
    do_sample=True,
    # 新增字段有默认值，保证向后兼容
    top_p=1.0,      # v4.0 新增
    top_k=50,       # v4.2 新增
)

pipe = pipeline("text-generation", model="gpt2", generation_config=config)

# 方案5: 输出模式协商
result = pipe("Once upon a time", return_full_text=True)
# return_full_text 控制是否包含 prompt，旧版本默认 False，新版本可选

// Vertex AI 的 protobuf 契约定义（简化）
message PredictRequest {
  string endpoint = 1;
  google.protobuf.Value instances = 2;
  google.protobuf.Struct parameters = 3;  // 可扩展参数，不影响旧客户端
}

message PredictResponse {
  repeated google.protobuf.Value predictions = 1;
  // 新增字段不影响旧客户端——旧客户端忽略未知字段
  DeployedModelRef deployed_model_ref = 2;  // v2 新增
}

from google.cloud import aiplatform

# 方案7: 运行时能力探测
client = aiplatform.gapic.PredictionServiceClient()
endpoint = client.get_endpoint(name=endpoint_path)
supported_models = endpoint.deployed_models  # 探测可用模型