当前位置：首页 > news >正文

Qwen2.5-7B-Instruct保姆级教程：Streamlit中实现7B对话历史持久化到SQLite数据库

news 2026/7/7 23:15:53

Qwen2.5-7B-Instruct保姆级教程：Streamlit中实现7B对话历史持久化到SQLite数据库

1. 项目简介

今天给大家带来一个超级实用的教程：如何在Streamlit中为Qwen2.5-7B-Instruct大模型添加对话历史持久化功能。如果你之前已经体验过这个7B参数规模的旗舰模型，一定会被它的强大能力所折服——无论是逻辑推理、长文本创作，还是复杂代码编写，都远超那些轻量级模型。

但有一个痛点：每次重启服务，之前的对话记录就全没了！这对于需要连续讨论复杂问题的专业场景来说，简直是灾难。别担心，通过本教程，你将学会如何使用SQLite数据库永久保存对话历史，让AI助手真正记住你们的每一次交流。

为什么选择SQLite？

轻量级：单个文件数据库，无需安装复杂数据库服务
零配置：开箱即用，适合本地化部署场景
高性能：读写速度快，完全满足对话记录需求
易备份：整个数据库就是一个文件，备份迁移超级简单

2. 环境准备与快速部署

2.1 安装所需依赖

首先确保你已经安装了基础环境，然后添加SQLite相关的Python库：

pip install streamlit sqlalchemy datasets

SQLite是Python内置的数据库，不需要额外安装。我们使用SQLAlchemy来操作数据库，这样代码更加规范和专业。

2.2 创建数据库模型

在项目根目录创建一个新的Python文件database.py，用来定义数据表结构：

from sqlalchemy import create_engine, Column, Integer, String, Text, DateTime from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker from datetime import datetime import os # 创建数据库引擎（SQLite文件将保存在当前目录） engine = create_engine('sqlite:///chat_history.db', echo=False) Base = declarative_base() SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) class Conversation(Base): __tablename__ = "conversations" id = Column(Integer, primary_key=True, index=True) session_id = Column(String(50), index=True) # 会话ID，用于区分不同对话会话 role = Column(String(10)) # 角色：user或assistant message = Column(Text) # 消息内容 timestamp = Column(DateTime, default=datetime.now) # 时间戳 def __repr__(self): return f"<Conversation({self.role}: {self.message[:50]}...)>" # 创建数据表 def init_db(): Base.metadata.create_all(bind=engine) # 获取数据库会话 def get_db(): db = SessionLocal() try: yield db finally: db.close() # 初始化数据库 init_db()

3. 数据库集成到Streamlit应用

3.1 修改主程序文件

现在我们来修改主要的Streamlit应用文件，添加数据库功能：

import streamlit as st from transformers import AutoModelForCausalLM, AutoTokenizer import torch from database import get_db, Conversation from sqlalchemy.orm import Session import uuid # 生成唯一的会话ID if 'session_id' not in st.session_state: st.session_state.session_id = str(uuid.uuid4()) # 初始化模型和分词器（使用缓存） @st.cache_resource def load_model(): model_name = "Qwen/Qwen2.5-7B-Instruct" tokenizer = AutoTokenizer.from_pretrained(model_name) model = AutoModelForCausalLM.from_pretrained( model_name, device_map="auto", torch_dtype="auto" ) return model, tokenizer # 保存对话到数据库 def save_conversation(db: Session, role: str, message: str): conversation = Conversation( session_id=st.session_state.session_id, role=role, message=message ) db.add(conversation) db.commit() # 加载历史对话 def load_conversation_history(db: Session): conversations = db.query(Conversation).filter( Conversation.session_id == st.session_state.session_id ).order_by(Conversation.timestamp.asc()).all() history = [] for conv in conversations: history.append({"role": conv.role, "content": conv.message}) return history # 清空当前会话历史 def clear_current_session_history(db: Session): db.query(Conversation).filter( Conversation.session_id == st.session_state.session_id ).delete() db.commit() st.session_state.messages = [] st.success("当前会话历史已清空！") # 初始化消息历史 if "messages" not in st.session_state: st.session_state.messages = [] # 页面设置 st.set_page_config(page_title="Qwen2.5-7B智能对话", layout="wide") st.title("Qwen2.5-7B-Instruct 智能对话系统") st.caption("支持对话历史持久化保存 - 您的专业AI助手") # 侧边栏设置 with st.sidebar: st.header("⚙️ 控制台") temperature = st.slider("温度（创造力）", 0.1, 1.0, 0.7, 0.1) max_length = st.slider("最大回复长度", 512, 4096, 2048, 128) if st.button("🧹 强制清理显存"): torch.cuda.empty_cache() st.success("显存已清理！") if st.button("🗑️ 清空当前会话历史"): with next(get_db()) as db: clear_current_session_history(db) # 显示数据库信息 st.divider() st.subheader("💾 数据库信息") with next(get_db()) as db: total_messages = db.query(Conversation).count() session_messages = db.query(Conversation).filter( Conversation.session_id == st.session_state.session_id ).count() st.write(f"总消息数: {total_messages}") st.write(f"当前会话消息数: {session_messages}") # 加载模型 model, tokenizer = load_model() # 加载历史消息 if not st.session_state.messages: with next(get_db()) as db: st.session_state.messages = load_conversation_history(db) # 显示历史消息 for message in st.session_state.messages: with st.chat_message(message["role"]): st.markdown(message["content"]) # 用户输入处理 if prompt := st.chat_input("请输入您的问题或指令..."): # 显示用户消息 with st.chat_message("user"): st.markdown(prompt) # 保存用户消息到数据库 with next(get_db()) as db: save_conversation(db, "user", prompt) st.session_state.messages.append({"role": "user", "content": prompt}) # 准备模型输入 with st.chat_message("assistant"): message_placeholder = st.empty() message_placeholder.markdown("7B大脑正在高速运转...") # 构建对话历史 conversation_history = [] for msg in st.session_state.messages: conversation_history.append({"role": msg["role"], "content": msg["content"]}) # 生成回复 try: inputs = tokenizer.apply_chat_template( conversation_history, tokenize=True, add_generation_prompt=True, return_tensors="pt" ).to(model.device) generated_ids = model.generate( inputs, max_new_tokens=max_length, temperature=temperature, do_sample=True, pad_token_id=tokenizer.eos_token_id ) response = tokenizer.decode(generated_ids[0][len(inputs[0]):], skip_special_tokens=True) message_placeholder.markdown(response) # 保存助手回复到数据库 with next(get_db()) as db: save_conversation(db, "assistant", response) st.session_state.messages.append({"role": "assistant", "content": response}) except RuntimeError as e: if "CUDA out of memory" in str(e): error_msg = "💥 显存爆了！(OOM)\n\n建议操作：\n1. 点击「强制清理显存」\n2. 缩短输入文字\n3. 减少最大回复长度\n4. 考虑使用轻量版模型" message_placeholder.markdown(error_msg) else: message_placeholder.markdown(f"发生错误：{str(e)}")

4. 高级功能：会话管理

4.1 多会话支持

在实际使用中，你可能需要同时管理多个对话会话。让我们添加会话管理功能：

# 在侧边栏添加会话管理功能 with st.sidebar: st.divider() st.subheader("💬 会话管理") # 创建新会话 if st.button("🆕 创建新会话"): st.session_state.session_id = str(uuid.uuid4()) st.session_state.messages = [] st.rerun() # 显示当前会话ID st.write(f"当前会话ID: `{st.session_state.session_id[:8]}...`") # 会话历史浏览 st.divider() st.subheader("📋 会话历史") with next(get_db()) as db: # 获取所有会话列表 sessions = db.query(Conversation.session_id).distinct().all() session_list = [session[0] for session in sessions] selected_session = st.selectbox("选择会话", session_list, index=session_list.index(st.session_state.session_id) if st.session_state.session_id in session_list else 0) if selected_session != st.session_state.session_id: st.session_state.session_id = selected_session st.session_state.messages = load_conversation_history(db) st.rerun()

4.2 数据导出功能

添加数据导出功能，方便备份或分析对话记录：

# 添加导出功能到侧边栏 with st.sidebar: st.divider() st.subheader("📤 数据导出") if st.button("💾 导出当前会话"): with next(get_db()) as db: conversations = db.query(Conversation).filter( Conversation.session_id == st.session_state.session_id ).order_by(Conversation.timestamp.asc()).all() export_text = f"# 会话导出: {st.session_state.session_id}\n\n" for conv in conversations: export_text += f"**{conv.role}** ({conv.timestamp.strftime('%Y-%m-%d %H:%M:%S')}):\n{conv.message}\n\n" st.download_button( label="下载对话记录", data=export_text, file_name=f"conversation_{st.session_state.session_id[:8]}.md", mime="text/markdown" )

5. 实际使用效果

现在让我们启动服务，体验一下完整的对话历史持久化功能：

streamlit run your_app.py

使用流程：

首次启动时，系统会自动创建SQLite数据库文件chat_history.db
进行多轮对话，所有消息都会自动保存到数据库
即使重启服务，之前的对话记录仍然完整保留
可以通过会话管理功能切换不同的对话场景
随时导出对话记录进行备份或分析

实际测试效果：

对话记录保存完整，包含时间戳和角色信息
重启服务后历史对话完美恢复
多会话管理功能让不同主题对话互不干扰
导出功能生成规范的Markdown格式文件

6. 常见问题与解决方案

6.1 数据库文件过大

如果对话记录很多，数据库文件可能会变得很大。可以添加自动清理功能：

# 添加自动清理功能 def cleanup_old_conversations(db: Session, days_to_keep=30): from datetime import timedelta cutoff_date = datetime.now() - timedelta(days=days_to_keep) old_conversations = db.query(Conversation).filter( Conversation.timestamp < cutoff_date ).delete() db.commit() return old_conversations # 在适当的地方调用清理功能

6.2 性能优化建议

对于大量对话记录的情况，可以考虑以下优化：

# 添加索引提升查询性能 class Conversation(Base): # ... 其他字段不变 __table_args__ = ( Index('ix_session_timestamp', 'session_id', 'timestamp'), )

6.3 数据库备份

定期备份数据库文件：

import shutil from datetime import datetime def backup_database(): backup_name = f"chat_history_backup_{datetime.now().strftime('%Y%m%d_%H%M%S')}.db" shutil.copy2('chat_history.db', backup_name) return backup_name