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

安装

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

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”: “说话像个海盗。”}, { “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（服务账户令牌）

``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”: “你好！”}], ) ``

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 = “这张图片里有什么？” 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 = “这张图片里有什么？” 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=“向一个聪明五岁的孩子解释解构主义。” ) 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”: “说这是一个测试”, } ], model=“gpt-5.2”, )

asyncio.run(main()) ``

流式响应

我们支持使用服务器发送事件（SSE）进行流式响应。

``python from openai import OpenAI

client = OpenAI()

stream = client.responses.create( model=“gpt-5.2”, input=“写一个关于独角兽的一句话睡前故事。”, 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=“写一个关于独角兽的一句话睡前故事。”, 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)。

基于文本的基本示例：

``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": "说你好！"}],
        }
    )
    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 不会直接引发错误。

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

使用类型

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

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

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

非异步用法去掉 await。

``

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

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

``python 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) 实例，或一个 (文件名，内容，媒体类型) 元组。

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

关于 webhook 的更多信息，请参阅 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("响应完成：", 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 负载

在某些情况下，您可能希望直接验证 webh