Phi-3-mini-128k-instruct实战教程Chainlit集成数据库记录用户对话历史1. 教程目标与价值你是不是也遇到过这样的问题用Chainlit搭建了一个漂亮的AI对话前端和Phi-3-mini模型聊得挺开心但聊完之后对话记录就没了。想回顾一下刚才问了什么、模型怎么回答的或者想分析一下用户的使用习惯却发现数据无处可寻。今天这个教程就是来解决这个痛点的。我将手把手教你如何给基于vLLM部署的Phi-3-mini-128k-instruct模型和Chainlit前端加上一个数据库功能把每一次对话都完整地记录下来。学完这篇教程你将掌握如何快速搭建一个带对话历史记录的AI应用使用SQLite数据库存储用户对话的完整流程在Chainlit应用中实现对话的持久化存储一个可以立即上手的完整代码方案这个方案特别适合个人开发者想为自己的AI应用添加历史记录功能小团队需要分析用户与AI的交互数据任何希望保留对话记录以备后续查阅的场景2. 环境准备与快速部署2.1 确认基础环境在开始之前确保你的Phi-3-mini-128k-instruct模型已经通过vLLM成功部署并且Chainlit前端可以正常调用。如果你还没有完成这一步可以先用下面的命令检查一下# 检查模型服务是否正常运行 cat /root/workspace/llm.log如果看到模型加载成功的日志信息说明基础环境已经就绪。接下来我们需要为这个系统添加数据库功能。2.2 安装必要的Python包我们需要安装几个额外的Python包来实现数据库功能。打开终端执行以下命令pip install sqlalchemy chainlit这里简单解释一下这两个包的作用sqlalchemy一个强大的Python SQL工具包让我们可以用Python代码来操作数据库不用写复杂的SQL语句chainlit我们已经在用的前端框架这里确保版本兼容安装完成后我们可以开始搭建数据库部分了。3. 数据库设计与实现3.1 设计对话记录表首先我们来设计一下数据库表的结构。我们需要记录哪些信息呢一般来说一次完整的对话应该包括对话的基本信息什么时候开始的、谁发起的用户的问题用户具体问了什么模型的回答AI是怎么回答的一些额外信息比如对话的上下文关系基于这些需求我设计了一个简单的表结构。创建一个名为database.py的文件# database.py from sqlalchemy import create_engine, Column, Integer, String, Text, DateTime, ForeignKey from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker, relationship from datetime import datetime import os # 创建基类 Base declarative_base() class Conversation(Base): 对话会话表 - 记录每一次对话会话 __tablename__ conversations id Column(Integer, primary_keyTrue) session_id Column(String(100), uniqueTrue, nullableFalse) created_at Column(DateTime, defaultdatetime.utcnow) updated_at Column(DateTime, defaultdatetime.utcnow, onupdatedatetime.utcnow) # 建立与消息表的关系 messages relationship(Message, back_populatesconversation, cascadeall, delete-orphan) class Message(Base): 消息表 - 记录每一条具体的消息 __tablename__ messages id Column(Integer, primary_keyTrue) conversation_id Column(Integer, ForeignKey(conversations.id), nullableFalse) role Column(String(20), nullableFalse) # user 或 assistant content Column(Text, nullableFalse) created_at Column(DateTime, defaultdatetime.utcnow) # 建立与会话表的关系 conversation relationship(Conversation, back_populatesmessages) # 数据库初始化函数 def init_database(): 初始化数据库创建所有表 # 使用SQLite数据库文件名为chat_history.db database_url sqlite:///chat_history.db engine create_engine(database_url) # 创建所有表 Base.metadata.create_all(engine) # 创建会话工厂 SessionLocal sessionmaker(autocommitFalse, autoflushFalse, bindengine) return SessionLocal # 测试数据库连接 if __name__ __main__: SessionLocal init_database() print(✅ 数据库初始化成功) print( 数据库文件: chat_history.db)这个设计有两个表conversations表记录每一次对话会话比如你打开Chainlit开始聊天这就是一个会话messages表记录会话中的每一条消息包括你的提问和AI的回答3.2 数据库操作工具类有了表结构我们还需要一些工具函数来操作数据库。在同一个文件中添加以下代码# 继续在 database.py 中添加 class DatabaseManager: 数据库管理类封装常用的数据库操作 def __init__(self, session_local): self.SessionLocal session_local def create_conversation(self, session_id): 创建新的对话会话 db self.SessionLocal() try: conversation Conversation(session_idsession_id) db.add(conversation) db.commit() db.refresh(conversation) return conversation except Exception as e: db.rollback() print(f创建会话失败: {e}) return None finally: db.close() def add_message(self, conversation_id, role, content): 添加消息到指定会话 db self.SessionLocal() try: message Message( conversation_idconversation_id, rolerole, contentcontent ) db.add(message) db.commit() db.refresh(message) return message except Exception as e: db.rollback() print(f添加消息失败: {e}) return None finally: db.close() def get_conversation_messages(self, conversation_id, limit50): 获取指定会话的消息历史 db self.SessionLocal() try: messages db.query(Message).filter( Message.conversation_id conversation_id ).order_by(Message.created_at.asc()).limit(limit).all() # 格式化为Chainlit需要的格式 formatted_messages [] for msg in messages: formatted_messages.append({ role: msg.role, content: msg.content }) return formatted_messages finally: db.close() def get_conversation_by_session_id(self, session_id): 根据会话ID获取对话 db self.SessionLocal() try: conversation db.query(Conversation).filter( Conversation.session_id session_id ).first() return conversation finally: db.close() def get_all_conversations(self, limit100): 获取所有对话会话用于管理界面 db self.SessionLocal() try: conversations db.query(Conversation).order_by( Conversation.updated_at.desc() ).limit(limit).all() return conversations finally: db.close() # 全局数据库管理器实例 _session_local None _db_manager None def get_db_manager(): 获取数据库管理器单例 global _session_local, _db_manager if _session_local is None: _session_local init_database() _db_manager DatabaseManager(_session_local) return _db_manager这些函数让我们可以创建新的对话会话保存用户和AI的消息读取历史对话记录管理所有的对话会话4. 集成Chainlit与数据库4.1 修改Chainlit应用现在我们需要修改Chainlit应用让它能够使用我们刚刚创建的数据库。创建一个新的文件app_with_db.py# app_with_db.py import chainlit as cl from database import get_db_manager import aiohttp import json import asyncio # 初始化数据库管理器 db_manager get_db_manager() # vLLM API配置根据你的实际部署修改 VLLM_API_URL http://localhost:8000/v1/completions # 你的vLLM服务地址 MODEL_NAME Phi-3-mini-128k-instruct async def call_phi3_model(prompt, max_tokens500): 调用Phi-3-mini模型 headers { Content-Type: application/json } data { model: MODEL_NAME, prompt: prompt, max_tokens: max_tokens, temperature: 0.7, top_p: 0.9 } try: async with aiohttp.ClientSession() as session: async with session.post(VLLM_API_URL, headersheaders, jsondata) as response: if response.status 200: result await response.json() return result[choices][0][text].strip() else: return f模型调用失败: {response.status} except Exception as e: return f请求出错: {str(e)} cl.on_chat_start async def on_chat_start(): 聊天开始时的初始化 # 获取当前会话ID session_id cl.user_session.get(id) # 检查是否已有对话记录 conversation db_manager.get_conversation_by_session_id(session_id) if conversation: # 如果有历史记录加载并显示 messages db_manager.get_conversation_messages(conversation.id) # 发送欢迎消息并提示有历史记录 await cl.Message( contentf欢迎回来我们之前有{len(messages)}条对话记录。 ).send() # 将历史记录添加到会话中可选用于上下文 for msg in messages: if msg[role] user: cl.user_session.set(history, messages) else: # 创建新的对话记录 conversation db_manager.create_conversation(session_id) await cl.Message( content 你好我是基于Phi-3-mini模型的AI助手。我们的对话将被保存方便你随时回顾。 ).send() # 保存对话ID到用户会话 cl.user_session.set(conversation_id, conversation.id) cl.on_message async def on_message(message: cl.Message): 处理用户消息 # 获取当前对话ID conversation_id cl.user_session.get(conversation_id) if not conversation_id: # 如果没有对话ID创建新的 session_id cl.user_session.get(id) conversation db_manager.create_conversation(session_id) conversation_id conversation.id cl.user_session.set(conversation_id, conversation_id) # 1. 保存用户消息到数据库 db_manager.add_message(conversation_id, user, message.content) # 2. 准备提示词可以加入历史上下文 history cl.user_session.get(history, []) # 构建包含历史上下文的提示词 prompt build_prompt_with_history(message.content, history) # 3. 调用模型 msg cl.Message(content) await msg.send() # 模拟流式响应 response_text await call_phi3_model(prompt) # 逐步显示响应 for i in range(0, len(response_text), 20): await msg.stream_token(response_text[i:i20]) await asyncio.sleep(0.01) # 4. 保存AI响应到数据库 db_manager.add_message(conversation_id, assistant, response_text) # 5. 更新历史记录 if history: history.append({role: user, content: message.content}) history.append({role: assistant, content: response_text}) # 保持历史记录长度最近10轮对话 if len(history) 20: history history[-20:] else: history [ {role: user, content: message.content}, {role: assistant, content: response_text} ] cl.user_session.set(history, history) def build_prompt_with_history(user_input, history, max_history5): 构建包含历史上下文的提示词 prompt 你是一个有帮助的AI助手。请根据对话历史回答用户的问题。\n\n # 添加历史对话最近的几轮 if history: recent_history history[-(max_history*2):] if len(history) max_history*2 else history for msg in recent_history: if msg[role] user: prompt f用户: {msg[content]}\n else: prompt f助手: {msg[content]}\n prompt f\n用户: {user_input}\n助手: return prompt cl.on_chat_end async def on_chat_end(): 聊天结束时的清理工作 conversation_id cl.user_session.get(conversation_id) if conversation_id: print(f对话 {conversation_id} 已结束) # 这里可以添加更多的清理逻辑如统计对话时长等4.2 添加对话历史查看功能为了让用户能够查看自己的对话历史我们还可以添加一个简单的查询功能。在同一个文件中继续添加# 继续在 app_with_db.py 中添加 cl.password_auth_callback def auth_callback(username, password): 简单的身份验证可选 # 这里可以实现更复杂的认证逻辑 if username admin and password admin123: return cl.User(identifieradmin) return None cl.action_callback(view_history) async def on_view_history(action): 查看对话历史 conversation_id cl.user_session.get(conversation_id) if conversation_id: messages db_manager.get_conversation_messages(conversation_id) history_text ## 对话历史记录\n\n for i, msg in enumerate(messages, 1): role_icon if msg[role] user else history_text f**{i}. {role_icon} {msg[role].title()}:**\n history_text f{msg[content]}\n\n await cl.Message(contenthistory_text).send() else: await cl.Message(content暂无对话历史记录。).send() cl.action_callback(clear_history) async def on_clear_history(action): 清空当前对话历史仅从界面清除数据库保留 cl.user_session.set(history, []) await cl.Message(content当前会话的历史记录已清空。数据库中的记录仍然保留。).send() # 添加上下文菜单 cl.set_starters async def set_starters(): 设置聊天启动器 return [ cl.Starter( label查看对话历史, message我想看看我们之前的对话记录, icon/public/history.svg ), cl.Starter( label清空当前历史, message清空当前的对话上下文, icon/public/clear.svg ), ]5. 运行与测试5.1 启动应用现在让我们启动这个增强版的Chainlit应用。在终端中运行chainlit run app_with_db.py -w-w参数表示自动重新加载这样你修改代码后不需要手动重启服务。5.2 测试对话记录功能应用启动后打开浏览器访问Chainlit界面通常是http://localhost:8000。现在你可以开始新的对话输入一些问题比如介绍一下Phi-3-mini模型的特点查看历史记录在聊天界面中你应该能看到查看历史记录的选项测试持久化关闭浏览器重新打开应用看看之前的对话是否还在检查数据库使用SQLite浏览器工具打开chat_history.db文件查看数据是否正常存储5.3 验证数据存储为了确认数据确实保存到了数据库我们可以写一个简单的查询脚本# check_database.py from database import get_db_manager def check_database_content(): 检查数据库内容 db_manager get_db_manager() # 获取所有对话 conversations db_manager.get_all_conversations() print(f 数据库中有 {len(conversations)} 个对话会话) print( * 50) for i, conv in enumerate(conversations, 1): print(f\n对话 #{i}:) print(f 会话ID: {conv.session_id}) print(f 创建时间: {conv.created_at}) print(f 最后更新: {conv.updated_at}) # 获取该对话的所有消息 messages db_manager.get_conversation_messages(conv.id) print(f 消息数量: {len(messages)}) # 显示最近3条消息 for msg in messages[-3:]: role 用户 if msg[role] user else AI助手 print(f {role}: {msg[content][:50]}...) if __name__ __main__: check_database_content()运行这个脚本你会看到类似这样的输出 数据库中有 3 个对话会话 对话 #1: 会话ID: abc123 创建时间: 2024-01-15 10:30:00 最后更新: 2024-01-15 10:35:00 消息数量: 8 用户: 介绍一下Phi-3-mini模型的特点... AI助手: Phi-3-mini是一个38亿参数的轻量级模型... 用户: 它支持多长的上下文... 对话 #2: 会话ID: def456 创建时间: 2024-01-15 11:00:00 最后更新: 2024-01-15 11:05:00 消息数量: 4 ...6. 进阶功能与优化建议6.1 添加对话导出功能你可能希望用户能够导出他们的对话记录。我们可以添加一个导出功能# 在 app_with_db.py 中添加 cl.action_callback(export_history) async def on_export_history(action): 导出对话历史为文本文件 conversation_id cl.user_session.get(conversation_id) if conversation_id: messages db_manager.get_conversation_messages(conversation_id) # 构建导出文本 export_text f对话导出 - {datetime.now().strftime(%Y-%m-%d %H:%M:%S)}\n export_text * 50 \n\n for msg in messages: timestamp msg.get(created_at, ).strftime(%H:%M:%S) if hasattr(msg.get(created_at), strftime) else role 用户 if msg[role] user else AI助手 export_text f[{timestamp}] {role}:\n export_text f{msg[content]}\n\n # 创建文件并发送给用户 elements [ cl.File( name对话记录.txt, contentexport_text.encode(utf-8), displayinline ) ] await cl.Message( content这是你的对话记录导出文件, elementselements ).send()6.2 实现对话搜索功能如果对话记录很多搜索功能就很有用了# 在 database.py 的 DatabaseManager 类中添加 def search_messages(self, keyword, limit50): 搜索包含关键词的消息 db self.SessionLocal() try: messages db.query(Message).filter( Message.content.ilike(f%{keyword}%) ).order_by(Message.created_at.desc()).limit(limit).all() results [] for msg in messages: # 获取对应的会话信息 conv db.query(Conversation).filter( Conversation.id msg.conversation_id ).first() results.append({ conversation_id: msg.conversation_id, session_id: conv.session_id if conv else 未知, role: msg.role, content: msg.content, created_at: msg.created_at }) return results finally: db.close()6.3 添加对话统计功能了解用户的使用情况也很有价值# 在 database.py 的 DatabaseManager 类中添加 def get_conversation_stats(self, days7): 获取对话统计信息 db self.SessionLocal() try: from datetime import datetime, timedelta # 计算时间范围 end_date datetime.utcnow() start_date end_date - timedelta(daysdays) # 统计对话数量 total_conversations db.query(Conversation).count() recent_conversations db.query(Conversation).filter( Conversation.created_at start_date ).count() # 统计消息数量 total_messages db.query(Message).count() user_messages db.query(Message).filter(Message.role user).count() assistant_messages db.query(Message).filter(Message.role assistant).count() # 统计平均对话长度 avg_messages_per_conversation total_messages / total_conversations if total_conversations 0 else 0 return { total_conversations: total_conversations, recent_conversations: recent_conversations, total_messages: total_messages, user_messages: user_messages, assistant_messages: assistant_messages, avg_messages_per_conversation: round(avg_messages_per_conversation, 2) } finally: db.close()6.4 性能优化建议随着数据量增加你可能需要考虑一些优化数据库索引优化# 在表定义中添加索引 from sqlalchemy import Index class Message(Base): __tablename__ messages # ... 原有字段定义 ... # 添加索引 __table_args__ ( Index(idx_conversation_id, conversation_id), Index(idx_created_at, created_at), Index(idx_role, role), )分页查询当历史记录很多时不要一次性加载所有记录定期清理可以设置自动清理过期的对话记录使用更高效的数据库如果数据量很大可以考虑使用PostgreSQL或MySQL7. 总结回顾通过这个教程我们成功地为Phi-3-mini-128k-instruct模型和Chainlit前端添加了完整的对话历史记录功能。让我们回顾一下实现的关键点7.1 实现的核心功能数据库设计我们设计了两个表来存储对话数据一个记录会话信息一个记录具体的消息内容。这样的设计既清晰又灵活方便后续扩展。Chainlit集成通过修改Chainlit的回调函数我们在用户发送消息和接收AI回复时自动将数据保存到数据库。整个过程对用户是透明的体验很流畅。历史记录查看添加了查看历史记录的功能用户可以随时回顾之前的对话这对于学习、工作记录都很有帮助。数据持久化使用SQLite数据库数据会持久化保存即使重启应用之前的对话记录也不会丢失。7.2 实际应用价值这个方案的实际价值体现在几个方面对于个人用户可以回顾之前的对话避免重复提问重要的对话内容可以导出保存学习AI的使用技巧通过历史记录分析什么样的提问能得到更好的回答对于开发者可以分析用户的使用习惯优化AI应用收集对话数据用于模型改进为多轮对话提供上下文支持对于团队协作共享对话记录方便知识传递分析常见问题优化客服或支持流程7.3 下一步学习建议如果你对这个方案感兴趣还可以进一步探索添加用户系统为不同的用户创建独立的对话历史实现对话分享让用户可以分享有趣的对话给其他人添加标签功能给对话打上标签方便分类和检索集成向量数据库使用向量搜索实现更智能的对话检索添加分析仪表板可视化展示对话数据的统计信息7.4 遇到的常见问题在实现过程中你可能会遇到一些问题这里提供一些解决思路问题1数据库文件权限错误解决方案确保应用有写入当前目录的权限问题2Chainlit无法连接到数据库解决方案检查数据库文件路径确保在正确的目录下问题3历史记录加载缓慢解决方案实现分页加载不要一次性加载所有记录问题4对话上下文太长解决方案限制历史记录的长度只保留最近几轮对话7.5 最后的建议这个方案是一个起点你可以根据自己的需求进行调整和扩展。比如如果你需要更高的并发性能可以考虑使用PostgreSQL如果你需要存储文件或图片可以扩展消息表的结构如果你需要实时同步可以添加WebSocket支持最重要的是这个方案让你掌握了将AI应用与数据库集成的核心思路。无论你将来使用什么模型、什么前端框架这个思路都是通用的。现在你的Phi-3-mini应用不仅能够智能对话还能记住每一次交流这大大提升了它的实用价值。快去试试吧看看你的AI助手如何随着对话历史的积累变得越来越懂你获取更多AI镜像想探索更多AI镜像和应用场景访问 CSDN星图镜像广场提供丰富的预置镜像覆盖大模型推理、图像生成、视频生成、模型微调等多个领域支持一键部署。