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) 驱动的同步和异步客户端。

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

print(response.output_text)

先前标准（永久支持）的文本生成 API 是 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="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-4",
    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,
    }
)

视觉

使用图片 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}"},
            ],
        }
    ],
)

异步使用

只需导入 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="向一个聪明的五岁小孩解释反政教分离主义。"
    )
    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": "说这是测试",
                }
            ],
            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())

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": "说你好！"}],
            }
        )
        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)，连接会保持打开并仍然可用。这意味着您需要自行处理它，因为 SDK 在接收到 error 事件时_不会直接引发任何错误_。

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"将使用以下详细信息获取下一页：{first_page.next_page_info()}")
    next_page = await first_page.get_next_page()
    print(f"我们刚刚获取的项目数量：{len(next_page.data)}")

# 非异步用法请移除 `await`。

或者直接使用返回的数据：

first_page = await client.fine_tuning.jobs.list(
    limit=20,
)

print(f"下一页光标：{first_page.after}")  # => "下一页光标：..."
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": "多少钱？",
        }
    ],
    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 字符串（请勿先解析它）。在验证 webhook 是由 OpenAI 发送后，.unwrap() 方法将为您将此 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("响应已完成：", event.data)
        elif event.type == "response.failed":
            print("响应失败：", event.data)
        else:
            print("未处理的事件类型：", event.type)

        return "ok"
    except Exception as e:
        print("无效签名：", e)
        return "无效签名", 400

if __name__ == "__main__":
    app.run(port=8000)

直接验证 webhook 负载

在某些情况下，您可能希望将 webhook 验证与负载解析分开。如果您更喜欢分别处理这些步骤，我们提供了 client.webhooks.verify_signature() 方法来_仅验证_ webhook 请求的签名。与 .unwrap() 类似，如果签名无效，此方法将引发错误。