Python 调用大模型 API 实战:一套代码跑通 OpenAI、DeepSeek、智谱

2026 年，OpenAI GPT、DeepSeek、智谱 GLM 等主流大模型几乎都提供了 OpenAI API 兼容协议。大多数开发者仍然为每个平台单独写一套调用代码，维护成本高，切换困难。实际上，用一个 OpenAI SDK 配合不同的 base_url 和 api_key，就能跑通所有兼容平台。本文从 API Key 管理、统一调用封装，到流式输出、并发调用、重试容错和成本控制，给出一套可直接用于生产的完整方案。

一、API Key 管理：别硬编码

API Key 直接写在代码里是最常见的隐患。一旦提交到 Git 仓库，Key 会永久留在提交历史中，后续删除也无法彻底清除。

反例：硬编码 Key

from openai import OpenAI

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

正例：用 python-dotenv 从 .env 文件加载

OPENAI_API_KEY=sk-xxxxxxxxxxxxxxxxxxxxxxxx
DEEPSEEK_API_KEY=sk-yyyyyyyyyyyyyyyyyyyy
ZHIPU_API_KEY=zzzzzzzzzzzzzzzzzzzzzzzz

from dotenv import load_dotenv
import os

load_dotenv()  # 从 .env 文件加载环境变量到 os.environ

api_key = os.getenv("DEEPSEEK_API_KEY")

同时，在 .gitignore 中排除 .env 文件，防止密钥泄露：

.env
.env.*

安装依赖：pip install openai python-dotenv。

二、统一调用封装：一个 Client 切三家

OpenAI SDK 的 base_url 参数是切换平台的核心。DeepSeek 和智谱 GLM 都兼容 OpenAI Chat Completions 接口，只需替换 base_url、api_key 和 model 三个参数即可。

各平台配置如下：

封装一个统一的调用函数，通过 provider 参数切换：

from openai import OpenAI
from dotenv import load_dotenv
import os

load_dotenv()

PROVIDERS = {
    "openai": {
        "api_key": os.getenv("OPENAI_API_KEY"),
        "base_url": "https://api.openai.com/v1",
        "model": "gpt-4o",
    },
    "deepseek": {
        "api_key": os.getenv("DEEPSEEK_API_KEY"),
        "base_url": "https://api.deepseek.com",
        "model": "deepseek-v4-pro",
    },
    "zhipu": {
        "api_key": os.getenv("ZHIPU_API_KEY"),
        "base_url": "https://open.bigmodel.cn/api/paas/v4/",
        "model": "glm-4",
    },
}

def llm_chat(prompt, provider="deepseek", system="你是一个专业助手"):
    """统一调用函数，通过 provider 切换平台"""
    config = PROVIDERS[provider]
    client = OpenAI(
        api_key=config["api_key"],
        base_url=config["base_url"],
    )
    response = client.chat.completions.create(
        model=config["model"],
        messages=[
            {"role": "system", "content": system},
            {"role": "user", "content": prompt},
        ],
    )
    return response.choices[0].message.content

print(llm_chat("用一句话解释什么是向量数据库", provider="deepseek"))

print(llm_chat("用一句话解释什么是向量数据库", provider="zhipu"))

print(llm_chat("用一句话解释什么是向量数据库", provider="openai"))

切换平台只需改一个 provider 参数，调用逻辑完全一致。注意智谱 GLM 的 base_url 末尾带斜杠，漏掉可能导致部分 SDK 版本拼接路径出错。

三、流式响应：实现打字机效果

非流式调用需要等模型生成完整内容后才返回，用户等待时间长。流式响应通过 stream=True 参数，逐块返回内容，实现打字机效果。

from openai import OpenAI
from dotenv import load_dotenv
import os

load_dotenv()

def stream_chat(prompt, provider="deepseek", system="你是一个专业助手"):
    """流式调用，逐字打印"""
    config = PROVIDERS[provider]
    client = OpenAI(
        api_key=config["api_key"],
        base_url=config["base_url"],
    )
    stream = client.chat.completions.create(
        model=config["model"],
        messages=[
            {"role": "system", "content": system},
            {"role": "user", "content": prompt},
        ],
        stream=True,  # 开启流式输出
    )
    full_text = ""
    for chunk in stream:
        delta = chunk.choices[0].delta.content
        if delta:
            print(delta, end="", flush=True)
            full_text += delta
    print()  # 换行
    return full_text

stream_chat("写一个 Python 快速排序的实现并解释原理", provider="deepseek")

flush=True 确保每个字符立即输出到终端，不经过缓冲区。Web 应用中，将 delta 通过 SSE（Server-Sent Events）推送到前端即可实现同样的逐字渲染效果。

四、并发调用：asyncio 批量提速

批量处理场景下，同步逐条调用是性能瓶颈。100 条数据逐条请求，每条平均 3 秒，总耗时约 5 分钟。

同步调用（慢）：

import time

prompts = [f"总结以下文本的要点：文本{i}" for i in range(100)]

start = time.time()
results = [llm_chat(p, provider="deepseek") for p in prompts]
print(f"同步耗时：{time.time() - start:.1f}s")  # 约 300s

异步并发调用（快）：

import asyncio
from openai import AsyncOpenAI

async def async_chat(prompt, provider="deepseek", system="你是一个专业助手"):
    config = PROVIDERS[provider]
    client = AsyncOpenAI(
        api_key=config["api_key"],
        base_url=config["base_url"],
    )
    response = await client.chat.completions.create(
        model=config["model"],
        messages=[
            {"role": "system", "content": system},
            {"role": "user", "content": prompt},
        ],
    )
    return response.choices[0].message.content

async def async_batch_chat(prompts, provider="deepseek", max_concurrency=10):
    """批量并发调用，Semaphore 控制并发数"""
    semaphore = asyncio.Semaphore(max_concurrency)

    async def limited_chat(prompt):
        async with semaphore:  # 超过并发数时自动等待
            return await async_chat(prompt, provider)

    results = await asyncio.gather(*[limited_chat(p) for p in prompts])
    return results

prompts = [f"总结以下文本的要点：文本{i}" for i in range(100)]
start = time.time()
results = asyncio.run(async_batch_chat(prompts, provider="deepseek", max_concurrency=10))
print(f"异步并发耗时：{time.time() - start:.1f}s")  # 约 30s

Semaphore(10) 将并发数限制在 10，避免触发平台的速率限制。同步 100 条约 300 秒，异步并发约 30 秒，提速 10 倍。实际并发数需根据各平台速率限制调整，DeepSeek 和智谱的默认限制通常在每分钟数十到数百次请求，超出会返回 429 错误。

五、重试与容错：tenacity 处理 429/超时

大模型 API 调用中常见的错误有三类：429 限流、5xx 服务器错误、网络超时。直接调用遇到这些错误会立即中断，需要手动重试。

tenacity 库提供 @retry 装饰器，支持指数退避重试：

from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
from openai import (
    OpenAI, RateLimitError, APIConnectionError,
    APITimeoutError, InternalServerError
)
from dotenv import load_dotenv
import os

load_dotenv()

@retry(
    stop=stop_after_attempt(5),  # 最多 5 次尝试（1 次初始 + 4 次重试）
    wait=wait_exponential(multiplier=2, min=2, max=60),  # 指数退避：2s, 4s, 8s, 16s
    retry=retry_if_exception_type(
        (RateLimitError, APIConnectionError, APITimeoutError, InternalServerError)
    ),
)
def retry_chat(prompt, provider="deepseek", system="你是一个专业助手"):
    """带自动重试的调用封装"""
    config = PROVIDERS[provider]
    client = OpenAI(
        api_key=config["api_key"],
        base_url=config["base_url"],
        timeout=30,  # 请求超时 30 秒
    )
    response = client.chat.completions.create(
        model=config["model"],
        messages=[
            {"role": "system", "content": system},
            {"role": "user", "content": prompt},
        ],
    )
    return response.choices[0].message.content

result = retry_chat("解释什么是 RAG 检索增强生成", provider="deepseek")

异常分类处理策略：

400 和 401 类错误重试也不会成功，不列入重试范围，直接抛出。wait_exponential 的退避序列为 2s → 4s → 8s → 16s，给服务端足够的恢复时间。

六、Token 计数与成本控制

大模型按 Token 计费，调用前估算 Token 数和成本，能避免预算超支。

tiktoken 是 OpenAI 的 Token 计数库，支持 cl100k_base（GPT-4 系列）和 o200k_base（GPT-4o 系列）等编码：

import tiktoken

def count_tokens(text, encoding_name="cl100k_base"):
    """计算文本的 Token 数"""
    encoding = tiktoken.get_encoding(encoding_name)
    return len(encoding.encode(text))

def estimate_cost(input_tokens, output_tokens, provider="deepseek"):
    """估算单次调用成本（单位：元/百万Token，价格以官方为准）"""
    pricing = {
        "openai":   {"input": 17.5, "output": 70.0},   # gpt-4o
        "deepseek": {"input": 2.0,  "output": 8.0},     # deepseek-v4-pro
        "zhipu":    {"input": 0.5,  "output": 0.5},     # glm-4-flash
    }
    price = pricing[provider]
    cost = (input_tokens * price["input"] + output_tokens * price["output"]) / 1_000_000
    return cost

prompt = "请详细解释 Transformer 架构中自注意力机制的原理"
input_tokens = count_tokens(prompt)
output_tokens = 500  # 假设输出 500 Token

cost = estimate_cost(input_tokens, output_tokens, provider="deepseek")
print(f"输入 Token：{input_tokens}，预估成本：{cost:.6f} 元")

def batch_budget_check(prompts, provider="deepseek", budget=1.0, est_output=500):
    """批量任务预算检查，超预算时提前终止"""
    total_cost = 0
    for i, prompt in enumerate(prompts):
        input_tokens = count_tokens(prompt)
        total_cost += estimate_cost(input_tokens, est_output, provider)
        if total_cost > budget:
            print(f"第 {i+1} 条时预算超限，预估总成本 {total_cost:.4f} 元")
            return False
    print(f"预估总成本：{total_cost:.4f} 元，在预算内")
    return True

prompts = [f"分析以下文本的关键信息：文本{i}" for i in range(1000)]
batch_budget_check(prompts, provider="deepseek", budget=1.0)

批量任务执行前先跑一遍 batch_budget_check，能在不实际调用 API 的情况下评估成本，避免意外超支。tiktoken 对中文的计数与实际可能有少量偏差，智谱和 DeepSeek 各自的 Tokenizer 也略有不同，最终计费以平台返回的 usage 字段为准。

生产级调用检查清单

将以上方案整合，一套生产级的大模型调用代码应满足以下条件：

• [ ] API Key 不硬编码，通过 .env 环境变量加载

• [ ] .env 已加入 .gitignore

• [ ] 统一封装调用函数，通过 provider 参数切换平台

• [ ] 支持流式输出，提升用户体验

• [ ] 批量任务使用 asyncio 并发，Semaphore 控制并发数

• [ ] 重试机制覆盖 429/5xx/超时，指数退避策略

• [ ] 400/401 类错误不重试，直接抛出

• [ ] Token 计数与成本预估，批量任务执行前做预算