OpenAI API实战：5分钟快速搭建你的第一个AI聊天机器人（Python版）
·
从零到一：用Python构建你的第一个智能对话助手
还记得第一次与ChatGPT对话时那种奇妙的感受吗？仿佛打开了一扇通往智能世界的大门。如今，作为开发者，我们不再仅仅是这些强大AI模型的用户，更可以成为创造者——通过OpenAI的API，将这种智能对话能力集成到我们自己的应用中。无论你是想为你的产品添加一个智能客服，还是想打造一个个性化的学习伙伴，甚至是开发一个创意写作助手，这一切都变得触手可及。
这篇文章就是为你准备的。如果你是一名Python开发者，对AI应用开发充满好奇，但面对API文档和密钥管理感到无从下手，那么请跟随我的脚步。我将带你从最基础的环境搭建开始，一步步构建一个功能完整的聊天机器人。在这个过程中，我不会仅仅停留在“如何调用API”的表面操作，而是会深入探讨一些实际开发中真正重要的问题：如何安全地管理密钥？如何处理复杂的对话上下文？如何优化响应质量？这些经验都来自我过去一年里在多个AI项目中踩过的坑和总结的最佳实践。
1. 环境准备与密钥安全：打好坚实的基础
在开始编写任何代码之前，我们需要确保开发环境配置正确且安全。这听起来可能有些枯燥，但我见过太多项目因为初期的基础工作没做好，导致后期出现各种难以调试的问题，甚至发生密钥泄露的安全事故。
1.1 Python环境与依赖安装
首先，确保你使用的是Python 3.7或更高版本。我强烈推荐使用虚拟环境来管理项目依赖，这样可以避免不同项目之间的包版本冲突。
# 创建项目目录并进入
mkdir my_chatbot && cd my_chatbot
# 创建虚拟环境（以venv为例）
python -m venv venv
# 激活虚拟环境
# 在Windows上：
venv\Scripts\activate
# 在macOS/Linux上：
source venv/bin/activate
接下来安装必要的包。除了官方的openai库，我通常还会安装几个辅助工具：
pip install openai python-dotenv requests
这里简单说明一下每个包的作用：
openai: 官方SDK，封装了API调用python-dotenv: 用于从.env文件加载环境变量，这是管理敏感信息的最佳实践requests: 虽然openai库已经封装了HTTP请求，但在某些高级场景下直接使用requests会更灵活
注意：如果你在安装过程中遇到SSL证书相关的问题，特别是在某些企业网络环境下，可能需要配置代理或更新证书。不过根据我的经验，大多数情况下使用最新版本的pip和Python都能顺利安装。
1.2 API密钥的安全管理：绝对不能忽视的一环
这是整个项目中最关键的安全环节。我看到过太多开发者把API密钥硬编码在代码中，然后不小心把代码上传到GitHub——这相当于把信用卡密码贴在公告栏上。
绝对不要这样做：
# 危险！永远不要硬编码密钥
openai.api_key = "sk-你的真实密钥"
正确的做法是使用环境变量。创建一个名为.env的文件（确保它被添加到.gitignore中），然后在其中添加：
OPENAI_API_KEY=sk-你的真实密钥
然后在代码中这样读取：
import os
from dotenv import load_dotenv
# 加载.env文件中的环境变量
load_dotenv()
# 安全地获取API密钥
api_key = os.getenv("OPENAI_API_KEY")
if not api_key:
raise ValueError("请在.env文件中设置OPENAI_API_KEY环境变量")
但环境变量管理也有几个层级，根据你的应用场景选择：
| 环境 | 推荐方案 | 优点 | 缺点 |
|---|---|---|---|
| 本地开发 | .env文件 + python-dotenv | 简单易用，隔离性好 | 需要确保.env不被提交到版本控制 |
| 生产服务器 | 系统环境变量 | 无需额外文件，操作系统级别管理 | 需要服务器配置权限 |
| 云平台 | 平台密钥管理服务（如AWS Secrets Manager） | 最高安全性，自动轮换 | 平台绑定，学习成本 |
| 客户端应用 | 后端代理转发 | 密钥完全不暴露在客户端 | 需要维护后端服务 |
对于大多数项目，我建议采用这样的分层策略：开发时使用.env文件，部署到生产环境时使用云平台的密钥管理服务。如果你正在开发的是Web应用，那么必须通过后端服务器来调用API，而不是在前端JavaScript中直接使用API密钥。
这里有一个实际项目中我常用的密钥验证脚本，可以在项目启动时检查配置是否正确：
import openai
from openai import OpenAIError
def validate_api_key(api_key):
"""验证API密钥是否有效"""
client = openai.OpenAI(api_key=api_key)
try:
# 尝试调用一个简单的、低成本的API来验证密钥
response = client.models.list()
# 检查是否有可用的模型
if len(response.data) > 0:
print("✅ API密钥验证成功")
return True
else:
print("❌ API密钥有效但无法获取模型列表")
return False
except OpenAIError as e:
print(f"❌ API密钥验证失败: {e}")
return False
# 使用示例
if __name__ == "__main__":
from dotenv import load_dotenv
import os
load_dotenv()
api_key = os.getenv("OPENAI_API_KEY")
if validate_api_key(api_key):
print("可以开始开发了！")
else:
print("请检查API密钥配置")
这个验证脚本有几个好处：首先，它能在项目启动早期发现问题，避免在开发到一半时才发现密钥有问题；其次，它使用了models.list()这个低成本API，不会产生不必要的费用；最后，它提供了清晰的错误信息，帮助你快速定位问题。
2. 第一个聊天机器人：从"Hello World"开始
现在环境准备好了，密钥也安全地配置好了，让我们来写第一个真正的聊天机器人。虽然OpenAI的API文档提供了基本的示例，但我想带你走得更远一些——不仅仅是让代码运行起来，还要理解每个参数的意义，以及如何根据实际需求调整它们。
2.1 最基本的对话实现
让我们从一个最简单的例子开始，这个例子展示了API调用的核心结构：
import openai
from dotenv import load_dotenv
import os
# 加载环境变量
load_dotenv()
# 初始化客户端
client = openai.OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
def simple_chat(prompt):
"""最基本的聊天函数"""
try:
response = client.chat.completions.create(
model="gpt-3.5-turbo", # 使用gpt-3.5-turbo模型，性价比高
messages=[
{"role": "user", "content": prompt}
],
max_tokens=150, # 限制响应长度
temperature=0.7 # 控制创造性，0-2之间，越高越随机
)
# 提取助手的回复
reply = response.choices[0].message.content
return reply.strip()
except Exception as e:
return f"抱歉，出错了: {str(e)}"
# 测试一下
if __name__ == "__main__":
user_input = "用Python写一个简单的斐波那契数列生成函数"
print(f"你: {user_input}")
print(f"助手: {simple_chat(user_input)}")
运行这个脚本，你应该能看到AI助手给出的Python代码。但这段代码有几个可以立即改进的地方：
- 错误处理不够细致：OpenAI API可能返回多种类型的错误，比如认证失败、额度不足、请求超时等
- 没有上下文记忆：每次对话都是独立的，AI不记得之前的对话内容
- 响应格式固定：我们可能希望AI以特定格式（如JSON、Markdown）回复
2.2 添加系统指令和对话历史
在实际应用中，我们通常希望AI扮演特定的角色，并且能够记住对话上下文。这就需要使用system角色和维护消息历史。
class ChatBot:
def __init__(self, system_prompt=None, model="gpt-3.5-turbo"):
"""初始化聊天机器人"""
load_dotenv()
self.client = openai.OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
self.model = model
self.conversation_history = []
# 设置系统提示，定义AI的角色和行为
if system_prompt:
self.conversation_history.append({
"role": "system",
"content": system_prompt
})
else:
# 默认的系统提示
self.conversation_history.append({
"role": "system",
"content": "你是一个有帮助的AI助手。回答要简洁、准确、有用。"
})
def chat(self, user_message, max_tokens=500, temperature=0.7):
"""进行一轮对话"""
# 添加用户消息到历史
self.conversation_history.append({
"role": "user",
"content": user_message
})
try:
response = self.client.chat.completions.create(
model=self.model,
messages=self.conversation_history,
max_tokens=max_tokens,
temperature=temperature,
top_p=0.9, # 另一种控制随机性的方式
frequency_penalty=0.5, # 降低重复内容
presence_penalty=0.5 # 鼓励谈论新话题
)
# 获取助手回复
assistant_reply = response.choices[0].message.content
# 添加到历史记录
self.conversation_history.append({
"role": "assistant",
"content": assistant_reply
})
# 返回本次回复和token使用情况
return {
"reply": assistant_reply,
"usage": {
"prompt_tokens": response.usage.prompt_tokens,
"completion_tokens": response.usage.completion_tokens,
"total_tokens": response.usage.total_tokens
}
}
except openai.RateLimitError:
return {"error": "请求过于频繁，请稍后再试"}
except openai.APIConnectionError:
return {"error": "网络连接错误，请检查网络"}
except openai.APIError as e:
return {"error": f"API错误: {str(e)}"}
def clear_history(self):
"""清空对话历史，但保留系统提示"""
system_prompt = self.conversation_history[0]
self.conversation_history = [system_prompt]
def get_history_summary(self):
"""获取对话历史摘要（用于调试或日志）"""
summary = []
for msg in self.conversation_history:
role = msg["role"]
# 只显示内容的前50个字符
preview = msg["content"][:50] + "..." if len(msg["content"]) > 50 else msg["content"]
summary.append(f"{role}: {preview}")
return "\n".join(summary)
# 使用示例
if __name__ == "__main__":
# 创建一个专门用于代码审查的机器人
code_reviewer = ChatBot(
system_prompt="""你是一个资深的Python代码审查专家。你的任务是：
1. 仔细分析提供的Python代码
2. 指出潜在的性能问题、安全漏洞和代码风格问题
3. 提供具体的改进建议
4. 用Markdown格式返回，包含代码示例
请保持专业但友好的语气。"""
)
# 测试代码审查
test_code = """
def process_data(data_list):
result = []
for item in data_list:
if item > 10:
result.append(item * 2)
return result
"""
print("正在审查代码...")
result = code_reviewer.chat(f"请审查这段Python代码：\n```python\n{test_code}\n```")
if "error" not in result:
print("审查结果：")
print(result["reply"])
print(f"\nToken使用情况：{result['usage']}")
else:
print(f"错误：{result['error']}")
这个改进版的聊天机器人有几个关键特性：
- 角色定义：通过系统提示，我们可以让AI扮演特定角色
- 对话历史管理：自动维护完整的对话上下文
- 错误处理：针对不同类型的API错误提供不同的处理
- 使用统计：返回token使用情况，便于监控成本和优化
2.3 理解关键参数：不只是temperature
很多教程只提到temperature参数，但实际上OpenAI的Chat Completions API提供了多个控制生成质量的参数。理解这些参数对于构建高质量的聊天机器人至关重要。
temperature (温度)
- 范围：0.0到2.0
- 作用：控制输出的随机性
- 低值（如0.2）：输出更加确定性和一致，适合代码生成、事实回答
- 高值（如0.8-1.2）：输出更加创造性，适合创意写作、头脑风暴
- 我的经验：对于技术问答，0.3-0.5效果最好；对于创意任务，0.7-1.0更合适
top_p (核采样)
- 范围：0.0到1.0
- 作用：控制词汇选择的集中程度
- 低值（如0.1）：只考虑概率最高的词汇
- 高值（如0.9）：考虑更广泛的词汇
- 注意：通常只调整temperature或top_p中的一个，不要同时调整
frequency_penalty (频率惩罚)
- 范围：-2.0到2.0
- 作用：惩罚重复出现的词汇
- 正值：降低重复内容出现的概率
- 我的经验：对于长对话，设置0.5-1.0可以减少重复
presence_penalty (存在惩罚)
- 范围：-2.0到2.0
- 作用：惩罚已经出现过的主题
- 正值：鼓励谈论新话题
- 我的经验：在多轮对话中设置0.3-0.6可以让对话更加丰富
为了帮助你理解这些参数的实际效果，我创建了一个对比表格：
| 参数组合 | 适用场景 | 示例配置 | 输出特点 |
|---|---|---|---|
| 技术问答 | 代码生成、技术解答 | temperature=0.3, top_p=0.9 | 准确、一致、可预测 |
| 创意写作 | 故事生成、营销文案 | temperature=0.8, frequency_penalty=0.5 | 多样、有创意、避免重复 |
| 客服对话 | 客户支持、问答 | temperature=0.5, presence_penalty=0.3 | 平衡、友好、覆盖多话题 |
| 头脑风暴 | 创意构思、方案设计 | temperature=1.0, presence_penalty=0.8 | 发散、新颖、跳出框架 |
在实际项目中，我通常会根据任务类型创建不同的参数预设：
class ParameterPresets:
"""参数预设集合"""
@staticmethod
def technical_qa():
"""技术问答"""
return {
"temperature": 0.3,
"top_p": 0.9,
"frequency_penalty": 0.2,
"presence_penalty": 0.1,
"max_tokens": 800
}
@staticmethod
def creative_writing():
"""创意写作"""
return {
"temperature": 0.8,
"top_p": 0.95,
"frequency_penalty": 0.5,
"presence_penalty": 0.4,
"max_tokens": 1000
}
@staticmethod
def code_generation():
"""代码生成"""
return {
"temperature": 0.2,
"top_p": 0.9,
"frequency_penalty": 0.1,
"presence_penalty": 0.0,
"max_tokens": 1500
}
# 使用示例
preset = ParameterPresets.technical_qa()
response = client.chat.completions.create(
model="gpt-3.5-turbo",
messages=messages,
**preset # 解包参数
)
3. 进阶功能：构建实用的聊天机器人
基本的聊天功能已经实现了，但在真实的应用场景中，我们通常需要更多高级功能。让我分享几个在实际项目中特别有用的进阶技巧。
3.1 流式响应：提升用户体验
当AI需要生成较长内容时，等待完整的响应可能会让用户感到不耐烦。流式响应（Streaming）可以让用户看到AI思考的过程，就像ChatGPT网页版那样。
import time
def stream_chat_response(client, messages, model="gpt-3.5-turbo"):
"""流式输出AI的回复"""
try:
# 创建流式响应
stream = client.chat.completions.create(
model=model,
messages=messages,
stream=True, # 关键参数：启用流式
temperature=0.7,
max_tokens=500
)
full_response = []
print("AI: ", end="", flush=True)
# 逐块处理响应
for chunk in stream:
if chunk.choices[0].delta.content is not None:
content = chunk.choices[0].delta.content
print(content, end="", flush=True)
full_response.append(content)
# 模拟人类打字速度
time.sleep(0.02)
print() # 换行
return "".join(full_response)
except Exception as e:
print(f"\n流式响应出错: {e}")
return None
# 使用示例
if __name__ == "__main__":
load_dotenv()
client = openai.OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
messages = [
{"role": "system", "content": "你是一个有耐心的老师，用简单易懂的方式解释复杂概念。"},
{"role": "user", "content": "请用简单的语言解释什么是机器学习？"}
]
print("你: 请用简单的语言解释什么是机器学习？")
response = stream_chat_response(client, messages)
流式响应不仅提升了用户体验，还有几个实际的好处：
- 降低感知延迟：用户立即看到响应开始，感觉更快
- 节省内存：对于很长的响应，不需要一次性加载到内存
- 实时处理：可以在生成过程中就进行部分处理
3.2 函数调用：让AI执行具体操作
OpenAI的API支持函数调用（Function Calling），这可能是最强大的功能之一。它允许AI决定何时调用你提供的函数，并返回结构化的参数。
让我们看一个实际的例子：创建一个可以查询天气和做计算的智能助手。
import json
import math
from datetime import datetime
class FunctionCallingBot:
def __init__(self):
load_dotenv()
self.client = openai.OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
self.conversation_history = []
# 定义可用的函数
self.available_functions = {
"get_current_time": self.get_current_time,
"calculate_expression": self.calculate_expression,
"get_weather_info": self.get_weather_info,
}
# 函数的描述（AI需要知道这些信息）
self.functions = [
{
"name": "get_current_time",
"description": "获取当前的时间和日期",
"parameters": {
"type": "object",
"properties": {
"timezone": {
"type": "string",
"description": "时区，例如'Asia/Shanghai'或'America/New_York'",
"enum": ["Asia/Shanghai", "America/New_York", "UTC"]
}
},
"required": ["timezone"]
}
},
{
"name": "calculate_expression",
"description": "计算数学表达式",
"parameters": {
"type": "object",
"properties": {
"expression": {
"type": "string",
"description": "数学表达式，例如'2 + 3 * 4'或'sqrt(16)'"
}
},
"required": ["expression"]
}
},
{
"name": "get_weather_info",
"description": "获取天气信息（模拟函数）",
"parameters": {
"type": "object",
"properties": {
"city": {
"type": "string",
"description": "城市名称"
},
"date": {
"type": "string",
"description": "日期，格式为YYYY-MM-DD"
}
},
"required": ["city"]
}
}
]
# 实际的功能函数
def get_current_time(self, timezone):
"""获取指定时区的当前时间"""
from datetime import datetime, timezone as tz
import pytz
try:
tz_obj = pytz.timezone(timezone)
current_time = datetime.now(tz_obj)
return f"{timezone}的当前时间是: {current_time.strftime('%Y-%m-%d %H:%M:%S')}"
except Exception as e:
return f"无法获取{timezone}的时间: {str(e)}"
def calculate_expression(self, expression):
"""计算数学表达式（注意：实际使用中应该更安全地处理）"""
try:
# 安全地计算表达式
allowed_names = {
k: v for k, v in math.__dict__.items()
if not k.startswith("_")
}
# 使用eval但要限制可用的函数
result = eval(expression, {"__builtins__": {}}, allowed_names)
return f"{expression} = {result}"
except Exception as e:
return f"计算错误: {str(e)}"
def get_weather_info(self, city, date=None):
"""模拟获取天气信息"""
# 在实际应用中，这里会调用真实的天气API
weather_data = {
"Beijing": {"temp": "22°C", "condition": "晴朗", "humidity": "45%"},
"Shanghai": {"temp": "25°C", "condition": "多云", "humidity": "65%"},
"New York": {"temp": "18°C", "condition": "小雨", "humidity": "75%"}
}
if city in weather_data:
info = weather_data[city]
date_str = f"在{date}" if date else "今天"
return f"{city}{date_str}的天气：温度{info['temp']}，{info['condition']}，湿度{info['humidity']}"
else:
return f"抱歉，没有{city}的天气信息"
def process_message(self, user_input):
"""处理用户输入，可能涉及函数调用"""
# 添加用户消息到历史
self.conversation_history.append({"role": "user", "content": user_input})
# 第一次API调用：让AI决定是否需要调用函数
response = self.client.chat.completions.create(
model="gpt-3.5-turbo",
messages=self.conversation_history,
functions=self.functions,
function_call="auto", # 让AI自动决定
temperature=0.3
)
response_message = response.choices[0].message
# 检查AI是否想调用函数
if response_message.function_call:
# 获取函数名和参数
function_name = response_message.function_call.name
function_args = json.loads(response_message.function_call.arguments)
print(f"🤖 AI决定调用函数: {function_name}")
print(f" 参数: {function_args}")
# 调用对应的函数
if function_name in self.available_functions:
function_to_call = self.available_functions[function_name]
function_response = function_to_call(**function_args)
print(f"✅ 函数返回: {function_response}")
# 将函数响应添加到对话历史
self.conversation_history.append(response_message) # AI的函数调用请求
self.conversation_history.append({
"role": "function",
"name": function_name,
"content": function_response
})
# 第二次API调用：让AI基于函数结果生成最终回复
second_response = self.client.chat.completions.create(
model="gpt-3.5-turbo",
messages=self.conversation_history,
temperature=0.3
)
final_reply = second_response.choices[0].message.content
self.conversation_history.append({
"role": "assistant",
"content": final_reply
})
return final_reply
else:
return "抱歉，请求的函数不可用"
else:
# 不需要调用函数，直接返回AI的回复
ai_reply = response_message.content
self.conversation_history.append({
"role": "assistant",
"content": ai_reply
})
return ai_reply
# 使用示例
if __name__ == "__main__":
bot = FunctionCallingBot()
test_queries = [
"现在北京的时间是多少？",
"计算一下15的平方根加上20除以4等于多少？",
"上海明天的天气怎么样？",
"请介绍一下Python的列表推导式"
]
for query in test_queries:
print(f"\n你: {query}")
print("-" * 50)
response = bot.process_message(query)
print(f"助手: {response}")
print("-" * 50)
函数调用的强大之处在于：
- 扩展AI的能力：让AI可以执行实际的操作，而不仅仅是生成文本
- 结构化数据：函数参数和返回值都是结构化的，便于程序处理
- 决策能力：AI可以自主决定何时调用哪个函数
在实际项目中，你可以用这个功能实现各种复杂的功能，比如：
- 数据库查询
- 发送邮件
- 调用其他API
- 文件操作
- 等等
3.3 上下文管理：处理长对话的挑战
随着对话的进行，消息历史会越来越长。OpenAI的模型有token限制（gpt-3.5-turbo通常是4096个token，gpt-4最多可达128K），我们需要智能地管理上下文。
class SmartContextManager:
def __init__(self, max_tokens=3000, model="gpt-3.5-turbo"):
"""
智能上下文管理器
Args:
max_tokens: 最大token数（留出空间给新回复）
model: 使用的模型，用于估算token数
"""
self.max_tokens = max_tokens
self.model = model
self.conversation_history = []
self.system_prompt = None
def add_message(self, role, content):
"""添加消息到历史"""
self.conversation_history.append({"role": role, "content": content})
def set_system_prompt(self, prompt):
"""设置系统提示"""
self.system_prompt = prompt
# 如果这是第一条消息，或者系统提示改变了，重新开始
if not self.conversation_history or self.conversation_history[0]["role"] != "system":
self.conversation_history = [{"role": "system", "content": prompt}]
else:
self.conversation_history[0] = {"role": "system", "content": prompt}
def estimate_tokens(self, text):
"""粗略估算文本的token数（实际应该使用tiktoken库）"""
# 简单估算：英文大约1个token=4个字符，中文大约1个token=2个字符
# 注意：这是粗略估算，实际应该使用OpenAI的tiktoken库
chinese_chars = sum(1 for c in text if '\u4e00' <= c <= '\u9fff')
other_chars = len(text) - chinese_chars
return chinese_chars // 2 + other_chars // 4
def get_current_token_count(self):
"""计算当前对话历史的token数"""
total = 0
for message in self.conversation_history:
total += self.estimate_tokens(message["content"])
# 加上角色标识的token
total += 5
return total
def smart_trim_history(self):
"""智能修剪对话历史"""
current_tokens = self.get_current_token_count()
if current_tokens <= self.max_tokens:
return self.conversation_history
print(f"⚠️ 对话历史过长 ({current_tokens} tokens)，正在智能修剪...")
# 保留系统提示
trimmed_history = [self.conversation_history[0]] if self.conversation_history and self.conversation_history[0]["role"] == "system" else []
# 策略1：优先保留最近的消息
recent_messages = self.conversation_history[-10:] # 保留最近10条
# 策略2：如果还是太长，尝试总结早期对话
temp_history = trimmed_history + recent_messages
temp_tokens = sum(self.estimate_tokens(msg["content"]) + 5 for msg in temp_history)
if temp_tokens > self.max_tokens:
# 需要更激进的修剪：总结早期对话
old_messages = self.conversation_history[1:-10] # 排除系统和最近的消息
if old_messages:
# 创建总结提示
summary_prompt = f"""请用一段话总结以下对话的主要内容，保留关键信息：
{' '.join([f"{msg['role']}: {msg['content'][:100]}..." for msg in old_messages])}
总结："""
# 在实际应用中，这里应该调用AI来生成总结
# 为了简化，我们只取前几条消息的关键信息
summary = "之前的对话讨论了：" + "，".join([msg["content"][:50] for msg in old_messages[:3]]) + "等话题。"
# 添加总结作为系统提示的一部分
if trimmed_history: # 有系统提示
trimmed_history[0]["content"] += f"\n\n对话历史摘要：{summary}"
else:
trimmed_history.append({"role": "system", "content": f"对话历史摘要：{summary}"})
# 确保不超过token限制
final_history = trimmed_history + recent_messages
final_tokens = sum(self.estimate_tokens(msg["content"]) + 5 for msg in final_history)
# 如果还是太长，逐步删除最旧的非系统消息
while final_tokens > self.max_tokens and len(final_history) > (2 if trimmed_history else 1):
# 删除最旧的非系统消息
for i in range(len(final_history)):
if final_history[i]["role"] != "system":
deleted_msg = final_history.pop(i)
final_tokens -= self.estimate_tokens(deleted_msg["content"]) + 5
break
print(f"✅ 修剪完成，现在有 {final_tokens} tokens")
return final_history
def add_and_trim(self, role, content):
"""添加消息并自动修剪历史"""
self.add_message(role, content)
return self.smart_trim_history()
# 使用示例：模拟长对话
if __name__ == "__main__":
import time
context_manager = SmartContextManager(max_tokens=1000) # 设置较小的限制以便测试
context_manager.set_system_prompt("你是一个有帮助的助手。")
# 模拟长对话
topics = [
"Python中的列表和元组有什么区别？",
"能详细解释一下列表推导式吗？",
"那生成器表达式呢？它和列表推导式有什么不同？",
"在Python中如何处理异常？",
"try-except和try-finally有什么区别？",
"什么是上下文管理器？with语句是怎么工作的？",
"能给我一个自定义上下文管理器的例子吗？",
"Python的装饰器是什么？怎么使用？",
"带参数的装饰器怎么写？",
"异步编程在Python中是怎么实现的？",
"asyncio库的主要组件有哪些？",
"协程和普通函数有什么区别？",
"Python的类型提示有什么好处？",
"如何使用mypy进行类型检查？",
"Python 3.10中的match语句怎么用？",
"结构模式匹配和if-elif-else有什么区别？",
"Python的内存管理机制是怎样的？",
"引用计数和垃圾回收是怎么工作的？",
"如何避免内存泄漏？",
"Python的多线程和多进程有什么区别？"
]
for i, topic in enumerate(topics):
print(f"\n[{i+1}] 用户: {topic}")
# 添加用户消息并修剪历史
messages = context_manager.add_and_trim("user", topic)
# 模拟AI回复
ai_reply = f"这是关于'{topic}'的回复。"
context_manager.add_and_trim("assistant", ai_reply)
print(f" 助手: {ai_reply}")
print(f" 当前token数: {context_manager.get_current_token_count()}")
time.sleep(0.1) # 稍微延迟以便观察
这个智能上下文管理器实现了几个关键功能：
- Token估算：虽然这里使用了简单的估算方法，但在实际项目中应该使用
tiktoken库进行精确计算 - 智能修剪：当对话历史过长时，自动修剪旧消息
- 对话总结：对于被修剪的旧对话，可以生成摘要保留关键信息
- 优先级保留：总是保留系统提示和最近的对话
在实际应用中，你可能还需要考虑：
- 重要消息标记：让用户或系统标记重要消息，避免被修剪
- 主题分割：根据对话主题自动分割上下文
- 向量化存储：将历史对话存储到向量数据库，需要时检索相关部分
4. 实战项目：构建一个多功能聊天机器人
现在让我们把所有知识整合起来，构建一个完整的、功能丰富的聊天机器人。这个机器人将具备：
- 多角色切换能力
- 上下文管理
- 函数调用
- 流式响应
- 使用统计和成本控制
import json
import time
from typing import Dict, List, Optional, Any
from dataclasses import dataclass, asdict
from enum import Enum
class BotRole(Enum):
"""机器人角色枚举"""
GENERAL_ASSISTANT = "general"
CODE_TUTOR = "code_tutor"
CREATIVE_WRITER = "creative_writer"
TECHNICAL_SUPPORT = "tech_support"
@dataclass
class ConversationStats:
"""对话统计"""
total_messages: int = 0
total_tokens: int = 0
total_cost: float = 0.0
start_time: float = None
def __post_init__(self):
if self.start_time is None:
self.start_time = time.time()
def add_message(self, prompt_tokens: int, completion_tokens: int, model: str):
"""添加消息统计"""
self.total_messages += 1
self.total_tokens += prompt_tokens + completion_tokens
# 根据模型计算成本（价格可能变化，这里使用近似值）
cost_per_1k = {
"gpt-3.5-turbo": 0.0015, # 输入
"gpt-3.5-turbo-16k": 0.003,
"gpt-4": 0.03,
"gpt-4-turbo": 0.01
}
model_key = model if model in cost_per_1k else "gpt-3.5-turbo"
cost = (prompt_tokens / 1000) * cost_per_1k[model_key] + \
(completion_tokens / 1000) * (cost_per_1k[model_key] * 2)
self.total_cost += cost
def get_session_duration(self):
"""获取会话持续时间"""
return time.time() - self.start_time
def get_summary(self):
"""获取统计摘要"""
duration = self.get_session_duration()
return {
"total_messages": self.total_messages,
"total_tokens": self.total_tokens,
"total_cost_usd": round(self.total_cost, 4),
"session_duration_seconds": round(duration, 2),
"avg_tokens_per_message": round(self.total_tokens / max(self.total_messages, 1), 2)
}
class MultiFunctionChatBot:
"""多功能聊天机器人"""
def __init__(self, api_key: str, default_model: str = "gpt-3.5-turbo"):
self.client = openai.OpenAI(api_key=api_key)
self.default_model = default_model
self.current_role = BotRole.GENERAL_ASSISTANT
self.stats = ConversationStats()
# 角色配置
self.role_configs = {
BotRole.GENERAL_ASSISTANT: {
"system_prompt": "你是一个有帮助的、知识渊博的AI助手。回答要准确、有用、友好。",
"temperature": 0.7,
"max_tokens": 1000
},
BotRole.CODE_TUTOR: {
"system_prompt": """你是一个耐心的编程导师。你的任务是：
1. 用简单易懂的方式解释编程概念
2. 提供可运行的代码示例
3. 指出常见错误和最佳实践
4. 鼓励学习者自己尝试
请用Markdown格式返回代码，并添加适当的注释。""",
"temperature": 0.3,
"max_tokens": 1500
},
BotRole.CREATIVE_WRITER: {
"system_prompt": """你是一个富有创造力的作家。你的任务是：
1. 根据提示创作故事、诗歌或文案
2. 使用生动的语言和丰富的想象力
3. 保持风格一致
4. 提供多个创意选项
请尽情发挥你的创造力！""",
"temperature": 0.9,
"max_tokens": 2000
},
BotRole.TECHNICAL_SUPPORT: {
"system_prompt": """你是一个专业的技术支持专家。你的任务是：
1. 准确诊断技术问题
2. 提供逐步的解决方案
3. 解释技术原理但不使用过多术语
4. 提供预防措施和最佳实践
请保持专业、耐心、清晰。""",
"temperature": 0.4,
"max_tokens": 1200
}
}
# 初始化对话历史
self.conversation_history = []
self._update_system_prompt()
# 可用函数
self.functions = self._define_functions()
self.available_functions = self._register_functions()
def _define_functions(self):
"""定义可用函数"""
return [
{
"name": "switch_role",
"description": "切换机器人的角色",
"parameters": {
"type": "object",
"properties": {
"role": {
"type": "string",
"enum": ["general", "code_tutor", "creative_writer", "tech_support"],
"description": "要切换到的角色"
}
},
"required": ["role"]
}
},
{
"name": "get_conversation_stats",
"description": "获取当前对话的统计信息",
"parameters": {
"type": "object",
"properties": {},
"required": []
}
},
{
"name": "clear_conversation",
"description": "清空对话历史（但保留当前角色设置）",
"parameters": {
"type": "object",
"properties": {
"confirm": {
"type": "boolean",
"description": "确认是否清空"
}
},
"required": ["confirm"]
}
},
{
"name": "export_conversation",
"description": "导出当前对话历史",
"parameters": {
"type": "object",
"properties": {
"format": {
"type": "string",
"enum": ["json", "text", "markdown"],
"description": "导出格式"
}
},
"required": ["format"]
}
}
]
def _register_functions(self):
"""注册函数实现"""
return {
"switch_role": self._switch_role,
"get_conversation_stats": self._get_conversation_stats,
"clear_conversation": self._clear_conversation,
"export_conversation": self._export_conversation
}
def _update_system_prompt(self):
"""更新系统提示"""
config = self.role_configs[self.current_role]
system_message = {"role": "system", "content": config["system_prompt"]}
# 更新或添加系统消息
if self.conversation_history and self.conversation_history[0]["role"] == "system":
self.conversation_history[0] = system_message
else:
self.conversation_history.insert(0, system_message)
def _switch_role(self, role: str):
"""切换角色"""
role_map = {
"general": BotRole.GENERAL_ASSISTANT,
"code_tutor": BotRole.CODE_TUTOR,
"creative_writer": BotRole.CREATIVE_WRITER,
"tech_support": BotRole.TECHNICAL_SUPPORT
}
if role not in role_map:
return f"未知角色: {role}。可用角色: {', '.join(role_map.keys())}"
self.current_role = role_map[role]
self._update_system_prompt()
role_names = {
BotRole.GENERAL_ASSISTANT: "通用助手",
BotRole.CODE_TUTOR: "编程导师",
BotRole.CREATIVE_WRITER: "创意作家",
BotRole.TECHNICAL_SUPPORT: "技术支持"
}
return f"已切换到 {role_names[self.current_role]} 模式。"
def _get_conversation_stats(self):
"""获取对话统计"""
stats = self.stats.get_summary()
stats["current_role"] = self.current_role.value
stats["history_length"] = len(self.conversation_history) - 1 # 排除系统消息
# 格式化为友好字符串
summary = f"""对话统计：
• 消息总数: {stats['total_messages']}
• Token总数: {stats['total_tokens']}
• 估计成本: ${stats['total_cost_usd']}
• 会话时长: {stats['session_duration_seconds']}秒
• 当前角色: {stats['current_role']}
• 历史长度: {stats['history_length']}条消息"""
return summary
def _clear_conversation(self, confirm: bool):
"""清空对话历史"""
if not confirm:
return "取消清空操作。"
# 保留系统消息
system_msg = self.conversation_history[0] if self.conversation_history and self.conversation_history[0]["role"] == "system" else None
if system_msg:
self.conversation_history = [system_msg]
else:
self.conversation_history = []
return "对话历史已清空。"
def _export_conversation(self, format: str):
"""导出对话历史"""
if format == "json":
return json.dumps(self.conversation_history, ensure_ascii=False, indent=2)
elif format == "text":
lines = []
for msg in self.conversation_history:
role = {"system": "系统", "user": "用户", "assistant": "助手"}.get(msg["role"], msg["role"])
lines.append(f"{role}: {msg['content']}")
return "\n\n".join(lines)
elif format == "markdown":
lines = ["# 对话记录", ""]
for msg in self.conversation_history:
role = {"system": "💻 系统", "user": "👤 用户", "assistant": "🤖 助手"}.get(msg["role"], msg["role"])
lines.append(f"**{role}**\n\n{msg['content']}\n\n---\n")
return "\n".join(lines)
else:
return f"不支持的格式: {format}"
def chat(self, user_input: str, stream: bool = False, use_functions: bool = True):
"""主聊天方法"""
# 添加用户消息
self.conversation_history.append({"role": "user", "content": user_input})
# 获取当前角色的配置
config = self.role_configs[self.current_role]
try:
# 准备API参数
api_params = {
"model": self.default_model,
"messages": self.conversation_history,
"temperature": config["temperature"],
"max_tokens": config["max_tokens"],
"top_p": 0.9,
"frequency_penalty": 0.5,
"presence_penalty": 0.5
}
# 如果需要函数调用，添加相关参数
if use_functions:
api_params["functions"] = self.functions
api_params["function_call"] = "auto"
# 流式响应
if stream:
return self._stream_response(api_params)
# 普通响应
response = self.client.chat.completions.create(**api_params)
response_message = response.choices[0].message
# 处理函数调用
if hasattr(response_message, 'function_call') and response_message.function_call:
return self._handle_function_call(response_message)
# 普通文本回复
assistant_reply = response_message.content
# 更新对话历史
self.conversation_history.append({
"role": "assistant",
"content": assistant_reply
})
# 更新统计
self.stats.add_message(
response.usage.prompt_tokens,
response.usage.completion_tokens,
self.default_model
)
return {
"type": "text",
"content": assistant_reply,
"usage": {
"prompt_tokens": response.usage.prompt_tokens,
"completion_tokens": response.usage.completion_tokens,
"total_tokens": response.usage.total_tokens
}
}
except openai.RateLimitError:
return {"type": "error", "content": "请求过于频繁，请稍后再试。"}
except openai.APIConnectionError:
return {"type": "error", "content": "网络连接错误，请检查网络。"}
except openai.APIError as e:
return {"type": "error", "content": f"API错误: {str(e)}"}
except Exception as e:
return {"type": "error", "content": f"未知错误: {str(e)}"}
def _stream_response(self, api_params):
"""处理流式响应"""
try:
# 移除stream参数，因为我们要手动设置
api_params.pop("functions", None)
api_params.pop("function_call", None)
api_params["stream"] = True
stream = self.client.chat.completions.create(**api_params)
full_response = []
for chunk in stream:
if chunk.choices[0].delta.content is not None:
content = chunk.choices[0].delta.content
full_response.append(content)
yield {"type": "chunk", "content": content}
# 完成流式响应后，添加到历史记录
assistant_reply = "".join(full_response)
self.conversation_history.append({
"role": "assistant",
"content": assistant_reply
})
# 注意：流式响应不返回usage统计
yield {"type": "complete", "content": assistant_reply}
except Exception as e:
yield {"type": "error", "content": f"流式响应错误: {str(e)}"}
def _handle_function_call(self, response_message):
"""处理函数调用"""
function_name = response_message.function_call.name
function_args = json.loads(response_message.function_call.arguments)
# 添加到历史记录
self.conversation_history.append(response_message)
# 调用函数
if function_name in self.available_functions:
function_to_call = self.available_functions[function_name]
function_response = function_to_call(**function_args)
# 添加函数响应到历史
self.conversation_history.append({
"role": "function",
"name": function_name,
"content": str(function_response)
})
# 获取AI的最终回复
final_response = self.client.chat.completions.create(
model=self.default_model,
messages=self.conversation_history,
temperature=self.role_configs[self.current_role]["temperature"],
max_tokens=self.role_configs[self.current_role]["max_tokens"]
)
assistant_reply = final_response.choices[0].message.content
# 更新历史记录
self.conversation_history.append({
"role": "assistant",
"content": assistant_reply
})
# 更新统计（需要估算第一次调用的token使用）
# 这里简化处理，实际应该记录第一次调用的token使用
self.stats.add_message(
final_response.usage.prompt_tokens,
final_response.usage.completion_tokens,
self.default_model
)
return {
"type": "function",
"function_name": function_name,
"function_args": function_args,
"function_response": function_response,
"content": assistant_reply
}
else:
return {"type": "error", "content": f"未知函数: {function_name}"}
def get_conversation_preview(self, max_messages=5):
"""获取对话预览"""
preview = []
for msg in self.conversation_history[-max_messages:]:
role_icon = {
"system": "💻",
"user": "👤",
"assistant": "🤖",
"function": "⚙️"
}.get(msg["role"], msg["role"])
content_preview = msg["content"][:50] + "..." if len(msg["content"]) > 50 else msg["content"]
preview.append(f"{role_icon} {content_preview}")
return "\n".join(preview)
# 使用示例
def demo_chatbot():
"""演示聊天机器人的使用"""
import os
from dotenv import load_dotenv
load_dotenv()
api_key = os.getenv("OPENAI_API_KEY")
if not api_key:
print("请设置OPENAI_API_KEY环境变量")
return
# 创建机器人实例
bot = MultiFunctionChatBot(api_key=api_key)
print("=" * 60)
print("多功能聊天机器人已启动！")
print("支持命令：")
print(" /stats - 查看统计信息")
print(" /role <角色名> - 切换角色（general/code_tutor/creative_writer/tech_support）")
print(" /clear - 清空对话历史")
print(" /export <格式> - 导出对话（json/text/markdown）")
print(" /exit - 退出")
print("=" * 60)
# 演示不同角色
test_cases = [
("切换到编程导师模式", "/role code_tutor"),
("请教Python的装饰器", "请用简单的例子解释Python装饰器的工作原理"),
("查看统计", "/stats"),
("切换到创意作家模式", "/role creative_writer"),
("写一首诗", "写一首关于秋天的短诗"),
("导出对话", "/export markdown"),
("清空历史", "/clear true"),
("最后问一个问题", "什么是机器学习？")
]
for description, input_text in test_cases:
print(f"\n{'='*40}")
print(f"测试: {description}")
print(f"输入: {input_text}")
print(f"{'='*40}")
if input_text.startswith("/"):
# 处理命令
if input_text.startswith("/role"):
role = input_text.split()[1] if len(input_text.split()) > 1 else "general"
result = bot._switch_role(role)
print(f"系统: {result}")
elif input_text.startswith("/stats"):
result = bot._get_conversation_stats()
print(f"统计: \n{result}")
elif input_text.startswith("/clear"):
confirm = "true" in input_text.lower()
result = bot._clear_conversation(confirm)
print(f"系统: {result}")
elif input_text.startswith("/export"):
fmt = input_text.split()[1] if len(input_text.split()) > 1 else "text"
result = bot._export_conversation(fmt)
print(f"导出结果（前500字符）:\n{result[:500]}...")
continue
# 普通聊天
print(f"你: {input_text}")
response = bot.chat(input_text, stream=False)
if response["type"] == "text":
print(f"助手: {response['content'][:200]}...")
elif response["type"] == "function":
print(f"⚙️ 调用了函数: {response['function_name']}")
print(f" 参数: {response['function_args']}")
print(f" 结果: {response['function_response']}")
print(f" 最终回复: {response['content'][:200]}...")
elif response["type"] == "error":
print(f"错误: {response['content']}")
# 显示最终统计
print(f"\n{'='*60}")
print("对话结束，最终统计：")
stats = bot.stats.get_summary()
for key, value in stats.items():
print(f" {key}: {value}")
print("=" * 60)
if __name__ == "__main__":
demo_chatbot()
这个完整的聊天机器人示例展示了如何将前面讨论的所有概念整合到一个实用的系统中。它具备了生产级聊天机器人需要的许多特性：
- 多角色支持：用户可以随时切换机器人的角色和行为模式
- 完整的函数调用：机器人可以执行各种操作，如切换角色、获取统计、清空历史等
- 流式响应支持：提供更好的用户体验
- 使用统计：跟踪token使用和成本
- 健壮的错误处理：处理各种API错误和异常
- 对话管理：提供导出、清空等功能
在实际部署这样的系统时，你还需要考虑：
- 持久化存储：将对话历史保存到数据库
- 用户认证：支持多用户，隔离对话历史
- 速率限制：防止API被过度调用
- 缓存机制：缓存常见问题的回答以减少API调用
- 监控告警：监控API使用情况和错误率
构建一个完整的AI聊天机器人项目就像搭积木，从最简单的API调用开始，逐步添加上下文管理、函数调用、流式响应等高级功能。每个功能都解决了一个实际问题：上下文管理让对话更连贯，函数调用扩展了AI的能力边界，流式响应提升了用户体验。
我建议你在理解这些基础概念后，根据自己的需求进行定制和扩展。比如，如果你要构建一个客服机器人，可能需要集成知识库检索；如果要构建编程助手，可能需要添加代码执行沙箱。OpenAI的API提供了强大的基础能力，而如何将这些能力组合成有用的应用，则完全取决于你的创造力和对用户需求的理解。
记住，最好的学习方式是在理解原理的基础上动手实践。从一个小功能开始，逐步迭代，你会发现自己能够构建出越来越复杂的AI应用。在这个过程中，你不仅会掌握技术细节，更会培养出解决实际问题的思维方式——这才是作为开发者最宝贵的财富。
这里是“一人公司”的成长家园。我们提供从产品曝光、技术变现到法律财税的全栈内容，并连接云服务、办公空间等稀缺资源，助你专注创造，无忧运营。
更多推荐
23
0- 0
已为社区贡献1条内容
所有评论(0)