Python OpenAI

openai 是一个强大的 Python 库，用于与 OpenAI 的一系列模型和服务进行交互。

openai 封装了所有 RESTful API 调用，让开发者能轻松地在自己的 Python 应用中集成强大的 AI 能力，例如自然语言处理、图像生成和语音识别等。

主要功能：

• 文本生成：使用 GPT-4 或 GPT-5 等模型生成文章、代码、摘要、对话等。
• 图像生成：通过 DALL-E 模型根据文本描述创建图像。
• 嵌入（Embeddings）：将文本转换成向量表示，常用于语义搜索、文本分类和聚类等任务。
• 语音转文本：使用 Whisper 模型将音频文件转录成文本。
• 微调（Fine-tuning）：通过提供自己的数据集来训练一个更具针对性的模型。
• 助手（Assistants）API：构建能够理解上下文、调用工具并进行长期交互的复杂应用。

openai 开源地址：https://github.com/openai/openai-python

如何使用？

环境要求：

• Python 版本：3.9 及以上。
• 依赖库：httpx（默认）、aiohttp（可选）、websockets（Realtime API 需）。

首先，你需要用 pip 安装 openai 库：

pip install openai

或

pip3 install openai

然后需要去 OpenAI 官网 https://platform.openai.com/ 注册账号，并在 API 密钥页面生成一个 API Key。

查看安装版本:

import openai
print(openai.__version__)  # 输出当前 SDK 版本

参数说明：

第三方模型

我们国内目前访问 openai 还是有点麻烦，国内很多也支持 openai，比如 DeepSeek、千问、GLM。

讯飞星辰 MaaS

访问讯飞星辰 MaaS 讯飞大模型集合平台，目前有免费的大模型产品可以用。

鼠标移到免费的模型上，然后，选择 API 调用，设置名称及授权的应用：

看了下协议支持，现在只有 OpenAI 的兼容协议：

我们写个 Python 文件测试看看：

填好对应的 api_key、api_base、model 三个参数，然后就可以执行了：

Hello! How can I help you today?

更完整的调用：

注意以代码你要设置为你申请的 api_key 和 model_id：

api_key = "xxx" # 这里配置要修改
api_base = "http://maas-api.cn-huabei-1.xf-yun.com/v2"
model_id = " xxxx" # 这里配置要修改

输出结果如下：

********* 1. 普通非流式调用 *********
--- 非流式输出 ---
你好！我是通义千问（Qwen），是由阿里巴巴集团通义实验室独立开发的大语言模型。

很高兴与你交流！我可以作为你的智能伙伴，在多个领域为你提供支持和帮助，例如：

*   **内容创作与编辑**：协助你撰写文章、邮件、报告、故事，或进行文本润色和翻译。
*   **逻辑推理与问题解决
...

DeepSeek

DeepSeek API 完全兼容 OpenAI 的 API 格式，只需修改少量配置，即可直接使用 OpenAI SDK 或兼容工具访问 DeepSeek API。

使用方式

接下来我们使用 OpenAI SDK 访问百炼服务上的通义千问模型。

非流式调用示例

运行代码可以获得以下结果：

我是通义千问，阿里巴巴集团旗下的通义实验室自主研发的超大规模语言模型。我可以帮助你回答问题、创作文字，比如写故事、写公文、写邮件、写剧本、逻辑推理、编程等等，还能表达观点，玩等。如果你有任何问题或需要帮助，欢迎随时告诉我！

流式调用示例

运行代码可以获得以下结果：

我是通义千问，阿里巴巴集团旗下的通义实验室自主研发的超大规模语言模型。我可以帮助你回答问题、创作文字，比如写故事、写公文、写邮件、写剧本、逻辑推理、编程等等，还能表达观点，玩等。如果你有任何问题或需要帮助，欢迎随时告诉我！

特色功能

视觉（Vision）功能

支持图片输入（URL 或 Base64 编码），实现多模态交互。

图片 URL 输入:

Base64 编码图片输入:

异步使用（Async usage）

替换 OpenAI 为 AsyncOpenAI，并通过 await 调用 API：

基础异步示例:

异步场景优化（aiohttp 后端），安装扩展版本：

pip install openai[aiohttp]

aiohttp 后端优化:

流式响应（Streaming responses）

基于 Server Side Events (SSE) 实时获取响应，同步 / 异步接口一致。

同步流式示例:

异步流式示例:

实时 API（Realtime API）

低延迟多模态对话（文本 / 音频），基于 WebSocket 实现，依赖 websockets 库。

基础文本示例:

实时 API 错误处理:

参考手册

以下表格包含的内容为：

• 核心初始化：同步用 OpenAI、异步用 AsyncOpenAI、Azure 用 AzureOpenAI，推荐通过环境变量配置 API Key。
• 文本生成：新版优先用 responses.create()，经典场景用 chat.completions.create()，流式输出需设置 stream=True。
• 高级能力：支持多模态（图片输入）、实时 API、文件上传/微调，错误处理需捕获 APIError 子类，超时/重试可灵活配置。

安装与导入

安装官方 SDK：

pip install --upgrade openai

导入核心类：

from openai import OpenAI, AsyncOpenAI, AzureOpenAI

创建客户端

创建标准 OpenAI 客户端：

from openai import OpenAI

client = OpenAI(api_key="YOUR_API_KEY")

通过环境变量创建客户端：

import os
from openai import OpenAI

os.environ["OPENAI_API_KEY"] = "YOUR_API_KEY"
client = OpenAI()

设置请求超时与重试次数：

client = OpenAI(
    timeout=30,
    max_retries=2
)

文本生成（Responses API）

最基础的文本生成请求：

response = client.responses.create(
    model="gpt-5.2",
    input="用一句话解释什么是 Python"
)

print(response.output_text)

通过 instructions 约束模型行为：

response = client.responses.create(
    model="gpt-5.2",
    instructions="你是一个严谨的 Python 教程作者",
    input="解释什么是列表推导式"
)

print(response.output_text)

多轮对话

使用 input 数组构建上下文对话：

response = client.responses.create(
    model="gpt-5.2",
    input=[
        {"role": "user", "content": "Python 是什么？"},
        {"role": "assistant", "content": "Python 是一门高级编程语言。"},
        {"role": "user", "content": "它适合做什么？"}
    ]
)

print(response.output_text)

结构化输出（JSON Schema）

要求模型返回符合 Schema 的 JSON 数据：

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)

工具调用（Function Calling）

定义工具并让模型自动决定是否调用：

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）

同步流式输出：

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="")

异步流式输出：

stream = await client.responses.create(
    model="gpt-5.2",
    input="解释什么是生成式 AI",
    stream=True
)

async for event in stream:
    print(event)

异步调用（AsyncOpenAI）

在 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())

实时 API（Realtime）

使用 WebSocket 进行低延迟实时对话：

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 = client.images.generate(
    model="gpt-image-1",
    prompt="一只在写代码的猫",
    size="1024x1024"
)

print(image.data[0].url)

嵌入生成（Embeddings）

生成文本向量嵌入：

embedding = client.embeddings.create(
    model="text-embedding-3-small",
    input="Hello world"
)

print(embedding.data[0].embedding)

音频转文本（Speech 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）

将文本转换为语音文件：

speech = client.audio.speech.create(
    model="gpt-4o-mini-tts",
    voice="alloy",
    input="你好，欢迎使用 OpenAI"
)

with open("output.mp3", "wb") as f:
    f.write(speech)

文件管理（Files API）

上传文件：

from pathlib import Path

client.files.create(
    file=Path("data.jsonl"),
    purpose="fine-tune"
)

列出文件：

files = client.files.list()
for f in files:
    print(f.id, f.filename)

删除文件：

client.files.delete(file_id="file-xxx")

模型微调（Fine-tuning Jobs）

创建微调任务：

job = client.fine_tuning.jobs.create(
    training_file="file-xxx",
    model="gpt-3.5-turbo"
)

print(job.id, job.status)

Webhook 验证

验证并解析 Webhook 请求：

event = client.webhooks.unwrap(request_body, request_headers)

仅验证签名：

client.webhooks.verify_signature(request_body, request_headers)

错误处理

捕获 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)

获取失败请求的 Request ID：

try:
    client.responses.create(model="gpt-5.2", input="test")
except openai.APIStatusError as exc:
    print(exc.request_id)

Azure OpenAI 兼容

初始化 Azure OpenAI 客户端：

from openai import AzureOpenAI

client = AzureOpenAI(
    api_version="2023-07-01-preview",
    azure_endpoint="https://xxx.openai.azure.com"
)

版本与日志

查看 SDK 版本：

import openai
print(openai.__version__)

开启 SDK 调试日志：

export OPENAI_LOG=debug

更多 API 参考：https://github.com/openai/openai-python/blob/main/api.md