简介这是一份面向Python开发者与AI应用入门者的多平台API调用示例源码包聚焦于用统一代码风格接入Baichuan、ChatGLM、Deepseek、Kimi、MChat、X元象、mistral、字节、文心一言、紫东太初、腾讯、讯飞、通义等主流大模型服务解决不同厂商接口认证、请求参数与返回解析差异带来的上手成本问题。压缩包共22个文件全部为py源码整体约21KB按平台分目录组织涵盖对话生成、Token处理、语音与文本交互等典型调用场景便于按需查阅与二次改造。目前已有347人学习下载。读者可直接参考各平台示例完成密钥配置、请求构造与响应解析快速搭建聊天机器人或智能问答原型并对比不同厂商接口风格积累多模型接入与排错经验。1. 多模型 API 调用的统一封装从十三家 SDK 里抽出一套能跑的代码手上同时维护三个以上 AI 项目的工程师大概率都经历过这种场面需求方临时要把文心一言换成 Deepseek你翻出半年前的代码发现鉴权方式、消息体结构、流式返回格式全不一样改完一处漏一处联调一整天。这个标题讲的就是把 Baichuan、ChatGLM、Deepseek、Kimi、MChat、Token、X元象、mistral、字节、文心一言、紫东太初、腾讯、讯飞、通义这十三家平台的调用方式收敛到一套 Python 封装里让切换模型变成改一个字符串。它解决的不是怎么调某一个模型而是怎么让十几个模型在同一套代码里共存而不互相打架。适合手里有多个模型账号、需要做对比测试或灰度切换的开发者也适合刚入门 Python、想通过一个真实项目把 requests、异常处理、流式解析练熟的人。2. 十三家平台的接口差异到底在哪先看清再动手2.1 鉴权方式分三派决定了封装的第一层抽象把十三家平台的鉴权方式摊开看本质上只有三种模式。第一种是 Bearer TokenDeepseek、Kimi、mistral、X元象、通义走这条路请求头里带Authorization: Bearer sk-xxx最省事。第二种是 AK/SK 签名腾讯混元和讯飞星火属于这类需要拿 SecretKey 对参数做 HMAC 签名签名串拼错一个字符就返回鉴权失败而且两家的签名算法细节还不一样。第三种是百度文心一言的 access_token 机制要先用 API Key 和 Secret Key 换一个有效期 30 天的 token再拿这个 token 去调接口token 过期不刷新就会突然全线报错。我一般会在封装层定义一个AuthProvider基类把如何生成请求头抽成一个方法每个平台实现自己的子类。这样上层调用代码完全不用关心底下是 Bearer 还是签名切换平台时只换 Provider 实例。import hmac import hashlib import base64 import time from abc import ABC, abstractmethod class AuthProvider(ABC): abstractmethod def build_headers(self) - dict: 返回该平台需要的完整请求头 pass class BearerAuth(AuthProvider): def __init__(self, api_key: str): self.api_key api_key def build_headers(self) - dict: return { Authorization: fBearer {self.api_key}, Content-Type: application/json } class TencentSignAuth(AuthProvider): def __init__(self, secret_id: str, secret_key: str, region: str ap-guangzhou): self.secret_id secret_id self.secret_key secret_key self.region region def build_headers(self) - dict: timestamp int(time.time()) # 腾讯混元的签名串需要按固定顺序拼接顺序错了直接 401 sign_str fhost:hunyuan.tencentcloudapi.com\ntimestamp:{timestamp} signature base64.b64encode( hmac.new( self.secret_key.encode(), sign_str.encode(), hashlib.sha1 ).digest() ).decode() return { Authorization: fTC3-HMAC-SHA256 Credential{self.secret_id}, X-TC-Timestamp: str(timestamp), X-TC-Region: self.region, Content-Type: application/json }BearerAuth直接拼头TencentSignAuth演示了签名类平台的核心逻辑时间戳参与签名、签名串顺序敏感、region 影响路由。实际接入时腾讯的签名算法比这里复杂需要按官方文档补全 CanonicalRequest 部分但抽象层的意义已经体现出来了——上层不需要知道这些差异。2.2 消息体结构OpenAI 格式正在成为事实标准十三家里有九家的消息体可以直接用 OpenAI 的{model: ..., messages: [{role: user, content: ...}]}格式包括 Deepseek、Kimi、mistral、X元象、通义、字节豆包。剩下四家需要转换文心一言用messages但字段名是role和content的变体ChatGLM 早期版本用prompt单字符串讯飞星火用parameter.chat嵌套结构腾讯混元用Messages数组但 role 取值是user/assistant没有system。封装层要做的是定义一个内部统一的消息格式然后在每个平台的 Adapter 里做一次转换。我习惯用 dataclass 定义内部消息避免到处传裸 dict。from dataclasses import dataclass, field from typing import List, Literal dataclass class Message: role: Literal[system, user, assistant] content: str dataclass class ChatRequest: messages: List[Message] model: str temperature: float 0.7 max_tokens: int 2048 stream: bool False def to_openai_format(req: ChatRequest) - dict: 适用于 Deepseek/Kimi/mistral/通义等九家 return { model: req.model, messages: [{role: m.role, content: m.content} for m in req.messages], temperature: req.temperature, max_tokens: req.max_tokens, stream: req.stream } def to_spark_format(req: ChatRequest) - dict: 讯飞星火system 消息要单独放到 parameter 里 system_msgs [m.content for m in req.messages if m.role system] chat_msgs [m for m in req.messages if m.role ! system] return { header: {app_id: YOUR_APP_ID}, parameter: { chat: { temperature: req.temperature, max_tokens: req.max_tokens } }, payload: { message: { text: [ {role: m.role, content: m.content} for m in chat_msgs ] } } }to_openai_format是通用路径to_spark_format展示了讯飞的特殊处理system 消息不能混在 message 数组里要提到 parameter 层。这种差异如果不提前抽象每接一家就要改一次业务代码。2.3 流式返回SSE 解析的三种坑流式输出是刚需但十三家的 SSE 格式不统一。Deepseek 和 Kimi 返回data: {...}后跟空行mistral 返回data: {...}但结束标记是data: [DONE]讯飞星火返回的是带status字段的 JSON 且没有data:前缀。更麻烦的是文心一言流式返回里每个 chunk 的字段名和一次性返回不一样。解析层要做的第一件事是统一成逐 token 产出文本的生成器接口。我一般写一个StreamParser每个平台实现自己的parse_line方法上层只拿 yield 出来的文本。import json from typing import Generator class OpenAIStreamParser: 适用于 Deepseek/Kimi/mistral/通义等 def parse(self, response) - Generator[str, None, None]: for line in response.iter_lines(): if not line: continue line line.decode(utf-8) if line.startswith(data: ): data line[6:] if data [DONE]: break try: chunk json.loads(data) delta chunk[choices][0][delta] if content in delta: yield delta[content] except (json.JSONDecodeError, KeyError): # 有些平台会插入心跳包或空 chunk跳过即可 continue class SparkStreamParser: 讯飞星火没有 data: 前缀靠 status 判断结束 def parse(self, response) - Generator[str, None, None]: for line in response.iter_lines(): if not line: continue try: chunk json.loads(line.decode(utf-8)) status chunk[header][status] if status 2: break content chunk[payload][choices][text][0][content] yield content except (json.JSONDecodeError, KeyError, IndexError): continue两个 Parser 的差异集中在三点前缀有无、结束标记形式、内容字段路径。OpenAIStreamParser里对JSONDecodeError和KeyError的捕获不是偷懒是实际跑下来发现部分平台会在流中间插入空行或心跳包不捕获就会中断整个流。3. 统一调用层怎么写一个 Client 类管十三家3.1 注册表模式用字符串映射到 Adapter最直接的做法是写一个字典key 是平台名value 是对应的 Adapter 类。调用时传平台名Client 自动实例化正确的 Adapter。这样新增一家平台只需要加一个类加一行注册不用动 Client 本身。from typing import Dict, Type class BaseAdapter: def __init__(self, auth: AuthProvider, endpoint: str): self.auth auth self.endpoint endpoint def chat(self, req: ChatRequest) - str: raise NotImplementedError def chat_stream(self, req: ChatRequest): raise NotImplementedError class DeepseekAdapter(BaseAdapter): def chat(self, req: ChatRequest) - str: import requests payload to_openai_format(req) resp requests.post( self.endpoint, headersself.auth.build_headers(), jsonpayload, timeout60 ) resp.raise_for_status() return resp.json()[choices][0][message][content] def chat_stream(self, req: ChatRequest): import requests payload to_openai_format(req) payload[stream] True resp requests.post( self.endpoint, headersself.auth.build_headers(), jsonpayload, streamTrue, timeout60 ) parser OpenAIStreamParser() yield from parser.parse(resp) ADAPTER_REGISTRY: Dict[str, Type[BaseAdapter]] { deepseek: DeepseekAdapter, kimi: KimiAdapter, mistral: MistralAdapter, ernie: ErnieAdapter, spark: SparkAdapter, hunyuan: HunyuanAdapter, # ... 其余平台 } class AIClient: def __init__(self, platform: str, auth: AuthProvider, endpoint: str): adapter_cls ADAPTER_REGISTRY.get(platform) if not adapter_cls: raise ValueError(f不支持的平台: {platform}) self.adapter adapter_cls(auth, endpoint) def chat(self, req: ChatRequest) - str: return self.adapter.chat(req) def chat_stream(self, req: ChatRequest): return self.adapter.chat_stream(req)ADAPTER_REGISTRY是核心新增平台时只加类加注册项。AIClient构造时做一次平台名校验避免运行时才发现拼错。chat和chat_stream分别对应非流式和流式返回类型不同但调用方式一致。3.2 参数映射表temperature 和 max_tokens 的边界差异同一个参数在不同平台的有效范围不一样。Deepseek 的 temperature 支持 0 到 2Kimi 是 0 到 1文心一言是 0.01 到 1讯飞星火是 0 到 1 但步长 0.01。max_tokens 的上限差异更大从 2048 到 8192 不等。如果业务层传了一个 1.5 的 temperature 给 Kimi有的平台会报错有的会静默截断到 1行为不一致。我一般会在 Adapter 里加一个参数校验和裁剪逻辑把超出范围的值 clamp 到边界同时打一条 warning 日志。这样业务层不用为每个平台单独设参数但日志里能看到实际生效的值。import logging logger logging.getLogger(__name__) PARAM_LIMITS { deepseek: {temperature: (0, 2), max_tokens: (1, 8192)}, kimi: {temperature: (0, 1), max_tokens: (1, 4096)}, ernie: {temperature: (0.01, 1), max_tokens: (1, 2048)}, spark: {temperature: (0, 1), max_tokens: (1, 4096)}, } def clamp_params(platform: str, req: ChatRequest) - ChatRequest: limits PARAM_LIMITS.get(platform) if not limits: return req temp_min, temp_max limits[temperature] if not (temp_min req.temperature temp_max): old req.temperature req.temperature max(temp_min, min(temp_max, req.temperature)) logger.warning( %s temperature %.2f 超出范围 [%.2f, %.2f]已调整为 %.2f, platform, old, temp_min, temp_max, req.temperature ) tok_min, tok_max limits[max_tokens] if not (tok_min req.max_tokens tok_max): old req.max_tokens req.max_tokens max(tok_min, min(tok_max, req.max_tokens)) logger.warning( %s max_tokens %d 超出范围 [%d, %d]已调整为 %d, platform, old, tok_min, tok_max, req.max_tokens ) return reqPARAM_LIMITS是一张静态表实际接入时按各家文档填。clamp_params在 Adapter 的chat入口调用一次保证发给平台的值一定合法。warning 日志很重要否则业务层以为传了 1.5 实际生效 1.0调参调半天找不到原因。3.3 重试与降级429 和超时怎么处理多平台调用的一个隐性收益是可以做降级。主平台返回 429 或连续超时自动切到备用平台。但降级不能无脑做因为不同平台的输出质量不一样业务上未必能接受。我一般把降级做成可配置的默认关闭需要时打开。import time import requests from requests.exceptions import Timeout, HTTPError def call_with_retry(adapter, req, max_retries3, backoff1.0): last_exc None for attempt in range(max_retries): try: return adapter.chat(req) except HTTPError as e: if e.response.status_code 429: # 限流等一会儿再试 wait backoff * (2 ** attempt) logger.warning(触发限流%.1f 秒后重试, wait) time.sleep(wait) last_exc e elif e.response.status_code 500: # 服务端错误短暂等待后重试 time.sleep(backoff) last_exc e else: # 4xx 客户端错误重试没意义直接抛 raise except Timeout as e: logger.warning(请求超时第 %d 次重试, attempt 1) last_exc e time.sleep(backoff) raise last_exc429 用指数退避500 用固定间隔其他 4xx 直接抛不浪费时间。max_retries默认 3 次实际跑下来大部分限流在第二次重试就能过。注意backoff不要设太小有些平台的限流窗口是秒级的退避 0.5 秒以下基本没用。4. 避坑与排查十三家平台踩过的真实坑4.1 坑一文心一言 token 过期不报错返回空内容现象调用文心一言接口返回 200但result字段是空字符串没有任何错误信息。原因access_token 有效期 30 天过期后接口不一定返回 401有时返回 200 加空结果。解决在 Adapter 里缓存 token 并记录获取时间每次调用前检查是否超过 25 天超过就重新获取。同时判断返回内容为空时主动抛异常不要静默返回空串。4.2 坑二Kimi 流式返回的 chunk 里 content 可能不存在现象解析 Kimi 流式返回时KeyError: content。原因Kimi 在流开始时先发一个只含 role 的 chunk结束时发一个只含 finish_reason 的 chunk这两个 chunk 的 delta 里没有 content 字段。解决用delta.get(content)而不是delta[content]取不到就跳过。这个坑在 Deepseek 上不存在所以从 Deepseek 切到 Kimi 时容易翻车。4.3 坑三讯飞星火的 app_id 和 api_key 是两回事现象按文档填了 api_key 和 api_secret调用返回 401。原因讯飞星火需要三个凭证——app_id、api_key、api_secret其中 app_id 在请求体的 header 里api_key 和 api_secret 用于生成鉴权 URL。只填两个必失败。解决确认控制台上三个值都拿到app_id 放在header.app_idapi_key 和 api_secret 拼鉴权 URL。4.4 坑四腾讯混元的签名串里 host 必须和 endpoint 一致现象签名算法按文档实现但一直返回AuthFailure.SignatureFailure。原因签名串里的 host 字段必须和实际请求的 endpoint 完全一致包括区域前缀。比如请求hunyuan.tencentcloudapi.com但签名串里写了hunyuan.ap-guangzhou.tencentcloudapi.com签名就对不上。解决把 endpoint 的 host 抽成常量签名和请求共用同一个值不要手写两遍。4.5 坑五通义千问的 max_tokens 包含输入 token现象设了 max_tokens2048但输出只有几百字就截断了。原因通义的 max_tokens 是输入加输出的总上限不是仅输出。输入 prompt 长了输出空间就被压缩。解决估算输入 token 数中文大约 1 字 1.5 token用max_tokens - 输入估算作为实际输出上限或者直接调大 max_tokens 留足余量。5. 进阶用配置文件驱动多平台切换与 A/B 对比把平台配置从代码里抽到 YAML 文件是让这套封装真正好用的最后一步。配置文件里写清楚每个平台的 endpoint、鉴权类型、参数上限代码只读配置不写死。这样换平台、加平台、调参数都不用改 Python 代码。# platforms.yaml platforms: deepseek: endpoint: https://api.deepseek.com/v1/chat/completions auth_type: bearer api_key_env: DEEPSEEK_API_KEY limits: temperature: [0, 2] max_tokens: [1, 8192] kimi: endpoint: https://api.moonshot.cn/v1/chat/completions auth_type: bearer api_key_env: KIMI_API_KEY limits: temperature: [0, 1] max_tokens: [1, 4096] spark: endpoint: wss://spark-api.xf-yun.com/v3.1/chat auth_type: spark_sign app_id_env: SPARK_APP_ID api_key_env: SPARK_API_KEY api_secret_env: SPARK_API_SECRET limits: temperature: [0, 1] max_tokens: [1, 4096]读取配置后动态构建 AdapterAPI Key 从环境变量取不写进配置文件。这样配置文件可以进版本控制密钥不会泄露。import os import yaml def load_platforms(config_path: str platforms.yaml) - dict: with open(config_path, r, encodingutf-8) as f: config yaml.safe_load(f) platforms {} for name, cfg in config[platforms].items(): auth_type cfg[auth_type] if auth_type bearer: api_key os.environ.get(cfg[api_key_env]) if not api_key: raise EnvironmentError(f缺少环境变量 {cfg[api_key_env]}) auth BearerAuth(api_key) elif auth_type spark_sign: auth SparkSignAuth( app_idos.environ[cfg[app_id_env]], api_keyos.environ[cfg[api_key_env]], api_secretos.environ[cfg[api_secret_env]] ) else: raise ValueError(f未知鉴权类型: {auth_type}) platforms[name] { auth: auth, endpoint: cfg[endpoint], limits: cfg.get(limits, {}) } return platforms有了配置驱动做 A/B 对比就很简单同一个 prompt 并发发给多个平台收集结果做对比。我一般用concurrent.futures开线程池每个平台一个任务超时设 60 秒谁先返回先记录。from concurrent.futures import ThreadPoolExecutor, as_completed def compare_platforms(prompt: str, platform_names: list, platforms: dict): req ChatRequest( messages[Message(roleuser, contentprompt)], modeldefault, temperature0.7, max_tokens1024 ) results {} with ThreadPoolExecutor(max_workerslen(platform_names)) as executor: futures {} for name in platform_names: cfg platforms[name] client AIClient(name, cfg[auth], cfg[endpoint]) futures[executor.submit(client.chat, req)] name for future in as_completed(futures, timeout90): name futures[future] try: results[name] future.result() except Exception as e: results[name] f[失败] {type(e).__name__}: {e} return resultsas_completed带 timeout 参数防止某个平台卡死拖垮整个对比。失败的结果不抛异常而是记成字符串这样一次对比能看到所有平台的成败情况不用一个个重跑。这套封装我前后迭代了四版最大的教训是不要试图一开始就支持全部十三家。先接两家把抽象层跑通再一家一家加每加一家就补一条参数限制和一条避坑记录。我现在的习惯是每接一个新平台先写一个最小调用脚本验证鉴权和返回格式跑通了再往 Adapter 里塞。这样出问题时排查范围小不会在抽象层和平台差异之间来回猜。希望帮到你。本文还有配套的精品资源点击获取
