OpenAI Python API 库
OpenAI Python 库提供了从任何 Python 3.7+ 应用程序方便地访问 OpenAI REST API 的方法。该库包含所有请求参数和响应字段的类型定义,并提供由 httpx 驱动的同步和异步客户端。
该库根据我们的 OpenAPI 规范 使用 Stainless 生成。
文档
REST API 文档可以在 platform.openai.com 上找到。该库的完整 API 可以在 api.md 中找到。
安装
[!IMPORTANT] SDK 在 v1 中被重写,于 2023 年 11 月 6 日发布。请参阅 v1 迁移指南,其中包括自动更新代码的脚本。
# 从 PyPI 安装
pip install openai
使用
该库的完整 API 可以在 api.md 中找到。
import os
from openai import OpenAI
client = OpenAI(
# 这是默认值,可以省略
api_key=os.environ.get("OPENAI_API_KEY"),
)
chat_completion = client.chat.completions.create(
messages=[
{
"role": "user",
"content": "Say this is a test",
}
],
model="gpt-3.5-turbo",
)
虽然你可以提供一个 api_key
关键字参数,但我们建议使用 python-dotenv 将 OPENAI_API_KEY="My API Key"
添加到 .env
文件中,以确保你的 API 密钥不会存储在源代码控制中。
轮询助手
在与 API 交互时,一些操作(如启动运行和向向量存储添加文件)是异步的,需要时间完成。SDK 包含助手函数,这些函数将轮询状态直到其达终止状态,然后返回结果对象。如果 API 方法的操作可从轮询中受益,将有相应的以 '_and_poll' 结尾的方法版本。
例如,要创建一个运行并轮询直到其达终止状态,你可以运行:
run = client.beta.threads.runs.create_and_poll(
thread_id=thread.id,
assistant_id=assistant.id,
)
有关运行生命周期的更多信息,请参见 运行生命周期文档
批量上传助手
在创建和操作向量存储时,你可以使用轮询助手来监控操作状态。为了方便,我们还提供了批量上传助手,以允许你同时上传多个文件。
sample_files = [Path("sample-paper.pdf"), ...]
batch = await client.vector_stores.file_batches.upload_and_poll(
store.id,
files=sample_files,
)
流助手
SDK 还包括处理流和处理传入事件的助手。
with client.beta.threads.runs.stream(
thread_id=thread.id,
assistant_id=assistant.id,
instructions="Please address the user as Jane Doe. The user has a premium account.",
) as stream:
for event in stream:
# 打印来自文本增量事件的文本
if event.type == "thread.message.delta" and event.data.delta.content:
print(event.data.delta.content[0].text)
关于流助手的更多信息可以在专用文档中找到:helpers.md
异步使用
只需导入 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:
chat_completion = await client.chat.completions.create(
messages=[
{
"role": "user",
"content": "Say this is a test",
}
],
model="gpt-3.5-turbo",
)
asyncio.run(main())
同步和异步客户端之间的功能是相同的。
流响应
我们提供了使用服务器端事件 (SSE) 流响应的支持。
from openai import OpenAI
client = OpenAI()
stream = client.chat.completions.create(
model="gpt-4",
messages=[{"role": "user", "content": "Say this is a test"}],
stream=True,
)
for chunk in stream:
print(chunk.choices[0].delta.content or "", end="")
异步客户端使用了完全相同的接口。
from openai import AsyncOpenAI
client = AsyncOpenAI()
async def main():
stream = await client.chat.completions.create(
model="gpt-4",
messages=[{"role": "user", "content": "Say this is a test"}],
stream=True,
)
async for chunk in stream:
print(chunk.choices[0].delta.content or "", end="")
asyncio.run(main())
模块级客户端
[!IMPORTANT] 我们强烈建议实例化客户端实例,而不是依赖全局客户端。
我们还公开了一个全局客户端实例,其访问方式与 v1 之前的版本类似。
import openai
# 可选;默认为 `os.environ['OPENAI_API_KEY']`
openai.api_key = '...'
# 所有客户端选项可以像 `OpenAI` 实例化对手一样配置
openai.base_url = "https://..."
openai.default_headers = {"x-foo": "true"}
completion = openai.chat.completions.create(
model="gpt-4",
messages=[
{
"role": "user",
"content": "How do I output all files in a directory using Python?",
},
],
)
print(completion.choices[0].message.content)
API 与标准客户端实例的 API 完全相同。
这旨在在 REPL 或笔记本中用于更快的迭代,不建议在应用程序代码中使用。
我们建议你始终在应用程序代码中实例化一个客户端(例如,使用 client = OpenAI()
),因为:
- 很难推理客户端选项在哪配置
- 没有可能在不可能引起竞争条件的情况下更改某些客户端选项
- 对测试目的而言更难模拟
- 不可能控制网络连接的清理
使用类型
嵌套的请求参数是 TypedDict。响应是 Pydantic 模型,它们还提供以下助手方法:
- 序列化回 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,
):
# 在这里处理作业
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()
completion = client.chat.completions.create(
messages=[
{
"role": "user",
"content": "Can you generate an example json object describing a fruit?",
}
],
model="gpt-3.5-turbo-1106",
response_format={"type": "json_object"},
)
文件上传
对应文件上传的请求参数可以作为 bytes
,PathLike
实例或(文件名、内容、媒体类型)的元组传递。
from pathlib import Path
from openai import OpenAI
client = OpenAI()
client.files.create(
file=Path("input.jsonl"),
purpose="fine-tune",
)
异步客户端使用的接口完全相同。如果你传递一个 PathLike
实例,文件内容将自动异步读取。
错误处理
当库无法连接到 API 时(例如,由于网络连接问题或超时),将引发 openai.APIConnectionError
的子类。
当 API 返回非成功状态码(即 4xx 或 5xx 响应)时,将引发 openai.APIStatusError
的子类,其中包含 status_code
和 response
属性。
所有错误都继承自 openai.APIError
。
import openai
from openai import OpenAI
client = OpenAI()
try:
client.fine_tuning.jobs.create(
model="gpt-3.5-turbo",
training_file="file-abc123",
)
except openai.APIConnectionError as e:
print("服务器无法被访问")
print(e.__cause__) # 一个底层异常,可能在 httpx 内部引发。
except openai.RateLimitError as e:
print("收到 429 状态码;我们应该稍作等待。")
except openai.APIStatusError as e:
print("收到其他非 200 范围内的状态码")
print(e.status_code)
print(e.response)
错误代码如下:
状态码 | 错误类型 |
---|---|
400 | BadRequestError |
401 | AuthenticationError |
403 | PermissionDeniedError |
404 | NotFoundError |
422 | UnprocessableEntityError |
429 | RateLimitError |
>=500 | InternalServerError |
N/A | APIConnectionError |
重试
某些错误会默认重试 2 次,带有短暂的指数退避。 连接错误(例如,由于网络连接问题)、408 请求超时、409 冲突、429 速率限制和 >=500 内部错误会默认重试。
你可以使用 max_retries
选项配置或禁用重试设置:
from openai import OpenAI
# 配置所有请求的默认值:
client = OpenAI(
# 默认值是 2
max_retries=0,
)
# 或者,配置每个请求:
client.with_options(max_retries=5).chat.completions.create(
messages=[
{
"role": "user",
"content": "How can I get the name of the current day in Node.js?",
}
],
model="gpt-3.5-turbo",
)
超时
默认情况下,请求在 10 分钟后超时。你可以使用 timeout
选项配置它,
它接受一个浮点数或 httpx.Timeout
对象:
from openai import OpenAI
# 配置所有请求的默认值:
client = OpenAI(
# 20 秒(默认是 10 分钟)
timeout=20.0,
)
# 更精细的控制:
client = OpenAI(
timeout=httpx.Timeout(60.0, read=5.0, write=10.0, connect=2.0),
)
# 每个请求重写:
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-3.5-turbo",
)
超时时将抛出 APITimeoutError
。
请注意,超时的请求会 默认重试两次。
高级
日志记录
我们使用标准库 logging
模块。
你可以通过将环境变量 OPENAI_LOG
设置为 debug
来启用日志记录。
$ export OPENAI_LOG=debug
如何分辨 None
表示 null
还是缺失
在 API 响应中,一个字段可能显式为 null
,或者完全缺失;在这两种情况下,其值在该库中为 None
。你可以通过 .model_fields_set
区分这两种情况:
if response.my_field is None:
if 'my_field' not in response.model_fields_set:
print('得到类似 {} 的 json,没有 "my_field" 键存在。')
else:
print('得到类似 {"my_field": null} 的 json。')
访问原始响应数据(例如,标头)
“原始”响应对象可以通过在任何HTTP方法调用前缀 .with_raw_response.
来访问,例如,
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-3.5-turbo",
)
print(response.headers.get('X-My-Header'))
completion = response.parse() # 获取 `chat.completions.create()` 将返回的对象
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
对象。
with client.chat.completions.with_streaming_response.create(
messages=[
{
"role": "user",
"content": "Say this is a test",
}
],
model="gpt-3.5-turbo",
) as response:
print(response.headers.get("X-My-Header"))
for line in response.iter_lines():
print(line)
需要上下文管理器以确保响应能够可靠地关闭。
制作自定义/未文档化的请求
该库经过类型标注,方便访问有文档记录的API。
如果需要访问未记录的端点、参数或响应属性,仍然可以使用该库。
未文档化的端点
要对未记录的端点发起请求,可以使用 client.get
、client.post
和其他HTTP方法。客户端上的选项(例如重试)将在发出此请求时被尊重。
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
一样访问额外字段。您还可以将Pydantic模型中的所有额外字段作为字典获取,即 response.model_extra
。
配置HTTP客户端
您可以直接覆盖 httpx客户端 以根据您的用例进行自定义,包括:
- 支持代理
- 自定义传输
- 其他 高级 功能
from openai import OpenAI, DefaultHttpxClient
client = OpenAI(
# 或者使用 `OPENAI_BASE_URL` 环境变量
base_url="http://my.test.server.example.com:8083",
http_client=DefaultHttpxClient(
proxies="http://my.test.proxy.example.com",
transport=httpx.HTTPTransport(local_address="0.0.0.0"),
),
)
您也可以通过使用 with_options()
根据每个请求自定义客户端:
client.with_options(http_client=DefaultHttpxClient(...))
管理HTTP资源
默认情况下,该库将在客户端 垃圾回收 时关闭底层HTTP连接。您可以使用 .close()
方法手动关闭客户端,或者使用退出时关闭的上下文管理器。
微软Azure OpenAI
要使用此库与 Azure OpenAI,请使用 AzureOpenAI
类代替 OpenAI
类。
[!重要] Azure API的形状与核心API形状不同,这意味着响应/参数的静态类型不总是正确的。
from openai import AzureOpenAI
# 从环境变量AZURE_OPENAI_API_KEY获取API密钥
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", # 例如 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
使用微软Entra ID(以前称为Azure Active Directory)的客户端示例可以在 这里 找到。
版本控制
该软件包通常遵循 SemVer 规范,尽管某些向后不兼容的更改可能会作为次要版本发布:
- 仅影响静态类型的更改,不会破坏运行时行为。
- 技术上是公共但不是为外部使用而设计或记录的库内部更改。 (如果您依赖于这样的内部,请打开GitHub问题告知我们)。
- 我们预计实际上不会影响大多数用户的更改。
我们非常重视向后兼容性,并努力确保您可以依靠顺利的升级体验。
我们非常期待您的反馈;请打开一个 问题 提出问题、漏洞或建议。
要求
Python 3.7或更高版本。