OpenAI: 迁移到 HTTPX2
摘要
OpenAI Python 库已更新，现支持 HTTPX2，带来了改进的 API 访问体验，包括类型定义、同步与异步客户端支持，以及全新的认证方法。
查看缓存全文
缓存时间: 2026/08/28 12:27
openai/openai-python
来源：https://github.com/openai/openai-python
OpenAI Python API 库
PyPI 版本
OpenAI Python 库为任何 Python 3.10+ 应用程序提供便捷访问 OpenAI REST API 的方式。 该库包含所有请求参数和响应字段的类型定义，并提供同步和异步客户端，由 HTTPX2 (https://httpx2.pydantic.dev/) 提供支持。 它基于我们的 OpenAPI 规范 (https://github.com/openai/openai-openapi) 生成。
文档
REST API 文档可在 platform.openai.com 找到。 本库的完整 API 文档位于 api.md。
安装
# 从 PyPI 安装
pip install openai
用法
本库的完整 API 文档位于 api.md。
与 OpenAI 模型交互的主要 API 是 Responses API (https://platform.openai.com/docs/api-reference/responses)。 你可以使用以下代码从模型生成文本：
import os
from openai import OpenAI
client = OpenAI(
# 这是默认值，可以省略
api_key=os.environ.get("OPENAI_API_KEY"),
)
response = client.responses.create(
model="gpt-5.5",
instructions="You are a coding assistant that talks like a pirate.",
input="How do I check if a Python object is an instance of a class?",
)
print(response.output_text)
生成文本的上一个标准（将无限期支持）是 Chat Completions API (https://platform.openai.com/docs/api-reference/chat)。 你可以使用该 API 通过以下代码从模型生成文本：
from openai import OpenAI
client = OpenAI()
completion = client.chat.completions.create(
model="gpt-5.5",
messages=[
{"role": "developer", "content": "Talk like a pirate."},
{
"role": "user",
"content": "How do I check if a Python object is an instance of a class?",
},
],
)
print(completion.choices[0].message.content)
虽然你可以提供 api_key 关键字参数，但我们建议使用 python-dotenv 将 OPENAI_API_KEY="My API Key" 添加到你的 .env 文件中，这样你的 API 密钥就不会存储在源代码控制中。
在此处获取 API 密钥 (https://platform.openai.com/settings/organization/api-keys)。
工作负载身份验证
对于像云托管的 Kubernetes、Azure 和 Google Cloud Platform 这样的安全、自动化环境，你可以使用工作负载身份验证，结合来自云身份提供商的短期令牌，而不是长期有效的 API 密钥。
Kubernetes（服务账户令牌）
from openai import OpenAI
from openai.auth import k8s_service_account_token_provider
client = OpenAI(
workload_identity={
"identity_provider_id": "idp-123",
"service_account_id": "sa-456",
"provider": k8s_service_account_token_provider(
"/var/run/secrets/kubernetes.io/serviceaccount/token"
),
},
)
response = client.chat.completions.create(
model="gpt-5.5",
messages=[{"role": "user", "content": "Hello!"}],
)
Azure（托管身份）
from openai import OpenAI
from openai.auth import azure_managed_identity_token_provider
client = OpenAI(
workload_identity={
"identity_provider_id": "idp-123",
"service_account_id": "sa-456",
"provider": azure_managed_identity_token_provider(
resource="https://management.azure.com/",
),
},
)
Google Cloud Platform（计算引擎元数据）
from openai import OpenAI
from openai.auth import gcp_id_token_provider
client = OpenAI(
workload_identity={
"identity_provider_id": "idp-123",
"service_account_id": "sa-456",
"provider": gcp_id_token_provider(audience="https://api.openai.com/v1"),
},
)
自定义主体令牌提供程序
from openai import OpenAI
def get_custom_token() -> str:
return "your-jwt-token"
client = OpenAI(
workload_identity={
"identity_provider_id": "idp-123",
"service_account_id": "sa-456",
"provider": {
"token_type": "jwt",
"get_token": get_custom_token,
},
}
)
你还可以自定义令牌刷新缓冲时间（默认为过期前 1200 秒（20 分钟））：
from openai import OpenAI
from openai.auth import k8s_service_account_token_provider
client = OpenAI(
workload_identity={
"identity_provider_id": "idp-123",
"service_account_id": "sa-456",
"provider": k8s_service_account_token_provider("/var/token"),
"refresh_buffer_seconds": 120.0,
}
)
X.509 工作负载身份（双向 TLS）
对于 X.509 工作负载身份联合，请在 HTTPX2 客户端上配置客户端证书和服务器信任，然后仅将身份提供程序和服务账户 ID 传递给 SDK：
import os
import ssl
from openai import OpenAI, DefaultHttpx2Client
from openai.auth import x509_workload_identity
tls_context = ssl.create_default_context(
cafile=os.getenv("OPENAI_MTLS_CA_BUNDLE"),
)
tls_context.load_cert_chain(
certfile=os.environ["OPENAI_MTLS_CERTIFICATE_CHAIN"],
keyfile=os.environ["OPENAI_MTLS_PRIVATE_KEY"],
password=os.getenv("OPENAI_MTLS_PRIVATE_KEY_PASSWORD"),
)
client = OpenAI(
workload_identity=x509_workload_identity(
identity_provider_id=os.environ["OPENAI_IDENTITY_PROVIDER_ID"],
service_account_id=os.environ["OPENAI_SERVICE_ACCOUNT_ID"],
# refresh_buffer_seconds=120.0,
),
http_client=DefaultHttpx2Client(
verify=tls_context,
follow_redirects=False,
),
)
当未设置 base_url 和 OPENAI_BASE_URL 时，X.509 模式默认为 https://mtls.api.openai.com/v1。相同的配置 HTTP 客户端会向固定的 mTLS 令牌交换端点和 API 呈现其证书。令牌是延迟交换的，会被缓存，并自动刷新。证书文件、私钥、密码、服务器信任、代理和轮换仍然是应用程序和传输层的关注点。X.509 API 请求需要 HTTPS，并且必须保持在配置的 API 源上。有效的 HTTP Host 权限必须与该源匹配。不能在 X.509 认证旁边向 API 发送提供程序 API 密钥和仅代理标头。令牌交换不会继承 API 请求钩子、身份验证或 Cookie。身份设置在客户端构造时捕获；创建新客户端以更改身份。Azure 客户端不支持 X.509 工作负载身份。对于异步请求，请使用 AsyncOpenAI 和 DefaultAsyncHttpx2Client。请参阅完整的同步部署切换示例和异步部署切换示例，它们使用应用程序拥有的 OPENAI_AUTH_MODE 环境变量来选择 API 密钥或 X.509 身份验证。X.509 工作负载身份目前支持 HTTP API；不包括实时和 WebSocket。
视觉
使用图像 URL：
prompt = "What is in this image?"
img_url = "https://upload.wikimedia.org/wikipedia/commons/thumb/d/d5/2023_06_08_Raccoon1.jpg/1599px-2023_06_08_Raccoon1.jpg"
response = client.responses.create(
model="gpt-5.5",
input=[
{
"role": "user",
"content": [
{"type": "input_text", "text": prompt},
{"type": "input_image", "image_url": f"{img_url}"},
],
}
],
)
使用 Base64 编码字符串的图像：
import base64
from openai import OpenAI
client = OpenAI()
prompt = "What is in this image?"
with open("path/to/image.png", "rb") as image_file:
b64_image = base64.b64encode(image_file.read()).decode("utf-8")
response = client.responses.create(
model="gpt-5.5",
input=[
{
"role": "user",
"content": [
{"type": "input_text", "text": prompt},
{"type": "input_image", "image_url": f"data:image/png;base64,{b64_image}"},
],
}
],
)
异步用法
只需导入 AsyncOpenAI 代替 OpenAI，并在每个 API 调用中使用 await：
import os
import asyncio
from openai import AsyncOpenAI
client = AsyncOpenAI(
# 这是默认值，可以省略
api_key=os.environ.get("OPENAI_API_KEY"),
)
async def main() -> None:
response = await client.responses.create(
model="gpt-5.5",
input="Explain disestablishmentarianism to a smart five year old."
)
print(response.output_text)
asyncio.run(main())
同步和异步客户端之间的功能是相同的。
使用 aiohttp
默认情况下，异步客户端使用 HTTPX2。为了获得更好的并发性能，你也可以使用 aiohttp 作为 HTTPX2 传输。
你可以通过安装 aiohttp 来启用它：
# 从 PyPI 安装
pip install openai[aiohttp]
然后你可以通过使用 http_client=DefaultAioHttpClient() 实例化客户端来启用它：
import os
import asyncio
from openai import DefaultAioHttpClient
from openai import AsyncOpenAI
async def main() -> None:
async with AsyncOpenAI(
api_key=os.environ.get("OPENAI_API_KEY"),
# 这是默认值，可以省略
http_client=DefaultAioHttpClient(),
) as client:
chat_completion = await client.chat.completions.create(
messages=[
{
"role": "user",
"content": "Say this is a test",
}
],
model="gpt-5.5",
)
asyncio.run(main())
HTTPX2 迁移
HTTPX2 是默认的 HTTP 客户端。如果你配置了自定义 HTTP 客户端、传输、超时、身份验证处理程序、事件钩子或请求模拟，请参阅 HTTPX2 迁移指南。
流式响应
我们使用服务器发送事件 (SSE) 提供对流式响应的支持。
from openai import OpenAI
client = OpenAI()
stream = client.responses.create(
model="gpt-5.5",
input="Write a one-sentence bedtime story about a unicorn.",
stream=True,
)
for event in stream:
print(event)
异步客户端使用完全相同的接口。
import asyncio
from openai import AsyncOpenAI
client = AsyncOpenAI()
async def main():
stream = await client.responses.create(
model="gpt-5.5",
input="Write a one-sentence bedtime story about a unicorn.",
stream=True,
)
async for event in stream:
print(event)
asyncio.run(main())
实时 API
实时 API 使你能够构建低延迟、多模式的对话体验。它目前通过 WebSocket 连接支持文本和音频作为输入和输出，以及函数调用 (https://platform.openai.com/docs/guides/function-calling)。
在底层，SDK 使用 websockets (https://websockets.readthedocs.io/en/stable/) 库来管理连接。
实时 API 通过客户端发送的事件和服务器发送的事件组合来工作。客户端可以发送事件来更新会话配置或发送文本和音频输入。服务器事件确认音频响应何时完成，或何时收到模型的文本响应。完整的事件参考可在此处找到 (https://platform.openai.com/docs/api-reference/realtime-client-events)，指南可在此处找到 (https://platform.openai.com/docs/guides/realtime)。
基于文本的基本示例：
import asyncio
from openai import AsyncOpenAI
async def main():
client = AsyncOpenAI()
async with client.realtime.connect(model="gpt-realtime-2") as connection:
await connection.session.update(
session={"type": "realtime", "output_modalities": ["text"]}
)
await connection.conversation.item.create(
item={
"type": "message",
"role": "user",
"content": [{"type": "input_text", "text": "Say hello!"}],
}
)
await connection.response.create()
async for event in connection:
if event.type == "response.output_text.delta":
print(event.delta, flush=True, end="")
elif event.type == "response.output_text.done":
print()
elif event.type == "response.done":
break
asyncio.run(main())
然而，实时 API 的真正魔力在于处理音频输入/输出，请参阅此 TUI 脚本示例 (https://github.com/openai/openai-python/blob/main/examples/realtime/push_to_talk_app.py) 以获取一个完整的示例。
实时错误处理
每当发生错误时，实时 API 将发送一个 error 事件 (https://platform.openai.com/docs/guides/realtime-model-capabilities#error-handling)，连接将保持打开并可用。这意味着你需要自己处理它，因为当 error 事件发生时，SDK 不会直接引发任何错误。
client = AsyncOpenAI()
async with client.realtime.connect(model="gpt-realtime-2") as connection:
...
async for event in connection:
if event.type == 'error':
print(event.error.type)
print(event.error.code)
print(event.error.event_id)
print(event.error.message)
使用类型
嵌套的请求参数是 TypedDicts (https://docs.python.org/3/library/typing.html#typing.TypedDict)。 响应是 Pydantic 模型 (https://docs.pydantic.dev)，它们也提供辅助方法，例如：
- 序列化回 JSON，
model.to_json() - 转换为字典，
model.to_dict()
类型化的请求和响应在编辑器中提供自动完成和文档。如果你想在 VS Code 中看到类型错误以帮助更早地发现错误，请将 python.analysis.typeCheckingMode 设置为 basic。
分页
OpenAI API 中的列表方法是分页的。本库为每个列表响应提供了自动分页迭代器，因此你不必手动请求后续页面：
from openai import OpenAI
client = OpenAI()
all_jobs = []
# 根据需要自动获取更多页面。
for job in client.fine_tuning.jobs.list(
limit=20,
):
# 在这里对 job 执行某些操作
all_jobs.append(job)
print(all_jobs)
或者，异步地：
import asyncio
from openai import AsyncOpenAI
client = AsyncOpenAI()
async def main() -> None:
all_jobs = []
# 遍历所有页面的项，根据需要发出请求。
async for job in client.fine_tuning.jobs.list(
limit=20,
):
all_jobs.append(job)
print(all_jobs)
asyncio.run(main())
或者，你可以使用 .has_next_page()、.next_page_info() 或 .get_next_page() 方法对处理页面进行更细粒度的控制：
first_page = await client.fine_tuning.jobs.list(
limit=20,
)
if first_page.has_next_page():
print(f"will fetch next page using these details: {first_page.next_page_info()}")
next_page = await first_page.get_next_page()
print(f"number of items we just fetched: {len(next_page.data)}")
# 对于非异步用法，请移除 `await`。
或者直接使用返回的数据：
first_page = await client.fine_tuning.jobs.list(
limit=20,
)
print(f"next page cursor: {first_page.after}")
# => "next page cursor: ..."
for job in first_page.data:
print(job.id)
# 对于非异步用法，请移除 `await`。
嵌套参数
嵌套参数是字典，使用 TypedDict 进行类型化，例如：
from openai import OpenAI
client = OpenAI()
response = client.responses.create(
input=[
{
"role": "user",
"content": "How much ?",
}
],
model="gpt-5.5",
text={"format": {"type": "json_object"}},
)
文件上传
对应文件上传的请求参数可以作为 bytes、PathLike (https://docs.python.org/3/library/os.html#os.PathLike) 实例或 (filename, contents, media type) 元组传递。
from pathlib import Path
from openai import OpenAI
client = OpenAI()
client.files.create(
file=Path("input.jsonl"),
purpose="fine-tune",
)
异步客户端使用完全相同的接口。如果你传递 PathLike (https://docs.python.org/3/library/os.html#os.PathLike) 实例，文件内容将自动异步读取。
Webhook 验证
验证 webhook 签名是 可选但鼓励的。有关 webhook 的更多信息，请参阅 API 文档 (https://platform.openai.com/docs/guides/webhooks)。
解析 webhook 载荷
对于大多数用例，你可能希望同时验证 webhook 并解析载荷。为此，我们提供 client.webhooks.unwrap() 方法，它解析 webhook 请求并验证它是否由 OpenAI 发送。如果签名无效，此方法将引发错误。注意，body 参数必须是从服务器发送的原始 JSON 字符串（不要先解析它）。.unwrap() 方法将在验证 webhook 是由 OpenAI 发送后，将此 JSON 解析为事件对象。
from openai import OpenAI
from flask import Flask, request
app = Flask(__name__)相似文章
@OpenAIDevs: https://x.com/OpenAIDevs/status/2054973436280418521
OpenAI Developers 账户在 X 上发布了一条更新。具体内容未知，但很可能与 OpenAI 开发者工具或服务相关。
@OpenAI: https://x.com/OpenAI/status/2095527557924082061
OpenAI 发布了一条包含链接的推文，可能分享关于其人工智能技术的新闻或更新。
@ClementDelangue：OpenAI 今天在 HF 上新发布了一个模型！
OpenAI 今天在 Hugging Face 发布了一个新模型。
OpenAI Agents API
OpenAI的Agents API提供了一个托管平台，用于构建AI代理应用程序，具有会话、编排以及用于代码执行和工具交互的沙箱环境等功能。
OpenAI o1 和开发者新工具
OpenAI 向 API 发布 o1 模型，具备生产就绪的功能，包括函数调用、结构化输出、视觉能力，以及比 o1-preview 低 60% 的延迟。其他开发者工具包括 Realtime API 改进、偏好微调，以及新的 Go 和 Java SDK。