Python 接入多种 AI 模型实战：一个 SDK 调用所有模型

背景

项目里需要同时用多个AI模型：Claude写代码、GPT做文档、DeepSeek跑批量任务。如果每家都单独对接，光维护配置就很头疼。

解决方案：用一个兼容 OpenAI SDK 的中转服务，统一接口调用所有模型。

环境准备

pip install openai

基础接入

from openai import OpenAI

# 中转站配置（兼容 OpenAI SDK）
client = OpenAI(
    api_key="你的API-Key",
    base_url="https://你的中转站地址/v1"
)

# 调用 Claude
response = client.chat.completions.create(
    model="claude-sonnet-4-6",
    messages=[
        {"role": "system", "content": "你是一个专业的Python开发者"},
        {"role": "user", "content": "用Python实现一个简单的缓存装饰器"}
    ]
)
print(response.choices[0].message.content)

切换模型只需要改 model 参数，其他代码完全一样：

# 调用 GPT-5.5
response = client.chat.completions.create(
    model="gpt-5.5",
    messages=[{"role": "user", "content": "写一份产品介绍"}]
)

# 调用 DeepSeek
response = client.chat.completions.create(
    model="deepseek-v4-flash",
    messages=[{"role": "user", "content": "翻译这段话"}]
)

# 调用 Qwen
response = client.chat.completions.create(
    model="qwen3.7-max",
    messages=[{"role": "user", "content": "写一篇中文周报"}]
)

流式输出

stream = client.chat.completions.create(
    model="claude-sonnet-4-6",
    messages=[{"role": "user", "content": "写一个FastAPI接口"}],
    stream=True
)

for chunk in stream:
    if chunk.choices[0].delta.content:
        print(chunk.choices[0].delta.content, end="", flush=True)

按任务选模型

实际项目中，不同任务用不同模型效果更好，成本也更优：

MODEL_MAP = {
    "code":      "claude-sonnet-4-6",   # 写代码
    "fast":      "gpt-5-mini",           # 快速回复
    "reasoning": "deepseek-r1",          # 推理任务
    "batch":     "deepseek-v4-flash",    # 批量处理
    "chinese":   "qwen3.7-max",          # 中文写作
}

def ai(task_type: str, prompt: str, system: str = "") -> str:
    model = MODEL_MAP.get(task_type, "gpt-5-mini")
    messages = []
    if system:
        messages.append({"role": "system", "content": system})
    messages.append({"role": "user", "content": prompt})
    
    resp = client.chat.completions.create(model=model, messages=messages)
    return resp.choices[0].message.content

# 用法
code = ai("code", "用Python写一个并发下载器")
summary = ai("chinese", "总结这段文本: ...", system="用200字以内概括")
batch_result = ai("batch", "翻译: Hello World")

错误处理

import time
from openai import APIError, RateLimitError, APITimeoutError

def ai_safe(prompt: str, model: str = "claude-sonnet-4-6", retries: int = 3) -> str:
    for attempt in range(retries):
        try:
            resp = client.chat.completions.create(
                model=model,
                messages=[{"role": "user", "content": prompt}],
                timeout=30
            )
            return resp.choices[0].message.content
        except RateLimitError:
            wait = 2 ** attempt
            print(f"限流，等待 {wait}s...")
            time.sleep(wait)
        except APITimeoutError:
            print(f"超时，重试 {attempt+1}/{retries}")
        except APIError as e:
            print(f"API错误: {e}")
            if attempt == retries - 1:
                raise
    raise Exception("重试次数用完")

成本监控

# 各模型参考价格（每百万token，人民币）
PRICING = {
    "claude-sonnet-4-6":  {"input": 2.87,  "output": 18.35},
    "gpt-5.5":            {"input": 3.45,  "output": 28.71},
    "deepseek-v4-flash":  {"input": 0.097, "output": 0.193},
    "qwen3.7-max":        {"input": 0.287, "output": 1.656},
}

def calc_cost(model, input_tokens, output_tokens):
    price = PRICING.get(model, {"input": 10, "output": 30})
    return (input_tokens * price["input"] + output_tokens * price["output"]) / 1_000_000

# 记录每次调用
total_cost = 0
resp = client.chat.completions.create(
    model="claude-sonnet-4-6",
    messages=[{"role": "user", "content": prompt}]
)
cost = calc_cost("claude-sonnet-4-6", resp.usage.prompt_tokens, resp.usage.completion_tokens)
total_cost += cost
print(f"本次调用: ¥{cost:.4f} (累计: ¥{total_cost:.4f})")

各中转站 base_url 参考

所有中转站都兼容 OpenAI SDK，切换只需改 base_url 和 api_key。

批量处理示例

def batch_translate(texts: list[str]) -> list[str]:
    """批量翻译，用便宜模型"""
    results = []
    for text in texts:
        resp = client.chat.completions.create(
            model="deepseek-v4-flash",  # 最便宜
            messages=[{"role": "user", "content": f"翻译成英文: {text}"}],
            max_tokens=200
        )
        results.append(resp.choices[0].message.content)
    return results

# 用法
texts = ["你好世界", "人工智能", "深度学习"]
translations = batch_translate(texts)

注意事项

1. 先小额测试：别上来就充大额，先验证计费和稳定性
2. 验证token计费：用固定prompt对比官方消耗，偏差>10%有问题
3. 注意隐私：敏感数据不要走中转站
4. 准备备用方案：主力中转站挂了能快速切换

代码基于Python 3.10+和openai SDK 1.x。价格来自2026年5-6月实测，可能随时调整。