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

API(Application Programming Interface)是应用程序之间通信的桥梁,通过 API,你的代码可以直接向 AI 模型发送请求并获取回复,不需要手动打开网页复制粘贴。
想象一下这些场景:
你的电商网站自动为每件商品生成描述文案。
你的笔记应用自动总结用户输入的长文。
你的客服系统自动分类和回复用户咨询。
这些功能都可以通过 AI API 实现。
本模块的目标是让你从"只会用 AI 聊天"变成"能让 AI 在你的代码里干活"。
在写代码之前,先搞懂几个核心概念。
API 就是一套约定好的对话方式,我们按特定格式发送请求,对方按特定格式返回结果。
用点餐打个比方:
你走进餐厅,拿到菜单(API 文档),知道可以点什么、怎么点(请求格式)。
你对服务员说:一份宫保鸡丁,少辣(发送 API 请求)。
厨房做好菜给你端上来(返回 API 响应)。
在这个过程中,你不需要进厨房,不需要知道菜怎么做的——这就是 API 的价值: 封装复杂细节,只暴露简单接口 。
REST(Representational State Transfer)是目前最流行的 API 设计风格。
概括来看,就是用 HTTP 协议的方式来通信:
| HTTP 方法 | 用途 | AI API 中的例子 |
|---|---|---|
| POST | 创建资源 | 发送聊天请求、生成图片 |
| GET | 获取资源 | 查询模型列表、获取账单信息 |
| DELETE | 删除资源 | 删除对话历史 |
AI API 绝大多数是 POST 请求,因为我们要"发送内容给 AI 处理"。
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 是调用 AI API 最常用的语言,生态完善,库丰富。
先检查你电脑上有没有安装 Python:
# 检查 Python 版本(Windows、macOS、Linux 通用)python --version# 或者python3 --version
推荐使用 Python 3.9 或更高版本。
如果还没安装,去 python.org 下载最新稳定版。
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 是最经典、生态最完善的 AI API,很多其他厂商也兼容它的格式。
API Key 是你的通行证,证明你的身份,记录你的用量,获取步骤:
1. 访问 platform.openai.com 注册账号
2. 进入 API Keys 页面
3. 点击 "Create new secret key"
4. 复制保存这个 Key(只显示一次!)
重要: 永远不要把 API Key 提交到公开的代码仓库里 。一旦泄露,别人可能用你的额度产生费用。
先安装 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 调用大模型:
# 初始化客户端
# 注意: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:创意写作、头脑风暴
Claude 是 Anthropic 公司的大模型,以安全性和长文本处理能力著称。
两者都能做文本生成,但设计理念有些区别:
| 对比项 | OpenAI API | Claude API |
|---|---|---|
| 消息结构 | system、user、assistant 交替 | system 单独设,user、assistant 交替 |
| 上下文长度 | 16k-128k 不等 | Claude 3 系列支持 200k |
| 参数设计 | temperature、top_p 等 | temperature、top_p 等,概念类似 |
先安装 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 这三个参数:
# 初始化客户端
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)
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:
"""获取对话历史"""
returnself.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 条
iflen(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
exceptExceptionas 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 =[]
forchunkin response:
# 提取当前 chunk 的内容
ifchunk.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()
用户体验建议:只要不是要求"必须等完整结果"的场景,优先使用流式输出。让用户看到进度,能显著提升体验。
AI API 是按使用量计费的,用得越多,花得越多。
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。
# 文件路径: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)
returnself.cache.get(key)
defset(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# 错误处理与指数退避重试
importtime
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_codeand500<= 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)}"
}
exceptExceptionas 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)
exceptExceptionas e:
print(f"最终失败:{e}")
现在把前面学的知识整合起来,做一个完整的命令行聊天工具。
# 文件路径:runoob_chatbot.py# RUNOOB AI 聊天机器人 - 完整命令行工具
importsys
import json
importos
fromdatetimeimportdatetime
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
}
ifos.path.exists(config_file):
withopen(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"):
"""保存配置文件"""
withopen(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"]
iflen(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()
ifself.config["stream"]:
returnself._chat_stream()
else:
returnself._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
# 统计用量
ifhasattr(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)
forchunkin response:
ifchunk.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):
"""保存对话历史"""
ifnot 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()
}
withopen(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()
whileTrue:
try:
user_input =input("你:").strip()
ifnot user_input:
continue
# 处理命令
if user_input.startswith("/"):
cmd= user_input.lower()
ifcmdin["/quit","/exit","/q"]:
print("再见!")
break
elifcmd=="/help":
print_help()
elifcmd=="/clear":
bot.clear_history()
print("对话历史已清空")
elifcmd=="/stats":
bot.show_stats()
elifcmd=="/save":
filename = bot.save_conversation()
print(f"对话已保存到:{filename}")
elifcmd.startswith("/model "):
model = user_input[7:].strip()
bot.config["model"]= model
bot._save_config()
print(f"模型已切换为:{model}")
elifcmd.startswith("/temp "):
try:
temp =float(user_input[6:].strip())
if0<= temp <=2:
bot.config["temperature"]= temp
bot._save_config()
print(f"Temperature 已设置为:{temp}")
else:
print("Temperature 必须在 0-2 之间")
exceptValueError:
print("请输入有效的数字")
elifcmd.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:
# 普通聊天
ifnot bot.config["stream"]:
print("AI:", end="")
bot.chat(user_input)
print()
exceptKeyboardInterrupt:
print("n输入 /quit 退出,或继续聊天")
exceptExceptionas 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再见!