openai/openai-python

Source: https://github.com/openai/openai-python

OpenAI Python API 库

PyPI 版本) (https://pypi.org/project/openai/)

OpenAI Python 库为任何 Python 3.9+ 应用程序提供了便捷的 OpenAI REST API 访问方式。该库包含了所有请求参数和响应字段的类型定义，并提供了由 httpx (https://github.com/encode/httpx) 支持的同步和异步客户端。

它是由 Stainless (https://stainlessapi.com/) 根据我们的 OpenAPI 规范 (https://github.com/openai/openai-openapi) 生成的。

文档

REST API 文档可在 platform.openai.com (https://platform.openai.com/docs/api-reference) 找到。该库的完整 API 可在 api.md 中找到。

安装

``sh

从 PyPI 安装

pip install openai ``

使用

该库的完整 API 可在 api.md 中找到。

与 OpenAI 模型交互的主要 API 是 Responses API (https://platform.openai.com/docs/api-reference/responses)。你可以使用以下代码从模型生成文本。

``python import os from openai import OpenAI

client = OpenAI( # 这是默认值，可以省略 api_key=os.environ.get(“OPENAI_API_KEY”), )

response = client.responses.create( model=“gpt-5.2”, 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 从模型生成文本。

``python from openai import OpenAI

client = OpenAI()

completion = client.chat.completions.create( model=“gpt-5.2”, 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 (https://pypi.org/project/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 (服务账户令牌)

``python from openai import OpenAI from openai.auth import k8s_service_account_token_provider

client = OpenAI( workload_identity={ “client_id”: “your-client-id”, “identity_provider_id”: “idp-123”, “service_account_id”: “sa-456”, “provider”: k8s_service_account_token_provider( “/var/run/secrets/kubernetes.io/serviceaccount/token” ), }, organization=“org-xyz”, project=“proj-abc”, )

response = client.chat.completions.create( model=“gpt-4”, messages=[{“role”: “user”, “content”: “Hello!”}], ) ``

Azure (托管身份)

``python from openai import OpenAI from openai.auth import azure_managed_identity_token_provider

client = OpenAI( workload_identity={ “client_id”: “your-client-id”, “identity_provider_id”: “idp-123”, “service_account_id”: “sa-456”, “provider”: azure_managed_identity_token_provider( resource=“https://management.azure.com/”, ), }, ) ``

Google Cloud Platform (计算引擎元数据)

``python from openai import OpenAI from openai.auth import gcp_id_token_provider

client = OpenAI( workload_identity={ “client_id”: “your-client-id”, “identity_provider_id”: “idp-123”, “service_account_id”: “sa-456”, “provider”: gcp_id_token_provider(audience=“https://api.openai.com/v1”), }, ) ``

自定义主体令牌提供商

``python from openai import OpenAI

def get_custom_token() -> str: return “your-jwt-token”

client = OpenAI( workload_identity={ “client_id”: “your-client-id”, “identity_provider_id”: “idp-123”, “service_account_id”: “sa-456”, “provider”: { “token_type”: “jwt”, “get_token”: get_custom_token, }, } ) ``

你也可以自定义令牌刷新缓冲区（默认为过期前 1200 秒（20 分钟））：

``python from openai import OpenAI from openai.auth import k8s_service_account_token_provider

client = OpenAI( workload_identity={ “client_id”: “your-client-id”, “identity_provider_id”: “idp-123”, “service_account_id”: “sa-456”, “provider”: k8s_service_account_token_provider(“/var/token”), “refresh_buffer_seconds”: 120.0, } ) ``

视觉

使用图片 URL：

``python 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.2”, input=[ { “role”: “user”, “content”: [ {“type”: “input_text”, “text”: prompt}, {“type”: “input_image”, “image_url”: f“{img_url}“}, ], } ], ) ``

使用 base64 编码字符串形式的图片：

``python 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.2”, 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：

``python 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.2”, input=“Explain disestablishmentarianism to a smart five year old.” ) print(response.output_text)

asyncio.run(main()) ``

同步客户端和异步客户端之间的功能否则是相同的。

使用 aiohttp

默认情况下，异步客户端使用 httpx 进行 HTTP 请求。然而，为了提高并发性能，你也可以使用 aiohttp 作为 HTTP 后端。

你可以通过安装 aiohttp 来启用此功能：

``sh

从 PyPI 安装

pip install openai[aiohttp] ``

然后通过使用 http_client=DefaultAioHttpClient() 实例化客户端来启用它：

``python 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.2”, )

asyncio.run(main()) ``

流式响应

我们提供通过服务器发送事件（SSE）支持流式响应。

``python from openai import OpenAI

client = OpenAI()

stream = client.responses.create( model=“gpt-5.2”, input=“Write a one-sentence bedtime story about a unicorn.”, stream=True, )

for event in stream: print(event) ``

异步客户端使用完全相同的接口。

``python import asyncio from openai import AsyncOpenAI

client = AsyncOpenAI()

async def main(): stream = await client.responses.create( model=“gpt-5.2”, 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)。

基于文本的基本示例：

``py import asyncio from openai import AsyncOpenAI

async def main(): client = AsyncOpenAI()

async with client.realtime.connect(model="gpt-realtime") 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 不会直接抛出错误。

``py client = AsyncOpenAI()

async with client.realtime.connect(model=“gpt-realtime”) 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) ``

使用类型

嵌套请求参数是 TypedDict (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 中的列表方法是分页的。

该库为每个列表响应提供了自动分页迭代器，因此你不必手动请求后续页面：

``python 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) ``

或者，异步地：

``python 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() 方法以更细粒度的方式控制页面操作：

``python 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。

``

或者直接处理返回的数据：

``python 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 进行类型化，例如：

``python from openai import OpenAI

client = OpenAI()

response = client.chat.responses.create( input=[ { “role”: “user”, “content”: “How much ?”, } ], model=“gpt-5.2”, response_format={“type”: “json_object”}, ) ``

文件上传

对应于文件上传的请求参数可以作为 bytes、PathLike (https://docs.python.org/3/library/os.html#os.PathLike) 实例或 (filename, contents, media type) 元组传递。

``python 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 签名是 可选但推荐的。

有关 webhooks 的更多信息，请参阅 API 文档 (https://platform.openai.com/docs/guides/webhooks)。

解析 webhook 负载

对于大多数用例，你可能希望同时验证 webhook 并解析负载。为了实现这一点，我们提供了方法 client.webhooks.unwrap()，它解析 webhook 请求并验证它是否由 OpenAI 发送。如果签名无效，此方法将引发错误。

注意，body 参数必须是服务器发送的原始 JSON 字符串（不要先解析它）。.unwrap() 方法将在验证 webhook 是由 OpenAI 发送后，将此 JSON 解析为事件对象。

``python from openai import OpenAI from flask import Flask, request

app = Flask(name) client = OpenAI() # 默认使用 OPENAI_WEBHOOK_SECRET 环境变量

@app.route(“/webhook”, methods=[“POST”]) def webhook(): request_body = request.get_data(as_text=True)

try:
    event = client.webhooks.unwrap(request_body, request.headers)

    if event.type == "response.completed":
        print("Response completed:", event.data)
    elif event.type == "response.failed":
        print("Response failed:", event.data)
    else:
        print("Unhandled event type:", event.type)

    return "ok"
except Exception as e:
    print("Invalid signature:", e)
    return "Invalid signature", 400

if name == “main”: app.run(port=8000) ``

直接验证 webhook 负载

在某些情况下，你可能希望直接验证 webhook