openai/openai-python

来源：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 中找到。

安装

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

工作负载身份认证

对于云管理的 Kubernetes、Azure 和 Google Cloud Platform 等安全的自动化环境，你可以使用工作负载身份认证，使用来自云身份提供商的短期令牌，而不是长期 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,
    }
)

视觉

使用图像 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())

流式响应

我们提供了使用服务器发送事件 (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())

实时 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") 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") 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.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 负载

在某些情况下，你可能想单独验证 webh