概述

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

安装

pip install openai

快速开始

from openai import OpenAI

client = OpenAI(
    api_key=os.environ.get("OPENAI_API_KEY"),
)

response = client.responses.create(
    model="gpt-5.2",
    instructions="You are a helpful assistant.",
    input="Hello, world!",
)

print(response.output_text)

认证方式

API Key认证

from openai import OpenAI

client = OpenAI(
    api_key="your-api-key"
)

工作负载身份认证

适用于Kubernetes、Azure和Google Cloud Platform等云环境：

from openai import OpenAI
from openai.auth import k8s_service_account_token_provider

client = OpenAI(
    workload_identity={
        "client_id": "your-client-id",
        "identity_provider_id": "idp-123",
        "service_account_id": "sa-456",
        "provider": k8s_service_account_token_provider(
            "/var/run/secrets/kubernetes.io/serviceaccount/token"
        ),
    },
)

图像识别

支持输入图像进行视觉理解：

from openai import OpenAI

client = OpenAI()

response = client.responses.create(
    model="gpt-5.2",
    input=[
        {
            "role": "user",
            "content": [
                {"type": "input_text", "text": "这张图片里有什么?"},
                {"type": "input_image", "image_url": "https://example.com/image.jpg"},
            ],
        }
    ],
)

异步调用

import asyncio
from openai import AsyncOpenAI

client = AsyncOpenAI()

async def main():
    response = await client.responses.create(
        model="gpt-5.2",
        input="Explain quantum computing"
    )
    print(response.output_text)

asyncio.run(main())

流式输出

from openai import OpenAI

client = OpenAI()

stream = client.responses.create(
    model="gpt-5.2",
    input="Write a story",
    stream=True,
)

for event in stream:
    print(event)

实时API

支持低延迟多模态对话：

import asyncio
from openai import AsyncOpenAI

async def main():
    client = AsyncOpenAI()

    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": "你好!"}]
            }
        )
        await conn.response.create()

        async for event in conn:
            if event.type == "response.output_text.delta":
                print(event.delta, end="")
            elif event.type == "response.done":
                break

asyncio.run(main())

主要API端点

错误处理

from openai import OpenAI
from openai import RateLimitError, APIError

client = OpenAI()

try:
    response = client.responses.create(
        model="gpt-5.2",
        input="Hello"
    )
except RateLimitError:
    print("请求频率超限，请稍后重试")
except APIError as e:
    print(f"API错误: {e.message}")