简介这是一份面向Python开发者与AI应用入门者的多平台API调用示例源码包聚焦于用统一代码风格接入Baichuan、ChatGLM、Deepseek、Kimi、MChat、X元象、mistral、字节、文心一言、紫东太初、腾讯、讯飞、通义等主流大模型服务解决不同厂商接口认证、请求构造与响应解析差异较大的问题。资源共22个文件全部为py脚本压缩包约21KB按平台分目录组织涵盖对话生成、Token处理、语音识别等典型调用场景便于按需查阅与二次改造。目前已有347人学习下载。读者可参考各平台示例快速完成API密钥配置、请求参数构建与返回数据解析理解GET/POST、JSON格式、超时与错误处理等关键细节并借助预设模板搭建聊天机器人或智能问答应用框架适合希望横向对比多家AI接口、快速验证功能的开发者。1. 从一份 Python 调用各家 AI 的源码说起多模型接入到底难在哪手上有一份 Python 调用各家 AI 的示例源码覆盖 Baichuan、ChatGLM、Deepseek、Kimi、MChat、Token、X元象、mistral、字节、文心一言、紫东太初、腾讯、讯飞、通义。第一次拿到这种集合式代码包的人通常有两种反应一种是觉得不过就是换个 URL 换个 key另一种是接了三家之后发现每家的鉴权、流式返回、参数命名全不一样直接翻车。这份源码真正值钱的地方不是它调通了某一家而是它把十几家厂商的差异摊平成了同一套调用骨架让你在 Python 里用接近一致的方式去接 Deepseek、Kimi、ChatGLM 这些主流模型也能顺手把文心一言、讯飞、通义这类国内平台串起来。它适合两类人一类是刚入门、想照着跑通第一个大模型 API 的 Python 学习者另一类是手里已经有业务、需要做多模型兜底或比价的中高级开发者。下面我按自己接这些接口的顺序把选型、骨架、参数和坑一条条讲清楚。2. 多模型接入的骨架怎么搭从统一接口到厂商适配层2.1 为什么不要每家写一份独立脚本新手最容易犯的错是给 Deepseek 写一个deepseek_demo.py给 Kimi 写一个kimi_demo.py给 ChatGLM 再写一个。跑通没问题但一旦你要做「同一个问题问三家、对比回答质量」或者要做失败自动切换这些散落的脚本就会变成维护灾难。常见做法是抽一层统一接口定义一个chat(messages, model, **kwargs)的抽象把「构造请求 → 发请求 → 解析响应 → 处理流式」这四步固定下来厂商差异全部下沉到各自的适配类里。这样上层业务只认一个函数换模型只改一个字符串。这份源码的目录结构基本就是这个思路一个base.py定义基类和通用返回结构然后每家一个文件继承它。你拿到手第一件事不是急着跑而是先看基类里约定了哪些字段——通常是model、messages、temperature、max_tokens、stream这几个。看懂这层约定后面十几家就是填空题。2.2 统一请求体的最小实现下面是我一般会写的基类骨架去掉了具体厂商细节只保留共性。你可以直接抄进自己的项目当起点。# base_client.py import requests from typing import List, Dict, Iterator class BaseAIClient: def __init__(self, api_key: str, base_url: str, timeout: int 60): self.api_key api_key self.base_url base_url self.timeout timeout def _headers(self) - Dict[str, str]: # 大多数厂商用 Bearer少数用自定义头子类覆盖 return { Authorization: fBearer {self.api_key}, Content-Type: application/json, } def _build_payload(self, messages: List[Dict], model: str, **kwargs) - Dict: # 统一入参子类按需增删字段 return { model: model, messages: messages, temperature: kwargs.get(temperature, 0.7), max_tokens: kwargs.get(max_tokens, 2048), stream: kwargs.get(stream, False), } def chat(self, messages: List[Dict], model: str, **kwargs) - str: payload self._build_payload(messages, model, **kwargs) resp requests.post( self.base_url, headersself._headers(), jsonpayload, timeoutself.timeout, ) resp.raise_for_status() return self._parse_response(resp.json()) def _parse_response(self, data: Dict) - str: # 默认按 OpenAI 兼容格式解析子类覆盖 return data[choices][0][message][content]这段代码的关键在三个方法的分工_headers管鉴权_build_payload管请求体_parse_response管响应解析。参数上timeout默认给 60 秒是因为 Deepseek、Kimi 这类在长文本生成时首包可能就要十几秒设太短会误判超时max_tokens给 2048 是个保守值通义和文心一言对上限有各自限制超了会直接报参数错误。temperature默认 0.7 是通用对话的稳妥值做代码生成时我一般调到 0.2 以下。2.3 流式返回为什么要单独处理非流式和流式的差别不只是「一个字一个字蹦出来」的体验问题。流式接口的响应体是 SSEServer-Sent Events格式一行行data: {...}最后以data: [DONE]结束。如果你用resp.json()去解析流式响应必然报错。所以基类里streamTrue时必须走另一条路径。# 流式解析放在基类里 def chat_stream(self, messages: List[Dict], model: str, **kwargs) - Iterator[str]: payload self._build_payload(messages, model, streamTrue, **kwargs) with requests.post( self.base_url, headersself._headers(), jsonpayload, timeoutself.timeout, streamTrue, ) as resp: resp.raise_for_status() for line in resp.iter_lines(): if not line: continue line line.decode(utf-8) if line.startswith(data: ): line line[6:] if line.strip() [DONE]: break chunk self._parse_stream_chunk(line) if chunk: yield chunk这里有个血泪经验iter_lines()默认按字节切中文多字节字符可能被切断所以一定要decode(utf-8)之后再判断前缀。另外不同厂商的结束标记不一样Deepseek 和 Kimi 用[DONE]但有些平台是空行结束或者自定义字段_parse_stream_chunk要留给子类覆盖。参数上streamTrue传给requests.post是为了不让它一次性把响应读进内存长回答时这个区别很明显。3. 各家 API 的鉴权与参数差异Deepseek、Kimi、ChatGLM 怎么接3.1 鉴权方式的三种流派接了这么多家鉴权基本分三类。第一类是标准 Bearer TokenDeepseek、Kimi、mistral、X元象大多如此请求头Authorization: Bearer sk-xxx就完事。第二类是「先换 token 再调用」典型是讯飞和部分腾讯接口需要先用 API Key 和 Secret 做一次签名或换取临时 token再拿 token 去请求token 还有有效期。第三类是文心一言这种走的是access_token拼接在 URL 上的方式需要先用 AK/SK 换 access_token。这三类的差异必须在适配层消化掉不能让业务代码感知。我的做法是在每个子类的__init__里完成鉴权准备把最终要用的 token 或签名函数存成实例属性_headers或_build_url直接取用。这样上层调用永远是client.chat(...)不关心背后是哪种流派。3.2 Deepseek 与 Kimi 的接入要点Deepseek 和 Kimi 都提供 OpenAI 兼容接口这意味着它们的请求体和响应格式跟 OpenAI 几乎一致_parse_response可以直接复用默认实现。差异主要在base_url和模型名上。Deepseek 的对话模型和推理模型要分开指定推理模型返回里会多一个reasoning_content字段如果你只取content思维链部分就丢了。Kimi 这边要注意的是长上下文它的模型对超长输入有分段计费逻辑max_tokens设太大不一定被接受。# deepseek_client.py from base_client import BaseAIClient class DeepSeekClient(BaseAIClient): def _parse_response(self, data): msg data[choices][0][message] # 推理模型会带 reasoning_content普通对话没有 if msg.get(reasoning_content): return msg[reasoning_content] \n msg[content] return msg[content]参数说明reasoning_content是推理类模型特有的普通对话模型返回里没有这个 key所以用.get()而不是直接索引避免 KeyError。如果你不需要思维链直接返回content即可但做复杂推理任务时保留它有助于排查模型是不是「想歪了」。3.3 ChatGLM 与通义的参数命名坑ChatGLM 和通义虽然也提供兼容接口但参数命名上有自己的习惯。比如有的平台把max_tokens叫max_new_tokens有的把temperature和top_p的默认值设得跟 OpenAI 不一样。更隐蔽的是部分平台对messages里role的取值有额外要求比如必须交替出现 user 和 assistant连续两个 user 会报错。我的处理方式是在子类的_build_payload里做一次字段映射和校验把统一入参翻译成厂商要的名字同时检查 messages 序列是否合法。这一步多写十几行能省掉后面大量「为什么这家报 400 那家正常」的排查时间。通义这边还要注意它的流式返回里finish_reason字段出现的位置和 OpenAI 不完全一致解析时要容错。4. 把十几家串起来跑批量调用、对比与失败切换的落地4.1 用工厂模式管理客户端十几家客户端如果每次都手动 new代码会又臭又长。常见做法是写一个注册表加工厂函数用模型名或厂商名做 key返回对应客户端实例。# factory.py from deepseek_client import DeepSeekClient from kimi_client import KimiClient from chatglm_client import ChatGLMClient REGISTRY { deepseek: DeepSeekClient, kimi: KimiClient, chatglm: ChatGLMClient, } def get_client(provider: str, api_key: str, **kwargs): cls REGISTRY.get(provider) if not cls: raise ValueError(f未知厂商: {provider}) return cls(api_keyapi_key, **kwargs)逻辑说明REGISTRY把厂商名映射到类新增一家只需加一行。get_client做一次存在性校验避免拼错厂商名时静默失败。参数上**kwargs透传给具体客户端方便传base_url或timeout这类厂商特有配置。4.2 同一问题多模型对比有了工厂做多模型对比就是循环的事。但要注意并发和限流十几家同时打很容易触发某家的 QPS 限制。我一般用线程池控制并发数在 3 到 5并且给每个请求加独立超时。from concurrent.futures import ThreadPoolExecutor, as_completed def compare(question, providers): messages [{role: user, content: question}] results {} with ThreadPoolExecutor(max_workers3) as pool: futures { pool.submit(get_client(p, KEY[p]).chat, messages, MODEL[p]): p for p in providers } for fut in as_completed(futures): provider futures[fut] try: results[provider] fut.result() except Exception as e: results[provider] f[失败] {e} return results参数说明max_workers3是经验值既能并行又不至于把免费额度瞬间打光。as_completed保证谁先返回谁先处理不会因为某家慢而卡住整体。异常被捕获后写进结果而不是让整个对比崩掉这样你能一眼看出是哪家挂了。4.3 失败切换与重试策略生产环境里单模型不可用是常态。失败切换的核心是主模型失败后按预设顺序尝试备用模型并且区分「可重试」和「不可重试」错误。429限流和 5xx 可以重试401鉴权失败和 400参数错误重试没意义直接换模型或报错。def chat_with_fallback(messages, chain, retries2): last_err None for provider in chain: client get_client(provider, KEY[provider]) for attempt in range(retries): try: return provider, client.chat(messages, MODEL[provider]) except requests.HTTPError as e: code e.response.status_code if code in (401, 400): break # 换下一家 last_err e except requests.Timeout as e: last_err e raise RuntimeError(f全部失败: {last_err})这里chain是备用顺序列表比如[deepseek, kimi, chatglm]。retries2表示同一家最多试两次避免在明显故障的厂商上浪费时间。注意 401 和 400 直接 break 跳出内层循环换下一家这是区分错误类型的关键不然会在一个鉴权错误的模型上反复重试。5. 避坑与排查多模型接入最常见的 5 个翻车点5.1 现象某家一直超时其他家正常原因通常有两个一是这家服务端确实慢二是你的timeout设得太短而这家首包延迟本来就高。排查时先把timeout临时调到 120 秒单独测这家如果还是超时基本是网络或服务端问题如果调大就好了说明是参数问题。解决方式是按厂商设置不同超时别用全局一个值。5.2 现象流式输出中文乱码或截断原因是没有正确处理多字节字符的边界或者iter_lines的 chunk 大小不合适。解决是在解码后再判断前缀并且对不完整的 JSON 做缓冲拼接。如果某家返回的 SSE 格式里data:后面没有空格你的startswith(data: )就会漏掉要兼容data:和data:两种。5.3 现象同样的 messages一家正常一家报 400原因是不同厂商对 messages 结构要求不同比如 role 必须交替、content 不能为空、system 消息位置有要求。排查时把报错那家的请求体打印出来逐字段对照它的文档。解决是在_build_payload里做规范化过滤空 content、确保 role 合法、必要时合并连续同角色消息。5.4 现象token 用量对不上账单比预期高原因是流式返回里 usage 字段可能只在最后一个 chunk 出现或者某些厂商根本不返回 usage。如果你按请求次数估算会严重低估。解决是显式收集每个响应里的 usage没有的话用本地 tokenizer 估算并且对长上下文模型特别留意输入 token 往往比输出贵。5.5 现象并发一高就大面积 429原因是没做限流或者限流阈值设得比厂商给的还高。排查时看错误码分布如果 429 集中在某几家就是这几家的配额问题。解决是给每家单独配一个信号量或令牌桶把并发压到配额以下并且把 429 纳入重试逻辑配合指数退避。6. 进阶把这份源码改造成你自己的模型路由层跑通示例只是起点真正有价值的是把它变成你自己的模型路由层。我一般会加三个东西。第一是配置外置把 API Key、base_url、模型名、超时、并发数全部写进一个 YAML 或环境变量代码里不出现硬编码这样换 key 不用改代码。第二是加一层缓存对相同问题和相同模型的请求做结果缓存尤其是做对比测试时能省掉大量重复调用。第三是加可观测性记录每次调用的厂商、模型、耗时、token 用量、是否命中缓存、是否走了 fallback这些数据积累下来你才能回答「Kimi 和 Deepseek 哪个更适合我的场景」这种问题而不是靠感觉。验证改造是否成功有个简单办法写一个测试脚本对同一批问题跑一遍全链路检查每个厂商的返回是否都能正确解析、失败切换是否按预期触发、缓存是否生效。我自己的习惯是每次新增一家厂商先单独跑通非流式和流式两条路径再接入路由层跑一次对比确认没有回归。这套流程走下来十几家模型对你来说就不再是十几个孤立的接口而是一个可以随时调度、随时替换的资源池。希望帮到你。本文还有配套的精品资源点击获取