好的,技术同仁。面对50+轮的长对话需求,网页界面是瓶颈所在。我们来构建一个专业、可控的解决方案。
**核心思路**:弃用网页UI,通过API自主掌控上下文,配合智能压缩与持久化存储。
### 🚀 第一步:建立基础API通信框架
1. **获取API密钥**
* 访问 OpenAI API 平台 (platform.openai.com)
* 生成并保管好你的 `API Key`。
2. **创建基础Python脚本**
* 创建一个新的Python文件,例如 `professional_chat.py`。
* 安装必要的库:`pip install openai`
```python
# professional_chat.py - 基础版本
import openai
import json
from datetime import datetime
# 配置API密钥
client = openai.OpenAI(api_key='你的API密钥')
# 初始化对话历史,可预设系统角色
conversation_history = [
{"role": "system", "content": "你是一个资深的软件架构师和技术专家,善于进行深度的技术讨论和问题排查。"}
]
def chat_with_gpt(user_input, history):
"""核心通信函数"""
history.append({"role": "user", "content": user_input})
try:
response = client.chat.completions.create(
model="gpt-4", # 或 "gpt-4-turbo-preview"
messages=history,
max_tokens=2000, # 控制单次回复长度
temperature=0.7
)
assistant_reply = response.choices[0].message.content
history.append({"role": "assistant", "content": assistant_reply})
return assistant_reply
except Exception as e:
return f"API调用错误: {str(e)}"
# 测试基础功能
if __name__ == "__main__":
while True:
user_input = input("\nYou: ")
if user_input.lower() in ['exit', 'quit']:
break
reply = chat_with_gpt(user_input, conversation_history)
print(f"\nAssistant: {reply}")
print(f"当前对话轮数: {len(conversation_history)//2}")
```
### 🔧 第二步:实现智能上下文管理
在基础脚本上增强,添加上下文压缩逻辑。
```python
# 添加上下文管理器类
class ContextManager:
def __init__(self, max_rounds=50, compression_ratio=0.6):
self.max_rounds = max_rounds
self.compression_ratio = compression_ratio
self.important_messages = set() # 存储重要信息的索引
def compress_history(self, full_history):
"""智能压缩对话历史"""
if len(full_history) <= self.max_rounds * 2: # 乘以2因为每条记录包含user和assistant
return full_history
# 策略:保留系统消息 + 重要消息 + 最近对话
system_msg = [msg for msg in full_history if msg["role"] == "system"]
recent_msgs = full_history[-int(self.max_rounds * self.compression_ratio * 2):]
# 提取被标记为重要的历史消息
important_msgs = []
for idx, msg in enumerate(full_history):
if idx in self.important_messages and msg not in recent_msgs:
important_msgs.append(msg)
# 合并并去重
compressed_history = system_msg + important_msgs + recent_msgs
# 简单的去重逻辑(根据实际需求可增强)
seen = set()
unique_history = []
for msg in compressed_history:
msg_str = json.dumps(msg, sort_keys=True)
if msg_str not in seen:
seen.add(msg_str)
unique_history.append(msg)
print(f"[系统提示] 上下文已压缩: {len(full_history)} -> {len(unique_history)} 条消息")
return unique_history
def mark_important(self, message_index):
"""标记重要消息"""
self.important_messages.add(message_index)
def auto_mark_important(self, message):
"""基于关键词自动标记重要消息"""
important_keywords = ['架构', '设计', '代码', '配置', '方案', '总结', '结论', '决定']
if any(keyword in message.get('content', '') for keyword in important_keywords):
return True
return False
# 集成到主流程中
context_manager = ContextManager(max_rounds=50)
def enhanced_chat_with_gpt(user_input, history):
"""增强版的通信函数,带上下文管理"""
# 在添加用户消息前检查是否需要压缩
if len(history) >= context_manager.max_rounds * 2:
history = context_manager.compress_history(history)
history.append({"role": "user", "content": user_input})
# 自动标记可能重要的用户消息
if context_manager.auto_mark_important(history[-1]):
context_manager.mark_important(len(history) - 1)
try:
response = client.chat.completions.create(
model="gpt-4",
messages=history,
max_tokens=2000,
temperature=0.7
)
assistant_reply = response.choices[0].message.content
history.append({"role": "assistant", "content": assistant_reply})
# 自动标记可能重要的助手回复
if context_manager.auto_mark_important(history[-1]):
context_manager.mark_important(len(history) - 1)
return assistant_reply, history
except Exception as e:
return f"API调用错误: {str(e)}", history
# 更新主循环
conversation_history = [
{"role": "system", "content": "你是一个资深的软件架构师和技术专家。"}
]
if __name__ == "__main__":
print("=== 专业技术对话系统(支持50+轮长对话)===")
while True:
user_input = input("\nYou: ")
if user_input.lower() in ['exit', 'quit']:
# 退出前保存对话
with open(f"conversation_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json", 'w', encoding='utf-8') as f:
json.dump(conversation_history, f, ensure_ascii=False, indent=2)
print("对话已保存!")
break
if user_input == '/save':
# 手动保存命令
with open(f"conversation_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json", 'w', encoding='utf-8') as f:
json.dump(conversation_history, f, ensure_ascii=False, indent=2)
print("对话已手动保存!")
continue
reply, conversation_history = enhanced_chat_with_gpt(user_input, conversation_history)
print(f"\nAssistant: {reply}")
print(f"当前对话轮数: {(len(conversation_history)-1)//2}") # 减去系统消息
```
### 📁 第三步:实现对话持久化与检索
添加对话的保存和加载功能,实现真正的长周期对话。
```python
# 对话存储管理类
class ConversationStore:
def __init__(self, storage_path="./conversations"):
self.storage_path = storage_path
os.makedirs(storage_path, exist_ok=True)
def save_conversation(self, history, session_id=None):
"""保存对话到文件"""
if session_id is None:
session_id = f"session_{datetime.now().strftime('%Y%m%d_%H%M%S')}"
filename = f"{self.storage_path}/{session_id}.json"
with open(filename, 'w', encoding='utf-8') as f:
json.dump({
"session_id": session_id,
"created_at": datetime.now().isoformat(),
"history": history
}, f, ensure_ascii=False, indent=2)
return session_id
def load_conversation(self, session_id):
"""从文件加载对话"""
filename = f"{self.storage_path}/{session_id}.json"
try:
with open(filename, 'r', encoding='utf-8') as f:
data = json.load(f)
return data["history"]
except FileNotFoundError:
print(f"对话会话 {session_id} 不存在")
return None
def list_conversations(self):
"""列出所有保存的对话"""
sessions = []
for file in os.listdir(self.storage_path):
if file.endswith('.json'):
sessions.append(file[:-5]) # 移除.json后缀
return sessions
# 集成存储功能
conversation_store = ConversationStore()
# 在主循环中添加加载功能
def load_previous_session():
"""加载之前的对话会话"""
sessions = conversation_store.list_conversations()
if sessions:
print("\n可加载的对话会话:")
for i, session in enumerate(sessions[-5:]): # 显示最近5个
print(f"{i+1}. {session}")
choice = input("\n选择要加载的会话编号 (直接回车创建新会话): ")
if choice.isdigit() and 1 <= int(choice) <= len(sessions[-5:]):
selected_session = sessions[-5:][int(choice)-1]
return conversation_store.load_conversation(selected_session)
return [
{"role": "system", "content": "你是一个资深的软件架构师和技术专家。"}
]
# 更新主函数
if __name__ == "__main__":
print("=== 专业技术对话系统(支持50+轮长对话)===")
# 加载或创建新会话
conversation_history = load_previous_session()
current_session_id = None
while True:
user_input = input("\nYou: ").strip()
# 处理特殊命令
if user_input.lower() in ['exit', 'quit']:
current_session_id = conversation_store.save_conversation(conversation_history, current_session_id)
print(f"对话已保存!会话ID: {current_session_id}")
break
elif user_input == '/save':
current_session_id = conversation_store.save_conversation(conversation_history, current_session_id)
print(f"对话已手动保存!会话ID: {current_session_id}")
continue
elif user_input == '/new':
current_session_id = conversation_store.save_conversation(conversation_history, current_session_id)
conversation_history = [
{"role": "system", "content": "你是一个资深的软件架构师和技术专家。"}
]
print("已创建新对话会话!")
continue
elif user_input == '/list':
sessions = conversation_store.list_conversations()
print("\n所有对话会话:")
for session in sessions[-10:]:
print(f" - {session}")
continue
# 正常对话处理
reply, conversation_history = enhanced_chat_with_gpt(user_input, conversation_history)
print(f"\nAssistant: {reply}")
print(f"当前对话轮数: {(len(conversation_history)-1)//2}")
```
### 🎯 第四步:部署与优化建议
1. **环境配置**
```bash
# 创建虚拟环境
python -m venv chat_env
source chat_env/bin/activate # Linux/Mac
# 或 chat_env\Scripts\activate # Windows
# 安装依赖
pip install openai python-dotenv
```
2. **安全配置**
* 创建 `.env` 文件存储API密钥:
```
OPENAI_API_KEY=你的实际API密钥
```
* 在代码中通过 `os.getenv('OPENAI_API_KEY')` 读取
3. **性能调优参数**
```python
# 在ContextManager初始化时调整这些参数
context_manager = ContextManager(
max_rounds=75, # 根据需求调整
compression_ratio=0.7, # 压缩强度
)
```
### 💡 使用工作流建议
1. **日常使用**:运行脚本,选择加载最近会话或创建新会话
2. **重要讨论**:使用 `/save` 手动保存关键节点
3. **主题切换**:使用 `/new` 开始全新主题讨论
4. **历史追溯**:使用 `/list` 查看所有历史会话
这个方案让你完全掌控对话上下文,不再受网页界面限制,同时保持了对话的连贯性和重要信息的持久化。
需要进一步优化某个特定功能,我们可以继续深入讨论!