Python OpenAI
openai is a powerful Python library for interacting with OpenAI's suite of models and services.
openai wraps all RESTful API calls, allowing developers to easily integrate powerful AI capabilities—such as natural language processing, image generation, and speech recognition—into their Python applications.
Key features:
- Text generation: Use models such as GPT-4 or GPT-5 to generate articles, code, summaries, conversations, and more.
- Image generation: Create images from text descriptions using DALL-E models.
- Embeddings: Convert text into vector representations, commonly used for tasks such as semantic search, text classification, and clustering.
- Speech to text: Use the Whisper model to transcribe audio files into text.
- Fine-tuning: Train a more targeted model by providing your own dataset.
- Assistants API: Build complex applications that can understand context, call tools, and perform long-term interactions.
openai open-source repository:https://github.com/openai/openai-python

How to use?
Requirements:
- Python version: 3.9 or above.
- Dependencies: httpx (default), aiohttp (optional), websockets (required for the Realtime API).
First, you need to install the openai library with pip:
pip install openai 或 pip3 install openai
Then go to the official OpenAI websitehttps://platform.openai.com/ to register an account and generate an API Key on the API keys page.
Check the installed version:
import openai print(openai.__version__) # 输出当前 SDK 版本
Example
from openai import OpenAI
client = OpenAI(
# This is the default and can be omitted
api_key="The API key you applied for",
)
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)
Parameter description:
| Parameter | Required | Type | Description |
|---|---|---|---|
api_key |
Yes | str |
The OpenAI key you applied for |
model |
Yes | str |
Specifies the OpenAI model used, determining capability, reasoning level, and cost |
instructions |
no | str |
System Prompt, defining the model's identity, behavioral norms, and expression style, taking priority overinput |
input |
Yes | str / list |
User input content, describing the specific question or task |
Third-party models
Accessing OpenAI from China is still somewhat troublesome. Many domestic providers also support OpenAI compatibility, such as DeepSeek, Qwen, and GLM.
DeepSeek
The DeepSeek API is fully compatible with OpenAI's API format. You only need to modify a few configuration settings to directly use the OpenAI SDK or compatible tools to access the DeepSeek API.
| Parameter | Value / Description |
|---|---|
| base_url | Required, fixed value:https://api.deepseek.com(You may also usehttps://api.deepseek.com/v1, only for OpenAI compatibility; v1 is unrelated to model versions) |
| api_key | Required. You need to first apply for a dedicated API Key on the DeepSeek official website (application URL:https://platform.deepseek.com/) |
| model | Required. The model to set:
|
Example
from openai import OpenAI
client = OpenAI(
api_key=os.environ.get('DEEPSEEK_API_KEY'), # Recommended to configure via environment variables; you can also hardcode them (not recommended)
base_url="https://api.deepseek.com")
response = client.chat.completions.create(
model="deepseek-v4-flash",
messages=[
{"role": "system", "content": "You are a helpful assistant"},
{"role": "user", "content": "Hello"},
],
stream=False # stream=False non-streaming (returned at once), stream=True streaming (returned in real time)
)
print(response.choices[0].message.content)
Usage
Next, we use the OpenAI SDK to access the Qwen models on the Bailian service.
Non-streaming call example
Example
import os
def get_response():
client = OpenAI(
api_key="sk-xxx", # Use your Alibaba Cloud Bailian API Key
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1", # Fill in the base_url of the DashScope SDK
)
completion = client.chat.completions.create(
model="qwen-plus", # Here we use qwen-plus as an example; you can change the model name as needed. Model list: https://help.aliyun.com/zh/model-studio/getting-started/models
messages=[{'role': 'system', 'content': 'You are a helpful assistant.'},
{'role': 'user', 'content': 'Who are you?'}]
)
# json data
#print(completion.model_dump_json())
print(completion.choices[0].message.content)
if __name__ == '__main__':
get_response()
Running the code produces the following result:
I am Qwen, an ultra-large-scale language model independently developed by the Tongyi Laboratory under Alibaba Group. I can help you answer questions and create text, such as writing stories, official documents, emails, scripts, logical reasoning, programming, and more. I can also express opinions and play games. If you have any questions or need help, feel free to tell me at any time!
Streaming call example
Example
def get_response():
client = OpenAI(
api_key="sk-xxx",
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
)
completion = client.chat.completions.create(
model="qwen-plus",
messages=[
{'role': 'system', 'content': 'You are a helpful assistant.'},
{'role': 'user', 'content': 'Who are you?'}
],
stream=True,
stream_options={"include_usage": True}
)
for chunk in completion:
# chunk may not contain choices or delta
if hasattr(chunk, "choices") and len(chunk.choices) > 0:
choice = chunk.choices[0]
if hasattr(choice, "delta") and hasattr(choice.delta, "content"):
print(choice.delta.content, end='', flush=True)
if __name__ == '__main__':
get_response()
Running the code produces the following result:
I am Qwen, an ultra-large-scale language model independently developed by the Tongyi Laboratory under Alibaba Group. I can help you answer questions and create text, such as writing stories, official documents, emails, scripts, logical reasoning, programming, and more. I can also express opinions and play games. If you have any questions or need help, feel free to tell me at any time!
Featured Features
Vision capability
Supports image input (URL or Base64-encoded) to enable multimodal interaction.
Image URL input:
Example
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": img_url},
],
}
],
)
print(response.output_text)
Base64-encoded image input:
Example
from openai import OpenAI
client = OpenAI()
prompt = "What is in this image?"
# Read a local image and encode it as Base64
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}"},
],
}
],
)
Asynchronous usage (Async usage)
Replace OpenAI with AsyncOpenAI and call the API with await:
Basic async example:
Example
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="Explain disestablishmentarianism to a smart five year old."
)
print(response.output_text)
asyncio.run(main())
For async scenario optimization (aiohttp backend), install the extended version:
pip install openai[aiohttp]
aiohttp backend optimization:
Example
import asyncio
from openai import DefaultAioHttpClient, AsyncOpenAI
async def main() -> None:
# Context manager ensures resources are released
async with AsyncOpenAI(
api_key=os.environ.get("OPENAI_API_KEY"),
http_client=DefaultAioHttpClient(), # Enable the aiohttp backend
) as client:
chat_completion = await client.chat.completions.create(
messages=[{"role": "user", "content": "Say this is a test"}],
model="gpt-5.2",
)
asyncio.run(main())
Streaming responses
Based on Server Side Events (SSE) to obtain responses in real time; synchronous and asynchronous interfaces are consistent.
Synchronous streaming example:
Example
client = OpenAI()
stream = client.responses.create(
model="gpt-5.2",
input="Write a one-sentence bedtime story about a unicorn.",
stream=True, # Enable streaming output
)
for event in stream:
print(event) # Print the response segment by segment
Asynchronous streaming example:
Example
from openai import AsyncOpenAI
client = AsyncOpenAI()
async def main():
stream = await client.responses.create(
model="gpt-5.2",
input="Write a one-sentence bedtime story about a unicorn.",
stream=True,
)
async for event in stream:
print(event)
asyncio.run(main())
Realtime API
Low-latency multimodal conversation (text / audio), implemented over WebSocket, requires the websockets library.
Basic text example:
Example
from openai import AsyncOpenAI
async def main():
client = AsyncOpenAI()
# Establish a realtime connection
async with client.realtime.connect(model="gpt-realtime") as connection:
# Update session configuration (text-only output)
await connection.session.update(
session={"type": "realtime", "output_modalities": ["text"]}
)
# Send a user message
await connection.conversation.item.create(
item={
"type": "message",
"role": "user",
"content": [{"type": "input_text", "text": "Say hello!"}],
}
)
# Trigger the model response
await connection.response.create()
# Listen for realtime events
async for event in connection:
if event.type == "response.output_text.delta":
print(event.delta, flush=True, end="") # Print text fragments in real time
elif event.type == "response.output_text.done":
print() # Add a newline when the response ends
elif event.type == "response.done":
break # Terminate listening
asyncio.run(main())
Realtime API error handling:
Example
from openai import AsyncOpenAI
client = AsyncOpenAI()
async def main():
async with client.realtime.connect(model="gpt-realtime") as connection:
async for event in connection:
if event.type == 'error':
# Catch and handle error events (the connection will not be disconnected)
print(f"Error type: {event.error.type}")
print(f"Error code: {event.error.code}")
print(f"Error message: {event.error.message}")
asyncio.run(main())
Reference Manual
The following table contains:
- Core initialization: for synchronous use,
OpenAIfor asynchronous use,AsyncOpenAIfor Azure use,AzureOpenAIit is recommended to configure the API Key via environment variables. - Text generation: prefer the new version,
responses.create()for classic scenarios, usechat.completions.create()streaming output requires settingstream=True。 - Advanced capabilities: support multimodal (image input), realtime API, file upload/fine-tuning; error handling must catch
APIErrorsubclasses, and timeout/retry can be flexibly configured.
Installation and import
Install the official SDK:
pip install --upgrade openai
Import core classes:
from openai import OpenAI, AsyncOpenAI, AzureOpenAI
Creating a client
Create a standard OpenAI client:
from openai import OpenAI client = OpenAI(api_key="YOUR_API_KEY")
Create a client via environment variables:
import os from openai import OpenAI os.environ["OPENAI_API_KEY"] = "YOUR_API_KEY" client = OpenAI()
Set request timeout and retry count:
client = OpenAI(
timeout=30,
max_retries=2
)
Text generation (Responses API)
The most basic text generation request:
response = client.responses.create(
model="gpt-5.2",
input="用一句话解释什么是 Python"
)
print(response.output_text)
Constrain model behavior via instructions:
response = client.responses.create(
model="gpt-5.2",
instructions="你是一个严谨的 Python 教程作者",
input="解释什么是列表推导式"
)
print(response.output_text)
Multi-turn conversation
Use the input array to build contextual conversation:
response = client.responses.create(
model="gpt-5.2",
input=[
{"role": "user", "content": "Python 是什么?"},
{"role": "assistant", "content": "Python 是一门高级编程语言。"},
{"role": "user", "content": "它适合做什么?"}
]
)
print(response.output_text)
Structured output (JSON Schema)
Ask the model to return JSON data conforming to the Schema:
response = client.responses.create(
model="gpt-5.2",
response_format={
"type": "json_schema",
"json_schema": {
"name": "tech_summary",
"schema": {
"type": "object",
"properties": {
"title": {"type": "string"},
"summary": {"type": "string"}
},
"required": ["title", "summary"]
}
}
},
input="介绍 asyncio"
)
print(response.output_parsed)
Tool calling (Function Calling)
Define tools and let the model automatically decide whether to call them:
tools = [
{
"type": "function",
"function": {
"name": "get_weather",
"description": "获取城市天气",
"parameters": {
"type": "object",
"properties": {
"city": {"type": "string"}
},
"required": ["city"]
}
}
}
]
response = client.responses.create(
model="gpt-5.2",
tools=tools,
input="北京今天天气怎么样?"
)
print(response.output)
Streaming output (Streaming)
Synchronous streaming output:
with client.responses.stream(
model="gpt-5.2",
input="写一段 Python 示例代码"
) as stream:
for event in stream:
if event.type == "response.output_text.delta":
print(event.delta, end="")
Asynchronous streaming output:
stream = await client.responses.create(
model="gpt-5.2",
input="解释什么是生成式 AI",
stream=True
)
async for event in stream:
print(event)
Asynchronous calls (AsyncOpenAI)
Use the asynchronous client in asyncio:
import asyncio
from openai import AsyncOpenAI
client = AsyncOpenAI()
async def main():
response = await client.responses.create(
model="gpt-5.2",
input="解释什么是事件循环"
)
print(response.output_text)
asyncio.run(main())
Realtime API (Realtime)
Use WebSocket for low-latency realtime conversations:
async with client.realtime.connect(model="gpt-realtime") as conn:
await conn.session.update(
session={"type": "realtime", "output_modalities": ["text"]}
)
await conn.conversation.item.create(
item={
"type": "message",
"role": "user",
"content": [{"type": "input_text", "text": "你好"}]
}
)
async for event in conn:
print(event)
Image generation
Generate an image from text:
image = client.images.generate(
model="gpt-image-1",
prompt="一只在写代码的猫",
size="1024x1024"
)
print(image.data[0].url)
Embeddings generation
Generate text vector embeddings:
embedding = client.embeddings.create(
model="text-embedding-3-small",
input="Hello world"
)
print(embedding.data[0].embedding)
Speech to Text
Convert an audio file to text:
audio_file = open("speech.mp3", "rb")
transcript = client.audio.transcriptions.create(
model="whisper-1",
file=audio_file
)
print(transcript.text)
Text to Speech
Convert text to a speech file:
speech = client.audio.speech.create(
model="gpt-4o-mini-tts",
voice="alloy",
input="你好,欢迎使用 OpenAI"
)
with open("output.mp3", "wb") as f:
f.write(speech)
File management (Files API)
Upload a file:
from pathlib import Path
client.files.create(
file=Path("data.jsonl"),
purpose="fine-tune"
)
List files:
files = client.files.list()
for f in files:
print(f.id, f.filename)
Delete a file:
client.files.delete(file_id="file-xxx")
Model fine-tuning (Fine-tuning Jobs)
Create a fine-tuning job:
job = client.fine_tuning.jobs.create(
training_file="file-xxx",
model="gpt-3.5-turbo"
)
print(job.id, job.status)
Webhook verification
Verify and parse the Webhook request:
event = client.webhooks.unwrap(request_body, request_headers)
Verify only the signature:
client.webhooks.verify_signature(request_body, request_headers)
Error handling
Catch exceptions thrown by the SDK:
import openai
try:
client.responses.create(
model="gpt-5.2",
input="test"
)
except openai.RateLimitError as e:
print("触发速率限制:", e)
except openai.APIConnectionError as e:
print("连接失败:", e)
Get the Request ID of a failed request:
try:
client.responses.create(model="gpt-5.2", input="test")
except openai.APIStatusError as exc:
print(exc.request_id)
Azure OpenAI compatibility
Initialize the Azure OpenAI client:
from openai import AzureOpenAI
client = AzureOpenAI(
api_version="2023-07-01-preview",
azure_endpoint="https://xxx.openai.azure.com"
)
Versions and changelog
Check the SDK version:
import openai print(openai.__version__)
Enable SDK debug logging:
export OPENAI_LOG=debug
Other ExtensionsMore API references:https://github.com/openai/openai-python/blob/main/api.md