Python 调用大模型 API 完整示例

2026-06-14

Python 调用大模型 API 有两条路：用 requests 直发 HTTP，或用 openai SDK（兼容所有 OpenAI 格式平台）。推荐生产项目优先用 SDK，内置重试、流式与异步支持；快速脚本或定制需求可用 requests。

这两条路不是纯粹的口味问题，选错了是要还债的。比如你团队里已经有个统一的 HTTP 客户端封装（带链路追踪、统一日志），硬塞一个 openai SDK 进去反而破坏了原本的可观测性，这时候 requests 拼一下更省事；但如果你要做的是一个稍微正式点的服务，涉及重试、超时、流式，自己拿 requests 从零造轮子基本上就是把 SDK 已经踩过的坑重新踩一遍。下面这张表是我自己在几个项目里踩出来的经验，供你对照：

场景	推荐方案	原因
临时脚本、一次性跑批	`requests`	不需要额外依赖，几行代码就能跑
已有统一 HTTP 客户端封装	`requests`	复用现有的日志/追踪/熔断逻辑
正式服务，长期维护	`openai` SDK	重试、流式、类型提示都是现成的，少踩坑
需要异步高并发	`openai` SDK（`AsyncOpenAI`）	官方维护的异步客户端，比自己包一层 `requests` 靠谱
对接非 OpenAI 格式的私有协议	`requests`	SDK 假设了 OpenAI 的请求/响应结构，格式不兼容时改起来比较别扭

环境准备

pip install openai requests

把 API key 写入环境变量，绝不硬编码：

export OPENAI_API_KEY="sk-xxx"
export OPENAI_BASE_URL="https://api.lidayun.com/v1"  # 换成你的端点

这里有两个容易被忽略的细节。第一，base_url 结尾要不要带斜杠：openai SDK 内部会自动拼接路径（比如 /chat/completions），带不带尾斜杠通常都能兼容，但如果你是自己拼 URL（走 requests 那条路），多一个斜杠就会拼出 //chat/completions 这种双斜杠路径，有些网关会 404，有些又能兼容，行为不统一，建议养成习惯：base_url 一律不带尾斜杠。第二，OPENAI_BASE_URL 这个环境变量名是社区约定俗成的，openai SDK 本身默认只认 OPENAI_API_KEY，base_url 需要你显式从环境变量读出来传进 OpenAI(...) 构造函数，就像下面代码里写的那样——不要以为设了环境变量 SDK 就自动会用，它不会。

方式一：requests 直发 HTTP

适合不想引入额外依赖、或需要完全控制请求头的场景：

import os
import requests

API_KEY = os.environ["OPENAI_API_KEY"]
BASE_URL = os.environ.get("OPENAI_BASE_URL", "https://api.openai.com/v1")

def chat(messages: list[dict], model: str = "gpt-4o-mini") -> str:
    resp = requests.post(
        f"{BASE_URL}/chat/completions",
        headers={
            "Authorization": f"Bearer {API_KEY}",
            "Content-Type": "application/json",
        },
        json={
            "model": model,
            "messages": messages,
            "max_tokens": 1024,
        },
        timeout=30,
    )
    resp.raise_for_status()          # 非 2xx 直接抛异常
    return resp.json()["choices"][0]["message"]["content"]

# 单轮调用
reply = chat([{"role": "user", "content": "用 Python 实现快速排序"}])
print(reply)

raise_for_status() 会把 4xx/5xx 转成 HTTPError，便于统一捕获。这一步很多人图省事直接省略，后果是接口挂了、返回一段 HTML 错误页时，代码会在 resp.json()["choices"][0]["message"]["content"] 这一行炸出一个 KeyError 或 JSONDecodeError，排查起来完全摸不着头脑——你以为是解析逻辑写错了，其实根本没走到解析那一步，是上游服务本身就没返回合法响应。养成习惯：凡是网络请求，先判断状态码再解析 body，这条铁律在任何 HTTP 调用场景里都成立，不只是大模型 API。

timeout=30 这个参数也值得多说两句。requests 默认是不设超时的，也就是说如果对端网络异常、连接建立后卡住不返回，你的程序会一直死等，线上服务这么写基本等于埋了个雷。30 秒对大部分对话场景够用，但如果你的 prompt 很长、或者模型在思考复杂问题（比如推理类模型），生成时间可能超过 30 秒，这时候要么把超时调大到 60~120 秒，要么改用流式接口边生成边接收（不用等全部生成完），流式的具体做法可以看流式输出 SSE：原理与各语言实现，这里就不重复展开了。

如果你要频繁调用，requests.post 每次都会新建一个 TCP 连接，握手开销白白浪费。改用 Session 复用连接，尤其是批量跑数据的场景，能明显省时间：

session = requests.Session()
session.headers.update({
    "Authorization": f"Bearer {API_KEY}",
    "Content-Type": "application/json",
})

def chat_with_session(messages, model="gpt-4o-mini"):
    resp = session.post(
        f"{BASE_URL}/chat/completions",
        json={"model": model, "messages": messages, "max_tokens": 1024},
        timeout=30,
    )
    resp.raise_for_status()
    return resp.json()["choices"][0]["message"]["content"]

Session 会复用底层的 TCP 连接（keep-alive），批量调用几百上千次的场景下，这个改动能省下不少握手时间，比单纯优化代码逻辑更立竿见影。

方式二：openai SDK（推荐）

openai 包兼容所有支持 Chat Completions 格式的平台，只需改 base_url：

import os
from openai import OpenAI

client = OpenAI(
    api_key=os.environ["OPENAI_API_KEY"],
    base_url=os.environ.get("OPENAI_BASE_URL", "https://api.openai.com/v1"),
)

def chat(messages: list[dict], model: str = "gpt-4o-mini") -> str:
    resp = client.chat.completions.create(
        model=model,
        messages=messages,
        max_tokens=1024,
    )
    return resp.choices[0].message.content

print(chat([{"role": "user", "content": "解释什么是 token"}]))

OpenAI(...) 这个客户端对象建议在模块顶层实例化一次，全局复用，不要在每个函数调用里重新 OpenAI(...)。原因和上面 Session 复用是一个道理：客户端内部维护了连接池，反复创建等于反复丢弃这个连接池，白白浪费握手开销。另外 SDK 自带的重试机制默认是 max_retries=2，也就是说你什么都不配置的情况下，遇到网络抖动它已经会自动重试两次，这也是为什么很多人觉得”SDK 比 requests 稳”——不是错觉，是真的帮你兜了一层底。想调整这个次数，在实例化时传 OpenAI(max_retries=5) 就行；想完全禁用自动重试（比如你要自己实现更精细的退避策略，见下文），传 max_retries=0。

max_tokens 这个参数经常被人想当然地设成一个很大的数图省事，但要注意它限制的是”本次回复最多生成多少 token”，不是输入长度。设得过大在有些平台会被直接拒绝（超出模型上限），设得过小则会导致回复被硬生生截断，一句话说到一半戛然而止——排查这个问题的方法很简单，看返回体里的 finish_reason 字段，如果是 "length" 就说明是被 max_tokens 截断的，不是模型不想说了，调大这个值就行；如果是 "stop" 才是模型正常说完。

多轮对话：维护 messages 历史

多轮对话的关键是把每轮的 user + assistant 消息都追加到 messages：

def multi_turn_chat():
    history = [{"role": "system", "content": "你是一个 Python 编程助手。"}]

    while True:
        user_input = input("你: ").strip()
        if user_input.lower() in ("quit", "exit", "退出"):
            break

        history.append({"role": "user", "content": user_input})
        reply = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=history,
            max_tokens=512,
        )
        assistant_msg = reply.choices[0].message.content
        history.append({"role": "assistant", "content": assistant_msg})
        print(f"助手: {assistant_msg}\n")

        # 超过 20 条时丢弃最早的用户/助手轮次（保留 system）
        if len(history) > 21:
            history = [history[0]] + history[-20:]

multi_turn_chat()

这段代码里最关键的一行是 history.append(...)，很多人写多轮对话第一次踩坑就是忘了把 assistant 的回复也存回 history——只存了 user 的问题，下一轮再问模型时，模型完全不知道自己上一轮说过什么，表现出来就是”这模型怎么什么都不记得”，其实是你自己没喂给它历史，模型本身是无状态的，每次调用都是一张白纸，“记忆”完全是靠你在客户端把历史拼回 messages 数组里实现的。

第 106~108 行那段截断逻辑也值得说清楚为什么这么写。多轮对话如果不加限制，history 会无限增长，每一轮请求都要把全部历史重新发一遍给模型，这意味着：一是费用会越滚越大（几乎所有平台按输入 + 输出的总 token 数计费，历史越长单次请求越贵）；二是迟早会撞到模型的 context 窗口上限，报错通常长这样：This model's maximum context length is 128000 tokens...，届时请求直接被拒绝，而不是优雅降级。代码里选择”保留 system + 最近 20 条”是个简单粗暴但好用的策略，代价是模型会”忘记”更早之前聊过的内容；如果你的场景需要长期记忆（比如客服机器人要记住用户很久前提过的诉求），更好的做法是把旧历史摘要成一两句话压缩存起来，而不是整段丢弃，具体摘要压缩的做法我在这篇文章里没有展开，你可以按同样的截断思路自己加一层”超过阈值就调用模型做一次摘要”的逻辑。

错误处理与重试

生产环境需处理限流（429）和临时服务异常（500/503），建议指数退避：

import time
from openai import RateLimitError, APIStatusError

def chat_with_retry(messages, model="gpt-4o-mini", max_retries=3):
    for attempt in range(max_retries):
        try:
            resp = client.chat.completions.create(
                model=model,
                messages=messages,
                max_tokens=1024,
            )
            return resp.choices[0].message.content
        except RateLimitError:
            wait = 2 ** attempt          # 1s, 2s, 4s…
            print(f"限流，{wait}s 后重试…")
            time.sleep(wait)
        except APIStatusError as e:
            if e.status_code >= 500 and attempt < max_retries - 1:
                time.sleep(2 ** attempt)
            else:
                raise
    raise RuntimeError("重试耗尽")

这段重试逻辑区分了两种性质完全不同的错误，不能一锅烩。RateLimitError（对应 HTTP 429）说明请求本身没问题，只是你调用太频繁了，服务端主动限流，这种情况”等一等再试”通常就能成功；APIStatusError 里 5xx 的部分说明服务端自己临时出了问题（比如后端某个实例挂了正在重启），这种也适合重试；但如果是 4xx 里的其他错误，比如 401（认证失败）或 400（请求参数不合法），重试是没有意义的——key 不对重试一百次还是不对，参数格式错了重试一百次还是错，这类错误应该直接往上抛，让人去修代码或者配置，而不是在重试循环里空转浪费时间。代码里 except APIStatusError 那段专门判断了 e.status_code >= 500 才重试，就是这个考虑。

2 ** attempt 这种写法叫指数退避，第一次等 1 秒、第二次 2 秒、第三次 4 秒，间隔越来越长。为什么不固定等 1 秒重试三次？因为如果是服务端真的在过载，所有客户端都在同一时刻疯狂重试，只会让过载更严重，指数退避加上一点随机抖动（业界叫 jitter，比如 2 ** attempt + random.uniform(0, 1)）能把重试请求在时间上错开，减轻服务端压力，这是分布式系统里的标准做法，不是大模型 API 独有的技巧。

实际调用时你大概率会遇到的几个报错，直接照这个表查：

报错现象	HTTP 状态码	根因	怎么修
`AuthenticationError: Incorrect API key provided`	401	key 错误、已过期或被吊销	重新生成 key，检查环境变量有没有多余空格/换行
`RateLimitError: Rate limit reached`	429	调用频率或并发超出限额	指数退避重试，或申请提高限额
`APIConnectionError` / `ConnectTimeout`	无（网络层）	DNS 解析失败、代理没配对、网络不通	先 `curl` 一下 `base_url` 确认能通
`This model's maximum context length is...`	400	messages 总 token 超出模型上限	截断历史或做摘要压缩，见上文多轮对话部分
返回内容被截断，没说完就停	200（正常）	`finish_reason` 是 `"length"`	调大 `max_tokens`
中文输出乱码或问号	200（正常）	终端/文件没用 UTF-8 编码打印或写入	Windows 下用 `print` 前设 `PYTHONIOENCODING=utf-8`，写文件时显式 `open(..., encoding="utf-8")`

高并发场景下，同步的 chat_with_retry 一次只能等一个请求返回，想要真正的并发，得用 SDK 自带的异步客户端 AsyncOpenAI，配合 asyncio.gather 一次性发出去多个请求：

import asyncio
from openai import AsyncOpenAI

async_client = AsyncOpenAI(
    api_key=os.environ["OPENAI_API_KEY"],
    base_url=os.environ.get("OPENAI_BASE_URL", "https://api.openai.com/v1"),
)

async def chat_async(messages, model="gpt-4o-mini"):
    resp = await async_client.chat.completions.create(
        model=model, messages=messages, max_tokens=1024,
    )
    return resp.choices[0].message.content

async def batch_chat(prompt_list):
    tasks = [chat_async([{"role": "user", "content": p}]) for p in prompt_list]
    return await asyncio.gather(*tasks, return_exceptions=True)

# results = asyncio.run(batch_chat(["问题1", "问题2", "问题3"]))

注意 asyncio.gather 里加了 return_exceptions=True，这样即使某一个请求失败（比如触发了限流），也不会让整批任务全部中断，失败的那一项在返回列表里会是个异常对象，你自己判断类型做处理即可，正常项照常拿到字符串结果。没有这个参数，只要有一个任务抛异常，其余还没跑完的任务会被直接取消，这是很多人第一次写并发调用最容易踩的坑。

并发数也不是越高越好，大部分平台对单个 key 有并发上限（比如同时最多 N 个请求在处理），超过这个数只会换来更多的 429，实际项目里建议用 asyncio.Semaphore 卡住并发上限，比如 Semaphore(5) 表示同一时刻最多 5 个请求在飞，其余的排队等前面的完成再发，这样既能提速又不会把限流打爆。

费用这块很多人上生产前根本没估算过，等账单出来才傻眼。大致的估算方法是：先拿你的典型 prompt 跑一次，用 resp.usage.prompt_tokens 和 resp.usage.completion_tokens 拿到这次调用实际消耗的输入、输出 token 数，再乘以你的日调用量、乘以对应的单价（单价以你所用平台当前公布的价目为准，各平台差异较大，这里不写死具体数字），就能大致估出月成本量级。多轮对话场景要格外注意，因为每一轮都会把历史重新发一遍，实际消耗的输入 token 会比你感觉的”只问了一句话”高得多，这也是上文强调历史截断的另一个现实原因——不只是为了不超 context，也是为了控制成本。

常见问题

requests 和 openai SDK 哪个更好？ 生产项目推荐 SDK：内置重试策略、流式支持、类型提示更完整。requests 适合轻量脚本或需要完全控制请求细节的场景。

调用报 401 错误怎么排查？ 先确认环境变量已正确 export；在 Python 内 print(os.environ.get("OPENAI_API_KEY")) 确认读到值；再确认 key 尚未过期或被吊销。

messages 历史太长导致 context 超限怎么办？ 保留 system prompt + 最近 N 轮对话，丢弃中间历史；或对历史做摘要后压缩成一条 user 消息。

同步代码要不要改成异步？ 不是所有场景都值得。如果你只是写个人用的小脚本、一次调用一次，同步的 openai SDK 完全够用，改异步反而增加了代码复杂度（async/await 到处传染）；只有当你需要”同时发出去几十上百个请求、等它们并发完成”这种场景（比如批量处理一批文档），异步才划算，日常单条调用场景没必要为了异步而异步。

本地测试没问题，上线后偶尔超时怎么办？ 先看是不是网络出口的问题——本地开发机和服务器出网的网络质量、到 API 服务商的物理距离都可能不一样，服务器如果部署在境外机房访问境内平台（或者反过来）延迟会明显变高。排查思路：先用 curl -w "%{time_total}\n" -o /dev/null -s <base_url> 测一下裸的网络延迟，如果这个数字本身就很大，说明问题不在你的代码里，是网络链路的问题，得从换机房、加代理或者换个延迟更低的接入点入手，而不是一味调大 timeout 参数糊弄过去。

用第三方中转平台调用，和直连官方 API 在代码上有什么区别？ 几乎没有区别——只要中转平台兼容 OpenAI 的 Chat Completions 协议，你需要改的只有 base_url 和对应的 api_key，本文里所有的 chat、multi_turn_chat、chat_with_retry 函数都不用改一行。这也是为什么”兼容 OpenAI 协议”在这个生态里这么重要：换供应商基本等于换一个环境变量，代码零改动。

更多接入细节见大模型 API 接入完全指南与接入教程专题。如需流式逐 token 输出，参考流式输出 SSE：原理与各语言实现。需要一套 key 跨多个模型？申请力达云聚合 API 内测。