项目文章

搜索 TinyTools

OpenAI官方的Python的API库

29.4k 文档演示 Github仓库

OpenAI Python API library

OpenAI Python 库支持从任何 Python 3.8+ 应用程序便捷地访问 OpenAI REST API。该库包含所有请求参数和响应字段的类型定义，并提供由 httpx 支持的同步和异步客户端。

文档

REST API 文档可在 platform.openai.com 上找到。此库的完整 API 可在 api.md 中找到。

安装

sh 复制代码

# 从 PyPI 安装
pip install openai

使用

此库的完整 API 可在 api.md 中找到。

与 OpenAI 模型交互的主要 API 是 Responses API。您可以使用以下代码从模型生成文本。

python 复制代码

import os
from openai import OpenAI

client = OpenAI(
    # This is the default and can be omitted
    api_key=os.environ.get("OPENAI_API_KEY"),
)

response = client.responses.create(
    model="gpt-4o",
    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。您可以使用该 API 通过以下代码从模型生成文本。

python 复制代码

from openai import OpenAI

client = OpenAI()

completion = client.chat.completions.create(
    model="gpt-4o",
    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 将 OPENAI_API_KEY="My API Key" 添加到您的 .env 文件中，这样您的 API 密钥就不会存储在源代码管理中。在此处获取 API 密钥。

愿景

带有图片网址：

python 复制代码

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-4o-mini",
    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 = "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-4o-mini",
    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(
    # This is the default and can be omitted
    api_key=os.environ.get("OPENAI_API_KEY"),
)


async def main() -> None:
    response = await client.responses.create(
        model="gpt-4o", input="Explain disestablishmentarianism to a smart five year old."
    )
    print(response.output_text)


asyncio.run(main())

同步客户端和异步客户端的功能在其他方面完全相同。

使用 aiohttp

默认情况下，异步客户端使用 httpx 进行 HTTP 请求。但是，为了提高并发性能，您也可以使用 aiohttp 作为 HTTP 后端。

您可以通过安装 aiohttp 来启用此功能：

sh 复制代码

# 从 PyPI 安装
pip install openai[aiohttp]

然后，您可以通过使用 http_client=DefaultAioHttpClient() 实例化客户端来启用此功能：

python 复制代码

import asyncio
from openai import DefaultAioHttpClient
from openai import AsyncOpenAI


async def main() -> None:
    async with AsyncOpenAI(
        api_key="My 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-4o",
        )


asyncio.run(main())

流式响应

我们支持使用服务器端事件 (SSE) 进行流式响应。

python 复制代码

from openai import OpenAI

client = OpenAI()

stream = client.responses.create(
    model="gpt-4o",
    input="Write a one-sentence bedtime story about a unicorn.",
    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-4o",
        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 连接进行函数调用。

SDK 在底层使用 websockets 库来管理连接。

Realtime API 通过客户端发送事件和服务器发送事件的组合来工作。客户端可以发送事件来执行诸如更新会话配置或发送文本和音频输入等操作。服务器事件用于确认音频响应何时完成，或何时收到来自模型的文本响应。完整的事件参考可在此处找到，指南可在此处找到。

基于文本的基本示例：

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={'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.text.delta':
                print(event.delta, flush=True, end="")

            elif event.type == 'response.text.done':
                print()

            elif event.type == "response.done":
                break

asyncio.run(main())

然而，Realtime API 的真正魔力在于处理音频输入/输出，请参阅此示例 TUI 脚本获取完整示例。

Realtime 错误处理

每当发生错误时，Realtime API 都会发送一个 error 事件，并且连接将保持打开状态并可用。这意味着您需要自行处理，因为当 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。响应是 Pydantic 模型，它还提供以下辅助方法：

序列化回 JSON，model.to_json()
转换为字典，model.to_dict()

类型化的请求和响应在编辑器中提供自动完成和文档功能。如果您想在 VS Code 中查看类型错误以帮助更早地发现错误，请将 python.analysis.typeCheckingMode 设置为 basic。

分页

OpenAI API 中的列表方法是分页的。

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

python 复制代码

from openai import OpenAI

client = OpenAI()

all_jobs = []
# Automatically fetches more pages as needed.
for job in client.fine_tuning.jobs.list(
    limit=20,
):
    # Do something with job here
    all_jobs.append(job)
print(all_jobs)

或者，异步地：

python 复制代码

import asyncio
from openai import AsyncOpenAI

client = AsyncOpenAI()


async def main() -> None:
    all_jobs = []
    # Iterate through items across all pages, issuing requests as needed.
    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"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)}")

# Remove `await` for non-async usage.

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

python 复制代码

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)

# Remove `await` for non-async usage.

嵌套参数

嵌套参数是字典，使用 TypedDict 进行类型化，例如：

python 复制代码

from openai import OpenAI

client = OpenAI()

response = client.chat.responses.create(
    input=[
        {
            "role": "user",
            "content": "How much ?",
        }
    ],
    model="gpt-4o",
    response_format={"type": "json_object"},
)

文件上传

与文件上传对应的请求参数可以作为 bytes、PathLike 实例或 (文件名, 内容, 媒体类型) 元组传递。

python 复制代码

from pathlib import Path
from openai import OpenAI

client = OpenAI()

client.files.create(
    file=Path("input.jsonl"),
    purpose="fine-tune",
)

异步客户端使用完全相同的接口。如果您传递一个 PathLike 实例，文件内容将自动异步读取。

Webhook 验证

验证 Webhook 签名是_可选的，但建议这样做_。

有关 Webhook 的更多信息，请参阅 API 文档。

解析 Webhook 负载

在大多数情况下，您可能希望同时验证 Webhook 和解析负载。为此，我们提供了方法 client.webhooks.unwrap()，该方法会解析 Webhook 请求并验证其是否由 OpenAI 发送。如果签名无效，此方法将引发错误。

请注意，“body”参数必须是从服务器发送的原始 JSON 字符串（无需先解析）。在验证 Webhook 是否由 OpenAI 发送后，“.unwrap()”方法会将此 JSON 解析为事件对象。

python 复制代码

from openai import OpenAI
from flask import Flask, request

app = Flask(__name__)
client = OpenAI()  # OPENAI_WEBHOOK_SECRET environment variable is used by default


@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 的验证与负载解析分开进行。如果您希望单独处理这些步骤，我们提供了方法 client.webhooks.verify_signature() 来仅验证webhook 请求的签名。与 .unwrap() 类似，如果签名无效，此方法会抛出错误。

请注意，body 参数必须是服务器发送的原始 JSON 字符串（无需先解析）。验证签名后，您需要解析 body。

python 复制代码

import json
from openai import OpenAI
from flask import Flask, request

app = Flask(__name__)
client = OpenAI()  # OPENAI_WEBHOOK_SECRET environment variable is used by default


@app.route("/webhook", methods=["POST"])
def webhook():
    request_body = request.get_data(as_text=True)

    try:
        client.webhooks.verify_signature(request_body, request.headers)

        # Parse the body after verification
        event = json.loads(request_body)
        print("Verified event:", event)

        return "ok"
    except Exception as e:
        print("Invalid signature:", e)
        return "Invalid signature", 400


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

错误处理

当库无法连接到 API（例如，由于网络连接问题或超时）时，会引发 openai.APIConnectionError 的子类。

当 API 返回不成功状态代码（即 4xx 或 5xx 响应）时，会引发 openai.APIStatusError 的子类，其中包含 status_code 和 response 属性。

所有错误都继承自 openai.APIError。

python 复制代码

import openai
from openai import OpenAI

client = OpenAI()

try:
    client.fine_tuning.jobs.create(
        model="gpt-4o",
        training_file="file-abc123",
    )
except openai.APIConnectionError as e:
    print("The server could not be reached")
    print(e.__cause__)  # an underlying Exception, likely raised within httpx.
except openai.RateLimitError as e:
    print("A 429 status code was received; we should back off a bit.")
except openai.APIStatusError as e:
    print("Another non-200-range status code was received")
    print(e.status_code)
    print(e.response)

错误代码如下：

状态码	错误类型
400	`BadRequestError`
401	`AuthenticationError`
403	`PermissionDeniedError`
404	`NotFoundError`
422	`UnprocessableEntityError`
429	`RateLimitError`
>=500	`InternalServerError`
N/A	`APIConnectionError`

请求 ID

有关调试请求的更多信息，请参阅这些文档

SDK 中的所有对象响应都提供了一个 _request_id 属性，该属性从 x-request-id 响应标头中添加，以便您可以快速记录失败的请求并将其报告给 OpenAI。

python 复制代码

response = await client.responses.create(
    model="gpt-4o-mini",
    input="Say 'this is a test'.",
)
print(response._request_id)  # req_123

请注意，与其他使用 _ 前缀的属性不同，_request_id 属性
是公开的。除非另有说明，否则所有其他以 _ 为前缀的属性、
方法和模块均为私有的。

[!IMPORTANT]
如果您需要访问失败请求的请求 ID，则必须捕获 APIStatusError 异常

python 复制代码

import openai

try:
    completion = await client.chat.completions.create(
        messages=[{"role": "user", "content": "Say this is a test"}], model="gpt-4"
    )
except openai.APIStatusError as exc:
    print(exc.request_id)  # req_123
    raise exc

重试

某些错误默认会自动重试 2 次，并采用短暂的指数退避机制。
连接错误（例如，由于网络连接问题）、408 请求超时、409 冲突、
429 速率限制以及 >=500 内部错误均默认重试。

您可以使用 max_retries 选项配置或禁用重试设置：

python 复制代码

from openai import OpenAI

# Configure the default for all requests:
client = OpenAI(
    # default is 2
    max_retries=0,
)

# Or, configure per-request:
client.with_options(max_retries=5).chat.completions.create(
    messages=[
        {
            "role": "user",
            "content": "How can I get the name of the current day in JavaScript?",
        }
    ],
    model="gpt-4o",
)

超时

默认情况下，请求会在 10 分钟后超时。您可以使用 timeout 选项进行配置，
该选项接受浮点数或 httpx.Timeout 对象：

python 复制代码

from openai import OpenAI

# Configure the default for all requests:
client = OpenAI(
    # 20 seconds (default is 10 minutes)
    timeout=20.0,
)

# More granular control:
client = OpenAI(
    timeout=httpx.Timeout(60.0, read=5.0, write=10.0, connect=2.0),
)

# Override per-request:
client.with_options(timeout=5.0).chat.completions.create(
    messages=[
        {
            "role": "user",
            "content": "How can I list all files in a directory using Python?",
        }
    ],
    model="gpt-4o",
)

超时时会抛出 APITimeoutError。

请注意，超时的请求默认会重试两次。

高级

日志记录

我们使用标准库 logging 模块。

您可以通过将环境变量 OPENAI_LOG 设置为 info 来启用日志记录。

shell 复制代码

$ export OPENAI_LOG=info

或者设置为 debug 以获取更详细的日志记录。

如何判断 `None` 是代表 `null` 还是缺失

在 API 响应中，某个字段可能明确为 null，也可能完全缺失；无论哪种情况，在本库中，其值都为 None。您可以使用 .model_fields_set 来区分这两种情况：

py 复制代码

if response.my_field is None:
  if 'my_field' not in response.model_fields_set:
    print('Got json like {}, without a "my_field" key present at all.')
  else:
    print('Got json like {"my_field": null}.')

访问原始响应数据（例如标题）

可以通过在任何 HTTP 方法调用前添加“.with_raw_response.”来访问“原始”响应对象，例如，

py 复制代码

from openai import OpenAI

client = OpenAI()
response = client.chat.completions.with_raw_response.create(
    messages=[{
        "role": "user",
        "content": "Say this is a test",
    }],
    model="gpt-4o",
)
print(response.headers.get('X-My-Header'))

completion = response.parse()  # get the object that `chat.completions.create()` would have returned
print(completion)

这些方法返回一个 LegacyAPIResponse 对象。这是一个遗留类，我们将在下一个主要版本中对其进行略微修改。

对于同步客户端，除了 content 和 text 将是方法而不是属性之外，其余部分基本相同。在
异步客户端中，所有方法都将是异步的。

我们将提供迁移脚本，迁移过程总体来说应该会
顺利进行。

`.with_streaming_response`

上述接口会在您发出请求时立即读取完整的响应主体，但这并不总是您想要的。

要流式传输响应主体，请使用 .with_streaming_response，它需要上下文管理器，并且仅在调用 .read()、.text()、.json()、.iter_bytes()、.iter_text()、.iter_lines() 或 .parse() 后读取响应主体。在异步客户端中，这些都是异步方法。

因此，.with_streaming_response 方法会返回不同的 APIResponse 对象，而异步客户端会返回一个 AsyncAPIResponse 对象。

python 复制代码

with client.chat.completions.with_streaming_response.create(
    messages=[
        {
            "role": "user",
            "content": "Say this is a test",
        }
    ],
    model="gpt-4o",
) as response:
    print(response.headers.get("X-My-Header"))

    for line in response.iter_lines():
        print(line)

需要上下文管理器才能可靠地关闭响应。

发出自定义/未记录的请求

此库的类型是为了方便访问已记录的 API。

如果您需要访问未记录的端点、参数或响应属性，仍然可以使用该库。

未记录的端点

要向未记录的端点发出请求，您可以使用 client.get、client.post 和其他
http 动词发出请求。发出此请求时，客户端上的选项（例如重试）将被考虑。

py 复制代码

import httpx

response = client.post(
    "/foo",
    cast_to=httpx.Response,
    body={"my_param": True},
)

print(response.headers.get("x-foo"))

未记录的请求参数

如果您想显式发送额外参数，可以使用 extra_query、extra_body 和 extra_headers 请求选项来实现。

未记录的响应属性

要访问未记录的响应属性，您可以访问额外字段，例如 response.unknown_prop。
您还可以使用 response.model_extra 以字典形式获取 Pydantic 模型上的所有额外字段。

配置 HTTP 客户端

您可以直接覆盖 httpx 客户端，根据您的用例进行自定义，包括：

支持代理
自定义传输
其他高级功能

python 复制代码

import httpx
from openai import OpenAI, DefaultHttpxClient

client = OpenAI(
    # Or use the `OPENAI_BASE_URL` env var
    base_url="http://my.test.server.example.com:8083/v1",
    http_client=DefaultHttpxClient(
        proxy="http://my.test.proxy.example.com",
        transport=httpx.HTTPTransport(local_address="0.0.0.0"),
    ),
)

您还可以使用“with_options()”根据每个请求自定义客户端：

python 复制代码

client.with_options(http_client=DefaultHttpxClient(...))

管理 HTTP 资源

默认情况下，每当客户端被垃圾回收时，库都会关闭底层 HTTP 连接。您可以根据需要使用 .close() 方法手动关闭客户端，或者使用在退出时自动关闭的上下文管理器。

py 复制代码

from openai import OpenAI

with OpenAI() as client:
  # make requests here
  ...

# HTTP client is now closed

Microsoft Azure OpenAI

要将此库与 Azure OpenAI 一起使用，请使用 AzureOpenAI
类而不是 OpenAI 类。

[!IMPORTANT]
Azure API 结构与核心 API 结构不同，这意味着响应/参数的静态类型
并不总是正确的。

py 复制代码

from openai import AzureOpenAI

# gets the API Key from environment variable AZURE_OPENAI_API_KEY
client = AzureOpenAI(
    # https://learn.microsoft.com/azure/ai-services/openai/reference#rest-api-versioning
    api_version="2023-07-01-preview",
    # https://learn.microsoft.com/azure/cognitive-services/openai/how-to/create-resource?pivots=web-portal#create-a-resource
    azure_endpoint="https://example-endpoint.openai.azure.com",
)

completion = client.chat.completions.create(
    model="deployment-name",  # e.g. gpt-35-instant
    messages=[
        {
            "role": "user",
            "content": "How do I output all files in a directory using Python?",
        },
    ],
)
print(completion.to_json())

除了基础版“OpenAI”客户端提供的选项外，还提供以下选项：

“azure_endpoint”（或“AZURE_OPENAI_ENDPOINT”环境变量）
“azure_deployment”
“api_version”（或“OPENAI_API_VERSION”环境变量）
“azure_ad_token”（或“AZURE_OPENAI_AD_TOKEN”环境变量）
“azure_ad_token_provider”

此处提供了使用 Microsoft Entra ID（以前称为 Azure Active Directory）客户端的示例。

版本控制

此软件包通常遵循 SemVer 约定，但某些向后不兼容的更改可能会以次要版本发布：

仅影响静态类型且不影响运行时行为的更改。
对库内部组件的更改，这些组件在技术上是公开的，但不打算或记录供外部使用。（如果您依赖此类内部组件，请提交 GitHub 问题告知我们。）
我们预计在实践中不会影响绝大多数用户的更改。

我们非常重视向后兼容性，并努力确保您能够享受顺畅的升级体验。

我们期待您的反馈；请提交 issue 提出问题、错误或建议。

确定已安装的版本

如果您已升级到最新版本，但仍未看到任何预期的新功能，则您的 Python 环境可能仍在使用旧版本。

您可以使用以下命令确定运行时使用的版本：

py 复制代码

import openai
print(openai.__version__)

要求

Python 3.8 或更高版本。

关于项目

Apache-2.0

Python

29,379

4429

349

2020-10-26

2025-11-27

OpenAI官方的Python的API库

OpenAI Python API library

文档

安装

使用

愿景

异步用法

使用 aiohttp

流式响应

Realtime API

Realtime 错误处理

使用类型

分页

嵌套参数

文件上传

Webhook 验证

解析 Webhook 负载

直接验证 webhook 负载

错误处理

请求 ID

重试

超时

高级

日志记录

如何判断 None 是代表 null 还是缺失

访问原始响应数据（例如标题）

.with_streaming_response

发出自定义/未记录的请求

未记录的端点

未记录的请求参数

未记录的响应属性

配置 HTTP 客户端

管理 HTTP 资源

Microsoft Azure OpenAI

版本控制

确定已安装的版本

要求

关于项目

增长趋势 - stars

如何判断 `None` 是代表 `null` 还是缺失

`.with_streaming_response`