1. 什么是 OpenAI？ #
OpenAI Python 库是官方提供的 Python 工具包，用于与 OpenAI 的各种 AI 服务进行交互。通过这个库，你可以轻松地在 Python 程序中使用 ChatGPT、图像生成、语音识别等强大的 AI 功能。
主要功能：
- 文本对话：与 ChatGPT 进行对话，获取智能回复
- 图像生成：使用 DALL-E 根据文字描述生成图片
- 语音转文字：使用 Whisper 将音频转换为文字
- 文本嵌入：将文本转换为向量，用于相似度计算和搜索
为什么使用这个库？
- 官方支持，稳定可靠
- API 简单易用，几行代码就能实现强大功能
- 功能全面，覆盖 OpenAI 所有主要服务
- 文档完善，社区支持好
适用场景：
- 构建智能聊天机器人
- 开发内容生成工具
- 实现智能问答系统
- 创建文本分析和处理工具
2. 前置知识补充 #
在开始使用之前，我们先了解一些基础概念，这些知识会帮助你更好地理解和使用 OpenAI 的服务。
2.1 什么是 API？ #
API（Application Programming Interface） 是应用程序编程接口的缩写。简单理解，API 就是两个程序之间通信的桥梁。
类比理解：
- 就像餐厅的服务员，你告诉服务员要什么菜（发送请求），服务员把菜端给你（返回结果）
- API 就是那个"服务员"，帮你向 OpenAI 的服务器发送请求，并接收返回的结果
在 OpenAI 中的应用：
- 你写代码调用 OpenAI 的 API
- API 把你的请求发送到 OpenAI 的服务器
- 服务器处理请求（比如生成文本、图片等）
- 结果通过 API 返回给你的程序
2.2 什么是 API Key？ #
API Key（API 密钥） 是一串用于身份验证的字符串，类似于密码。使用 OpenAI 的服务需要先获取 API Key。
为什么需要 API Key？
- 身份验证：证明你有权限使用服务
- 计费管理：OpenAI 根据 API Key 来计费
- 安全控制：防止未授权访问
如何获取 API Key？
- 访问 https://platform.openai.com/
- 注册或登录账号
- 进入 API Keys 页面
- 点击 "Create new secret key" 创建新的密钥
- 复制并保存密钥（只显示一次，务必保存好）
安全提示：
- 不要将 API Key 分享给他人
- 不要将 API Key 提交到代码仓库（如 GitHub）
- 如果泄露，立即删除并创建新的密钥
2.3 什么是模型（Model）？ #
模型是经过训练的人工智能程序，可以理解输入并生成输出。不同的模型有不同的能力和特点。
OpenAI 的主要模型：
文本对话模型：
gpt-4：最强大的模型，理解能力强，适合复杂任务gpt-4o：速度快，成本低，适合大多数应用
图像生成模型：
dall-e-3：最新的图像生成模型，质量高
语音识别模型：
whisper-1：语音转文字模型，支持多种语言
文本嵌入模型：
text-embedding-ada-002：将文本转换为向量
选择建议：
- 大多数应用使用
gpt-4o即可 - 需要更高精度时使用
gpt-4 - 根据具体需求选择合适的模型
2.4 什么是 Token？ #
Token 是模型处理文本的基本单位。对于英文，一个 token 大约等于 0.75 个单词；对于中文，一个 token 大约等于 1-2 个汉字。
为什么重要？
- OpenAI 按 token 数量计费
- 模型有最大 token 限制（如 4096 tokens）
- 需要控制输入和输出的 token 数量
估算方法：
- 英文：1000 tokens ≈ 750 个单词
- 中文：1000 tokens ≈ 500-700 个汉字
3. 安装 OpenAI Python 库 #
在开始使用之前，需要先安装 OpenAI Python 库。
3.1 Windows 系统安装 #
在 Windows PowerShell 或命令提示符中执行：
# 升级 pip 到最新版本（推荐）
python -m pip install --upgrade pip
# 安装 openai 库
python -m pip install openai注意事项：
- 如果系统中有多个 Python 版本，可能需要使用
python3代替python - 如果遇到权限问题，可以添加
--user参数：python -m pip install --user openai
3.2 macOS 系统安装 #
在 macOS 终端中执行：
# 升级 pip 到最新版本（推荐）
python3 -m pip install --upgrade pip
# 安装 openai 库
python3 -m pip install openai注意事项：
- macOS 系统通常需要使用
python3命令 - 如果提示找不到
python3，可能需要先安装 Python
3.3 验证安装 #
安装完成后，可以通过以下代码验证是否安装成功：
# 尝试导入 openai 库
try:
from openai import OpenAI
print("OpenAI 库安装成功！")
except ImportError:
print("安装失败，请检查安装过程")4. 初始化客户端 #
在使用 OpenAI 的服务之前，需要先创建一个客户端对象。客户端是连接你的程序和 OpenAI 服务器的桥梁。
4.1 获取 API Key #
在初始化客户端之前，你需要先获取 API Key（参考第 2.2 节）。
4.2 创建客户端（推荐方式：环境变量） #
最安全和推荐的方式是使用环境变量存储 API Key。
# 导入必要的模块
import os
from openai import OpenAI
# 方式 1：直接在代码中设置环境变量（仅用于测试，不推荐用于生产环境）
# 将 "your-api-key-here" 替换为你的实际 API Key
os.environ["OPENAI_API_KEY"] = "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"
# 创建客户端对象
# 如果不传入 api_key 参数，会自动从环境变量 OPENAI_API_KEY 读取
client = OpenAI()
# 测试客户端是否正常工作
print("客户端创建成功！")Windows 系统设置环境变量：
# 临时设置（当前会话有效）
$env:OPENAI_API_KEY = "your-api-key-here"
# 永久设置（需要管理员权限）
[System.Environment]::SetEnvironmentVariable("OPENAI_API_KEY", "your-api-key-here", "User")macOS 系统设置环境变量：
# 临时设置（当前会话有效）
export OPENAI_API_KEY="your-api-key-here"
# 永久设置（添加到 ~/.zshrc 或 ~/.bash_profile）
echo 'export OPENAI_API_KEY="your-api-key-here"' >> ~/.zshrc
source ~/.zshrc4.3 创建客户端（直接传入 API Key） #
如果不想使用环境变量，也可以直接传入 API Key。
# 导入 OpenAI 类
from openai import OpenAI
# 直接传入 API Key 创建客户端
# 将 "your-api-key-here" 替换为你的实际 API Key
client = OpenAI(api_key="your-api-key-here")
# 测试客户端
print("客户端创建成功！")注意： 这种方式虽然简单，但不推荐用于生产环境，因为 API Key 会暴露在代码中。
4.4 完整配置示例 #
如果需要更多配置选项，可以使用完整配置：
# 导入必要的模块
import os
from openai import OpenAI
# 创建客户端，包含所有配置选项
client = OpenAI(
api_key=os.getenv("OPENAI_API_KEY", "your-api-key-here"), # API Key
organization="your-org-id", # 组织 ID（可选，企业用户使用）
base_url="https://api.openai.com/v1", # API 基础 URL（通常不需要修改）
timeout=30.0, # 请求超时时间（秒）
)
print("客户端创建成功！")5. 与 ChatGPT 对话 #
5.1 基础对话示例 #
下面的代码演示如何发送一条消息并获取回复。
# 导入 OpenAI 类
from openai import OpenAI
from pprint import pprint
import os
os.environ["OPENAI_API_KEY"] = "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"
# 创建客户端（需要先设置环境变量 OPENAI_API_KEY，或直接传入 api_key）
client = OpenAI()
# 调用聊天补全 API
# chat.completions.create 是用于对话的方法
response = client.chat.completions.create(
model="gpt-4o", # 使用的模型，gpt-4o 速度快、成本低
messages=[
# messages 是一个消息列表，包含对话历史
{
"role": "user", # 角色：user 表示用户
"content": "1+1=?" # 消息内容
}
],
max_tokens=500, # 最大生成 token 数，控制回复长度
temperature=0.7 # 温度参数，控制回复的随机性（0-2，越大越随机）
)
pprint(response)
# 提取回复内容
# response.choices[0] 表示第一个选择（通常只有一个）
# .message.content 是消息内容
reply = response.choices[0].message.content
# 打印回复
print("ChatGPT 的回复：")
print(reply)
# 导入 pprint 模块用于美观地打印数据结构
from pprint import pprint
# 定义 chat_completion 字典，包含一次聊天补全的所有返回信息
chat_completion = {
# 聊天补全的唯一ID
'id': 'chatcmpl-Cej78G1Mw012YW4dKu61R7Z4Mu3Bq',
# choices 是模型生成的回复列表，这里只有一个回复
'choices': [
{
# 模型回复完成的原因
'finish_reason': 'stop',
# 当前回复的索引
'index': 0,
# 对数概率信息，这里为None
'logprobs': None,
# message 字典，包含具体回复内容和相关信息
'message': {
# 回复内容
'content': '1 + 1 equals 2.',
# 是否为拒答，这里为None
'refusal': None,
# 回复的角色（助理）
'role': 'assistant',
# 注释列表，这里为空
'annotations': [],
# 音频内容，这里为空
'audio': None,
# 函数调用信息，这里为空
'function_call': None,
# 工具调用信息，这里为空
'tool_calls': None
}
}
],
# 响应创建时间（Unix时间戳）
'created': 1763822574,
# 使用的模型名称
'model': 'gpt-4o-2024-08-06',
# 响应的对象类型
'object': 'chat.completion',
# 服务级别
'service_tier': 'default',
# 系统指纹
'system_fingerprint': 'fp_689bad8e9a',
# token 使用信息，记录本次生成请求消耗的 token 数量
'usage': {
# 回复时生成的 token 数
'completion_tokens': 8,
# 提示词消耗的 token 数
'prompt_tokens': 11,
# 总共消耗的 token 数
'total_tokens': 19,
# 详细的 completion token 统计
'completion_tokens_details': {
# 被接受的预测 token 数
'accepted_prediction_tokens': 0,
# 音频 token 数
'audio_tokens': 0,
# 推理 token 数
'reasoning_tokens': 0,
# 被拒绝的预测 token 数
'rejected_prediction_tokens': 0
},
# 详细的 prompt token 统计
'prompt_tokens_details': {
# 音频 token 数
'audio_tokens': 0,
# 已缓存的 token 数
'cached_tokens': 0
}
}
}
# 使用 pprint 以更友好的格式输出 chat_completion 字典
pprint(chat_completion, width=80, depth=3, sort_dicts=False)运行说明：
- 将代码保存为
first_chat.py - 确保已设置 API Key（环境变量或直接传入）
- 在终端运行：
python first_chat.py（Windows）或python3 first_chat.py（macOS） - 首次运行可能需要几秒钟，等待 API 响应
5.2 多轮对话示例 #
ChatGPT 可以记住对话历史，实现多轮对话。
# 导入 OpenAI 类
from openai import OpenAI
import os
os.environ["OPENAI_API_KEY"] = "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"
# 创建客户端
client = OpenAI()
# 存储对话历史
# 对话历史需要包含所有之前的消息
conversation_history = [
{
"role": "system", # system 角色用于设置助手的行为
"content": "你是一个友好的编程助手，擅长解释编程概念。"
}
]
# 第一轮对话
print("=== 第一轮对话 ===")
user_message_1 = "我叫张三"
conversation_history.append({
"role": "user",
"content": user_message_1
})
# 发送请求
response_1 = client.chat.completions.create(
model="gpt-4o",
messages=conversation_history,
max_tokens=300
)
# 获取回复
assistant_reply_1 = response_1.choices[0].message.content
print(f"用户：{user_message_1}")
print(f"助手：{assistant_reply_1}\n")
# 将助手的回复添加到对话历史
conversation_history.append({
"role": "assistant",
"content": assistant_reply_1
})
# 第二轮对话（基于之前的对话）
print("=== 第二轮对话 ===")
user_message_2 = "我叫什么?"
conversation_history.append({
"role": "user",
"content": user_message_2
})
# 发送请求（包含完整的对话历史）
response_2 = client.chat.completions.create(
model="gpt-4o",
messages=conversation_history,
max_tokens=300
)
# 获取回复
assistant_reply_2 = response_2.choices[0].message.content
print(f"用户：{user_message_2}")
print(f"助手：{assistant_reply_2}")运行说明：
- 运行后会看到两轮对话
- 第二轮的回复会基于第一轮的对话内容
- 这就是 ChatGPT 能够"记住"对话历史的原因
6. 核心功能 #
本节详细介绍 OpenAI 库的主要功能，包括文本对话、图像生成、语音识别等。
6.1 文本对话（Chat Completions） #
文本对话是最常用的功能，用于与 ChatGPT 进行交互。
6.1.1 基础参数说明 #
# 导入 OpenAI 类
from openai import OpenAI
# uv add python-dotenv
import dotenv
dotenv.load_dotenv()
# 创建客户端
client = OpenAI()
# 调用聊天补全 API，包含常用参数
response = client.chat.completions.create(
model="gpt-4o", # 模型名称
# messages 是消息列表，必须包含至少一条消息
messages=[
{
"role": "system", # system: 设置助手的行为和性格
"content": "你是一个有帮助的助手。"
},
{
"role": "user", # user: 用户的消息
"content": "请用简单的话解释什么是人工智能"
}
],
# max_tokens: 最大生成 token 数，控制回复长度
# 注意：输入和输出都计入 token 限制
max_tokens=500,
# temperature: 温度参数，控制回复的随机性
# 0.0: 非常确定，几乎总是相同回复
# 1.0: 平衡（推荐）
# 2.0: 非常随机，每次回复都不同
temperature=0.7,
# top_p: 核采样参数，与 temperature 类似，通常只设置一个
# top_p=0.9,
# n: 生成几个不同的回复（默认 1）
n=1,
# stream: 是否使用流式响应（默认 False）
# 流式响应可以实时显示生成的内容
stream=False
)
# 提取回复
message = response.choices[0].message.content
print(message)6.1.2 流式响应 #
流式响应可以实时显示生成的内容，提供更好的用户体验。
# 导入 OpenAI 类
from openai import OpenAI
# 导入 dotenv 用于加载环境变量
import dotenv
# 加载 .env 文件中的环境变量
dotenv.load_dotenv()
# 创建 OpenAI 客户端实例
client = OpenAI()
# 调用 ChatGPT API，开启流式输出模式
response = client.chat.completions.create(
# 指定使用的模型为 gpt-4o
model="gpt-4o",
# 提供消息列表，包含一条用户消息
messages=[
{
# 角色是 user，表示用户
"role": "user",
# 用户输入的内容，要求生成一个关于机器学习的短故事
"content": "请写一个关于机器学习的短故事"
}
],
# 设置生成回复的最大 token 数
max_tokens=500,
# 启用流式响应，实时返回生成内容
stream=True
)
# 提示用户 ChatGPT 正在生成内容
print("ChatGPT 正在生成回复：\n")
# 遍历流式返回的内容块
for chunk in response:
# 检查本块是否有新的内容
if chunk.choices[0].delta.content is not None:
# 实时输出内容块，end="" 保证不换行，flush=True 立即刷新
print(chunk.choices[0].delta.content, end="", flush=True)
# 回复完成后打印换行和结束信息
print("\n\n生成完成！")运行说明：
- 运行后会看到内容逐字显示，就像 ChatGPT 在实时打字
- 适合生成较长内容时使用
6.2 图像生成（DALL-E） #
使用 DALL-E 可以根据文字描述生成图片。
# 导入 OpenAI 类
from openai import OpenAI
import dotenv
dotenv.load_dotenv()
# 创建客户端
client = OpenAI()
# 生成图像
# images.generate 用于生成图像
response = client.images.generate(
model="dall-e-3", # 使用 DALL-E 3 模型
# prompt: 图像描述，越详细越好
prompt="一只穿着西装会编程的猫，坐在电脑前，数字艺术风格，高清",
# size: 图像尺寸
# dall-e-3 支持: "1024x1024", "1792x1024", "1024x1792"
size="1024x1024",
# quality: 图像质量
# "standard": 标准质量（默认）
# "hd": 高清质量（更贵）
quality="standard",
# n: 生成图像数量（dall-e-3 只支持 1）
n=1,
)
# 获取图像 URL
# response.data[0] 是第一个（也是唯一一个）图像
# .url 是图像的 URL 地址
image_url = response.data[0].url
# 打印图像 URL
print(f"生成的图片 URL: {image_url}")
print("\n提示：")
print("1. 图像 URL 会在 1 小时后过期")
print("2. 可以右键点击 URL 在浏览器中打开查看")
print("3. 或者使用 requests 库下载图像")下载图像示例：
# 导入必要的库
from openai import OpenAI
import requests
# 创建客户端
client = OpenAI()
# 生成图像
response = client.images.generate(
model="dall-e-3",
prompt="一只可爱的熊猫在竹林中，水彩画风格",
size="1024x1024",
quality="standard",
n=1,
)
# 获取图像 URL
image_url = response.data[0].url
# 下载图像
# 使用 requests 库下载图像内容
image_response = requests.get(image_url)
# 保存图像到本地文件
with open("generated_image.png", "wb") as f:
# 将图像内容写入文件
f.write(image_response.content)
print("图像已保存为 generated_image.png")注意： 需要先安装 requests 库：pip install requests（Windows）或 pip3 install requests（macOS）
6.3 语音转文字（Whisper） #
使用 Whisper 可以将音频文件转换为文字。
# 导入 OpenAI 类
from openai import OpenAI
import dotenv
#pip install opencc-python-reimplemented
from opencc import OpenCC # 添加简体繁体转换
dotenv.load_dotenv()
# 创建客户端
client = OpenAI()
# 打开音频文件
# 注意：将 "audio.mp3" 替换为你的实际音频文件路径
# 支持格式：mp3, mp4, mpeg, mpga, m4a, wav, webm
try:
with open("audio.mp3", "rb") as audio_file:
# 转录音频
# audio.transcriptions.create 用于将音频转换为文字
transcript = client.audio.transcriptions.create(
model="whisper-1", # 使用 Whisper 模型
# file: 音频文件对象
file=audio_file,
# response_format: 返回格式
# "text": 纯文本（默认）
# "json": JSON 格式
# "srt": 字幕格式
# "vtt": WebVTT 格式
response_format="text"
)
# 创建繁体转简体的转换器
cc = OpenCC('t2s') # t2s 表示繁体转简体
# 将转录结果转换为简体中文
simplified_text = cc.convert(transcript)
# 打印转录结果（简体中文）
print("转录结果：")
print(simplified_text)
except FileNotFoundError:
print("错误：找不到音频文件 audio.mp3")
print("请将音频文件放在当前目录，或修改文件路径")
except Exception as e:
print(f"处理音频时出错：{e}")音频翻译示例：
# 导入 OpenAI 类
from openai import OpenAI
# 创建客户端
client = OpenAI()
# 打开音频文件（可以是任何语言的音频）
try:
with open("chinese_audio.mp3", "rb") as audio_file:
# 翻译音频为英文
# audio.translations.create 用于将音频翻译为英文
translation = client.audio.translations.create(
model="whisper-1",
file=audio_file
)
# 打印翻译结果
print("翻译结果（英文）：")
print(translation.text)
except FileNotFoundError:
print("错误：找不到音频文件 chinese_audio.mp3")
except Exception as e:
print(f"处理音频时出错：{e}")6.4 文本嵌入（Embeddings） #
文本嵌入可以将文本转换为向量，用于相似度计算和搜索。
# 导入 OpenAI 类
from openai import OpenAI
# 创建客户端
client = OpenAI()
# 生成文本嵌入
# embeddings.create 用于生成文本嵌入向量
response = client.embeddings.create(
model="text-embedding-ada-002", # 嵌入模型
# input: 要嵌入的文本
# 可以是单个字符串或字符串列表
input="这是一个测试文本，用于生成嵌入向量。",
# encoding_format: 编码格式
# "float": 浮点数数组（默认）
# "base64": Base64 编码（较少使用）
encoding_format="float"
)
# 获取嵌入向量
# response.data[0] 是第一个文本的嵌入
# .embedding 是嵌入向量（一个数字数组）
embedding = response.data[0].embedding
# 打印嵌入向量信息
print(f"嵌入向量维度：{len(embedding)}")
print(f"前 10 个值：{embedding[:10]}")
# 批量生成嵌入
texts = [
"我喜欢编程",
"我爱写代码",
"今天天气很好"
]
# 可以一次处理多个文本
batch_response = client.embeddings.create(
model="text-embedding-ada-002",
input=texts, # 传入文本列表
encoding_format="float"
)
# 获取所有嵌入向量
print(f"\n批量处理 {len(texts)} 个文本：")
for i, text_embedding in enumerate(batch_response.data):
print(f"文本 {i+1} ({texts[i]}) 的嵌入向量维度：{len(text_embedding.embedding)}")7. 实现 #
7.1 实现聊天 #
7.1.1 invoke.py #
invoke.py
# 导入我们自己实现的简化版 OpenAI 客户端
from openai_client import OpenAI
# 导入 dotenv 库，用于加载 .env 文件内的环境变量
import dotenv
# 加载 .env 文件中的环境变量，使 OPENAI_API_KEY 可用
dotenv.load_dotenv()
# 创建 OpenAI 客户端实例，API key 会自动从环境变量读取
client = OpenAI()
# 调用 chat.completions.create 方法，向模型发送消息并获得回复
response = client.chat.completions.create(
model="gpt-4o", # 指定使用的模型名称
messages=[
{
"role": "user", # 消息角色为 user（用户）
"content": "1+1=?" # 消息内容
}
],
max_tokens=500, # 生成回复的最大 token 数
temperature=0.7 # 随机性温度参数
)
# 获取模型回复内容（只取第一个回复选项的消息内容）
reply = response.choices[0].message.content
# 打印 ChatGPT 的回复文本
print("ChatGPT 的回复：")
print(reply)7.1.2 openai_client.py #
openai_client.py
# 导入操作系统相关模块
import os
# 导入可选类型、列表、字典和任意类型用于类型注解
from typing import Optional, List, Dict, Any
# 导入 requests 库用于发送 HTTP 请求
import requests
# 定义 ChatCompletions 类，处理聊天相关的 API 调用
class ChatCompletions:
# 聊天完成资源类，处理聊天相关的 API 调用
"""聊天完成资源类，处理聊天相关的 API 调用"""
# 初始化函数，接收 OpenAI 客户端对象
def __init__(self, client: 'OpenAI'):
"""
初始化聊天完成资源
参数:
client: OpenAI 客户端实例
"""
# 保存 OpenAI 客户端实例到成员变量
self._client = client
# 创建聊天补全的 API 调用
def create(
self,
model: str,
messages: List[Dict[str, str]],
max_tokens: Optional[int] = None,
temperature: Optional[float] = None,
**kwargs
) -> 'ChatCompletionResponse':
"""
创建聊天完成请求
参数:
model: 模型名称，如 "gpt-4o"
messages: 消息列表，每个消息包含 "role" 和 "content"
max_tokens: 最大生成 token 数
temperature: 温度参数，控制随机性
**kwargs: 其他可选参数
返回:
ChatCompletionResponse: 包含模型回复的响应对象
"""
# 构建请求 URL，拼接 base_url 和接口路径
url = f"{self._client.base_url}/chat/completions"
# 创建请求体，包含模型名称和消息内容
body: Dict[str, Any] = {
"model": model,
"messages": messages,
}
# 如果 max_tokens 有值则添加到请求体
if max_tokens is not None:
body["max_tokens"] = max_tokens
# 如果 temperature 有值则添加到请求体
if temperature is not None:
body["temperature"] = temperature
# 添加其他额外参数到请求体
body.update(kwargs)
# 构建请求头，包含认证信息和内容类型
headers = {
"Authorization": f"Bearer {self._client.api_key}",
"Content-Type": "application/json",
}
# 发送 POST 请求到 OpenAI API
response = requests.post(
url,
headers=headers,
json=body,
timeout=self._client.timeout
)
# 如果响应状态码不为 200，则抛出异常
response.raise_for_status()
# 解析响应数据，返回 ChatCompletionResponse 对象
return ChatCompletionResponse(response.json())
# 定义 ChatResource 类，提供 chat 访问
class ChatResource:
# 聊天资源类，提供 chat 访问
"""聊天资源类，提供 chat 访问"""
# 初始化，接收 OpenAI 客户端对象
def __init__(self, client: 'OpenAI'):
"""
初始化聊天资源
参数:
client: OpenAI 客户端实例
"""
# 保存 OpenAI 客户端实例
self._client = client
# 定义 completions 属性，返回 ChatCompletions 实例
@property
def completions(self) -> ChatCompletions:
"""
获取聊天完成资源
返回:
ChatCompletions: 聊天完成资源实例
"""
# 返回 ChatCompletions 对象
return ChatCompletions(self._client)
# 定义 Message 类，表示一条对话消息
class Message:
# 消息类，表示聊天中的一条消息
"""消息类，表示聊天中的一条消息"""
# 初始化，data 为消息内容的字典
def __init__(self, data: Dict[str, Any]):
"""
初始化消息对象
参数:
data: 包含消息数据的字典
"""
# 设置消息角色，如 user、assistant、system
self.role = data.get("role") # 角色：user, assistant, system
# 设置消息正文内容
self.content = data.get("content") # 消息内容
# 定义 Choice 类，表示模型返回的一个回复选项
class Choice:
# 选择类，表示模型生成的一个回复选项
"""选择类，表示模型生成的一个回复选项"""
# 初始化，data 为该选项的字典
def __init__(self, data: Dict[str, Any]):
"""
初始化选择对象
参数:
data: 包含选择数据的字典
"""
# 获取选项索引
self.index = data.get("index") # 选择的索引
# 获取消息内容，生成 Message 实例
self.message = Message(data.get("message", {})) # 消息对象
# 获取完成原因
self.finish_reason = data.get("finish_reason") # 完成原因
# 定义 ChatCompletionResponse 类，解析返回的响应数据
class ChatCompletionResponse:
# 聊天完成响应类，包含模型的回复
"""聊天完成响应类，包含模型的回复"""
# 初始化，data 为 API 返回的数据
def __init__(self, data: Dict[str, Any]):
"""
初始化响应对象
参数:
data: API 返回的 JSON 数据
"""
# 响应 id
self.id = data.get("id") # 响应 ID
# 对象类型
self.object = data.get("object") # 对象类型
# 创建时间戳
self.created = data.get("created") # 创建时间戳
# 模型名称
self.model = data.get("model") # 使用的模型
# 解析 choices 字段，生成 Choice 实例列表
choices_data = data.get("choices", [])
self.choices = [Choice(choice_data) for choice_data in choices_data]
# 解析 usage 字段，记录 token 相关信息
usage_data = data.get("usage", {})
self.usage = {
"prompt_tokens": usage_data.get("prompt_tokens"),
"completion_tokens": usage_data.get("completion_tokens"),
"total_tokens": usage_data.get("total_tokens"),
}
# 重写 __repr__ 方法，返回响应对象的简要描述
def __repr__(self) -> str:
"""返回对象的字符串表示"""
return f"ChatCompletionResponse(id={self.id}, model={self.model}, choices={len(self.choices)})"
# 重写 __str__ 方法，调用 __repr__
def __str__(self) -> str:
"""返回对象的字符串表示"""
return self.__repr__()
# 定义 OpenAI 客户端类
class OpenAI:
# OpenAI 客户端类，用于与 OpenAI API 交互
"""OpenAI 客户端类，用于与 OpenAI API 交互"""
# 初始化函数，设置 api 密钥、基础 url 和超时时间
def __init__(
self,
api_key: Optional[str] = None,
base_url: str = "https://api.openai.com/v1",
timeout: float = 60.0
):
"""
初始化 OpenAI 客户端
参数:
api_key: API 密钥。如果为 None，则从环境变量 OPENAI_API_KEY 读取
base_url: API 基础 URL，默认为官方 OpenAI API
timeout: 请求超时时间（秒），默认 60 秒
"""
# 获取 API 密钥，优先使用参数，否则从环境变量获取
self.api_key = api_key or os.getenv("OPENAI_API_KEY")
# 如果未获取到密钥，抛出异常
if not self.api_key:
raise ValueError(
"API 密钥未设置。请设置 api_key 参数或环境变量 OPENAI_API_KEY"
)
# 移除 base_url 末尾的斜杠，设置基础 URL
self.base_url = base_url.rstrip("/") # 移除末尾的斜杠
# 设置请求超时时间
self.timeout = timeout
# 定义 chat 属性，返回 ChatResource 实例
@property
def chat(self) -> ChatResource:
"""
获取聊天资源
返回:
ChatResource: 聊天资源实例，可以访问 chat.completions.create()
"""
# 返回 ChatResource 对象
return ChatResource(self)7.2 流式输出 #
7.2.1. invoke.py #
invoke.py
# 导入我们自己实现的简化版 OpenAI 客户端
from openai_client import OpenAI
# 导入 dotenv 库，用于加载 .env 文件内的环境变量
import dotenv
# 加载 .env 文件中的环境变量，使 OPENAI_API_KEY 可用
dotenv.load_dotenv()
# 创建 OpenAI 客户端实例，API key 会自动从环境变量读取
client = OpenAI()
+# 调用 chat.completions.create 方法，启用流式输出
+print("ChatGPT 的回复（流式输出）：")
+stream = client.chat.completions.create(
model="gpt-4o", # 指定使用的模型名称
messages=[
{
"role": "user", # 消息角色为 user（用户）
+ "content": "介绍红楼梦" # 消息内容
}
],
max_tokens=500, # 生成回复的最大 token 数
+ temperature=0.7, # 随机性温度参数
+ stream=True # 启用流式输出
)
+# 遍历流式响应，逐个打印增量内容
+for chunk in stream:
+ # 检查是否有选择数据
+ if chunk.choices and len(chunk.choices) > 0:
+ # 获取增量内容（如果有）
+ delta_content = chunk.choices[0].delta.content
+ if delta_content:
+ # 打印增量内容，不换行，立即刷新
+ print(delta_content, end="", flush=True)
# 打印换行，结束流式输出
+print()7.2.2. openai_client.py #
openai_client.py
# 导入操作系统相关模块
import os
# 导入可选类型、列表、字典和任意类型用于类型注解
+from typing import Optional, List, Dict, Any, Iterator
# 导入 requests 库用于发送 HTTP 请求
import requests
+# 导入 json 模块用于解析 JSON 数据
+import json
# 定义 ChatCompletions 类，处理聊天相关的 API 调用
class ChatCompletions:
# 聊天完成资源类，处理聊天相关的 API 调用
"""聊天完成资源类，处理聊天相关的 API 调用"""
# 初始化函数，接收 OpenAI 客户端对象
def __init__(self, client: 'OpenAI'):
"""
初始化聊天完成资源
参数:
client: OpenAI 客户端实例
"""
# 保存 OpenAI 客户端实例到成员变量
self._client = client
# 创建聊天补全的 API 调用
def create(
self,
model: str,
messages: List[Dict[str, str]],
max_tokens: Optional[int] = None,
temperature: Optional[float] = None,
+ stream: bool = False,
**kwargs
+ ):
"""
创建聊天完成请求
参数:
model: 模型名称，如 "gpt-4o"
messages: 消息列表，每个消息包含 "role" 和 "content"
max_tokens: 最大生成 token 数
temperature: 温度参数，控制随机性
+ stream: 是否使用流式输出，默认为 False
**kwargs: 其他可选参数
返回:
+ 如果 stream=False: ChatCompletionResponse - 包含模型回复的响应对象
+ 如果 stream=True: Stream - 流式响应迭代器
"""
# 构建请求 URL，拼接 base_url 和接口路径
url = f"{self._client.base_url}/chat/completions"
# 创建请求体，包含模型名称和消息内容
body: Dict[str, Any] = {
"model": model,
"messages": messages,
}
# 如果 max_tokens 有值则添加到请求体
if max_tokens is not None:
body["max_tokens"] = max_tokens
# 如果 temperature 有值则添加到请求体
if temperature is not None:
body["temperature"] = temperature
+ # 添加流式输出参数
+ if stream:
+ body["stream"] = True
+
# 添加其他额外参数到请求体
body.update(kwargs)
# 构建请求头，包含认证信息和内容类型
headers = {
"Authorization": f"Bearer {self._client.api_key}",
"Content-Type": "application/json",
}
+ # 根据是否流式输出选择不同的处理方式
+ if stream:
+ # 流式输出：发送流式请求并返回流式响应对象
+ response = requests.post(
+ url,
+ headers=headers,
+ json=body,
+ timeout=self._client.timeout,
+ stream=True # 启用流式响应
+ )
+ response.raise_for_status()
+ return Stream(response)
+ else:
+ # 非流式输出：等待完整响应后返回
+ response = requests.post(
+ url,
+ headers=headers,
+ json=body,
+ timeout=self._client.timeout
+ )
+ response.raise_for_status()
+ return ChatCompletionResponse(response.json())
# 定义 ChatResource 类，提供 chat 访问
class ChatResource:
# 聊天资源类，提供 chat 访问
"""聊天资源类，提供 chat 访问"""
# 初始化，接收 OpenAI 客户端对象
def __init__(self, client: 'OpenAI'):
"""
初始化聊天资源
参数:
client: OpenAI 客户端实例
"""
# 保存 OpenAI 客户端实例
self._client = client
# 定义 completions 属性，返回 ChatCompletions 实例
@property
def completions(self) -> ChatCompletions:
"""
获取聊天完成资源
返回:
ChatCompletions: 聊天完成资源实例
"""
# 返回 ChatCompletions 对象
return ChatCompletions(self._client)
# 定义 Message 类，表示一条对话消息
class Message:
# 消息类，表示聊天中的一条消息
"""消息类，表示聊天中的一条消息"""
# 初始化，data 为消息内容的字典
def __init__(self, data: Dict[str, Any]):
"""
初始化消息对象
参数:
data: 包含消息数据的字典
"""
# 设置消息角色，如 user、assistant、system
self.role = data.get("role") # 角色：user, assistant, system
# 设置消息正文内容
self.content = data.get("content") # 消息内容
# 定义 Choice 类，表示模型返回的一个回复选项
class Choice:
# 选择类，表示模型生成的一个回复选项
"""选择类，表示模型生成的一个回复选项"""
# 初始化，data 为该选项的字典
def __init__(self, data: Dict[str, Any]):
"""
初始化选择对象
参数:
data: 包含选择数据的字典
"""
# 获取选项索引
self.index = data.get("index") # 选择的索引
# 获取消息内容，生成 Message 实例
self.message = Message(data.get("message", {})) # 消息对象
# 获取完成原因
self.finish_reason = data.get("finish_reason") # 完成原因
+# 定义 DeltaMessage 类，用于流式输出中的增量消息
+class DeltaMessage:
+ # 增量消息类，表示流式响应中的消息增量
+ """增量消息类，表示流式响应中的消息增量"""
+
+ # 初始化，data 为增量消息的字典
+ def __init__(self, data: Dict[str, Any]):
+ """
+ 初始化增量消息对象
+
+ 参数:
+ data: 包含增量消息数据的字典
+ """
+ # 获取增量内容（通常是部分文本）
+ self.content = data.get("content") # 增量内容
+ # 获取角色（如果有）
+ self.role = data.get("role") # 角色
+
+
+# 定义 DeltaChoice 类，用于流式输出中的增量选择
+class DeltaChoice:
+ # 增量选择类，表示流式响应中的选择增量
+ """增量选择类，表示流式响应中的选择增量"""
+
+ # 初始化，data 为增量选择的字典
+ def __init__(self, data: Dict[str, Any]):
+ """
+ 初始化增量选择对象
+
+ 参数:
+ data: 包含增量选择数据的字典
+ """
+ # 获取选项索引
+ self.index = data.get("index") # 选择的索引
+ # 获取增量消息
+ self.delta = DeltaMessage(data.get("delta", {})) # 增量消息
+ # 获取完成原因（如果有）
+ self.finish_reason = data.get("finish_reason") # 完成原因
+
+
+# 定义 StreamChunk 类，表示流式响应中的一个数据块
+class StreamChunk:
+ # 流式数据块类，表示流式响应中的一个数据块
+ """流式数据块类，表示流式响应中的一个数据块"""
+
+ # 初始化，data 为数据块的字典
+ def __init__(self, data: Dict[str, Any]):
+ """
+ 初始化流式数据块对象
+
+ 参数:
+ data: 包含数据块数据的字典
+ """
+ # 获取数据块 ID
+ self.id = data.get("id") # 数据块 ID
+ # 获取对象类型
+ self.object = data.get("object") # 对象类型
+ # 获取创建时间戳
+ self.created = data.get("created") # 创建时间戳
+ # 获取模型名称
+ self.model = data.get("model") # 使用的模型
+
+ # 解析增量选择列表
+ choices_data = data.get("choices", [])
+ self.choices = [DeltaChoice(choice_data) for choice_data in choices_data]
+
+
# 定义 ChatCompletionResponse 类，解析返回的响应数据
class ChatCompletionResponse:
# 聊天完成响应类，包含模型的回复
"""聊天完成响应类，包含模型的回复"""
# 初始化，data 为 API 返回的数据
def __init__(self, data: Dict[str, Any]):
"""
初始化响应对象
参数:
data: API 返回的 JSON 数据
"""
# 响应 id
self.id = data.get("id") # 响应 ID
# 对象类型
self.object = data.get("object") # 对象类型
# 创建时间戳
self.created = data.get("created") # 创建时间戳
# 模型名称
self.model = data.get("model") # 使用的模型
# 解析 choices 字段，生成 Choice 实例列表
choices_data = data.get("choices", [])
self.choices = [Choice(choice_data) for choice_data in choices_data]
# 解析 usage 字段，记录 token 相关信息
usage_data = data.get("usage", {})
self.usage = {
"prompt_tokens": usage_data.get("prompt_tokens"),
"completion_tokens": usage_data.get("completion_tokens"),
"total_tokens": usage_data.get("total_tokens"),
}
# 重写 __repr__ 方法，返回响应对象的简要描述
def __repr__(self) -> str:
"""返回对象的字符串表示"""
return f"ChatCompletionResponse(id={self.id}, model={self.model}, choices={len(self.choices)})"
# 重写 __str__ 方法，调用 __repr__
def __str__(self) -> str:
"""返回对象的字符串表示"""
return self.__repr__()
+# 定义 Stream 类，处理流式响应
+class Stream:
+ # 流式响应类，用于处理流式输出
+ """流式响应类，用于处理流式输出"""
+
+ # 初始化，接收 HTTP 响应对象
+ def __init__(self, response: requests.Response):
+ """
+ 初始化流式响应对象
+
+ 参数:
+ response: requests 的响应对象（已启用流式模式）
+ """
+ # 保存响应对象
+ self.response = response
+
+ # 定义迭代器方法，使对象可迭代
+ def __iter__(self) -> Iterator[StreamChunk]:
+ """
+ 迭代器方法，逐个返回流式数据块
+
+ 返回:
+ Iterator[StreamChunk]: 流式数据块的迭代器
+ """
+ # 逐行读取响应内容（SSE 格式）
+ try:
+ for line in self.response.iter_lines(decode_unicode=True):
+ # 跳过空行
+ if not line.strip():
+ continue
+
+ # SSE 格式：每行以 "data: " 开头
+ if line.startswith("data: "):
+ # 提取 JSON 数据部分
+ json_str = line[6:] # 移除 "data: " 前缀
+
+ # 如果遇到 [DONE] 标记，表示流式输出结束
+ if json_str.strip() == "[DONE]":
+ break
+
+ try:
+ # 解析 JSON 数据
+ data = json.loads(json_str)
+ # 创建并返回 StreamChunk 对象
+ yield StreamChunk(data)
+ except json.JSONDecodeError:
+ # JSON 解析失败，跳过这一行
+ continue
+ finally:
+ # 确保关闭响应连接
+ self.response.close()
+
+ # 定义上下文管理器入口
+ def __enter__(self):
+ """上下文管理器入口"""
+ return self
+
+ # 定义上下文管理器出口
+ def __exit__(self, exc_type, exc_val, exc_tb):
+ """上下文管理器出口，确保关闭连接"""
+ self.response.close()
+
+
# 定义 OpenAI 客户端类
class OpenAI:
# OpenAI 客户端类，用于与 OpenAI API 交互
"""OpenAI 客户端类，用于与 OpenAI API 交互"""
# 初始化函数，设置 api 密钥、基础 url 和超时时间
def __init__(
self,
api_key: Optional[str] = None,
base_url: str = "https://api.openai.com/v1",
timeout: float = 60.0
):
"""
初始化 OpenAI 客户端
参数:
api_key: API 密钥。如果为 None，则从环境变量 OPENAI_API_KEY 读取
base_url: API 基础 URL，默认为官方 OpenAI API
timeout: 请求超时时间（秒），默认 60 秒
"""
# 获取 API 密钥，优先使用参数，否则从环境变量获取
self.api_key = api_key or os.getenv("OPENAI_API_KEY")
# 如果未获取到密钥，抛出异常
if not self.api_key:
raise ValueError(
"API 密钥未设置。请设置 api_key 参数或环境变量 OPENAI_API_KEY"
)
# 移除 base_url 末尾的斜杠，设置基础 URL
self.base_url = base_url.rstrip("/") # 移除末尾的斜杠
# 设置请求超时时间
self.timeout = timeout
# 定义 chat 属性，返回 ChatResource 实例
@property
def chat(self) -> ChatResource:
"""
获取聊天资源
返回:
ChatResource: 聊天资源实例，可以访问 chat.completions.create()
"""
# 返回 ChatResource 对象
return ChatResource(self)8. 错误处理 #
在使用 API 时，可能会遇到各种错误。良好的错误处理可以让程序更加健壮。
8.1 常见错误类型 #
# 导入必要的类和异常
from openai import OpenAI, OpenAIError, RateLimitError, APIError, APIConnectionError
# 创建客户端
client = OpenAI()
# 安全的 API 调用函数
def safe_chat(user_message):
"""安全地调用聊天 API，包含错误处理"""
try:
# 尝试调用 API
response = client.chat.completions.create(
model="gpt-4o",
messages=[
{"role": "user", "content": user_message}
],
max_tokens=100
)
# 如果成功，返回回复
return response.choices[0].message.content
except RateLimitError:
# 请求频率超限错误
# 通常是因为发送请求太快或达到使用限额
print("错误：请求频率超限")
print("请稍后重试，或检查你的 API 使用限额")
return None
except APIConnectionError:
# API 连接错误
# 通常是网络问题
print("错误：无法连接到 OpenAI API")
print("请检查网络连接")
return None
except APIError as e:
# 其他 API 错误
# 可能是参数错误、模型不存在等
print(f"API 错误：{e}")
return None
except OpenAIError as e:
# OpenAI 库的其他错误
print(f"OpenAI 错误：{e}")
return None
except Exception as e:
# 其他未预期的错误
print(f"未知错误：{e}")
return None
# 测试错误处理
print("测试 1：正常请求")
result = safe_chat("你好")
if result:
print(f"回复：{result}\n")
print("测试 2：无效的 API Key（会触发错误）")
# 注意：这里会触发错误，用于演示错误处理
# 实际使用时确保 API Key 正确8.2 重试机制 #
对于临时性错误（如网络问题），可以实现重试机制。
# 导入必要的库
from openai import OpenAI, APIConnectionError, RateLimitError
import time
# 创建客户端
client = OpenAI()
def chat_with_retry(user_message, max_retries=3):
"""带重试机制的聊天函数"""
for attempt in range(max_retries):
try:
# 尝试调用 API
response = client.chat.completions.create(
model="gpt-4o",
messages=[
{"role": "user", "content": user_message}
],
max_tokens=100
)
# 如果成功，返回回复
return response.choices[0].message.content
except (APIConnectionError, RateLimitError) as e:
# 对于连接错误和频率限制错误，可以重试
if attempt < max_retries - 1:
# 计算等待时间（指数退避）
wait_time = 2 ** attempt
print(f"请求失败，{wait_time} 秒后重试... (尝试 {attempt + 1}/{max_retries})")
time.sleep(wait_time)
else:
# 最后一次尝试也失败
print(f"重试 {max_retries} 次后仍然失败：{e}")
return None
except Exception as e:
# 其他错误不重试
print(f"错误：{e}")
return None
return None
# 测试重试机制
result = chat_with_retry("你好")
if result:
print(f"回复：{result}")9. 实际应用示例 #
本节提供几个实际应用场景的完整示例，帮助你理解如何在实际项目中使用 OpenAI 库。
9.1 示例 1：智能客服机器人 #
创建一个简单的客服机器人，可以记住对话历史。
# 导入 OpenAI 类
from openai import OpenAI
# 定义客服机器人类
class ChatBot:
"""简单的客服机器人"""
def __init__(self, api_key=None):
"""初始化机器人"""
# 创建 OpenAI 客户端
if api_key:
self.client = OpenAI(api_key=api_key)
else:
self.client = OpenAI()
# 存储对话历史
self.conversation_history = []
def set_system_message(self, content):
"""设置系统消息（定义机器人的角色和行为）"""
# 系统消息通常放在对话历史的最前面
self.conversation_history = [
{
"role": "system",
"content": content
}
]
def chat(self, user_input):
"""与用户对话"""
# 将用户消息添加到对话历史
self.conversation_history.append({
"role": "user",
"content": user_input
})
# 调用 API 获取回复
response = self.client.chat.completions.create(
model="gpt-4o",
messages=self.conversation_history, # 传入完整对话历史
max_tokens=500,
temperature=0.7
)
# 获取助手回复
assistant_reply = response.choices[0].message.content
# 将助手回复添加到对话历史
self.conversation_history.append({
"role": "assistant",
"content": assistant_reply
})
# 返回回复
return assistant_reply
def clear_history(self):
"""清空对话历史"""
# 保留系统消息（如果有）
if self.conversation_history and self.conversation_history[0]["role"] == "system":
self.conversation_history = [self.conversation_history[0]]
else:
self.conversation_history = []
# 使用示例
if __name__ == "__main__":
# 创建机器人实例
# 注意：需要设置环境变量 OPENAI_API_KEY，或传入 api_key 参数
bot = ChatBot()
# 设置系统消息（定义机器人角色）
bot.set_system_message("你是一个友好的客服助手，擅长回答产品相关问题。")
# 开始对话
print("=== 客服机器人 ===")
print("输入 'quit' 退出\n")
while True:
# 获取用户输入
user_input = input("用户：")
# 检查是否退出
if user_input.lower() == "quit":
print("再见！")
break
# 获取机器人回复
reply = bot.chat(user_input)
print(f"助手：{reply}\n")9.2 示例 2：文本总结工具 #
创建一个工具，可以自动总结长文本。
# 导入 OpenAI 类
from openai import OpenAI
# 创建客户端
client = OpenAI()
def summarize_text(text, max_length=200):
"""总结文本内容"""
# 构建提示词
prompt = f"请用简洁的语言总结以下文本，不超过 {max_length} 字：\n\n{text}"
# 调用 API
response = client.chat.completions.create(
model="gpt-4o",
messages=[
{
"role": "user",
"content": prompt
}
],
max_tokens=max_length,
temperature=0.3 # 较低的温度，使总结更准确
)
# 返回总结
return response.choices[0].message.content
# 测试文本总结
long_text = """
人工智能是计算机科学的一个分支，它试图理解智能的实质，并生产出一种新的能以人类智能相似的方式做出反应的智能机器。
机器学习是人工智能的一个子领域。它使计算机能够在没有明确编程的情况下学习和改进。机器学习算法通过分析大量数据来识别模式，并基于这些模式做出预测或决策。
深度学习是机器学习的一个子集。它使用神经网络来模拟人脑的工作方式。深度学习在图像识别、自然语言处理等领域取得了突破性进展。
"""
print("原文：")
print(long_text)
print("\n" + "="*50 + "\n")
# 生成总结
summary = summarize_text(long_text, max_length=150)
print("总结：")
print(summary)9.3 示例 3：批量处理文本 #
当需要处理大量文本时，可以批量处理以提高效率。
# 导入必要的库
from openai import OpenAI
import time
# 创建客户端
client = OpenAI()
def batch_process_texts(texts, task="总结"):
"""批量处理文本"""
results = []
# 遍历每个文本
for i, text in enumerate(texts, 1):
try:
print(f"处理第 {i}/{len(texts)} 个文本...")
# 构建提示词
if task == "总结":
prompt = f"请总结以下文本：\n\n{text}"
elif task == "翻译":
prompt = f"请将以下文本翻译成英文：\n\n{text}"
else:
prompt = f"请处理以下文本：\n\n{text}"
# 调用 API
response = client.chat.completions.create(
model="gpt-4o",
messages=[
{
"role": "user",
"content": prompt
}
],
max_tokens=200
)
# 获取结果
result = response.choices[0].message.content
# 保存结果
results.append({
"original": text,
"result": result,
"status": "success"
})
# 添加延迟，避免请求过快
time.sleep(1)
except Exception as e:
# 处理错误
print(f"处理第 {i} 个文本时出错：{e}")
results.append({
"original": text,
"error": str(e),
"status": "error"
})
return results
# 测试批量处理
texts = [
"Python 是一种高级编程语言，广泛用于数据科学和机器学习。",
"机器学习是人工智能的一个分支，让计算机能够从数据中学习。",
"深度学习使用神经网络来模拟人脑的工作方式。"
]
print("开始批量处理...\n")
results = batch_process_texts(texts, task="总结")
print("\n处理结果：")
for i, result in enumerate(results, 1):
print(f"\n文本 {i}:")
if result["status"] == "success":
print(f" 原文：{result['original']}")
print(f" 总结：{result['result']}")
else:
print(f" 错误：{result['error']}")10. 最佳实践 #
根据实际应用经验，以下是一些使用建议和最佳实践。
10.1 API Key 安全管理 #
推荐做法：
# 导入必要的模块
import os
from openai import OpenAI
# 方式 1：从环境变量读取（推荐）
def get_client_from_env():
"""从环境变量创建客户端"""
api_key = os.getenv("OPENAI_API_KEY")
if not api_key:
raise ValueError("未找到 OPENAI_API_KEY 环境变量，请先设置")
return OpenAI(api_key=api_key)
# 方式 2：从配置文件读取（不提交到代码仓库）
def get_client_from_config():
"""从配置文件创建客户端"""
try:
# 从本地配置文件读取（不要提交到 Git）
with open("config.txt", "r") as f:
api_key = f.read().strip()
return OpenAI(api_key=api_key)
except FileNotFoundError:
raise ValueError("未找到配置文件 config.txt")
# 使用示例
try:
client = get_client_from_env()
print("客户端创建成功（从环境变量）")
except ValueError as e:
print(f"错误：{e}")10.2 成本控制 #
OpenAI 按 token 数量计费，控制 token 使用可以降低成本。
# 导入必要的模块
from openai import OpenAI
# 创建客户端
client = OpenAI()
def cost_aware_chat(user_message, use_gpt4=False):
"""成本感知的聊天函数"""
# 根据需求选择模型
if use_gpt4:
# GPT-4 更强大但更贵
model = "gpt-4"
max_tokens = 500 # 限制输出长度
else:
# gpt-4o 性价比高
model = "gpt-4o"
max_tokens = 1000
# 调用 API
response = client.chat.completions.create(
model=model,
messages=[
{
"role": "user",
"content": user_message
}
],
max_tokens=max_tokens, # 限制输出长度
temperature=0.7
)
return response.choices[0].message.content
# 使用示例
# 简单问题使用 gpt-4o
simple_reply = cost_aware_chat("什么是 Python？", use_gpt4=False)
print("gpt-4o 回复：")
print(simple_reply)
# 复杂问题使用 GPT-4
# complex_reply = cost_aware_chat("请详细解释量子计算原理", use_gpt4=True)10.3 参数调优建议 #
不同场景需要不同的参数设置。
# 导入必要的模块
from openai import OpenAI
# 创建客户端
client = OpenAI()
# 不同场景的参数配置
configs = {
"创意写作": {
"temperature": 0.9, # 高温度，更随机，更有创意
"max_tokens": 1000
},
"代码生成": {
"temperature": 0.2, # 低温度，更确定，代码更准确
"max_tokens": 2000
},
"问答": {
"temperature": 0.7, # 中等温度，平衡准确性和多样性
"max_tokens": 500
},
"总结": {
"temperature": 0.3, # 低温度，总结更准确
"max_tokens": 300
}
}
def chat_with_config(user_message, scenario="问答"):
"""根据场景使用不同配置"""
config = configs.get(scenario, configs["问答"])
response = client.chat.completions.create(
model="gpt-4o",
messages=[
{
"role": "user",
"content": user_message
}
],
temperature=config["temperature"],
max_tokens=config["max_tokens"]
)
return response.choices[0].message.content
# 使用示例
print("场景：问答")
reply = chat_with_config("什么是机器学习？", scenario="问答")
print(reply)11. 常见问题解答 #
在使用 OpenAI 库的过程中，可能会遇到一些问题。本节列出常见问题及解决方法。
问题 1：API Key 错误 #
问题描述： 提示 "Invalid API Key" 或类似错误。
原因：
- API Key 不正确或已过期
- API Key 未正确设置
解决方案：
- 检查 API Key 是否正确（没有多余空格）
- 确认环境变量设置正确
- 在 OpenAI 平台检查 API Key 状态
问题 2：请求频率超限 #
问题描述： 提示 "Rate limit exceeded"。
原因：
- 发送请求太快
- 达到使用限额
解决方案：
- 添加延迟（如
time.sleep(1)） - 检查你的 API 使用限额
- 考虑升级账户
问题 3：Token 限制错误 #
问题描述： 提示 "maximum context length" 或类似错误。
原因：
- 输入文本太长，超过模型限制
- 对话历史太长
解决方案：
- 缩短输入文本
- 清空或缩短对话历史
- 使用更大的模型（如 GPT-4）
问题 4：网络连接错误 #
问题描述： 无法连接到 OpenAI API。
原因：
- 网络问题
- 防火墙阻止
解决方案：
- 检查网络连接
- 检查防火墙设置
- 如果在国内，可能需要使用代理
问题 5：模型不存在错误 #
问题描述： 提示 "model not found"。
原因：
- 模型名称拼写错误
- 模型不可用或已废弃
解决方案：
- 检查模型名称（如
gpt-4o） - 查看 OpenAI 文档确认可用模型
- 使用
client.models.list()查看可用模型