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。

安装

# 从 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="你是一个像海盗一样说话的编程助手。",
    input="如何检查 Python 对象是否是某个类的实例？",
)

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": "像海盗一样说话。"},
        {
            "role": "user",
            "content": "如何检查 Python 对象是否是某个类的实例？",
        },
    ],
)

print(completion.choices[0].message.content)

虽然你可以提供 api_key 关键字参数， 但我们建议使用 python-dotenv (https://pypi.org/project/python-dotenv/) 将 OPENAI_API_KEY="你的 API 密钥" 添加到 .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 = "这张图片里有什么？"
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 = "这张图片里有什么？"
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}"},
            ],
        }
    ],
)

异步用法

只需将 OpenAI 替换为 AsyncOpenAI 导入，并在每次 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="向一个聪明的五岁孩子解释废除主义。"
    )
    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="写一句关于独角兽的睡前故事。",
    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="写一句关于独角兽的睡前故事。",
        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)

使用类型

嵌套请求参数是 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 中看到类型错误以帮助更早地发现 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 负载

在某些情况下，你可能希望直接验证 Webhook 负载。