用一套代码适配与调用多模型 API

前言

OpenAI 的 API 格式已经成为事实标准。国内的主流模型平台——DeepSeek、通义千问、豆包、硅基流动、零一万物——全都兼容这套格式。

这意味着写一次代码，换个 base_url 就能切模型。这篇文章讲清楚各平台的接入方式，以及如何封装一个统一的调用工具。

---

一、主流平台接入速查

OpenAI

from openai import OpenAI
client = OpenAI(
    api_key="sk-xxxxxxxx",
    base_url="https://api.openai.com/v1"  # 这是默认值，可省略
)

DeepSeek

platform.deepseek.com 注册充值，费用极低（约 OpenAI 的 1/10）。

client = OpenAI(
    api_key="sk-xxxxxxxx",
    base_url="https://api.deepseek.com"
)

可用模型：deepseek-chat（V3 通用）、deepseek-reasoner（R1 推理）

通义千问（阿里云百炼）

阿里云百炼 注册后申请 Key，新用户有免费额度。

client = OpenAI(
    api_key="sk-xxxxxxxx",
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"
)

可用模型：qwen-max、qwen-plus、qwen-turbo、qwen-long（长文本）

豆包（火山引擎）

火山方舟 注册后创建接入点。

client = OpenAI(
    api_key="xxxxxxxx",
    base_url="https://ark.cn-beijing.volces.com/api/v3",
)

豆包的模型 ID 不是模型名，而是接入点 ID（类似 ep-20250101xxxxx）。创建接入点 → 选模型 → 拿到 endpoint ID → 当 model 参数用。

硅基流动

国内外模型聚合平台，注册即送额度。适合不想每家都注册一遍的开发者。

client = OpenAI(
    api_key="sk-xxxxxxxx",
    base_url="https://api.siliconflow.cn/v1"
)

模型 ID 参考平台文档，支持 DeepSeek、Qwen、Llama、GLM 等。

零一万物（Yi 系列）

client = OpenAI(
    api_key="xxxxxxxx",
    base_url="https://api.lingyiwanwu.com/v1"
)

---

二、统一封装：一套代码切所有模型

import os
from openai import OpenAI
from dataclasses import dataclass
from typing import Optional, Generator

@dataclass
class ModelConfig:
    """模型接入配置"""
    api_key: str
    base_url: str
    model: str

# 预设配置
PRESETS = {
    "gpt-4o": ModelConfig(
        api_key=os.getenv("OPENAI_API_KEY", ""),
        base_url="https://api.openai.com/v1",
        model="gpt-4o"
    ),
    "deepseek": ModelConfig(
        api_key=os.getenv("DEEPSEEK_API_KEY", ""),
        base_url="https://api.deepseek.com",
        model="deepseek-chat"
    ),
    "deepseek-r1": ModelConfig(
        api_key=os.getenv("DEEPSEEK_API_KEY", ""),
        base_url="https://api.deepseek.com",
        model="deepseek-reasoner"
    ),
    "qwen-max": ModelConfig(
        api_key=os.getenv("QWEN_API_KEY", ""),
        base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
        model="qwen-max"
    ),
    "qwen-plus": ModelConfig(
        api_key=os.getenv("QWEN_API_KEY", ""),
        base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
        model="qwen-plus"
    ),
    "silicon-deepseek": ModelConfig(
        api_key=os.getenv("SILICON_API_KEY", ""),
        base_url="https://api.siliconflow.cn/v1",
        model="Pro/deepseek-ai/DeepSeek-V3"
    ),
}

class LLM:
    """统一的 LLM 调用客户端"""
    
    def __init__(self, preset: str = "deepseek"):
        config = PRESETS[preset]
        self.client = OpenAI(
            api_key=config.api_key,
            base_url=config.base_url
        )
        self.model = config.model
    
    def chat(self, prompt: str, system: str = "",
             temperature: float = 0.7, max_tokens: int = 2048) -> str:
        """单轮对话，返回文本"""
        messages = []
        if system:
            messages.append({"role": "system", "content": system})
        messages.append({"role": "user", "content": prompt})
        
        resp = self.client.chat.completions.create(
            model=self.model,
            messages=messages,
            temperature=temperature,
            max_tokens=max_tokens,
            stream=False
        )
        return resp.choices[0].message.content
    
    def chat_stream(self, prompt: str, system: str = "",
                    temperature: float = 0.7) -> Generator[str, None, None]:
        """流式对话"""
        messages = []
        if system:
            messages.append({"role": "system", "content": system})
        messages.append({"role": "user", "content": prompt})
        
        stream = self.client.chat.completions.create(
            model=self.model,
            messages=messages,
            temperature=temperature,
            stream=True
        )
        for chunk in stream:
            if chunk.choices[0].delta.content:
                yield chunk.choices[0].delta.content


# ===== 使用示例 =====
# 用 DeepSeek
llm = LLM("deepseek")
print(llm.chat("用一句话解释什么是 Docker"))

# 切到通义千问
llm = LLM("qwen-max")
print(llm.chat("介绍一下杭州"))

# 流式输出
llm = LLM("deepseek")
for text in llm.chat_stream("讲个笑话"):
    print(text, end="", flush=True)

---

三、多模型对比：同一问题问不同模型

def compare_models(question, presets=["deepseek", "qwen-max", "gpt-4o"]):
    """用同一个问题测试多个模型，对比回答"""
    for preset in presets:
        try:
            llm = LLM(preset)
            answer = llm.chat(question, temperature=0.3, max_tokens=500)
            print(f"===== {preset} =====")
            print(answer[:300])
            print()
        except Exception as e:
            print(f"===== {preset} ===== FAILED: {e}")
            print()

compare_models("用 Python 实现一个 LRU 缓存")

---

四、失败自动切换（Fallback 容灾网关）

在生产环境中，单一大模型服务商偶尔会发生网络抖动、429 频率超限或服务降级。因此高可用架构必须实现主备自动切换：

优化实现（预先初始化连接池，避免循环内重复建连）：

class LLMWithFallback:
    def __init__(self, primary: str, fallbacks: list[str]):
        self.primary = primary
        self.fallbacks = fallbacks
        self.model_names = [primary] + fallbacks
        # 预先创建各模型客户端实例，复用 HTTP 连接池
        self.clients = {name: LLM(name) for name in self.model_names}
    
    def chat(self, prompt: str, system: str = "", **kwargs) -> str:
        last_error = None
        for name in self.model_names:
            try:
                llm = self.clients[name]
                return llm.chat(prompt, system, **kwargs)
            except Exception as e:
                last_error = e
                print(f"[fallback] 模型 {name} 异常: {e}，尝试切换备选...")
        raise RuntimeError(f"所有备选模型均调用失败。最后一次报错: {last_error}")

# 使用示例
llm = LLMWithFallback("deepseek", ["qwen-max", "silicon-deepseek"])
print(llm.chat("介绍一下微积分的核心思想"))

---

五、成本控制

各平台价格对比（2025 年初，每百万 token）

平台	模型	输入价格(元)	输出价格(元)
DeepSeek	deepseek-chat	1	2
DeepSeek	deepseek-reasoner	4	16
通义千问	qwen-turbo	0.3	0.6
通义千问	qwen-max	20	60
硅基流动	DeepSeek-V3	1.3	1.3
OpenAI	GPT-4o	~18	~72
OpenAI	GPT-4o-mini	~1	~4

省钱套路：

• 简单任务用便宜的模型（qwen-turbo、GPT-4o-mini、deepseek-chat）
• 复杂推理才用贵的（deepseek-r1、qwen-max、GPT-4o）
• 开发调试阶段用硅基流动的免费额度

Token 消耗统计

class LLMWithUsage(LLM):
    def __init__(self, preset: str = "deepseek"):
        super().__init__(preset)
        self.total_input_tokens = 0
        self.total_output_tokens = 0
    
    def chat(self, prompt: str, system: str = "", **kwargs) -> str:
        messages = []
        if system:
            messages.append({"role": "system", "content": system})
        messages.append({"role": "user", "content": prompt})
        
        resp = self.client.chat.completions.create(
            model=self.model,
            messages=messages,
            stream=False,
            **kwargs
        )
        
        usage = resp.usage
        if usage:
            self.total_input_tokens += getattr(usage, "prompt_tokens", 0) or 0
            self.total_output_tokens += getattr(usage, "completion_tokens", 0) or 0
        
        return resp.choices[0].message.content
    
    def report(self):
        total_input = self.total_input_tokens
        total_output = self.total_output_tokens
        print(f"累计输入: {total_input} tokens")
        print(f"累计输出: {total_output} tokens")
        print(f"总计: {total_input + total_output} tokens")

---

以上这套封装足够日常开发使用。搞定了 API 调用之后，下一步就是把模型真正嵌入到业务——RAG、Agent、Function Calling 这些就水到渠成了。