OpenAI Python 程式庫可讓任何 Python 3.8+ 應用程式方便地存取 OpenAI REST API。該庫包含所有請求參數和回應欄位的類型定義,並提供由 httpx 提供支援的同步和非同步客戶端。
它是根據我們的 OpenAPI 規範與不銹鋼生成的。
REST API 文件可以在 platform.openai.com 上找到。該庫的完整 API 可以在 api.md 中找到。
重要的
SDK 在 v1 中進行了重寫,於 2023 年 11 月 6 日發布。
# install from PyPI
pip install openai
該庫的完整 API 可以在 api.md 中找到。
import os
from openai import OpenAI
client = OpenAI (
api_key = os . environ . get ( "OPENAI_API_KEY" ), # This is the default and can be omitted
)
chat_completion = client . chat . completions . create (
messages = [
{
"role" : "user" ,
"content" : "Say this is a test" ,
}
],
model = "gpt-4o" ,
)
雖然您可以提供api_key
關鍵字參數,但我們建議使用 python-dotenv 將OPENAI_API_KEY="My API Key"
新增至.env
檔案中,以便您的 API 金鑰不會儲存在原始程式碼管理中。
使用託管圖片:
response = client . chat . completions . create (
model = "gpt-4o-mini" ,
messages = [
{
"role" : "user" ,
"content" : [
{ "type" : "text" , "text" : prompt },
{
"type" : "image_url" ,
"image_url" : { "url" : f" { img_url } " },
},
],
}
],
)
將圖像作為 base64 編碼的字串:
response = client . chat . completions . create (
model = "gpt-4o-mini" ,
messages = [
{
"role" : "user" ,
"content" : [
{ "type" : "text" , "text" : prompt },
{
"type" : "image_url" ,
"image_url" : { "url" : f"data: { img_type } ;base64, { img_b64_str } " },
},
],
}
],
)
與 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 :
# Print the text from text delta events
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" ), # This is the default and can be omitted
)
async def main () -> None :
chat_completion = await client . chat . completions . create (
messages = [
{
"role" : "user" ,
"content" : "Say this is a test" ,
}
],
model = "gpt-4o" ,
)
asyncio . run ( main ())
同步和非同步客戶端之間的功能在其他方面是相同的。
我們使用伺服器端事件 (SSE) 提供對串流回應的支援。
from openai import OpenAI
client = OpenAI ()
stream = client . chat . completions . create (
messages = [
{
"role" : "user" ,
"content" : "Say this is a test" ,
}
],
model = "gpt-4o" ,
stream = True ,
)
for chunk in stream :
print ( chunk . choices [ 0 ]. delta . content or "" , end = "" )
非同步客戶端使用完全相同的介面。
import asyncio
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 ())
重要的
我們強烈建議實例化客戶端實例,而不是依賴全域客戶端。
我們也公開了一個全域客戶端實例,可以透過與 v1 之前的版本類似的方式存取該實例。
import openai
# optional; defaults to `os.environ['OPENAI_API_KEY']`
openai . api_key = '...'
# all client options can be configured just like the `OpenAI` instantiation counterpart
openai . base_url = "https://..."
openai . default_headers = { "x-foo" : "true" }
completion = openai . chat . completions . create (
model = "gpt-4o" ,
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()
),因為:
巢狀的請求參數是TypedDicts。反應是 Pydantic 模型,它還為以下內容提供輔助方法:
model.to_json()
model.to_dict()
鍵入的請求和回應在編輯器中提供自動完成和文件。如果您希望在 VS Code 中查看類型錯誤以幫助儘早發現錯誤,請將python.analysis.typeCheckingMode
設定為basic
。
OpenAI API 中的列表方法是分頁的。
該程式庫為每個清單回應提供自動分頁迭代器,因此您不必手動要求連續頁面:
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 )
或者,異步:
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()
方法對頁面進行更精細的控制:
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.
或直接處理傳回的資料:
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
鍵入,例如:
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-4o" ,
response_format = { "type" : "json_object" },
)
與檔案上傳相對應的請求參數可以作為bytes
、 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
實例,檔案內容將自動非同步讀取。
當程式庫無法連接到 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-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 |
不適用 | APIConnectionError |
有關調試請求的更多信息,請參閱這些文檔
SDK 中的所有物件回應都提供從x-request-id
回應標頭新增的_request_id
屬性,以便您可以快速記錄失敗的請求並將其報告回 OpenAI。
completion = await client . chat . completions . create (
messages = [{ "role" : "user" , "content" : "Say this is a test" }], model = "gpt-4"
)
print ( completion . _request_id ) # req_123
請注意,與使用_
前綴的其他屬性不同, _request_id
屬性是公共的。除非另有說明,否則所有其他_
前綴屬性、方法和模組都是私有的。
預設情況下,某些錯誤會自動重試 2 次,並具有短暫的指數退避。預設情況下,連線錯誤(例如,由於網路連線問題)、408 請求逾時、409 衝突、429 速率限制和 >=500 內部錯誤都會重試。
您可以使用max_retries
選項來配置或停用重試設定:
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
物件:
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
來啟用日誌記錄。
$ export OPENAI_LOG=info
或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 ( 'Got json like {}, without a "my_field" key present at all.' )
else :
print ( 'Got json like {"my_field": null}.' )
可以透過前綴.with_raw_response.
任何 HTTP 方法調用,例如
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
物件。
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 動詞發出請求。提出此請求時,將尊重客戶端的選項(例如重試)。
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 模型上的所有額外欄位作為字典取得。
您可以直接重寫 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 (
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 連線。如果需要,您可以使用.close()
方法手動關閉用戶端,或使用退出時關閉的上下文管理器。
若要將此程式庫與 Azure OpenAI 一起使用,請使用AzureOpenAI
類別而不是OpenAI
類別。
重要的
Azure API 形狀與核心 API 形狀不同,這表示回應/參數的靜態類型並不總是正確的。
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 約定,儘管某些向後不相容的變更可能會作為次要版本發布:
我們認真對待向後相容性,並努力確保您可以獲得流暢的升級體驗。
我們熱切希望得到您的回饋;請提出包含問題、錯誤或建議的問題。
如果您已升級到最新版本,但沒有看到任何您期望的新功能,那麼您的 python 環境可能仍在使用舊版本。
您可以透過以下方式確定運行時使用的版本:
import openai
print ( openai . __version__ )
Python 3.8 或更高版本。
請參閱貢獻文件。