AI API 开发

在 web 界面使用 AI 聊天工具很方便，但要把 AI 能力真正融入你自己的产品、工作流、自动化脚本中，你需要的是 API。

API（Application Programming Interface）是应用程序之间通信的桥梁，通过 API，你的代码可以直接向 AI 模型发送请求并获取回复，不需要手动打开网页复制粘贴。

想象一下这些场景：

你的电商网站自动为每件商品生成描述文案。
你的笔记应用自动总结用户输入的长文。
你的客服系统自动分类和回复用户咨询。
这些功能都可以通过 AI API 实现。

本模块的目标是让你从"只会用 AI 聊天"变成"能让 AI 在你的代码里干活"。

API 基础概念

在写代码之前，先搞懂几个核心概念。

什么是 API

API 就是一套约定好的对话方式，我们按特定格式发送请求，对方按特定格式返回结果。

用点餐打个比方：

你走进餐厅，拿到菜单（API 文档），知道可以点什么、怎么点（请求格式）。
你对服务员说：一份宫保鸡丁，少辣（发送 API 请求）。
厨房做好菜给你端上来（返回 API 响应）。

在这个过程中，你不需要进厨房，不需要知道菜怎么做的——这就是 API 的价值：封装复杂细节，只暴露简单接口。

REST API 基本原理

REST（Representational State Transfer）是目前最流行的 API 设计风格。

简单来说，就是用 HTTP 协议的方式来通信：

HTTP 方法	用途	AI API 中的例子
POST	创建资源	发送聊天请求、生成图片
GET	获取资源	查询模型列表、获取账单信息
DELETE	删除资源	删除对话历史

AI API 绝大多数是 POST 请求，因为我们要"发送内容给 AI 处理"。

JSON 格式入门

JSON（JavaScript Object Notation）是 API 通信中最常用的数据格式。

JSON 的语法很简单，就是键值对的组合。

实例

{
"model": "gpt-4",
"messages": [
{
"role": "user",
"content": "你好，请介绍一下 RUNOOB"
}
],
"temperature": 0.7,
"max_tokens": 1000
}

这就是一个典型的 AI API 请求体。

几个核心规则：

对象用 { } 包裹，数组用 [ ] 包裹。
字符串用双引号 "，不能用单引号。
键值对用冒号 : 分隔，键必须是字符串。
多个键值对用逗号 , 分隔。

JSON 看起来和 Python 字典很像，但语法上有区别，写代码时要注意。

开发环境准备

工欲善其事，必先利其器。

Python 安装与配置

Python 是调用 AI API 最常用的语言，生态完善，库丰富。

先检查你电脑上有没有安装 Python：

# 检查 Python 版本（Windows、macOS、Linux 通用）
python --version
# 或者
python3 --version

推荐使用 Python 3.9 或更高版本。

如果还没安装，去 python.org 下载最新稳定版。

pip 包管理

pip 是 Python 的包管理器，用来安装第三方库。

# 检查 pip 版本
pip --version
# 或者
pip3 --version

# 升级 pip 到最新版本
pip install --upgrade pip

# 配置国内镜像源（可选，提升下载速度）
pip config set global.index-url https://pypi.tuna.tsinghua.edu.cn/simple

代码编辑器选择

写 AI API 代码，推荐以下编辑器之一：

编辑器	特点	适用人群
VS Code	免费、插件丰富、轻量	大多数开发者
Cursor	内置 AI 辅助、可直接生成代码	想让 AI 帮忙写代码
PyCharm	功能最全、Python 支持最好	专业 Python 开发者

本教程的示例代码在任何编辑器中都能运行。

OpenAI API 入门

OpenAI API 是最经典、生态最完善的 AI API，很多其他厂商也兼容它的格式。

注册与获取 API Key

API Key 是你的通行证，证明你的身份，记录你的用量，获取步骤：

1. 访问 platform.openai.com 注册账号
2. 进入 API Keys 页面
3. 点击 "Create new secret key"
4. 复制保存这个 Key（只显示一次！）

重要：永远不要把 API Key 提交到公开的代码仓库里。一旦泄露，别人可能用你的额度产生费用。

第一个 API 调用

先安装 OpenAI 的官方 SDK：

pip install openai

现在写第一个 API 调用程序：

实例

# 文件路径：first_api_call.py
# 第一个 OpenAI API 调用示例

from openai import OpenAI

# 初始化客户端
# 注意：API Key 建议从环境变量读取，不要硬编码在代码里
# 这里为了演示方便直接写了，实际项目请用环境变量
client = OpenAI(
api_key="your-api-key-here" # 替换成你的 API Key
)

# 发送聊天请求
response = client.chat.completions.create(
model="gpt-3.5-turbo", # 选择模型
messages=[
{
"role": "user", # 角色：user 表示用户
"content": "请用一句话介绍 RUNOOB 菜鸟教程" # 用户输入的内容
}
]
)

# 打印完整响应（看看返回了什么）
print("完整响应：")
print(response)
print("\n" + "="*50 + "\n")

# 提取 AI 的回复
ai_reply = response.choices[0].message.content
print("AI 回复：")
print(ai_reply)

如果没有 OpenAI 的 api_key ，也可以使用国内替代，国内很多模型都兼容 OpenAI，比如 DeepSeek，可以去 https://platform.deepseek.com/api_keys 申请 api_key 然后以下代码替换你的 api_key 和 model，DeepSeek 支持的模型为 deepseek-v4-flash、deepseek-v4-pro（思考模式）即可使用 API 调用大模型：

实例

from openai import OpenAI

# 初始化客户端
# 注意：API Key 建议从环境变量读取，不要硬编码在代码里
# 这里为了演示方便直接写了，实际项目请用环境变量
client = OpenAI(
api_key="sk-xxxx", # 设置 api_key
base_url="https://api.deepseek.com") # 设置请求地址

# 发送聊天请求
response = client.chat.completions.create(
model="deepseek-v4-flash", # 选择模型
messages=[
{
"role": "user", # 角色：user 表示用户
"content": "请用一句话介绍 RUNOOB 菜鸟教程" # 用户输入的内容
}
]
)

# 打印完整响应（看看返回了什么）
print("完整响应：")
print(response)
print("\n" + "="*50 + "\n")

# 提取 AI 的回复
ai_reply = response.choices[0].message.content
print("AI 回复：")
print(ai_reply)

运行这个程序：

python first_api_call.py

如果一切正常，你会看到类似这样的输出：

完整响应：
ChatCompletion(
  id='chatcmpl-xxx',
  choices=[Choice(finish_reason='stop', index=0, message=ChatMessage(content='RUNOOB 菜鸟教程是一个专注于提供编程技术教程的学习平台...', role='assistant'))],
  model='gpt-3.5-turbo',
  usage=CompletionUsage(completion_tokens=30, prompt_tokens=18, total_tokens=48)
)

==================================================

AI 回复：
RUNOOB 菜鸟教程是一个专注于提供编程技术教程的学习平台，涵盖多种编程语言和技术领域。

恭喜！你已经成功调用了一次 AI API。

请求参数详解

OpenAI API 有很多参数可以调整，下面是最常用的几个：

参数名	类型	是否必填	说明	默认值
model	string	是	使用的模型 ID，如 gpt-3.5-turbo、gpt-4	无
messages	array	是	对话历史消息列表	无
temperature	number	否	采样温度，0-2 之间，越高越随机，越低越确定	1.0
max_tokens	integer	否	生成的最大 token 数	无限制
top_p	number	否	核采样参数，0-1 之间	1.0
stop	string/array	否	停止序列，遇到这些字符就停止生成	null

重点说一下 temperature，这是最常用的调节参数：

实例

# 文件路径：temperature_demo.py
# 演示 temperature 参数对输出的影响

from openai import OpenAI

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

def ask_ai(temperature_value: float, question: str) -> str:
"""指定 temperature 提问"""
response = client.chat.completions.create(
model="gpt-3.5-turbo",
messages=[{"role": "user", "content": question}],
temperature=temperature_value,
max_tokens=200
)
return response.choices[0].message.content

# 同一个问题，用不同的 temperature 问三次
question = "给 RUNOOB 起一个 slogan"

print("--- temperature=0（最确定，每次结果都差不多）---")
print(ask_ai(0.0, question))
print("\n--- temperature=0.7（平衡，推荐日常使用）---")
print(ask_ai(0.7, question))
print("\n--- temperature=1.8（最随机，每次结果差异大）---")
print(ask_ai(1.8, question))

一个经验法则：

temperature=0：事实回答、代码生成、需要精确结果的场景
temperature=0.7：日常对话、通用任务
temperature>1.0：创意写作、头脑风暴

Anthropic Claude API

Claude 是 Anthropic 公司的大模型，以安全性和长文本处理能力著称。

与 OpenAI API 的异同

两者都能做文本生成，但设计理念有些区别：

对比项	OpenAI API	Claude API
消息结构	system、user、assistant 交替	system 单独设，user、assistant 交替
上下文长度	16k-128k 不等	Claude 3 系列支持 200k
参数设计	temperature、top_p 等	temperature、top_p 等，概念类似

Claude Messages API 结构

先安装 Claude SDK：

pip install anthropic

Claude API 的基本调用：

实例

# 文件路径：claude_first_call.py
# Claude API 基础调用示例

import anthropic

# 初始化客户端
client = anthropic.Anthropic(
api_key="your-api-key-here" # 替换成你的 Claude API Key
)

# 发送消息请求
response = client.messages.create(
model="claude-3-sonnet-20240229", # 选择 Claude 模型
max_tokens=1024, # Claude 要求必须设置 max_tokens
messages=[
{
"role": "user",
"content": "请介绍一下 RUNOOB 菜鸟教程，用三句话"
}
]
)

# 打印响应
print("完整响应：")
print(response)
print("\n" + "="*50 + "\n")

# 提取回复
print("Claude 回复：")
print(response.content[0].text)

同样的我们可以替换为国内兼容的大模型，比如 DeepSeek，指定 api_key 、 base_url、model 这三个参数：

实例

import anthropic

# 初始化客户端
client = anthropic.Anthropic(
api_key="sk-xxx", # 替换成你的 API Key
base_url="https://api.deepseek.com/anthropic" # Claude API 的基础 URL
)

# 发送消息请求
response = client.messages.create(
model="deepseek-v4-flash", # 选择 Claude 模型
max_tokens=1024, # Claude 要求必须设置 max_tokens
messages=[
{
"role": "user",
"content": "请介绍一下 RUNOOB 菜鸟教程，用三句话"
}
]
)

# 打印响应
print("完整响应：")
print(response)
print("\n" + "="*50 + "\n")

# 提取回复
print("Claude 回复：")
print(response.content[0].thinking)

System Prompt 设置

Claude API 中，System Prompt 是一个独立的参数，不是放在 messages 数组里：

实例

# 文件路径：claude_system_prompt.py
# Claude System Prompt 使用示例

import anthropic

client = anthropic.Anthropic(api_key="your-api-key-here")

response = client.messages.create(
model="claude-3-sonnet-20240229",
max_tokens=1024,
# System Prompt 在这里设置
system="你是一个专业的技术文档编辑，回答要简洁准确，多用列表形式。",
messages=[
{
"role": "user",
"content": "学习 Python 应该掌握哪些核心知识点？"
}
]
)

print(response.content[0].text)

多轮对话管理

单轮对话简单，但真正有用的应用都需要"记住上下文"。

对话历史的维护方式

思路很简单：把之前的对话都存下来，每次请求一起发给 AI。

实例

# 文件路径：multi_turn_chat.py
# 多轮对话管理示例

from openai import OpenAI

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

class ChatBot:
"""一个简单的多轮对话机器人"""

def __init__(self, system_prompt: str = "你是一个乐于助人的助手"):
self.messages = [] # 用列表保存对话历史
# 先添加 system prompt
self.messages.append({
"role": "system",
"content": system_prompt
})

def chat(self, user_input: str) -> str:
"""发送消息并获取回复"""
# 添加用户消息到历史
self.messages.append({
"role": "user",
"content": user_input
})

# 调用 API
response = client.chat.completions.create(
model="gpt-3.5-turbo",
messages=self.messages
)

# 获取 AI 回复
ai_reply = response.choices[0].message.content

# 把 AI 回复也添加到历史
self.messages.append({
"role": "assistant",
"content": ai_reply
})

return ai_reply

def get_history(self) -> list:
"""获取对话历史"""
return self.messages

# 使用示例
if __name__ == "__main__":
bot = ChatBot(system_prompt="你是一个编程助手，专门回答关于 RUNOOB 和 Python 的问题")

print("=== 多轮对话演示 ===")

# 第一轮
r1 = bot.chat("什么是 Python？")
print("用户：什么是 Python？")
print("AI：", r1)
print()

# 第二轮（AI 能记住上一轮的话题）
r2 = bot.chat("它有什么特点？")
print("用户：它有什么特点？")
print("AI：", r2)
print()

# 第三轮
r3 = bot.chat("RUNOOB 上有 Python 教程吗？")
print("用户：RUNOOB 上有 Python 教程吗？")
print("AI：", r3)

关键点：对话历史是一个有序列表，严格按照 user-assistant-user-assistant 交替，每次请求都要把完整历史发过去，AI 才能理解上下文，历史会占用 token，所以需要定期清理。

上下文窗口超限处理

每个模型都有上下文窗口限制，比如 gpt-3.5-turbo 是 16k，gpt-4 是 8k 或 32k。

当历史对话太长时，会触发错误。

几种常见的处理策略：

实例

# 文件路径：context_management.py
# 上下文窗口管理策略示例

from openai import OpenAI

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

class ChatBotWithContextLimit:
"""带上下文限制的对话机器人"""

def __init__(self, system_prompt: str, max_messages: int = 10):
self.system_prompt = system_prompt
self.max_messages = max_messages # 最多保留多少条消息
self.messages = []

def _trim_messages(self):
"""裁剪消息列表，保留最近的消息"""
# 始终保留 system prompt，然后只保留最近的 max_messages 条
if len(self.messages) > self.max_messages:
# 保留最近的 N 条
self.messages = self.messages[-self.max_messages:]

def chat(self, user_input: str) -> str:
# 添加用户消息
self.messages.append({
"role": "user",
"content": user_input
})

# 裁剪历史
self._trim_messages()

# 构造完整请求（system prompt + 历史）
full_messages = [
{"role": "system", "content": self.system_prompt}
] + self.messages

try:
response = client.chat.completions.create(
model="gpt-3.5-turbo",
messages=full_messages
)
ai_reply = response.choices[0].message.content

self.messages.append({
"role": "assistant",
"content": ai_reply
})

return ai_reply

except Exception as e:
# 如果还是报错，可能需要更激进的裁剪
return f"出错了：{str(e)}"

# 策略二：用 AI 总结历史对话
def summarize_conversation(messages: list) -> str:
"""用 AI 总结历史对话，替代原历史"""
history_text = "\n".join([
f"{m['role']}: {m['content']}"
for m in messages
])

prompt = f"""请把以下对话总结成一段简洁的摘要，保留关键信息：

{history_text}

摘要："""

response = client.chat.completions.create(
model="gpt-3.5-turbo",
messages=[{"role": "user", "content": prompt}],
temperature=0
)

return response.choices[0].message.content

# 使用总结策略
print("--- 对话历史总结示例 ---")
sample_history = [
{"role": "user", "content": "我叫什么？"},
{"role": "assistant", "content": "抱歉，我不知道你叫什么，你可以告诉我。"},
{"role": "user", "content": "我叫小明，我在 RUNOOB 学习 Python。"},
{"role": "assistant", "content": "你好小明！很高兴认识你。RUNOOB 是个很好的学习平台。"},
{"role": "user", "content": "我想学习列表，能教教我吗？"},
{"role": "assistant", "content": "当然可以！Python 列表是..."},
]

summary = summarize_conversation(sample_history)
print("对话摘要：", summary)

总结一下上下文管理的方法：

策略	优点	缺点	适用场景
只保留最近 N 条	简单、快速	可能丢失重要信息	短对话、闲聊场景
用 AI 总结历史	保留语义信息	需要额外调用 API	长对话、重要信息多
滑动窗口	平衡简单和完整	实现稍复杂	一般场景通用

流式输出

在网页上用 AI 聊天时，你会看到文字一个字一个字地蹦出来——这就是流式输出。

什么是流式输出

普通模式：AI 生成完整回复后一次性返回，你可能要等几秒甚至几十秒。
流式模式：AI 生成一个 token 就发一个，用户体验更好。

技术上，流式输出使用 SSE（Server-Sent Events）协议，服务端持续推送数据。

实现打字机效果

实例

# 文件路径：streaming_demo.py
# 流式输出演示（打字机效果）

from openai import OpenAI

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

print("=== 流式输出演示 ===")
print("问：请写一段关于 RUNOOB 的介绍\n")
print("答：", end="", flush=True)

# 关键参数：stream=True
response = client.chat.completions.create(
model="gpt-3.5-turbo",
messages=[
{"role": "user", "content": "请写一段关于 RUNOOB 菜鸟教程的介绍"}
],
stream=True # 启用流式输出
)

collected_reply = []

for chunk in response:
# 提取当前 chunk 的内容
if chunk.choices[0].delta.content:
content = chunk.choices[0].delta.content
print(content, end="", flush=True) # 实时打印，不换行
collected_reply.append(content)

print() # 最后输出一个换行
print("\n" + "="*50)
print("完整回复：", "".join(collected_reply))

运行这个程序，你会看到文字一个字一个字地出现。

Claude 的流式输出用法类似：

实例

# 文件路径：claude_streaming.py
# Claude 流式输出示例

import anthropic

client = anthropic.Anthropic(api_key="your-api-key-here")

print("=== Claude 流式输出 ===")
print("问：如何学习编程？\n")
print("答：", end="", flush=True)

with client.messages.stream(
model="claude-3-sonnet-20240229",
max_tokens=1024,
messages=[{"role": "user", "content": "如何系统地学习编程？"}]
) as stream:
for text in stream.text_stream:
print(text, end="", flush=True)

print()

用户体验建议：只要不是要求"必须等完整结果"的场景，优先使用流式输出。让用户看到进度，能显著提升体验。

API 费用计算与优化

AI API 是按使用量计费的，用得越多，花得越多。

Token 计费原理

AI API 不是按字数或对话次数计费，而是按 Token，Token 是 AI 处理文本的基本单位。

一个 token 大约等于：

英文：0.75 个单词（或 4 个字母）
中文：1-2 个汉字

举个例子：

"Hello, 世界" — 大约 3-4 个 tokens
"RUNOOB 菜鸟教程很棒" — 大约 6-8 个 tokens

计费方式：

Prompt tokens（你发给 AI 的） + Completion tokens（AI 返回给你的）= 总 tokens

不同模型价格不同，比如 gpt-3.5-turbo 是 $0.0015/1k prompt tokens，$0.002/1k completion tokens。

如何控制 API 成本

实例

# 文件路径：cost_tracking.py
# API 费用追踪与控制示例

from openai import OpenAI

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

# 模型价格表（示例，实际价格请查官方文档）
# 单位：美元 / 1k tokens
MODEL_PRICES = {
"gpt-3.5-turbo": {
"prompt": 0.0015,
"completion": 0.002
},
"gpt-4": {
"prompt": 0.03,
"completion": 0.06
}
}

def chat_with_cost_tracking(model: str, messages: list):
"""带费用统计的聊天"""
response = client.chat.completions.create(
model=model,
messages=messages
)

# 从响应中获取 token 使用量
usage = response.usage
prompt_tokens = usage.prompt_tokens
completion_tokens = usage.completion_tokens
total_tokens = usage.total_tokens

# 计算费用
price = MODEL_PRICES.get(model, {"prompt": 0, "completion": 0})
prompt_cost = (prompt_tokens / 1000) * price["prompt"]
completion_cost = (completion_tokens / 1000) * price["completion"]
total_cost = prompt_cost + completion_cost

return {
"reply": response.choices[0].message.content,
"usage": {
"prompt_tokens": prompt_tokens,
"completion_tokens": completion_tokens,
"total_tokens": total_tokens
},
"cost": {
"prompt_cost": prompt_cost,
"completion_cost": completion_cost,
"total_cost": total_cost
}
}

# 使用示例
if __name__ == "__main__":
result = chat_with_cost_tracking(
model="gpt-3.5-turbo",
messages=[{"role": "user", "content": "介绍一下 RUNOOB"}]
)

print("AI 回复：", result["reply"])
print()
print("使用统计：")
print(f" Prompt tokens: {result['usage']['prompt_tokens']}")
print(f" Completion tokens: {result['usage']['completion_tokens']}")
print(f" Total tokens: {result['usage']['total_tokens']}")
print()
print("费用统计（美元）：")
print(f" Prompt cost: ${result['cost']['prompt_cost']:.6f}")
print(f" Completion cost: ${result['cost']['completion_cost']:.6f}")
print(f" Total cost: ${result['cost']['total_cost']:.6f}")

成本优化的几个实用技巧：

技巧	怎么做	节省比例
选择合适的模型	简单任务用 gpt-3.5-turbo，不用 gpt-4	90%+
限制 max_tokens	设合理的上限，避免 AI 说太多	视情况
裁剪上下文	只保留必要的历史对话	50%+
提示要简洁	system prompt 不要写得太长	视情况
结果缓存	相同问题直接返回缓存结果	视情况

缓存策略

对于重复的问题，缓存结果能省很多钱：

实例

# 文件路径：response_cache.py
# 简单的响应缓存实现

import hashlib
import json
from openai import OpenAI

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

class SimpleCache:
"""简单的内存缓存"""

def __init__(self):
self.cache = {}

def _make_key(self, model: str, messages: list) -> str:
"""根据请求内容生成缓存键"""
# 把关键信息序列化后取 hash
key_data = {
"model": model,
"messages": messages
}
key_str = json.dumps(key_data, sort_keys=True)
return hashlib.md5(key_str.encode()).hexdigest()

def get(self, model: str, messages: list):
"""获取缓存"""
key = self._make_key(model, messages)
return self.cache.get(key)

def set(self, model: str, messages: list, value: str):
"""设置缓存"""
key = self._make_key(model, messages)
self.cache[key] = value

# 带缓存的聊天函数
cache = SimpleCache()

def chat_with_cache(model: str, messages: list):
# 先查缓存
cached = cache.get(model, messages)
if cached:
print("（命中缓存，直接返回）")
return cached

# 缓存未命中，调用 API
response = client.chat.completions.create(
model=model,
messages=messages
)
reply = response.choices[0].message.content

# 存入缓存
cache.set(model, messages, reply)

return reply

# 测试
if __name__ == "__main__":
messages = [{"role": "user", "content": "RUNOOB 是什么网站？"}]

print("第一次调用...")
r1 = chat_with_cache("gpt-3.5-turbo", messages)
print(r1)
print()

print("第二次调用（相同问题）...")
r2 = chat_with_cache("gpt-3.5-turbo", messages)
print(r2)

错误处理与重试机制

网络请求总会出错，API 也不例外。健壮的代码必须处理各种异常情况。

常见错误类型

错误类型	HTTP 状态码	原因	处理方式
认证失败	401	API Key 错误或过期	检查 API Key
额度不足	402	账户没钱了	充值或换号
超限	429	请求太快或额度用完	等一会儿再试
服务器错误	5xx	AI 服务端出问题	重试

指数退避重试

遇到临时性错误（如 429、5xx），最常用的策略是指数退避重试：

第一次失败 → 等 1 秒 → 重试
第二次失败 → 等 2 秒 → 重试
第三次失败 → 等 4 秒 → 重试
第四次失败 → 等 8 秒 → 重试

每次等待时间翻倍，给服务恢复的机会。

实例

# 文件路径：error_handling.py
# 错误处理与指数退避重试

import time
from openai import OpenAI, APIError, RateLimitError, APIConnectionError

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

def chat_with_retry(
model: str,
messages: list,
max_retries: int = 3,
initial_wait: float = 1.0
):
"""带重试机制的聊天函数"""

retries = 0

while retries <= max_retries:
try:
response = client.chat.completions.create(
model=model,
messages=messages
)
return {
"success": True,
"reply": response.choices[0].message.content
}

except RateLimitError as e:
# 限速错误：等一会儿再试
retries += 1
if retries > max_retries:
return {
"success": False,
"error": f"重试次数用尽：{str(e)}"
}
wait_time = initial_wait * (2 ** (retries - 1))
print(f"触发限速，等待 {wait_time} 秒后重试...")
time.sleep(wait_time)

except APIConnectionError as e:
# 网络连接错误：重试
retries += 1
if retries > max_retries:
return {
"success": False,
"error": f"网络连接失败：{str(e)}"
}
wait_time = initial_wait * (2 ** (retries - 1))
print(f"网络错误，等待 {wait_time} 秒后重试...")
time.sleep(wait_time)

except APIError as e:
# API 错误：视情况重试
if e.status_code and 500 <= e.status_code < 600:
retries += 1
if retries > max_retries:
return {
"success": False,
"error": f"服务器错误：{str(e)}"
}
wait_time = initial_wait * (2 ** (retries - 1))
print(f"服务器错误，等待 {wait_time} 秒后重试...")
time.sleep(wait_time)
else:
# 非 5xx 错误，不重试
return {
"success": False,
"error": f"API 错误：{str(e)}"
}

except Exception as e:
# 其他错误，不重试
return {
"success": False,
"error": f"未知错误：{str(e)}"
}

# 使用示例
if __name__ == "__main__":
result = chat_with_retry(
model="gpt-3.5-turbo",
messages=[{"role": "user", "content": "你好，RUNOOB"}]
)

if result["success"]:
print("成功：", result["reply"])
else:
print("失败：", result["error"])

也可以用现成的库来简化重试逻辑，比如 tenacity，使用以下命令安装：

pip install tenacity

实例

# 文件路径：tenacity_demo.py
# 使用 tenacity 库简化重试逻辑

from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
from openai import OpenAI, RateLimitError, APIError

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

@retry(
stop=stop_after_attempt(3), # 最多重试 3 次
wait=wait_exponential(multiplier=1, min=1, max=10), # 指数退避：1, 2, 4... 最多 10 秒
retry=retry_if_exception_type((RateLimitError, APIError)) # 只对特定异常重试
)
def chat_with_tenacity(model: str, messages: list):
"""用 tenacity 装饰器实现重试"""
response = client.chat.completions.create(
model=model,
messages=messages
)
return response.choices[0].message.content

# 使用
try:
reply = chat_with_tenacity(
"gpt-3.5-turbo",
[{"role": "user", "content": "介绍一下 RUNOOB"}]
)
print(reply)
except Exception as e:
print(f"最终失败：{e}")

实战项目：命令行 AI 聊天工具

现在把前面学的知识整合起来，做一个完整的命令行聊天工具。

实例

# 文件路径：runoob_chatbot.py
# RUNOOB AI 聊天机器人 - 完整命令行工具

import sys
import json
import os
from datetime import datetime
from openai import OpenAI

class RunoobChatBot:
"""一个功能完整的命令行聊天机器人"""

def __init__(self, config_file: str = "config.json"):
# 加载配置
self.config = self._load_config(config_file)
self.client = OpenAI(api_key=self.config["api_key"])

# 对话状态
self.messages = []
self.total_tokens = 0
self.total_cost = 0.0

# 初始化 system prompt
self._init_system_prompt()

def _load_config(self, config_file: str) -> dict:
"""加载配置文件"""
default_config = {
"api_key": "your-api-key-here",
"model": "gpt-3.5-turbo",
"system_prompt": "你是一个乐于助人的 AI 助手，由 RUNOOB 提供技术支持。",
"max_history": 20,
"stream": True,
"temperature": 0.7
}

if os.path.exists(config_file):
with open(config_file, "r", encoding="utf-8") as f:
user_config = json.load(f)
default_config.update(user_config)

return default_config

def _save_config(self, config_file: str = "config.json"):
"""保存配置文件"""
with open(config_file, "w", encoding="utf-8") as f:
json.dump(self.config, f, indent=2, ensure_ascii=False)

def _init_system_prompt(self):
"""初始化 system prompt"""
self.messages = [
{
"role": "system",
"content": self.config["system_prompt"]
}
]

def _trim_history(self):
"""裁剪对话历史"""
# 保留 system prompt + 最近的 N 条对话
max_messages = self.config["max_history"]
if len(self.messages) > max_messages + 1:
self.messages = [self.messages[0]] + self.messages[-max_messages:]

def chat(self, user_input: str) -> str:
"""发送消息并获取回复"""
# 添加用户消息
self.messages.append({
"role": "user",
"content": user_input
})

# 裁剪历史
self._trim_history()

if self.config["stream"]:
return self._chat_stream()
else:
return self._chat_normal()

def _chat_normal(self) -> str:
"""普通模式（非流式）"""
response = self.client.chat.completions.create(
model=self.config["model"],
messages=self.messages,
temperature=self.config["temperature"]
)

ai_reply = response.choices[0].message.content

# 统计用量
if hasattr(response, "usage"):
self.total_tokens += response.usage.total_tokens
# 简单估算费用（gpt-3.5-turbo 价格）
self.total_cost += (response.usage.total_tokens / 1000) * 0.002

self.messages.append({
"role": "assistant",
"content": ai_reply
})

return ai_reply

def _chat_stream(self) -> str:
"""流式模式"""
response = self.client.chat.completions.create(
model=self.config["model"],
messages=self.messages,
temperature=self.config["temperature"],
stream=True
)

collected_chunks = []
print("AI：", end="", flush=True)

for chunk in response:
if chunk.choices[0].delta.content:
content = chunk.choices[0].delta.content
print(content, end="", flush=True)
collected_chunks.append(content)

print()

ai_reply = "".join(collected_chunks)

self.messages.append({
"role": "assistant",
"content": ai_reply
})

# 流式模式的 token 统计需要额外处理，这里简化了
# 实际项目可以调用 usage 接口或估算
self.total_tokens += len(ai_reply) // 2 # 粗略估算

return ai_reply

def save_conversation(self, filename: str = None):
"""保存对话历史"""
if not filename:
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
filename = f"conversation_{timestamp}.json"

data = {
"model": self.config["model"],
"total_tokens": self.total_tokens,
"estimated_cost": self.total_cost,
"messages": self.messages,
"saved_at": datetime.now().isoformat()
}

with open(filename, "w", encoding="utf-8") as f:
json.dump(data, f, indent=2, ensure_ascii=False)

return filename

def clear_history(self):
"""清空对话历史"""
self._init_system_prompt()
self.total_tokens = 0
self.total_cost = 0.0

def show_stats(self):
"""显示统计信息"""
print("\n" + "="*40)
print("对话统计")
print("="*40)
print(f"模型：{self.config['model']}")
print(f"消息数：{len(self.messages) - 1}")
print(f"总 tokens：{self.total_tokens}")
print(f"估算费用：${self.total_cost:.4f}")
print("="*40 + "\n")

def print_help():
"""打印帮助信息"""
print("\n" + "="*40)
print("RUNOOB AI 聊天工具 - 命令列表")
print("="*40)
print("/help - 显示帮助")
print("/clear - 清空对话历史")
print("/stats - 显示统计信息")
print("/save - 保存对话历史")
print("/model - 切换模型")
print("/temp - 调整 temperature")
print("/system - 修改系统提示")
print("/quit - 退出程序")
print("="*40 + "\n")

def main():
print("="*40)
print(" RUNOOB AI 聊天工具")
print("="*40)
print("输入 /help 查看命令，直接输入文字开始聊天\n")

bot = RunoobChatBot()

while True:
try:
user_input = input("你：").strip()

if not user_input:
continue

# 处理命令
if user_input.startswith("/"):
cmd = user_input.lower()

if cmd in ["/quit", "/exit", "/q"]:
print("再见！")
break

elif cmd == "/help":
print_help()

elif cmd == "/clear":
bot.clear_history()
print("对话历史已清空")

elif cmd == "/stats":
bot.show_stats()

elif cmd == "/save":
filename = bot.save_conversation()
print(f"对话已保存到：{filename}")

elif cmd.startswith("/model "):
model = user_input[7:].strip()
bot.config["model"] = model
bot._save_config()
print(f"模型已切换为：{model}")

elif cmd.startswith("/temp "):
try:
temp = float(user_input[6:].strip())
if 0 <= temp <= 2:
bot.config["temperature"] = temp
bot._save_config()
print(f"Temperature 已设置为：{temp}")
else:
print("Temperature 必须在 0-2 之间")
except ValueError:
print("请输入有效的数字")

elif cmd.startswith("/system "):
system_prompt = user_input[8:].strip()
bot.config["system_prompt"] = system_prompt
bot.clear_history()
bot._save_config()
print("System prompt 已更新，对话已重置")

else:
print("未知命令，输入 /help 查看帮助")

else:
# 普通聊天
if not bot.config["stream"]:
print("AI：", end="")

bot.chat(user_input)
print()

except KeyboardInterrupt:
print("\n输入 /quit 退出，或继续聊天")

except Exception as e:
print(f"出错了：{e}")

if __name__ == "__main__":
main()

创建一个配置文件：

{
    "api_key": "your-api-key-here",
    "model": "gpt-3.5-turbo",
    "system_prompt": "你是一个乐于助人的 AI 助手，由 RUNOOB 提供技术支持。",
    "max_history": 20,
    "stream": true,
    "temperature": 0.7
}

现在可以运行这个聊天工具了：

python runoob_chatbot.py

你会看到：

========================================
  RUNOOB AI 聊天工具
========================================
输入 /help 查看命令，直接输入文字开始聊天

你：你好
AI：你好！很高兴见到你。我是由 RUNOOB 提供技术支持的 AI 助手，有什么可以帮你的吗？

你：/stats

========================================
对话统计
========================================
模型：gpt-3.5-turbo
消息数：2
总 tokens：50
估算费用：$0.0001
========================================

你：/quit
再见！

返回顶部