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

import os
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:

  • deepseek-v4-flash: Corresponds to DeepSeek'snon-thinking mode, with fast response speed, suitable for routine Q&A.
  • deepseek-v4-pro: Corresponds to DeepSeek'sthinking mode, with stronger reasoning capability, suitable for solving complex problems.

Example

import os
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

from openai import OpenAI
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

from openai import OpenAI

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

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

import base64
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 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="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 os
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

from openai import OpenAI

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

import asyncio
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

import asyncio
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

import asyncio
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 catchAPIErrorsubclasses, 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

More API references:https://github.com/openai/openai-python/blob/main/api.md

Other Extensions