OpenAI 调用迁移 DeepSeek API 的兼容层实现

为什么需要兼容层

OpenAI SDK 的调用方式已经渗透到大量现有代码中。当需要把部分请求切到 DeepSeek API 时，直接修改业务代码会带来两个问题：一是改动面大、回归成本高；二是未来切换或回滚供应商时，又要重复修改。更稳妥的做法是在 SDK 与业务之间加一层薄封装，把供应商差异收敛到一处。

兼容层的目标不是完全模拟 OpenAI，而是让业务代码继续用熟悉的接口形态，同时把 base_url、api_key、模型名、请求参数和响应处理映射到目标供应商的约定上。需要强调的是，不同供应商的接口细节可能随版本变化，本文中的具体取值均为示例，实际以各供应商最新官方文档为准。

统一配置入口

第一步是把供应商配置从代码中抽离。建议用环境变量控制：

LLM_PROVIDER=deepseek
DEEPSEEK_API_KEY=sk-xxx
DEEPSEEK_BASE_URL=https://api.deepseek.com
OPENAI_API_KEY=sk-yyy
OPENAI_BASE_URL=https://api.openai.com/v1

注意：以上 base_url 和 api_key 均为示例值，实际地址和鉴权方式请以 DeepSeek 和 OpenAI 官方文档为准。

兼容层初始化时根据 LLM_PROVIDER 选择对应的 base_url 和 api_key。如果使用 OpenAI SDK，可以直接传入 base_url 参数；如果使用 REST，则拼接请求 URL。

import os
from openai import OpenAI

def build_client():
    provider = os.getenv("LLM_PROVIDER", "openai")
    if provider == "deepseek":
        return OpenAI(
            api_key=os.environ["DEEPSEEK_API_KEY"],
            base_url=os.environ["DEEPSEEK_BASE_URL"],
        )
    return OpenAI(
        api_key=os.environ["OPENAI_API_KEY"],
        base_url=os.environ.get("OPENAI_BASE_URL"),
    )

这样业务侧只依赖 build_client()，不关心底层是哪家。

模型名与参数映射

不同供应商的模型名不同，业务代码里不应硬编码。可以在兼容层维护一张映射表：

MODEL_MAP = {
    "openai": {
        "chat": "gpt-4o-mini",
        "reasoning": "gpt-4o",
    },
    "deepseek": {
        "chat": "deepseek-chat",
        "reasoning": "deepseek-reasoner",
    },
}

注意：以上模型名仅为示例，实际可用模型名请以各供应商官方文档为准。

业务调用时传入逻辑角色（如 chat、reasoning），兼容层再解析成实际模型名。参数方面，temperature、max_tokens、top_p 等常见字段通常可以直接透传，但要注意各家的取值范围和默认值可能不同。建议在兼容层做一次白名单过滤，只放行双方都支持的参数，避免因未知字段导致请求失败。

messages 格式与流式响应

OpenAI 的 messages 是 role + content 的数组。关于 DeepSeek 的对话接口是否采用完全相同的结构，目前尚未在本文中独立验证，建议读者以 DeepSeek 官方文档为准。在兼容层中，可以按以下原则处理：

系统消息的位置和数量限制可能不同，建议把 system 消息统一放在数组首位。
多模态内容（如图片）的支持范围不同，如果业务涉及图片输入，需要在兼容层判断当前供应商是否支持，不支持时提前报错或降级。

流式响应方面，OpenAI SDK 的 stream=True 返回的是迭代器，每个 chunk 包含 choices[0].delta.content。DeepSeek 的流式返回结构是否完全一致，同样需要以官方文档为准。兼容层应统一成一种内部事件格式：

def stream_chat(client, model, messages):
    resp = client.chat.completions.create(
        model=model,
        messages=messages,
        stream=True,
    )
    for chunk in resp:
        delta = chunk.choices[0].delta
        if delta and delta.content:
            yield delta.content

如果使用 REST，需要自己解析 SSE 的 data: 行，并处理 [DONE] 结束标记。建议把解析逻辑封装成生成器，业务侧只消费文本增量。

错误码与超时重试

不同供应商的错误码体系不完全一致，兼容层应把常见错误归类为几类：认证失败、限流、参数错误、服务端错误、超时。对于限流和服务端错误，可以配合指数退避重试；对于认证和参数错误，重试没有意义，应直接抛出。具体错误码和分类请以各供应商最新文档为准。

import time

def call_with_retry(fn, max_retries=3):
    for i in range(max_retries):
        try:
            return fn()
        except RateLimitError:
            time.sleep(2 ** i)
        except APIStatusError as e:
            if e.status_code >= 500:
                time.sleep(2 ** i)
            else:
                raise
    raise RuntimeError("retry exhausted")

超时设置也要显式指定。OpenAI SDK 支持 timeout 参数，建议根据业务场景设置合理值，流式请求可以适当放宽。

灰度路由与回滚

兼容层最大的价值在于可以按比例或按用户维度路由。例如：

def pick_provider(user_id):
    if user_id % 100 < 10:
        return "deepseek"
    return "openai"

这样可以把 10% 的流量切到 DeepSeek，观察错误率和延迟。如果指标异常，只需把比例调回 0，业务代码无需改动。回滚同样通过环境变量或配置中心完成，避免重新发布。

工程建议

把兼容层做成独立模块，不侵入业务逻辑。
所有供应商差异（base_url、模型名、错误码、流式解析）都收敛到这一层。
用环境变量或配置中心控制供应商和灰度比例。
记录每次请求的实际供应商和模型名，便于排查和计费。
对不支持的参数和内容类型提前校验，失败要给出明确错误信息。

兼容层不是一次性工作，而是随着供应商能力变化持续维护的边界。把差异控制住，业务代码才能稳定地跨供应商运行。