首页
看点啥
插画图片
首页 看点啥 AI API 开发

AI API 开发

2026-08-07 0

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

AI API 开发

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

想象一下这些场景:

  1. 你的电商网站自动为每件商品生成描述文案。

  2. 你的笔记应用自动总结用户输入的长文。

  3. 你的客服系统自动分类和回复用户咨询。

  4. 这些功能都可以通过 AI API 实现。

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

API 基础概念

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

什么是 API

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

用点餐打个比方:

  1. 你走进餐厅,拿到菜单(API 文档),知道可以点什么、怎么点(请求格式)。

  2. 你对服务员说:一份宫保鸡丁,少辣(发送 API 请求)。

  3. 厨房做好菜给你端上来(返回 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 请求体。

几个核心规则:

  1. 对象用 { } 包裹,数组用 [ ] 包裹。

  2. 字符串用双引号 ",不能用单引号。

  3. 键值对用冒号 : 分隔,键必须是字符串。

  4. 多个键值对用逗号 , 分隔。

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. 1. 访问 platform.openai.com 注册账号

  2. 2. 进入 API Keys 页面

  3. 3. 点击 "Create new secret key"

  4. 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 有很多参数可以调整,下面是最常用的几个:

参数名类型是否必填说明默认值
modelstring使用的模型 ID,如 gpt-3.5-turbo、gpt-4
messagesarray对话历史消息列表
temperaturenumber采样温度,0-2 之间,越高越随机,越低越确定1.0
max_tokensinteger生成的最大 token 数无限制
top_pnumber核采样参数,0-1 之间1.0
stopstring/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))

一个经验法则:

  1. temperature=0:事实回答、代码生成、需要精确结果的场景

  2. temperature=0.7:日常对话、通用任务

  3. temperature>1.0:创意写作、头脑风暴

Anthropic Claude API

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

与 OpenAI API 的异同

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

对比项OpenAI APIClaude 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:

"""获取对话历史"""

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 聊天时,你会看到文字一个字一个字地蹦出来——这就是流式输出。

什么是流式输出

  1. 普通模式:AI 生成完整回复后一次性返回,你可能要等几秒甚至几十秒。

  2. 流式模式: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()

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

API 费用计算与优化

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

Token 计费原理

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

一个 token 大约等于:

  1. 英文:0.75 个单词(或 4 个字母)

  2. 中文:1-2 个汉字

举个例子:

  1. "Hello, 世界" — 大约 3-4 个 tokens

  2. "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-490%+
限制 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 状态码原因处理方式
认证失败401API Key 错误或过期检查 API Key
额度不足402账户没钱了充值或换号
超限429请求太快或额度用完等一会儿再试
服务器错误5xxAI 服务端出问题重试

指数退避重试

遇到临时性错误(如 429、5xx),最常用的策略是指数退避重试:

  1. 第一次失败 → 等 1 秒 → 重试

  2. 第二次失败 → 等 2 秒 → 重试

  3. 第三次失败 → 等 4 秒 → 重试

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

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

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

实例

# 文件路径: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再见!
喜欢(0)

上一篇

AI 产品设计

AI 产品设计

下一篇

《光环战役进化》全骷髅头及终端机收集攻略 骷髅头及终端机位置一览

《光环战役进化》全骷髅头及终端机收集攻略 骷髅头及终端机位置一览
猜你喜欢