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

实例

import os
from openai import OpenAI

client = OpenAI(
    # This is the default and can be omitted
    api_key="你申请的 API key",
)

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)


参数说明：

参数名	是否必填	类型	作用说明
api_key	是	str	申请的 OpenAI key
model	是	str	指定使用的 OpenAI 大模型，决定能力、推理水平和成本
instructions	否	str	系统级指令（System Prompt），定义模型的身份、行为规范和表达风格，优先级高于 input
input	是	str / list	用户输入内容，描述具体问题或任务

---

第三方模型

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

讯飞星辰 MaaS

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



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



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



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

实例

from openai import OpenAI  

# 必填：从服务管控页面获取对应服务的APIKey和API Base
api_key = "上图中你申请的 APIKey"
api_base = "http://maas-api.cn-huabei-1.xf-yun.com/v2"

client = OpenAI(api_key=api_key, base_url=api_base)

response = client.chat.completions.create(
    model="上图中的 modelId",  # 模型名称，必填项
    messages=[
        {"role": "system", "content": "You are a helpful assistant"},
        {"role": "user", "content": "Hello"},
    ],
    stream=False    # stream=False 非流式（一次性返回）、stream=True 流式（实时返回）
)

print(response.choices[0].message.content)


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

Hello! How can I help you today?

更完整的调用：

实例

from openai import OpenAI  

# 必填：从服务管控页面获取对应服务的APIKey和API Base
api_key = "<YOUR_API_KEY>"
api_base = "http://maas-api.cn-huabei-1.xf-yun.com/v2"

client = OpenAI(api_key=api_key, base_url=api_base)

def unified_chat_test(model_id, messages, use_stream=False, extra_body={}):
    """
    一个统一的函数，用于演示多种调用场景。

    :param model_id: 要调用的模型ID。
    :param messages: 对话消息列表。
    :param use_stream: 是否使用流式输出。
    :param extra_body: 包含额外请求参数的字典，如 response_format。
    """
    try:
        response = client.chat.completions.create(
            model=model_id,
            messages=messages,
            stream=use_stream,
            temperature=0.7,
            max_tokens=4096,
            extra_headers={"lora_id": "0"},  # 调用微调大模型时,对应替换为模型服务卡片上的resourceId
            stream_options={"include_usage": True},
            extra_body=extra_body
        )

        if use_stream:
            # 处理流式响应
            full_response = ""
            print("--- 流式输出 ---")
            for chunk in response:
                if hasattr(chunk.choices[0].delta, 'content') and chunk.choices[0].delta.content:
                    content = chunk.choices[0].delta.content
                    print(content, end="", flush=True)
                    full_response += content
            print("\n\n--- 完整响应 ---")
            print(full_response)
        else:
            # 处理非流式响应
            print("--- 非流式输出 ---")
            message = response.choices[0].message
            print(message.content)

    except Exception as e:
        print(f"请求出错: {e}")

if __name__ == "__main__":
    model_id = "<YOUR_MODEL_ID>" # 必填：调用大模型时，对应为推理服务的模型卡片上对应的modelId

    # 1. 普通非流式调用
    print("*** 1. 普通非流式调用 ***")
    plain_messages = [{"role": "user", "content": "你好，请介绍一下自己。"}]
    unified_chat_test(model_id, plain_messages, use_stream=False)

    # 2. 普通流式调用
    print("\n*** 2. 普通流式调用 ***")
    stream_messages = [{"role": "user", "content": "写一首关于夏天的诗。"}]
    unified_chat_test(model_id, stream_messages, use_stream=True)

    # 3. JSON Mode 调用
    print("\n*** 3. JSON Mode 调用 ***")
    json_messages = [{"role": "user", "content": "请给我一个关于上海的JSON对象，包含城市名称(city)和人口数量(population)。"}]
    json_extra_body = {
        "response_format": {"type": "json_object"},
        "search_disable": True # JSON Mode下建议关闭搜索
    }
    unified_chat_test(model_id, json_messages, use_stream=False, extra_body=json_extra_body)

    # 4. 测试stop和前缀续写功能
    print("\n*** 4. 测试stop和前缀续写功能 ***")
    print("设置stop词: ['。', '！'] - 模型遇到句号或感叹号时会停止生成")
    stream_messages = [{"role": "user", "content": "给我解释下1加1等于多少。"}]
    unified_chat_test(model_id, stream_messages, use_stream=True, extra_body={"stop": ["。","！"],"continue_final_message":True})

    # 5. Tools/Function Calling 调用示例
    print("\n*** 5. Tools/Function Calling 调用示例 ***")
    tools = [
        {
            "type": "function",
            "function": {
                "name": "get_weather",
                "description": "获取指定城市的天气信息",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "location": {
                            "type": "string",
                            "description": "城市名称，例如：北京、上海"
                        },
                        "unit": {
                            "type": "string",
                            "enum": ["celsius", "fahrenheit"],
                            "description": "温度单位"
                        }
                    },
                    "required": ["location"]
                }
            }
        }
    ]
    tool_messages = [{"role": "user", "content": "北京今天天气怎么样？"}]
    response = client.chat.completions.create(
        model=model_id,
        messages=tool_messages,
        tools=tools,
        tool_choice="auto"
    )
    message = response.choices[0].message
    if message.tool_calls:
        print(f"模型请求调用工具: {message.tool_calls[0].function.name}")
        print(f"参数: {message.tool_calls[0].function.arguments}")
    else:
        print(message.content)


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

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

输出结果如下：

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

很高兴与你交流！我可以作为你的智能伙伴，在多个领域为你提供支持和帮助，例如：
  <strong>内容创作与编辑</strong>：协助你撰写文章、邮件、报告、故事，或进行文本润色和翻译。

• **逻辑推理与问题解决

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。

安装与导入

安装官方 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