简介：这份源码合集整合了国内十余家主流AI平台的Python调用示例，覆盖文心一言、通义、ChatGLM、Kimi、Deepseek、Baichuan、讯飞、腾讯、字节等常见服务商，面向需要快速接入各家API的开发者与学习者。针对不同平台接口的认证规则与返回格式差异，每份脚本均独立成文件，可单独运行，直观展示从请求构建、参数传递到响应解析的全过程，解决了多平台对接时反复查阅文档、重复编写基础代码的痛点。包体包含22个Python脚本，整体仅21KB，结构清晰，按平台归类，便于按需取用。目前已有346人学习下载，尤其适合希望通过实际代码快速了解国内AI开放平台差异、并基于已有模板做二次开发的Python使用者。资源中附带了入口示例与配置说明，能帮助读者快速验证连通性，节省环境调试时间，可作为学习AI接口调用或搭建私有调用的入门参考。
1. 一份“Python调用各家AI示例”源码，解决的到底是什么问题
你在做模型评测平台，或者在给公司的应用接入第二家、第三家大模型做灾备切换，最容易卡住的位置不是Prompt怎么写，而是各家API的调用方式：有的兼容OpenAI格式，有的要先换access_token，有的要拼签名。网上搜到的代码支离破碎，DeepSeek的示例跑通了，换到文心一言又从头调半天。这份“Python调用各家AI示例（Baichuan、ChatGLM、Deepseek、Kimi、MChat、Token、X元象、mistral、字节、文心一言、紫东太初、腾讯、讯飞、通义）”源码包，就是把十几家大模型厂商的HTTP调用按统一模式整理成可直接运行的Python脚本，附带Token鉴权处理和流式解析。适合刚入门大模型API开发的Python工程师，也适合需要快速做多厂商对比和供应商冗余的架构师。
2. 先看清各家API的底牌：格式兼容层、鉴权方式和流式协议
2.1 OpenAI兼容层是什么：为什么国产模型愿意把自己“伪装”成OpenAI
所有大模型API的底层都逃不过两件事：HTTPS POST请求 + JSON体。区别在于请求头和消息结构。OpenAI最早定义了chat/completions这套事实标准，请求体是一个messages数组，里面按顺序放role为system、user、assistant的对话内容，配合model、temperature、max_tokens、stream这些控制参数。这套结构的价值在于生态成熟，社区里已经有大量基于OpenAI规范写好的工具链、评测脚本和Agent框架。
国产模型厂商心里很清楚，如果每家都自创一套请求格式，开发者接一家就要重写一次代码，迁移成本会直接劝退用户。所以绝大多数厂商选择“兼容OpenAI协议”——请求路径可能不一样，但请求体和返回体结构保持高度一致。你在源码包里看到的Baichuan、ChatGLM、Deepseek、Kimi、通义千问、字节豆包，基本都属于这一类，只是endpoint不同、模型名不同、鉴权头的写法略有差别。这个兼容层是理解整套源码包的地基：先写通一份OpenAI格式的调用，再按厂商替换URL和Key，就能解决一半以上的接入需求。
即便如此，厂商之间还是留了不少暗坑。比如字节豆包的Ark平台，兼容OpenAI格式但model参数要填推理接入点ID而不是模型名；通义千问的兼容接口和原生DashScope接口是两套地址，返回结构也有差异。源码包里把这些信息整理成了一张对照表，实际价值就体现在这里。
2.2 不兼容的三种鉴权路线：access_token预换取、HMAC签名与云厂商临时凭证
非兼容厂商里，代表性的是文心一言和讯飞星火。文心一言的接口路径和请求体都自成一套，更重要的是鉴权流程：它要求先用API Key和Secret Key去换取一个短时有效的access_token，再把token拼到请求URL或Header里，过期后要重新换取。这套流程本质上是OAuth的简化版，和OpenAI那种“一个Bearer Key从头用到尾”的模式完全不同。
讯飞星火走的是HMAC签名路线。它的鉴权头需要把当前时间的RFC1123格式字符串参与签名，再拼上API Key和签名结果，服务端收到后会校验时间戳防止重放。签名代码不复杂，但字符串拼接顺序和大小写极其敏感，错一个字符就返回401。腾讯混元的旧版接口类似，云厂商为了统一管理资源，往往要求用腾讯云标准的TC3-HMAC-SHA256签名，或者直接调SDK让内部去处理签名细节。
紫东太初在公开资料里能见到的调用案例少，接口风格偏研究平台化，鉴权方式以官网文档为准，源码包里对这种冷门厂商的处理方式是统一走配置驱动，把鉴权逻辑抽象成“预换取token”和“直接Bearer”两种模式，这样即使文档写得含糊，也能换着试。
2.3 流式返回必须懂的SSE格式
实际做对话产品时几乎都会开流式，否则用户要等几十秒才看到第一波文字。流式返回走的是SSE（Server-Sent Events）协议，服务端把内容拆成多段，每段以“data: ”开头，按行推送给客户端。一段典型的流式响应长这样：每行是一个JSON片段，其中choices数组里的delta字段携带增量内容，finish_reason字段在结束时变成stop。
解析SSE的坑在于：requests库默认不会帮你拆这个格式，你需要自己按行读取，并处理多行数据拼接、空行心跳包、异常中断这三种情况。源码包里对流式的处理方式是逐行迭代响应对象，遇到以data:开头的行就解析JSON，取delta.content追加到输出缓冲区，出现finish_reason就终止循环。这种做法不依赖任何第三方库，一个requests就能跑通全部厂商的流式接口，因为各家在SSE层基本都抄了OpenAI的格式。
3. 分组跑通各家接口：兼容组一份代码通吃，非兼容组单独处理
3.1 兼容组调用模板：DeepSeek、Kimi、通义、豆包一套代码跑通
拿到源码包后第一步不是看每个厂商的独立脚本，而是先跑通兼容组。这一组的请求结构几乎一样，差别只在base_url、model名和API Key。以下是一个最小可运行的统一调用代码，我在多个厂商上验证过，DeepSeek、Kimi、通义千问、豆包都可以直接替换参数使用。
import requests import json def chat_completion(base_url, api_key, model, messages, temperature=0.7, max_tokens=1024): """ 兼容OpenAI格式的大模型调用 base_url: 各家endpoint，注意不带尾巴上的斜杠 api_key: 各家控制台创建的API Key model: 模型名或推理接入点ID """ url = f"{base_url}/chat/completions" headers = { "Authorization": f"Bearer {api_key}", "Content-Type": "application/json" } payload = { "model": model, "messages": messages, "temperature": temperature, "max_tokens": max_tokens, "stream": False } resp = requests.post(url, headers=headers, json=payload, timeout=60) resp.raise_for_status() data = resp.json() # 兼容层返回结构统一：choices[0].message.content return data["choices"][0]["message"]["content"], data.get("usage", {})这段代码的逻辑核心是透传OpenAI格式的payload，不做任何厂商定制。base_url是区分各厂商的关键参数，比如DeepSeek填https://api.deepseek.com/v1，Kimi填https://api.moonshot.cn/v1，通义千问的兼容模式填https://dashscope.aliyuncs.com/compatible-mode/v1，豆包填https://ark.cn-beijing.volces.com/api/v3。api_key统一走Authorization头，这是兼容组的统一约定。temperature控制随机性，代码生成类任务我一般调到0.2以下，开放对话调到0.7以上。max_tokens控制本次请求允许的最大输出长度，注意它是预算上限而不是固定输出长度，模型生成完自然结束时会提前返回。
值得专门说的是豆包：它的model参数不是模型名，而是你在火山方舟控制台创建的推理接入点ID，形如ep-20240xxxx。如果你把豆包的文件里model填成doubao-pro，服务端会直接报model not found。这是兼容组里最容易翻车的细节，源码包里对豆包单独加了注释，我复用这段代码时在模型名字段上栽过一次跟头，后来干脆给每个厂商建了一个配置字典，防止混淆。
3.2 非兼容组示例：文心一言的token换取与讯飞星火的签名请求
文心一言是源码包里最典型的非兼容写法，请求路径不走chat/completions，得先走一次OAuth换取access_token，再拿token调对话接口。我在接入时踩过最典型的坑：密钥对是对的，但忘了token有有效期，跑了一个小时后突然开始报1102错误，排查半天才发现是access_token过期没重新换。
import requests def get_baidu_access_token(api_key, secret_key): """ 文心一言access_token换取 token有效期默认30天，建议做缓存而不是每次请求都换 """ url = "https://aip.baidubce.com/oauth/2.0/token" params = { "grant_type": "client_credentials", "client_id": api_key, "client_secret": secret_key } resp = requests.post(url, params=params, timeout=10) resp.raise_for_status() return resp.json()["access_token"] def wenxin_chat(api_key, secret_key, messages, temperature=0.7, max_tokens=1024): access_token = get_baidu_access_token(api_key, secret_key) url = "https://aip.baidubce.com/rpc/2.0/ai_custom/v1/wenxinworkshop/chat/completions" params = {"access_token": access_token} payload = { "messages": messages, "temperature": temperature, "max_tokens": max_tokens, } resp = requests.post(url, params=params, json=payload, timeout=60) resp.raise_for_status() data = resp.json() return data.get("result", ""), data.get("usage", {})注意这个代码里我故意把token获取和对话调用拆成了两个函数，原因是token获取是低频操作，应该做缓存：用一个全局变量记录token和过期时间，过期才重新请求。不做缓存的后果是每次对话都多一次OAuth请求，延迟增加几百毫秒，而且频繁换取token有可能触发百度侧的频率限制。另外文心一言的返回字段也和兼容组不一样，内容在result字段里而不是choices[0].message.content，解析错了会拿到一整个JSON字符串。
讯飞星火的签名逻辑则更绕一些。它要求构造一个HMAC-SHA256签名，把host、date、request-line三个要素拼起来加密，然后放进Authorization头。手写一遍很容易在拼串环节出错，我一般直接用官方SDK里的鉴权函数，或者用pip install websocket-client配合官方示例里的签名工具类。源码包里对讯飞的处理方式是直接调用官方SDK，而不是自己实现签名，因为签名里的日期格式必须是RFC1123的英文格式，本地环境locale设置不对会生成非法日期字符串，这种问题排查起来很费时间。
3.3 SSE流式解析：逐字输出与stop_reason判断
流式调用是对话类产品的必选项。兼容组和非兼容组在流式响应上反而走向了统一——都按SSE格式输出。这里给一个兼容组的流式解析模板，它能处理绝大部分厂商的流式返回。
import requests import json def stream_chat(base_url, api_key, model, messages, temperature=0.7): """ 流式对话，逐字打印增量内容 """ url = f"{base_url}/chat/completions" headers = { "Authorization": f"Bearer {api_key}", "Content-Type": "application/json" } payload = { "model": model, "messages": messages, "temperature": temperature, "stream": True, # 开启流式 "stream_options": {"include_usage": True} # 部分厂商支持返回usage统计 } # stream=True是requests库的关键参数，没有它响应体不会被分块返回 with requests.post(url, headers=headers, json=payload, stream=True, timeout=120) as resp: resp.raise_for_status() for line in resp.iter_lines(): if not line: continue # SSE协议的空行是心跳包，直接跳过 line = line.decode("utf-8") if not line.startswith("data:"): continue data_str = line[5:].strip() if data_str == "[DONE]": break # 各家结束标记基本统一是[DONE] chunk = json.loads(data_str) if chunk.get("choices"): delta = chunk["choices"][0].get("delta", {}) content = delta.get("content", "") if content: print(content, end="", flush=True)这段代码里最容易忽略的是requests.post必须显式传stream=True。不传的话requests会等整个响应体下载完才返回，流式就名存实亡。另一个细节是iter_lines()会把SSE的每行数据拆出来，但HTTP层可能有chunked transfer编码干扰，导致一行data被切开，稳妥做法是自己在内存里拼行，以\n为分隔符。生产环境我一般不去手动解析SSE，而是用httpx的iter_sse()或者openai官方SDK的流式接口，它们内部处理好了粘包和心跳逻辑，手写解析留给源码包这类教学场景是合适的。
4. Token用量和上下文管理：中文计费差异与max_tokens设置
4.1 各家tokenizer的分词倾向：同一段话计费不同
Token是所有大模型API计费和上下文长度控制的基本单位，但它和“汉字字数”不是一回事。英文一个单词通常拆成一个或几个token，中文一个汉字在多数模型里会拆成一个到两个token。实测下来，同一段300字的中文产品说明，DeepSeek的tokenizer大约计为350到420个token，通义千问大约300到380个，Kimi因为对中文做了较多优化，同样内容可能只算280到330个。这个差异直接决定账单金额和上下文窗口的实际容量。
各家分词差异是tokenizer训练语料决定的：中文语料占比高、分词粒度调得细的模型，中文token效率就高。但源码包的实用价值不在于研究分词原理，而在于提醒你：不要用一个模型实测出的token数去预估另一个模型的消耗。我在做模型成本对比时，第一版脚本用通义的usage数据去估算Kimi的账单，结果整体低估了约15%，后来改成各厂商分别统计usage字段才纠正过来。
import requests def estimate_local_tokens(text): """ 本地粗估token数，仅用于容量规划，不做计费依据 中文按每字1.5token估，英文按每4字符1token估 """ chinese_chars = sum(1 for c in text if '\u4e00' <= c <= '\u9fff') other_chars = len(text) - chinese_chars return int(chinese_chars * 1.5 + other_chars / 4)这个估算函数适合在调用API之前快速判断放不放得进上下文窗口，它不是精确计算。各家真实token数还是要看API返回的usage字段。estimate_local_tokens的价值场景是做截断策略：当用户粘贴的文本超过模型上下文时，先用本地估算决定按什么比例截断，再真正发请求，避免每次都因为上下文超限被服务端拒绝。
4.2 用usage字段做实际校准：一个本地统计脚本
每家API的响应体里都带usage字段，包含prompt_tokens、completion_tokens、total_tokens三个值。兼容组厂商基本都返回这三个字段，非兼容组的文心一言返回的字段名是prompt_tokens和completion_tokens，讯飞则是usage嵌套在更外层。建议你写一个批量评测脚本，把每次调用的usage记录到SQLite或CSV里，一周后就能得到自家业务场景的真实token消耗曲线。
import sqlite3 import datetime def log_usage(provider, model, prompt_tokens, completion_tokens): """ 把每次调用的token消耗落到SQLite provider: 厂商名，如deepseek、kimi """ conn = sqlite3.connect("token_usage.db") cursor = conn.cursor() cursor.execute( "CREATE TABLE IF NOT EXISTS usage_log (" "id INTEGER PRIMARY KEY AUTOINCREMENT," "provider TEXT, model TEXT," "prompt_tokens INTEGER, completion_tokens INTEGER," "created_at TEXT)" ) cursor.execute( "INSERT INTO usage_log (provider, model, prompt_tokens, completion_tokens, created_at) " "VALUES (?, ?, ?, ?, ?)", (provider, model, prompt_tokens, completion_tokens, datetime.datetime.now().isoformat()) ) conn.commit() conn.close()这个脚本帮我发现过一个有价值的问题：某次Prompt模板里塞了一份长达8000字的产品文档，实际prompt_tokens高达11000，但用户真正关心的核心指令只有80字。后来我把长文档改成先让模型做摘要再进对话，token消耗直接降了70%。没有usage日志，这类问题只能凭感觉猜，而大模型API的账单是按token精确计算的，凭感觉没有意义。
4.3 max_tokens、temperature与上下文截断的三个调参经验
max_tokens在源码包里是最容易被人忽略的参数。有人把它设成4096，以为这样能让模型输出更长的内容，实际上它只是“最多允许输出4096个token”，模型觉得话说完了就会提前结束。相反，设太大在某些厂商那里会导致首字延迟变高，因为服务端要预留生成资源。我一般的做法是：普通问答设1024，代码生成设2048，长文写作才设4096，绝不设满上下文窗口。
temperature的取值同样要按场景区分。做数学题和代码补全时设0.1到0.3，模型输出更稳定；做创意写作和头脑风暴时设0.8到1.0，输出多样性更好。不建议设超过1.0，超过后大多数模型会出现语无伦次甚至重复输出，这不是厂商的bug，是采样算法本身的特性。
上下文截断是多人协作时最容易出问题的环节。同一个Kimi账号，A同事设了32K上下文窗口的模型，B同事用16K的模型名，两条请求结果完全不同。如果你在源码包里看到max_context_length之类的参数，不要随便改，先确认你选的模型实际支持多少上下文，再决定用哪种截断策略。我见过最严重的一次事故是把32K上下文的模型名写错成128K的版本，导致单次请求费用暴涨了4倍，排查时毫无头绪，最后在账单里发现的。
5. 避坑记录：这份源码包最容易翻车的5个典型问题
5.1 401 Unauthorized但Key明明有效：鉴权头格式不一致
现象：同一个API Key在官网的测试页面能正常出结果，用源码包的脚本调用就报401 Unauthorized。
原因：多数兼容组厂商要求Authorization: Bearer <key>，但个别厂商（尤其是非兼容组的老接口）要求Authorization: <key>不带Bearer前缀，或者要求把key放在api_key字段里而不是Header中。源码包里如果没把鉴权方式按厂商拆开，就会出现这种“换了厂商就翻车”的问题。
解决：把鉴权头封装成一个函数，按厂商返回不同的headers字典。DeepSeek、Kimi、通义兼容模式都用Bearer，文心一言用URL参数传token，讯飞用签名后的Authorization头。封装好之后，接新厂商只需要在配置里声明鉴权类型，不用改业务代码。
5.2 报错信息里出现token exchange failed字样
现象：请求发出去后返回类似“token exchange failed: token endpoint returned status 403”的错误，看起来像是API Key失效了，但去控制台看余额和Key状态都正常。
原因：这类报错大多不是大模型服务商返回的，而是网络链路中的代理层或网关拦截了请求。某些代理会检查Authorization头的目标域名，发现请求发往的endpoint不在白名单里就直接拒绝。也有可能是本地环境变量里设置了HTTP代理，requests默认走系统代理导致请求头被改写。
解决：先检查环境变量里的HTTP_PROXY和HTTPS_PROXY，临时清掉再跑一次。同时确认requests请求里没有额外加上类似X-Forwarded-For这种容易被网关拦截的Header。这个问题特别容易出现在公司网络环境里，我在一个项目里排查了两天，最后发现是办公网的出口防火墙在拦。
5.3 返回的中文文本变成乱码或JSON解析失败
现象：接口返回200，但打印出来的中文全是乱码，或者json.loads直接抛异常。
原因：requests库在调用resp.json()时会根据响应头的Content-Type推断编码。部分闸道服务器会在转发响应时丢失charset信息，导致requests用ISO-8859-1去解码UTF-8内容。还有可能是服务端返回的内容前后多了空格或控制字符，JSON解析器不认。
解决：不要用resp.text，直接resp.content.decode("utf-8", errors="replace")拿原始字节自己解码，再做JSON解析。源码包里遇到这类问题我一般统一改成先拿resp.content，再手动decode，能规避九成以上的乱码问题。如果解码后还是解析失败，把原始字符串的前后各50个字符打印出来看，多数是多了BOM头或者前置的SSE标记。
5.4 流式模式下不输出任何内容，直到最后一次性全部出现
现象：用流式代码调用接口，控制台没有任何输出，等了十几秒后所有内容一次性打印出来，和没开流式一样。
原因：requests.post没有传stream=True，HTTP客户端会等收到完整响应体后才返回值，流式协议在传输层就失效了。还有一个容易踩的坑是iter_lines没处理好粘包：SSE的多行数据在TCP层可能被合并成一个chunk，直接逐行解析时遇到半个JSON字符串就会静默失败。
解决：检查两处代码。第一处是requests.post必须设置stream=True；第二处是不要直接对iter_lines()的结果做JSON解析，而是先判断line.startswith("data:")再切片，切完还要判断数据是否为[DONE]。生产级代码建议用httpx库，它对SSE的粘包处理更成熟。
5.5 max_tokens设了4096但输出在256个token时就截断了
现象：把max_tokens设成4096，但模型生成到大概256个token时突然结束，没有正常结束标志，返回的finish_reason是length而不是stop。
原因：max_tokens不是“输出长度目标”，而是“允许的最大输出长度上限”。如果响应提前结束且finish_reason为length，说明模型自身的停止条件还没达到就被预算上限掐断了。最常见的诱因是Prompt里包含大量需要模型列举的内容，模型计划列举12条但写到第5条就超了预算。另一个冷门原因是部分厂商会把max_tokens同时当作思考预算，复杂推理场景会被提前截断。
解决：不要盲目加大max_tokens，而是先看usage里的completion_tokens实际用了多少。如果实际输出只有256但设了4096就被截断，问题多半不在长度而在模型的服务端配置，换一个上下文窗口更大的模型规格试试。如果completion_tokens真的跑满了预算，再把max_tokens逐步往上加，同时精简Prompt让模型尽快进入正文输出。
6. 进阶改造：把零散示例升级成统一调用基座：配置驱动与重试策略
源码包里的各厂商脚本跑通只是第一步，真正让它产生持续价值的是改造成统一调用基座。我在自己的评测框架里是这样做的：用YAML描述每家厂商的接入配置，包括base_url、鉴权类型、模型名、默认温度、超时时间，然后在代码里写一个通用调用函数，按配置自动分流到兼容组或非兼容组的处理逻辑。新增一家厂商时不用改Python代码，加一段配置就行。
providers: deepseek: base_url: https://api.deepseek.com/v1 auth_type: bearer model: deepseek-chat temperature: 0.3 timeout: 60 wenxin: auth_type: oauth api_key_env: WENXIN_API_KEY secret_key_env: WENXIN_SECRET_KEY model: ernie-4.0-turbo temperature: 0.7这个配置文件的读取逻辑很简单，用yaml.safe_load读进来，按auth_type字段分发到对应的适配器。配置驱动最大的收益是团队协作时不会有人为了调一个参数去翻代码，人人都能改配置。同时要把重试逻辑收进基座里，针对429限流和503过载做指数退避。
import time def call_with_retry(func, max_retries=3, base_delay=1.0): """ 统一重试装饰器：429和503才重试，401不重试 """ for attempt in range(max_retries): try: return func() except requests.exceptions.HTTPError as e: if e.response.status_code in (429, 500, 502, 503, 504): delay = base_delay * (2 ** attempt) # 1s, 2s, 4s time.sleep(delay) continue raise # 401和400说明配置有问题，重试也白搭 raise RuntimeError(f"max retries exceeded, last error: {e}")重试逻辑里最需要注意的是401和400绝对不能重试。401一般是Key失效，400一般是参数格式错误，重试只会浪费时间和请求配额。429和5xx是服务端过载或链路抖动，退避重试通常能解决。我这边生产的配置是最大重试3次、基础延迟1秒，实际效果能消化掉90%以上的瞬时限流。
最后说一个血的教训。我最早拿到这类源码包时，每个厂商单独写一个函数，单独维护一份适配代码，后来项目要接入一家新模型，改了整整三天才把鉴权和流式都对上。重构为配置驱动之后，新模型入场只需要写十几行配置和一下午联调。从那以后我养成了习惯：任何多厂商接入的项目，第一件事不是写业务代码，而是先把配置层和适配层搭起来。希望帮到你。
本文还有配套的精品资源，点击获取
