openai/openai-python

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

OpenAI Python API 库

PyPI 版本

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

它是基于我们的 OpenAPI 规范 (https://github.com/openai/openai-openapi) 并使用 Stainless (https://stainlessapi.com/) 生成的。

文档

REST API 文档可在 platform.openai.com (https://platform.openai.com/docs/api-reference) 找到。本库的完整 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.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 从模型生成文本。

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)。

Workload Identity 认证

对于像云托管的 Kubernetes、Azure 和 Google Cloud Platform 这样安全的自动化环境，你可以使用来自云身份提供商的短期令牌进行 workload identity 认证，而不是使用长期有效的 API 密钥。

Kubernetes (服务账户令牌)

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 (托管身份)

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 (计算引擎元数据)

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"),
    },
)

自定义主题令牌提供程序

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 分钟））：

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,
    }
)

视觉 (Vision)

使用图片 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.2",
    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.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：

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 来启用此功能：

# 从 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.2",
        )

asyncio.run(main())

流式响应

我们提供通过 Server Side Events (SSE) 支持流式响应。

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)

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

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())

Realtime API

Realtime API 使你能够构建低延迟、多模态的对话体验。它目前支持文本和音频作为输入和输出，并通过 WebSocket 连接支持函数调用 (https://platform.openai.com/docs/guides/function-calling)。

在底层，SDK 使用 websockets (https://websockets.readthedocs.io/en/stable/) 库来管理连接。

Realtime 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") 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())

然而，Realtime API 的真正魔力在于处理音频输入/输出，请参阅此示例 TUI 脚本 (https://github.com/openai/openai-python/blob/main/examples/realtime/push_to_talk_app.py) 以获取完整示例。

Realtime 错误处理

每当发生错误时，Realtime 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") 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 中看到类型错误以帮助尽早捕获 bug，请将 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.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) 元组传递。

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__)
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 负载

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