国产大模型 API 接入实战：用 OpenAI SDK 调通 Kimi K2.6、DeepSeek、Step 3.5 Flash（2026 完整代码）

qq_42720852

837人浏览 · 2026-04-23 11:51:48

qq_42720852 · 2026-04-23 11:51:48 发布

写在前面

最近 OpenAI 和 Anthropic 又涨价了，加上国内访问的稳定性问题，越来越多团队开始把生产流量切到国产大模型。但切的过程不是"换个 key"那么简单——每家厂商的 SDK、鉴权、错误码、限流策略都不一样，多 provider 并存的代码很快就会变成屎山。

这篇文章把过去一个月在生产环境实测过的 4 家国产大模型接入方案整理出来，全部用 OpenAI 官方 Python SDK 调通，附完整代码 + 错误处理模板。

数据采集 / 代码验证时间：2026 年 4 月 23 日。

涉及模型：

Kimi K2.6（Moonshot AI，4-20 刚发布，1T MoE / 256K context / SWE-Bench Pro 58.6）
DeepSeek V3.2（综合性价比天花板，输入 $0.14/M）
Step 3.5 Flash（StepFun，196B MoE / Apache 2.0 / AIME 2025 拿 97.3 分）
Qwen 3.6 Plus（阿里通义，1M 上下文）

一、为什么用 OpenAI SDK 调国产模型

简单说三个理由：

1. 国产厂商基本都做了 OpenAI 兼容接口。 Moonshot、DeepSeek、StepFun、阿里 DashScope、智谱、字节、腾讯混元——全部支持 OpenAI 格式的 /chat/completions 端点，只需要换 base_url 和 api_key。

2. 你的现有代码不用改。 如果项目里已经在用 openai 库（绝大多数 AI 项目都在用），换厂商只要改两行配置，业务逻辑完全不动。

3. 切换成本接近零，可以多家并存做 A/B。 同一个 prompt 跑 4 家比一遍，成本和延迟数据 1 小时就能拿到。

下面进入实战。

二、环境准备

pip install openai>=1.0.0

就这一个依赖。所有国产厂商都用这个 SDK 调，不需要装四遍。

三、Kimi K2.6（Moonshot AI）接入

Kimi K2.6 是 4 月 20 日刚发布的 1 万亿参数 MoE 模型，在 SWE-Bench Pro 上拿 58.6 分，高于 GPT-5.4 (xhigh) 的 57.7、Claude Opus 4.6 (max) 的 53.4。开源 + 便宜 + 跑分硬，是当前国产代码模型的首选。

注册： platform.moonshot.cn → 控制台拿 API key

代码：

from openai import OpenAI

client = OpenAI(
    api_key="sk-your-moonshot-key",
    base_url="https://api.moonshot.cn/v1",  # 国内端点
    # 或 base_url="https://api.moonshot.ai/v1"  # 海外端点
)

response = client.chat.completions.create(
    model="kimi-k2.6",
    messages=[
        {"role": "system", "content": "你是一个 Python 高级工程师。"},
        {"role": "user", "content": "用 Python 实现一个支持 LRU 淘汰的并发安全缓存，给完整代码。"},
    ],
    temperature=0.3,
    max_tokens=2048,
)

print(response.choices[0].message.content)

Kimi K2.6 关键参数：

参数	值	说明
上下文	256K	实测 180K 内 recall 准确率高
输入价	$0.60 / M tokens	列表价
输出价	$2.50 / M tokens	列表价
Cache hit	$0.16 / M tokens	长 system prompt 一定要用
多模态	支持图片 + 视频输入	text-out only
长程任务	12+ 小时自主运行	适合 coding agent

踩坑提示：

Tool call schema 偶尔有 trailing whitespace——如果用严格的 OpenAI 客户端校验工具调用 JSON，加一层 .strip() 兜底。
国内端点和海外端点是不同账号——别用海外 key 调国内端点，会 401。
Cache hit 价格比列表价低 4 倍——长 system prompt 务必启用 prompt caching。

四、DeepSeek V3.2 接入

DeepSeek 是国产里 OpenAI 兼容做得最早最干净的，也是综合性价比最稳的——输入 $0.14 / M、输出 $0.28 / M。

注册： platform.deepseek.com → API keys

代码：

from openai import OpenAI

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

response = client.chat.completions.create(
    model="deepseek-chat",  # V3 / V3.2 都走这个
    messages=[
        {"role": "user", "content": "解释 MoE 架构里的 expert routing 是什么。"},
    ],
    stream=False,
)

print(response.choices[0].message.content)

流式输出（生产推荐）：

stream = client.chat.completions.create(
    model="deepseek-chat",
    messages=[{"role": "user", "content": "..."}],
    stream=True,
)

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

踩坑提示：

deepseek-reasoner 是单独的 Reasoning 模型，按 reasoning_content + content 双字段返回，记得分别处理。
DeepSeek 的 cache hit 是 $0.07 / M，比 Kimi 还便宜，是当前 API 里 cache 价最低的。
限流比国际厂商更松，但偶尔会 5xx 抖动——加个指数退避就好。

五、Step 3.5 Flash（StepFun）接入

StepFun 这家公司比较低调，但 Step 3.5 Flash 是真便宜——$0.10 / $0.30 / M tokens，比 DeepSeek V3.2 还低。AIME 2025 跑了 97.3 分，数学和 STEM reasoning 是国产第一档。

注册： platform.stepfun.com → 创建 API key

代码：

from openai import OpenAI

client = OpenAI(
    api_key="sk-your-stepfun-key",
    base_url="https://api.stepfun.com/v1",
)

response = client.chat.completions.create(
    model="step-3-5-flash",  # 注意：用连字符不是点
    messages=[
        {"role": "user", "content": "证明 sqrt(2) 是无理数，给出严格的反证法步骤。"},
    ],
    temperature=0.1,  # 数学题建议低温
)

print(response.choices[0].message.content)

Step 3.5 Flash 关键参数：

参数	值	说明
总参数	196B	11B 激活（稀疏 MoE）
上下文	262K	单次输出可达 64K
协议	Apache 2.0	可商用、可自部署
AIME 2025	97.3	数学顶级
LiveCodeBench-V6	86.4%	代码也强
输入价	$0.10 / M tokens	当前最便宜国产
输出价	$0.30 / M tokens	—

六、Qwen 3.6 Plus（阿里通义）接入

Qwen 走的是阿里 DashScope 平台，OpenAI 兼容模式需要单独的 base_url：

注册： dashscope.aliyun.com → 开通 → 获取 API key

代码：

from openai import OpenAI
import os

client = OpenAI(
    api_key=os.environ["DASHSCOPE_API_KEY"],
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
)

response = client.chat.completions.create(
    model="qwen-max",  # 或 qwen-plus / qwen3-coder-plus / qwen3-vl-plus
    messages=[
        {"role": "user", "content": "用 React + TypeScript 写一个虚拟列表组件，处理 10 万行数据。"},
    ],
)

print(response.choices[0].message.content)

Qwen 系列模型选择（2026 Q2）：

模型	适用场景	特点
`qwen-max`	综合最强	旗舰版
`qwen3-coder-plus`	写代码	编码特化
`qwen3-vl-plus`	多模态	视觉 + 语言
`qwen-plus`	高性价比	日常调用
`qwen-turbo`	大并发	极速响应

踩坑提示：

DashScope 的限流按"调用次数 + 总 tokens"双维度算，别只看 RPM。
OpenAI 兼容模式不支持所有原生 DashScope 功能（比如 function calling 的某些扩展），高级功能要用 dashscope SDK。
Qwen 3.6 Plus 是 1M 上下文，但实测 700K 内才稳，超过会有 recall 衰减。

七、让代码不再屎山：抽象多 provider 调用

把上面四家代码合在一起，你会发现一个项目里要管 4 个 base_url、4 个 api_key、4 套限流和错误处理。生产环境很快就乱了。

简单的做法是写个工厂函数：

from openai import OpenAI
from typing import Literal
import os

PROVIDERS = {
    "kimi": {
        "base_url": "https://api.moonshot.cn/v1",
        "api_key_env": "MOONSHOT_API_KEY",
        "default_model": "kimi-k2.6",
    },
    "deepseek": {
        "base_url": "https://api.deepseek.com/v1",
        "api_key_env": "DEEPSEEK_API_KEY",
        "default_model": "deepseek-chat",
    },
    "step": {
        "base_url": "https://api.stepfun.com/v1",
        "api_key_env": "STEPFUN_API_KEY",
        "default_model": "step-3-5-flash",
    },
    "qwen": {
        "base_url": "https://dashscope.aliyuncs.com/compatible-mode/v1",
        "api_key_env": "DASHSCOPE_API_KEY",
        "default_model": "qwen-max",
    },
}

def get_client(provider: Literal["kimi", "deepseek", "step", "qwen"]):
    cfg = PROVIDERS[provider]
    return OpenAI(
        api_key=os.environ[cfg["api_key_env"]],
        base_url=cfg["base_url"],
    ), cfg["default_model"]


# 用法
client, model = get_client("kimi")
response = client.chat.completions.create(
    model=model,
    messages=[{"role": "user", "content": "Hello"}],
)

八、生产环境必备：重试、限流、降级模板

上面的代码只适合 demo。生产环境要处理：网络抖动、5xx、限流、余额不足、模型临时下线等异常。下面这套模板可以直接拿去用。

8.1 指数退避重试

import time
import random
from openai import OpenAI, RateLimitError, APIError, APITimeoutError

def chat_with_retry(client, model, messages, max_retries=3):
    for attempt in range(max_retries):
        try:
            return client.chat.completions.create(
                model=model,
                messages=messages,
                timeout=30,
            )
        except (RateLimitError, APITimeoutError, APIError) as e:
            if attempt == max_retries - 1:
                raise
            # 指数退避 + 随机抖动，避免雷鸣般重试
            wait = (2 ** attempt) + random.uniform(0, 1)
            print(f"[WARN] {type(e).__name__}, retry in {wait:.1f}s")
            time.sleep(wait)

8.2 多 provider fallback

业务关键路径上，单厂商挂了不能让整个系统停摆。写一个降级链：

def chat_with_fallback(messages, provider_chain=None):
    """
    按优先级尝试多个 provider，任意一个成功就返回。
    provider_chain: [("kimi", "kimi-k2.6"), ("deepseek", "deepseek-chat"), ...]
    """
    provider_chain = provider_chain or [
        ("deepseek", "deepseek-chat"),   # 首选：最便宜最稳
        ("kimi", "kimi-k2.6"),           # 备选 1
        ("qwen", "qwen-max"),            # 备选 2
    ]

    last_error = None
    for provider, model in provider_chain:
        try:
            client, _ = get_client(provider)
            return chat_with_retry(client, model, messages)
        except Exception as e:
            print(f"[FAIL] {provider}/{model}: {e}")
            last_error = e
            continue

    raise RuntimeError(f"All providers failed, last error: {last_error}")

8.3 按 token 预算限流

一些厂商是按 TPM（tokens per minute）限流，超了会 429。写个本地桶：

from collections import deque
import time
import threading

class TokenBudget:
    def __init__(self, tokens_per_minute):
        self.tpm = tokens_per_minute
        self.window = deque()  # (timestamp, tokens)
        self.lock = threading.Lock()

    def acquire(self, tokens):
        with self.lock:
            now = time.time()
            # 清理 60 秒前的记录
            while self.window and self.window[0][0] < now - 60:
                self.window.popleft()
            current = sum(t for _, t in self.window)
            if current + tokens > self.tpm:
                sleep = 60 - (now - self.window[0][0])
                time.sleep(max(0, sleep))
                return self.acquire(tokens)
            self.window.append((now, tokens))

# 用法：
# budget = TokenBudget(tokens_per_minute=100_000)
# budget.acquire(estimate_tokens(messages))
# response = client.chat.completions.create(...)

8.4 错误码快查表

错误	含义	怎么处理
`401 Unauthorized`	API key 无效或过期	检查 key，或者用错了端点（海外 key 调国内端点）
`429 Too Many Requests`	限流	指数退避 + TokenBudget
`400 Bad Request`	参数错误 / model 名错	检查 model 拼写（step-3-5-flash 不是 step-3.5-flash）
`500/502/503`	服务端抖动	指数退避，换 provider fallback
`context length exceeded`	超上下文	前端截断 / 摘要压缩
`content_filter`	命中内容审核	调整 prompt，避免敏感词

九、价格对比（2026 Q2 实测）

模型	输入 ($/M)	输出 ($/M)	Cache hit ($/M)	备注
Step 3.5 Flash	$0.10	$0.30	—	当前最便宜
DeepSeek V3.2	$0.14	$0.28	$0.07	cache 最便宜
Hunyuan-T1	~$0.20	~$0.80	—	推理特化
Qwen 3.6 Plus	$0.28	~$1.20	—	1M context
Kimi K2.6	$0.60	$2.50	$0.16	代码最强
GPT-5.4 (xhigh)	~$10	~$40	$0.50	国际旗舰
Claude Opus 4.6	~$15	~$75	—	Anthropic 旗舰