openai-python SDK 用法详解：Chat、流式与异步

2026-06-24

openai-python SDK 是调用大模型 API 的最主流方式：安装一个包即可访问 OpenAI 及所有兼容平台，内置流式、异步、自动重试，无需手写 HTTP 模板代码。本文从安装到生产用法逐步拆解。

如果你是刚从”直接用 requests.post 拼 JSON”转过来的，会明显感觉到这个 SDK 帮你省掉了什么：请求头拼装、超时重试、流式分片解析、异步客户端复用连接池——这些都封装好了，你只要关心业务参数。但也正因为封装得好，一旦出问题（比如超时不生效、重试次数对不上），排查起来反而容易两眼一抹黑，所以下面每个坑我都尽量给出真实报错文案，方便你直接搜关键词定位。

安装与初始化

pip install openai

装完先跑一下 python -c "import openai; print(openai.__version__)" 确认版本，生产项目建议在 requirements.txt 里锁死具体版本号（比如 openai==1.35.0），别写 openai>=1.0——这个 SDK 更新频率不低，字段名、异常类名都可能在小版本间调整，锁版本能省掉很多”昨天还好好的今天突然报错”的排查时间。

推荐用环境变量注入凭证，OpenAI() 构造时自动读取：

import os
from openai import OpenAI

client = OpenAI(
    api_key=os.environ["OPENAI_API_KEY"],       # sk-xxx
    base_url=os.environ.get("OPENAI_BASE_URL",  # 可选，切换兼容平台
                            "https://api.openai.com/v1"),
)

只改 base_url 即可切换至力达云、DeepSeek、Moonshot 等任何 OpenAI 兼容接口，其余代码零改动。这里有个新手容易踩的细节：base_url 末尾不要带多余的斜杠或 /chat/completions，SDK 自己会拼接路径。填成 https://xxx.com/v1/ 或 https://xxx.com/v1/chat/completions 都会导致请求路径变成 .../v1//chat/completions 或 .../v1/chat/completions/chat/completions，报出 404，看起来像是”这家平台不支持”，其实只是路径拼错了。排查这类问题的第一步永远是打印一下最终请求的 URL，而不是急着怀疑平台不兼容。

另外 OpenAI() 构造函数不会在实例化时发起任何网络请求，key 和 base_url 填错了不会立刻报错，要等到第一次真正调用 chat.completions.create 时才会暴露问题。所以写完初始化代码，一定要紧跟一次最简单的调用验证连通性，别攒到业务逻辑写完了才发现 key 从一开始就没配对。

基础 Chat Completions

最小可运行示例：

response = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[
        {"role": "system", "content": "你是一个简洁的技术助手。"},
        {"role": "user",   "content": "解释什么是 token。"},
    ],
    max_tokens=300,
    temperature=0.7,
)
print(response.choices[0].message.content)

逐个参数说一下容易踩坑的地方：

messages 是有序列表，顺序决定语义。system 角色的内容只在第一条生效影响力最大，放在中间或者重复多条 system 大概率被模型忽略或者互相打架，写一条精炼的系统提示词就够了。
max_tokens 限的是”本次回复”的 token 上限，不是对话总长度。如果设得太小（比如 20），模型话说到一半就被硬截断，你会看到 finish_reason 是 "length" 而不是 "stop"——这是排查”回答被莫名截断”问题的第一个检查点，很多人第一反应是怀疑模型能力不够，其实只是这个参数给小了。
temperature 越高越发散、越低越保守。做代码生成、结构化抽取这类需要稳定输出的任务，建议设到 0～0.3；做文案、头脑风暴这类需要多样性的任务，可以拉到 0.8～1.2（部分平台上限是 2）。不确定选多少的时候，先固定在 0.7 这个折中值跑通流程，再针对具体任务微调。

多轮对话只需把 response.choices[0].message 追加回 messages 列表再次请求即可：

messages = [{"role": "system", "content": "你是一个简洁的技术助手。"}]
messages.append({"role": "user", "content": "解释什么是 token。"})
resp1 = client.chat.completions.create(model="gpt-4o-mini", messages=messages)
messages.append(resp1.choices[0].message.model_dump())  # 把回复原样存回去
messages.append({"role": "user", "content": "那 embedding 呢？"})
resp2 = client.chat.completions.create(model="gpt-4o-mini", messages=messages)

这里有个成本上的坑必须提醒：每一轮请求都会把完整的 messages 历史重新发送给模型计费，这不是缓存续接，是从头重发。对话轮数越多，prompt_tokens 涨得越快，十轮对话之后单次请求的成本可能是第一轮的好几倍。生产环境做多轮对话，务必设计一个”历史截断”策略——比如只保留最近 N 轮，或者当累计 token 超过某个阈值时对早期对话做摘要压缩，否则账单会比你预想的涨得快很多。具体计费口径和单价以你所用平台的官方页面为准。

流式输出（Streaming）

流式输出可显著降低用户感知延迟，适合聊天应用：

stream = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[{"role": "user", "content": "写一首关于 API 的短诗。"}],
    stream=True,
)
for chunk in stream:
    delta = chunk.choices[0].delta.content or ""
    print(delta, end="", flush=True)
print()  # 换行

每个 chunk 携带一小片 content，拼接后即完整响应。

底层发生了什么：stream=True 之后，服务端不再是等模型把整段话生成完再一次性返回，而是用 SSE（Server-Sent Events） 协议，模型每生成几个 token 就往连接里推一个事件，SDK 内部把这些事件解析成一个个 chunk 对象供你迭代。这也是为什么流式输出能显著降低”首字延迟”——用户不用等整段话生成完，看到第一个字蹦出来的时间通常能从两三秒降到几百毫秒。

写流式代码时有两个常见 bug：

忘记判空。chunk.choices[0].delta.content 在流的开头（角色声明帧）和结尾（结束帧）经常是 None 而不是空字符串，直接 print(chunk.choices[0].delta.content, end="") 会打印出一堆 None 混在正文里。上面示例里 delta.content or "" 这个写法就是专门防这个的，千万别删掉。
只看 content 不看 finish_reason。每个 chunk 的 choices[0].finish_reason 平时是 None，只有最后一个 chunk 会带上具体原因（stop 正常结束、length 被 max_tokens 截断、content_filter 被内容审查拦截）。如果你的应用有”回答是否完整”的判断逻辑，一定要在流结束后检查这个字段，而不是简单地认为”流断了就是说完了”。

如果你要做的是”边流式接收边拼前端 SSE 转发”（比如后端是 FastAPI，要把这个流原样转发给浏览器），可以直接在 async for 里 yield f"data: {delta}\n\n"，不需要等 SDK 流结束再统一发送，这样才能真正把低延迟优势带到用户那一端；如果中间又加一层”攒够整段再发”，流式就白做了。

异步调用（AsyncOpenAI）

高并发场景用 AsyncOpenAI，与 asyncio / FastAPI 无缝集成：

import asyncio
from openai import AsyncOpenAI

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

async def ask(prompt: str) -> str:
    resp = await aclient.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}],
    )
    return resp.choices[0].message.content

asyncio.run(ask("你好"))

单个请求用 AsyncOpenAI 看不出优势，它的价值在并发批量调用的场景——比如你要给一千条商品描述批量生成摘要，同步方式一条条等着跑完可能要几十分钟，异步并发能压缩到几分钟。但这里千万别写成”无脑全量并发”：

import asyncio
from openai import AsyncOpenAI

aclient = AsyncOpenAI(api_key=os.environ["OPENAI_API_KEY"])

async def ask_one(sem: asyncio.Semaphore, prompt: str) -> str:
    async with sem:  # 控制同时在跑的请求数
        resp = await aclient.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": prompt}],
        )
        return resp.choices[0].message.content

async def batch_ask(prompts: list[str], concurrency: int = 5):
    sem = asyncio.Semaphore(concurrency)  # 同时最多 5 个请求在跑
    tasks = [ask_one(sem, p) for p in prompts]
    return await asyncio.gather(*tasks, return_exceptions=True)

这里的 asyncio.Semaphore 是关键：一千条 prompt 如果直接 asyncio.gather 全部丢出去，瞬间一千个连接同时打到平台，大概率会被限流直接拒掉一大片，报错文案通常长这样：

openai.RateLimitError: Error code: 429 - {'error': {'message': 'Rate limit reached ...'}}

用信号量把并发数控制在 5～20（具体取值取决于你所用平台给的速率限额，官方文档一般会写清楚每分钟请求数上限），既能跑得比同步快得多，又不会一次性打爆限额。return_exceptions=True 也很重要——不加这个参数，gather 里只要有一个任务抛异常，整批任务都会被这一个异常拖垮而中断，加上之后失败的请求会变成结果列表里的异常对象，你可以单独挑出来重试，不影响其他已经成功的结果。

错误处理与重试

异常类	场景	建议处理
`openai.RateLimitError`	超出速率限制	指数退避重试（SDK 默认 2 次）
`openai.AuthenticationError`	key 无效或过期	检查 key 是否正确注入
`openai.BadRequestError`	请求参数错误	检查 messages 格式、max_tokens
`openai.APIConnectionError`	网络不可达	检查 base_url 或代理配置

SDK 默认自动重试 2 次，可通过 max_retries=3 调整：

client = OpenAI(api_key="sk-xxx", max_retries=3)

再补充几个上表没细说的实操细节：

重试只对特定错误生效，不是所有失败都会自动重试。 SDK 内置的自动重试只针对 RateLimitError（429）、APIConnectionError（网络层失败）以及部分 5xx 服务端错误，采用的是指数退避（第一次等约 0.5 秒，之后逐次翻倍，叠加一点随机抖动避免多个客户端同时重试撞车）。而 AuthenticationError（401）、BadRequestError（400）这类”重试了也不会变好”的错误，SDK 不会自动重试——因为 key 错了重试一百次还是错的，这是设计上的合理取舍，不是 bug。

超时也要单独配置，默认值不一定适合你的场景：

client = OpenAI(api_key="sk-xxx", timeout=30.0, max_retries=3)

默认超时通常是 10 分钟量级，对大部分交互式场景太长了——用户不可能等 10 分钟看一个接口卡住。如果你的场景是同步接口返回给前端，建议把 timeout 设到 30～60 秒，超时了就该给用户一个”请求超时请重试”的提示，而不是让请求方一直傻等。反过来，如果是跑长文档摘要这种单次生成量很大的任务，超时给太短反而会打断本该正常完成的请求，需要按实际生成时长留够余量。

context_length_exceeded 是另一个高频报错，报错文案类似：

openai.BadRequestError: Error code: 400 - {'error': {'message': "This model's maximum context length is 128000 tokens ..."}}

根因是 messages 累加的历史 + 本次输入 + max_tokens 预留的输出空间，三者加起来超过了模型的上下文窗口。修法有两个方向：一是前面提到的历史截断/摘要压缩；二是检查 max_tokens 是不是设得偏大，把它调小到实际需要的输出长度也能腾出空间。判断到底是哪部分超了，可以先用 tiktoken 库离线估算一下 messages 序列化后的 token 数，比每次靠报错来试错快得多。

编码问题：如果你的 prompt 或返回内容里有中文，且请求日志/终端打印出现乱码或 \uXXXX 转义序列，通常不是 SDK 的问题，而是终端编码或者你自己序列化时用了 json.dumps(obj) 却没加 ensure_ascii=False。排查这类问题记得先确认是打印环节乱码还是数据本身乱码，两者修法完全不同。

常见问题

Q：SDK 版本 v0.x 与 v1.x 有何区别？ v1.0 于 2023 年底发布，API 接口彻底重构，不再兼容旧版。现有项目若仍用 import openai; openai.ChatCompletion.create(...) 写法，需迁移到 OpenAI() 客户端模式。

Q：如何查看本次请求消耗的 token 数？ response.usage 包含 prompt_tokens、completion_tokens、total_tokens，可用于成本核算与限速保护。

Q：temperature=0 时输出是否确定性的？ 接近确定，但底层浮点运算和负载均衡可能导致极小差异。需严格可复现时，同时设置 seed 参数（部分平台支持）。

Q：同步客户端和异步客户端该怎么选？ 简单判断标准：如果调用方本身就是同步代码（脚本、定时任务、Django 视图不用 async），用 OpenAI() 就好，没必要为了”看起来更高级”硬套异步；如果调用方是 FastAPI 这类原生异步框架，或者你需要批量并发请求，才用 AsyncOpenAI()。混用是最容易出问题的——在同步函数里手动 asyncio.run() 调异步客户端，或者在异步框架里调同步客户端阻塞事件循环，都会带来不必要的复杂度和性能损耗，没有中间地带，选一种贯穿到底。

Q：换成兼容平台后，SDK 参数完全通用吗？ messages、temperature、max_tokens、stream 这些核心参数是通用的，因为兼容平台都是照着 OpenAI 的接口协议实现的。但有两类参数要留意：一是部分平台不支持某些高级参数（比如 logprobs、seed、tool_choice 的某些取值），传了不支持的参数可能报 400 或者被静默忽略，具体支持范围以你所用平台的官方文档为准；二是模型名字符串必须换成目标平台自己的模型标识，不能沿用 gpt-4o-mini 这类 OpenAI 专属命名，这个坑在切换 base_url 时最容易漏改，报错通常是”模型不存在”而不是连接失败，看到这类报错先查模型名对不对，再怀疑其他环节。

Q：RateLimitError 一直重试还是报 429，是不是账号被封了？ 大概率不是被封，而是触发了限速阈值（每分钟请求数 RPM 或每分钟 token 数 TPM 达到上限）。先看错误信息里有没有具体的限额数字，再检查是不是短时间内并发请求打得太密——参考前面异步小节的信号量做法把并发降下来，通常就能解决。如果确认没超限额还持续报错，才需要联系平台客服核实账号状态。

延伸阅读：大模型 API 接入完全指南 · 接入教程 Hub · Python 调用大模型 API 完整示例 · 改 base_url 切换 OpenAI 兼容接口