Nanbeige4.1-3B Chainlit多会话管理支持并行对话、上下文隔离与历史回溯1. 引言你有没有遇到过这样的场景当你正在和AI助手讨论一个复杂的编程问题时突然想让它帮你翻译一段英文文档。如果直接在当前对话里提问AI很可能会把之前的代码上下文和翻译请求混在一起给出一个莫名其妙的回答。这就是传统单会话AI对话界面的痛点——所有对话都挤在同一个上下文里不同话题相互干扰想回溯之前的讨论也困难重重。今天我要介绍的这套方案完美解决了这个问题。我们基于Nanbeige4.1-3B这个强大的开源模型结合Chainlit前端框架实现了一个真正的多会话管理系统。你可以同时开启多个独立的对话窗口每个窗口都有自己的上下文和历史记录互不干扰还能随时回溯查看之前的对话内容。2. 为什么需要多会话管理在深入技术细节之前我们先来聊聊为什么多会话管理这么重要。2.1 传统单会话的局限性想象一下你正在用AI助手写代码你问“Python里怎么实现快速排序”AI回答了一段代码你又问“这段代码的时间复杂度是多少”AI分析了刚才的代码到这里都很正常。但如果你突然问“帮我翻译一下‘Hello World’”AI可能会一脸懵——它以为你在问排序算法的“Hello World”实现。2.2 多会话带来的好处有了多会话管理情况就完全不同了并行处理左边窗口写代码右边窗口查资料中间窗口翻译文档互不影响上下文隔离每个会话只看到自己的对话历史不会“串台”历史回溯随时可以回到之前的会话查看当时的讨论内容主题聚焦为不同的项目或任务创建专门的会话保持思路清晰2.3 我们的解决方案概览我们这套方案的核心是后端使用vLLM高效部署Nanbeige4.1-3B模型前端用Chainlit构建交互界面核心功能在Chainlit基础上实现了多会话管理、上下文隔离和历史回溯接下来我会带你一步步搭建这个系统并展示它的强大功能。3. 环境准备与快速部署3.1 系统要求在开始之前确保你的环境满足以下要求Python 3.8或更高版本至少8GB可用内存运行模型需要支持CUDA的GPU可选但推荐使用以获得更好性能基本的命令行操作知识3.2 一键部署脚本为了简化部署过程我准备了一个完整的部署脚本。创建一个名为deploy_nanbeige.sh的文件内容如下#!/bin/bash echo 开始部署Nanbeige4.1-3B多会话管理系统... # 创建项目目录 mkdir -p nanbeige-multisession cd nanbeige-multisession echo 1. 创建Python虚拟环境... python3 -m venv venv source venv/bin/activate echo 2. 安装依赖包... pip install torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cu118 pip install vllm pip install chainlit pip install fastapi uvicorn echo 3. 下载部署脚本... cat deploy_model.py EOF from vllm import LLM, SamplingParams import uvicorn from fastapi import FastAPI, HTTPException from pydantic import BaseModel from typing import List, Dict, Optional import asyncio import uuid from datetime import datetime # 初始化模型 print(正在加载Nanbeige4.1-3B模型...) llm LLM( modelNanbeige/Nanbeige4.1-3B, tensor_parallel_size1, gpu_memory_utilization0.8, max_model_len4096 ) # 会话管理类 class SessionManager: def __init__(self): self.sessions: Dict[str, Dict] {} self.session_histories: Dict[str, List[Dict]] {} def create_session(self, session_id: Optional[str] None) - str: 创建新会话 if session_id is None: session_id str(uuid.uuid4()) self.sessions[session_id] { id: session_id, created_at: datetime.now(), last_active: datetime.now(), context: [] } self.session_histories[session_id] [] return session_id def add_to_history(self, session_id: str, role: str, content: str): 添加对话到历史记录 if session_id not in self.session_histories: self.session_histories[session_id] [] entry { role: role, content: content, timestamp: datetime.now() } self.session_histories[session_id].append(entry) def get_session_history(self, session_id: str, limit: int 50) - List[Dict]: 获取会话历史记录 if session_id not in self.session_histories: return [] return self.session_histories[session_id][-limit:] def clear_session_context(self, session_id: str): 清空会话上下文但保留历史记录 if session_id in self.sessions: self.sessions[session_id][context] [] def delete_session(self, session_id: str): 删除会话 if session_id in self.sessions: del self.sessions[session_id] if session_id in self.session_histories: del self.session_histories[session_id] # 初始化会话管理器 session_manager SessionManager() # 创建FastAPI应用 app FastAPI(titleNanbeige4.1-3B多会话API) class ChatRequest(BaseModel): message: str session_id: Optional[str] None temperature: float 0.7 max_tokens: int 512 class ChatResponse(BaseModel): response: str session_id: str history: List[Dict] app.post(/chat) async def chat(request: ChatRequest): 处理聊天请求 try: # 获取或创建会话 if request.session_id is None or request.session_id not in session_manager.sessions: session_id session_manager.create_session(request.session_id) else: session_id request.session_id # 更新最后活跃时间 session_manager.sessions[session_id][last_active] datetime.now() # 构建提示词 session_context session_manager.sessions[session_id][context] context_messages \n.join([f{msg[role]}: {msg[content]} for msg in session_context[-10:]]) prompt f以下是对话历史 {context_messages} 用户{request.message} 助手 # 生成回复 sampling_params SamplingParams( temperaturerequest.temperature, max_tokensrequest.max_tokens, stop[用户, 助手, \n\n] ) outputs llm.generate([prompt], sampling_params) response outputs[0].outputs[0].text.strip() # 更新会话上下文和历史记录 session_manager.sessions[session_id][context].append({ role: user, content: request.message }) session_manager.sessions[session_id][context].append({ role: assistant, content: response }) # 添加到历史记录 session_manager.add_to_history(session_id, user, request.message) session_manager.add_to_history(session_id, assistant, response) # 获取最近的历史记录 recent_history session_manager.get_session_history(session_id, limit10) return ChatResponse( responseresponse, session_idsession_id, historyrecent_history ) except Exception as e: raise HTTPException(status_code500, detailstr(e)) app.get(/sessions) async def list_sessions(): 列出所有活跃会话 sessions [] for session_id, session_data in session_manager.sessions.items(): sessions.append({ id: session_id, created_at: session_data[created_at], last_active: session_data[last_active], message_count: len(session_manager.get_session_history(session_id)) }) return {sessions: sessions} app.get(/history/{session_id}) async def get_history(session_id: str, limit: int 50): 获取指定会话的历史记录 if session_id not in session_manager.session_histories: raise HTTPException(status_code404, detailSession not found) history session_manager.get_session_history(session_id, limit) return {session_id: session_id, history: history} app.delete(/session/{session_id}) async def delete_session(session_id: str): 删除指定会话 if session_id not in session_manager.sessions: raise HTTPException(status_code404, detailSession not found) session_manager.delete_session(session_id) return {message: fSession {session_id} deleted} if __name__ __main__: print(模型加载完成) print(API服务启动中...) uvicorn.run(app, host0.0.0.0, port8000) EOF echo 4. 创建Chainlit前端应用... cat chainlit_app.py EOF import chainlit as cl import aiohttp import asyncio from typing import Dict, List import json # API配置 API_URL http://localhost:8000 class MultiSessionManager: def __init__(self): self.active_sessions: Dict[str, Dict] {} self.current_session_id: str None async def create_session(self, session_name: str None) - str: 创建新会话 async with aiohttp.ClientSession() as session: async with session.post( f{API_URL}/chat, json{message: Hello, session_id: None} ) as response: if response.status 200: data await response.json() session_id data[session_id] self.active_sessions[session_id] { name: session_name or f会话_{len(self.active_sessions) 1}, created_at: cl.user_session.get(current_time), message_count: 0 } if self.current_session_id is None: self.current_session_id session_id return session_id else: raise Exception(f创建会话失败: {response.status}) async def switch_session(self, session_id: str): 切换到指定会话 if session_id in self.active_sessions: self.current_session_id session_id # 加载该会话的历史记录 await self.load_session_history(session_id) return True return False async def load_session_history(self, session_id: str): 加载会话历史记录 async with aiohttp.ClientSession() as session: async with session.get(f{API_URL}/history/{session_id}) as response: if response.status 200: data await response.json() history data.get(history, []) # 清空当前聊天界面 await cl.Message(content).send() # 重新显示历史消息 for msg in history: if msg[role] user: await cl.Message( contentmsg[content], author用户 ).send() else: await cl.Message( contentmsg[content], author助手 ).send() async def delete_session(self, session_id: str): 删除会话 async with aiohttp.ClientSession() as session: async with session.delete(f{API_URL}/session/{session_id}) as response: if response.status 200: if session_id in self.active_sessions: del self.active_sessions[session_id] if self.current_session_id session_id: self.current_session_id None if self.active_sessions: self.current_session_id next(iter(self.active_sessions.keys())) return True return False def get_session_list(self) - List[Dict]: 获取会话列表 sessions [] for session_id, session_data in self.active_sessions.items(): sessions.append({ id: session_id, name: session_data[name], is_current: session_id self.current_session_id, message_count: session_data[message_count] }) return sessions # 全局会话管理器 session_manager MultiSessionManager() cl.on_chat_start async def on_chat_start(): 聊天开始时的初始化 # 设置用户会话数据 cl.user_session.set(current_time, cl.Timestamp()) # 创建第一个会话 first_session_id await session_manager.create_session(主会话) # 显示欢迎消息 welcome_msg # Nanbeige4.1-3B 多会话管理系统 欢迎使用多会话AI助手你可以 ## 主要功能 1. **创建多个独立会话** - 不同话题互不干扰 2. **并行对话** - 同时处理多个任务 3. **上下文隔离** - 每个会话有自己的记忆 4. **历史回溯** - 随时查看之前的对话 ## 使用方法 - 点击左下角的新建会话按钮创建新窗口 - 使用会话列表切换不同对话 - 每个会话都是完全独立的 现在开始你的第一个对话吧 await cl.Message(contentwelcome_msg).send() # 显示会话管理侧边栏 await update_session_sidebar() cl.on_message async def on_message(message: cl.Message): 处理用户消息 if session_manager.current_session_id is None: await cl.Message(content请先创建一个会话).send() return # 显示用户消息 user_msg cl.Message(contentmessage.content, author用户) await user_msg.send() # 调用API获取回复 async with aiohttp.ClientSession() as session: try: async with session.post( f{API_URL}/chat, json{ message: message.content, session_id: session_manager.current_session_id, temperature: 0.7, max_tokens: 512 }, timeout30 ) as response: if response.status 200: data await response.json() response_text data[response] # 更新会话消息计数 if session_manager.current_session_id in session_manager.active_sessions: session_manager.active_sessions[session_manager.current_session_id][message_count] 1 # 显示助手回复 assistant_msg cl.Message(contentresponse_text, author助手) await assistant_msg.send() # 更新侧边栏 await update_session_sidebar() else: error_text await response.text() await cl.Message(contentf请求失败: {error_text}).send() except asyncio.TimeoutError: await cl.Message(content请求超时请稍后重试).send() except Exception as e: await cl.Message(contentf发生错误: {str(e)}).send() async def update_session_sidebar(): 更新会话管理侧边栏 sessions session_manager.get_session_list() # 创建会话管理界面 session_elements [] # 添加新建会话按钮 session_elements.append( cl.Button( namenew_session, label➕ 新建会话, variantprimary ) ) session_elements.append(cl.Divider()) # 添加会话列表 for session in sessions: session_label f{session[name]} if session[is_current]: session_label f {session_label} session_elements.append( cl.Button( namefswitch_{session[id]}, labelsession_label, variantsecondary if not session[is_current] else primary ) ) # 添加删除按钮非当前会话 if not session[is_current]: session_elements.append( cl.Button( namefdelete_{session[id]}, label️, sizesm, variantghost ) ) # 发送到侧边栏 await cl.Sidebar( title会话管理, contentsession_elements ).send() cl.on_button_click async def on_button_click(button: cl.Button): 处理按钮点击事件 if button.name new_session: # 创建新会话 session_id await session_manager.create_session() await session_manager.switch_session(session_id) await cl.Message(contentf已创建并切换到新会话).send() await update_session_sidebar() elif button.name.startswith(switch_): # 切换会话 session_id button.name.replace(switch_, ) if await session_manager.switch_session(session_id): await cl.Message(contentf已切换到会话: {session_manager.active_sessions[session_id][name]}).send() await update_session_sidebar() elif button.name.startswith(delete_): # 删除会话 session_id button.name.replace(delete_, ) session_name session_manager.active_sessions.get(session_id, {}).get(name, 未知会话) # 确认删除 confirm await cl.AskActionMessage( contentf确定要删除会话 {session_name} 吗, actions[ cl.Action(nameconfirm_delete, label确认删除, valuesession_id), cl.Action(namecancel_delete, label取消, valuecancel) ] ).send() if confirm and confirm.get(name) confirm_delete: if await session_manager.delete_session(session_id): await cl.Message(contentf已删除会话: {session_name}).send() await update_session_sidebar() if __name__ __main__: # 启动Chainlit应用 cl.run() EOF echo 5. 创建启动脚本... cat start_all.sh EOF #!/bin/bash echo 启动Nanbeige4.1-3B多会话管理系统... # 激活虚拟环境 source venv/bin/activate # 启动模型API服务后台运行 echo 启动模型API服务... python deploy_model.py model_api.log 21 MODEL_PID$! echo 等待模型服务启动... sleep 15 # 启动Chainlit前端 echo 启动Chainlit前端... chainlit run chainlit_app.py -w --port 7860 # 清理当Chainlit退出时也停止模型服务 kill $MODEL_PID 2/dev/null echo 服务已停止 EOF chmod x start_all.sh chmod x deploy_nanbeige.sh echo 部署完成 echo echo 使用方法 echo 1. 运行部署脚本./deploy_nanbeige.sh echo 2. 启动服务./start_all.sh echo 3. 打开浏览器访问http://localhost:7860 echo echo 注意首次运行需要下载模型可能需要一些时间请耐心等待。3.3 一键启动系统保存脚本后只需要几个简单的命令# 给脚本添加执行权限 chmod x deploy_nanbeige.sh # 运行部署脚本 ./deploy_nanbeige.sh # 启动所有服务 ./start_all.sh脚本会自动完成以下工作创建Python虚拟环境安装所有必要的依赖包下载并部署Nanbeige4.1-3B模型启动后端API服务启动Chainlit前端界面启动成功后打开浏览器访问http://localhost:7860就能看到多会话管理界面了。4. 多会话功能详解现在系统已经运行起来了让我们深入看看它的核心功能。4.1 创建和管理多个会话当你第一次打开界面时系统会自动创建一个主会话。但真正的强大之处在于你可以随时创建新的独立会话。创建新会话的几种方式点击新建会话按钮在侧边栏直接点击系统会自动创建一个新窗口使用快捷键我们可以在代码中添加快捷键支持比如CtrlN通过命令在聊天框输入/new_session 会话名称来创建每个新会话都会获得一个唯一的ID并且有自己独立的对话上下文历史记录界面状态4.2 并行对话实践让我们来看一个实际的使用场景。假设你同时需要处理三个任务任务1编程问题用户帮我写一个Python函数计算斐波那契数列 助手def fibonacci(n): if n 0: return [] elif n 1: return [0] elif n 2: return [0, 1] fib [0, 1] for i in range(2, n): fib.append(fib[i-1] fib[i-2]) return fib任务2翻译工作用户翻译这段英文Artificial Intelligence is transforming every industry 助手人工智能正在改变每一个行业任务3学习咨询用户机器学习中的过拟合是什么意思 助手过拟合是指模型在训练数据上表现很好但在新数据上表现差的现象...在传统单会话系统中这些对话会混在一起导致上下文混乱。但在我们的多会话系统中你可以在编程会话中讨论代码切换到翻译会话处理文档在学习会话中咨询概念问题随时切回任何一个会话继续之前的对话4.3 上下文隔离机制多会话的核心是上下文隔离。让我们看看这是如何实现的# 这是简化后的会话管理逻辑 class SessionManager: def __init__(self): # 每个会话有独立的上下文存储 self.session_contexts { session_1: [消息1, 消息2, 消息3], session_2: [消息A, 消息B], session_3: [消息X, 消息Y, 消息Z] } def get_context(self, session_id): # 只返回当前会话的上下文 return self.session_contexts.get(session_id, []) def add_message(self, session_id, message): # 消息只添加到指定会话 if session_id not in self.session_contexts: self.session_contexts[session_id] [] self.session_contexts[session_id].append(message)这种设计确保了会话A看不到会话B的对话内容每个会话的上下文长度独立管理可以针对不同会话设置不同的上下文长度限制4.4 历史回溯功能历史回溯让你可以随时查看之前的对话记录。系统会自动保存所有会话的完整历史# 历史记录存储结构 session_history { session_1: [ {role: user, content: 你好, timestamp: 2024-01-01 10:00:00}, {role: assistant, content: 你好有什么可以帮助你的, timestamp: 2024-01-01 10:00:01}, # ... 更多历史消息 ], session_2: [ # 另一个会话的独立历史 ] }你可以通过多种方式查看历史在侧边栏点击会话名称自动加载该会话的历史记录使用历史查看命令输入/history 会话ID查看指定历史导出历史记录支持导出为JSON或文本格式5. 实际应用场景展示多会话管理不仅仅是一个技术特性它在实际工作中能带来实实在在的效率提升。5.1 场景一多项目并行开发假设你同时在开发两个项目项目A电商网站会话1讨论商品推荐算法会话2处理订单系统的数据库设计会话3编写用户认证代码项目B数据分析工具会话4讨论数据可视化方案会话5优化数据处理性能会话6设计用户界面传统方式下你需要在不同话题间不断切换每次都要重新解释上下文。现在每个项目、每个子任务都有独立的会话切换时上下文自动恢复。5.2 场景二学习与研究在学习新技术或进行研究时基础概念会话记录所有基础概念和定义代码实践会话保存实验代码和运行结果问题解决会话记录遇到的问题和解决方案参考资料会话整理相关文档和链接这样组织信息查找和回顾都变得非常方便。5.3 场景三团队协作虽然当前是个人使用但我们可以扩展为团队协作模式# 扩展的团队会话管理 class TeamSessionManager: def __init__(self): self.team_sessions { team_project_a: { design: [设计讨论记录...], development: [开发进度...], testing: [测试结果...] }, team_project_b: { # 另一个项目的会话组 } }6. 高级功能与定制6.1 会话模板功能对于经常重复的任务可以创建会话模板# 会话模板配置 session_templates { code_review: { system_prompt: 你是一个代码审查助手请仔细检查代码并提出改进建议。, temperature: 0.3, context_length: 2000 }, creative_writing: { system_prompt: 你是一个创意写作助手请帮助我进行故事创作。, temperature: 0.8, context_length: 4000 }, technical_support: { system_prompt: 你是一个技术支持专家请帮助解决技术问题。, temperature: 0.5, context_length: 3000 } }使用模板创建会话# 在聊天框输入 /new_template code_review 代码审查会话6.2 会话合并与拆分有时候你可能需要合并相关的会话或者将一个复杂的会话拆分成多个合并会话将两个相关会话的历史记录合并拆分会话将会话中的不同话题分离到新的会话中6.3 会话导出与导入所有会话都可以导出为标准格式方便备份或分享# 导出会话数据 def export_session(session_id, formatjson): history session_manager.get_session_history(session_id) metadata session_manager.sessions[session_id] export_data { metadata: metadata, history: history, export_time: datetime.now(), model_version: Nanbeige4.1-3B } if format json: return json.dumps(export_data, indent2, defaultstr) elif format txt: # 转换为文本格式 text_output f会话: {metadata[name]}\n text_output f创建时间: {metadata[created_at]}\n\n for msg in history: text_output f{msg[role]}: {msg[content]}\n\n return text_output7. 性能优化建议7.1 内存管理多会话系统会占用更多内存这里有一些优化建议# 智能上下文管理 class SmartContextManager: def __init__(self, max_sessions10, max_context_per_session20): self.max_sessions max_sessions self.max_context_per_session max_context_per_session def cleanup_old_sessions(self): 清理不活跃的会话 if len(self.sessions) self.max_sessions: # 按最后活跃时间排序清理最旧的会话 sorted_sessions sorted( self.sessions.items(), keylambda x: x[1][last_active] ) for session_id, _ in sorted_sessions[:len(self.sessions) - self.max_sessions]: self.delete_session(session_id) def trim_context(self, session_id): 修剪过长的上下文 if session_id in self.session_contexts: context self.session_contexts[session_id] if len(context) self.max_context_per_session: # 保留最近的对话但保留重要的系统提示 self.session_contexts[session_id] context[-self.max_context_per_session:]7.2 响应速度优化为了确保多会话下的响应速度异步处理使用异步IO处理多个会话的请求缓存机制缓存常用的模型输出连接池复用数据库和API连接7.3 扩展性考虑如果需要在生产环境部署可以考虑数据库存储将会话数据存储到数据库而不是内存分布式部署支持多节点部署负载均衡用户认证添加用户系统支持多用户使用8. 总结通过将Nanbeige4.1-3B与Chainlit结合我们构建了一个真正实用的多会话AI对话系统。这个系统解决了传统单会话AI助手的核心痛点让AI对话变得更加高效和有条理。核心价值总结真正的并行处理同时处理多个任务互不干扰智能上下文管理每个会话保持独立的对话记忆完整的历史回溯随时查看和恢复之前的对话灵活的场景适配为不同用途创建专门的会话实际使用建议项目开发为每个项目创建独立会话保持代码讨论的连贯性学习研究按主题分类建立知识管理系统日常工作区分工作、学习、娱乐等不同场景团队协作虽然当前是个人版但架构支持扩展到团队使用下一步探索方向会话智能合并AI自动识别相关会话并建议合并跨会话知识共享在保护隐私的前提下允许会话间有限的知识传递移动端支持开发手机App随时随地管理会话插件系统支持第三方插件扩展功能这个多会话管理系统不仅提升了AI对话的实用性更为未来的AI助手发展提供了一个重要的方向——让AI更好地适应人类多任务、多场景的工作方式。获取更多AI镜像想探索更多AI镜像和应用场景访问 CSDN星图镜像广场提供丰富的预置镜像覆盖大模型推理、图像生成、视频生成、模型微调等多个领域支持一键部署。
Nanbeige4.1-3B Chainlit多会话管理：支持并行对话、上下文隔离与历史回溯
Nanbeige4.1-3B Chainlit多会话管理支持并行对话、上下文隔离与历史回溯1. 引言你有没有遇到过这样的场景当你正在和AI助手讨论一个复杂的编程问题时突然想让它帮你翻译一段英文文档。如果直接在当前对话里提问AI很可能会把之前的代码上下文和翻译请求混在一起给出一个莫名其妙的回答。这就是传统单会话AI对话界面的痛点——所有对话都挤在同一个上下文里不同话题相互干扰想回溯之前的讨论也困难重重。今天我要介绍的这套方案完美解决了这个问题。我们基于Nanbeige4.1-3B这个强大的开源模型结合Chainlit前端框架实现了一个真正的多会话管理系统。你可以同时开启多个独立的对话窗口每个窗口都有自己的上下文和历史记录互不干扰还能随时回溯查看之前的对话内容。2. 为什么需要多会话管理在深入技术细节之前我们先来聊聊为什么多会话管理这么重要。2.1 传统单会话的局限性想象一下你正在用AI助手写代码你问“Python里怎么实现快速排序”AI回答了一段代码你又问“这段代码的时间复杂度是多少”AI分析了刚才的代码到这里都很正常。但如果你突然问“帮我翻译一下‘Hello World’”AI可能会一脸懵——它以为你在问排序算法的“Hello World”实现。2.2 多会话带来的好处有了多会话管理情况就完全不同了并行处理左边窗口写代码右边窗口查资料中间窗口翻译文档互不影响上下文隔离每个会话只看到自己的对话历史不会“串台”历史回溯随时可以回到之前的会话查看当时的讨论内容主题聚焦为不同的项目或任务创建专门的会话保持思路清晰2.3 我们的解决方案概览我们这套方案的核心是后端使用vLLM高效部署Nanbeige4.1-3B模型前端用Chainlit构建交互界面核心功能在Chainlit基础上实现了多会话管理、上下文隔离和历史回溯接下来我会带你一步步搭建这个系统并展示它的强大功能。3. 环境准备与快速部署3.1 系统要求在开始之前确保你的环境满足以下要求Python 3.8或更高版本至少8GB可用内存运行模型需要支持CUDA的GPU可选但推荐使用以获得更好性能基本的命令行操作知识3.2 一键部署脚本为了简化部署过程我准备了一个完整的部署脚本。创建一个名为deploy_nanbeige.sh的文件内容如下#!/bin/bash echo 开始部署Nanbeige4.1-3B多会话管理系统... # 创建项目目录 mkdir -p nanbeige-multisession cd nanbeige-multisession echo 1. 创建Python虚拟环境... python3 -m venv venv source venv/bin/activate echo 2. 安装依赖包... pip install torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cu118 pip install vllm pip install chainlit pip install fastapi uvicorn echo 3. 下载部署脚本... cat deploy_model.py EOF from vllm import LLM, SamplingParams import uvicorn from fastapi import FastAPI, HTTPException from pydantic import BaseModel from typing import List, Dict, Optional import asyncio import uuid from datetime import datetime # 初始化模型 print(正在加载Nanbeige4.1-3B模型...) llm LLM( modelNanbeige/Nanbeige4.1-3B, tensor_parallel_size1, gpu_memory_utilization0.8, max_model_len4096 ) # 会话管理类 class SessionManager: def __init__(self): self.sessions: Dict[str, Dict] {} self.session_histories: Dict[str, List[Dict]] {} def create_session(self, session_id: Optional[str] None) - str: 创建新会话 if session_id is None: session_id str(uuid.uuid4()) self.sessions[session_id] { id: session_id, created_at: datetime.now(), last_active: datetime.now(), context: [] } self.session_histories[session_id] [] return session_id def add_to_history(self, session_id: str, role: str, content: str): 添加对话到历史记录 if session_id not in self.session_histories: self.session_histories[session_id] [] entry { role: role, content: content, timestamp: datetime.now() } self.session_histories[session_id].append(entry) def get_session_history(self, session_id: str, limit: int 50) - List[Dict]: 获取会话历史记录 if session_id not in self.session_histories: return [] return self.session_histories[session_id][-limit:] def clear_session_context(self, session_id: str): 清空会话上下文但保留历史记录 if session_id in self.sessions: self.sessions[session_id][context] [] def delete_session(self, session_id: str): 删除会话 if session_id in self.sessions: del self.sessions[session_id] if session_id in self.session_histories: del self.session_histories[session_id] # 初始化会话管理器 session_manager SessionManager() # 创建FastAPI应用 app FastAPI(titleNanbeige4.1-3B多会话API) class ChatRequest(BaseModel): message: str session_id: Optional[str] None temperature: float 0.7 max_tokens: int 512 class ChatResponse(BaseModel): response: str session_id: str history: List[Dict] app.post(/chat) async def chat(request: ChatRequest): 处理聊天请求 try: # 获取或创建会话 if request.session_id is None or request.session_id not in session_manager.sessions: session_id session_manager.create_session(request.session_id) else: session_id request.session_id # 更新最后活跃时间 session_manager.sessions[session_id][last_active] datetime.now() # 构建提示词 session_context session_manager.sessions[session_id][context] context_messages \n.join([f{msg[role]}: {msg[content]} for msg in session_context[-10:]]) prompt f以下是对话历史 {context_messages} 用户{request.message} 助手 # 生成回复 sampling_params SamplingParams( temperaturerequest.temperature, max_tokensrequest.max_tokens, stop[用户, 助手, \n\n] ) outputs llm.generate([prompt], sampling_params) response outputs[0].outputs[0].text.strip() # 更新会话上下文和历史记录 session_manager.sessions[session_id][context].append({ role: user, content: request.message }) session_manager.sessions[session_id][context].append({ role: assistant, content: response }) # 添加到历史记录 session_manager.add_to_history(session_id, user, request.message) session_manager.add_to_history(session_id, assistant, response) # 获取最近的历史记录 recent_history session_manager.get_session_history(session_id, limit10) return ChatResponse( responseresponse, session_idsession_id, historyrecent_history ) except Exception as e: raise HTTPException(status_code500, detailstr(e)) app.get(/sessions) async def list_sessions(): 列出所有活跃会话 sessions [] for session_id, session_data in session_manager.sessions.items(): sessions.append({ id: session_id, created_at: session_data[created_at], last_active: session_data[last_active], message_count: len(session_manager.get_session_history(session_id)) }) return {sessions: sessions} app.get(/history/{session_id}) async def get_history(session_id: str, limit: int 50): 获取指定会话的历史记录 if session_id not in session_manager.session_histories: raise HTTPException(status_code404, detailSession not found) history session_manager.get_session_history(session_id, limit) return {session_id: session_id, history: history} app.delete(/session/{session_id}) async def delete_session(session_id: str): 删除指定会话 if session_id not in session_manager.sessions: raise HTTPException(status_code404, detailSession not found) session_manager.delete_session(session_id) return {message: fSession {session_id} deleted} if __name__ __main__: print(模型加载完成) print(API服务启动中...) uvicorn.run(app, host0.0.0.0, port8000) EOF echo 4. 创建Chainlit前端应用... cat chainlit_app.py EOF import chainlit as cl import aiohttp import asyncio from typing import Dict, List import json # API配置 API_URL http://localhost:8000 class MultiSessionManager: def __init__(self): self.active_sessions: Dict[str, Dict] {} self.current_session_id: str None async def create_session(self, session_name: str None) - str: 创建新会话 async with aiohttp.ClientSession() as session: async with session.post( f{API_URL}/chat, json{message: Hello, session_id: None} ) as response: if response.status 200: data await response.json() session_id data[session_id] self.active_sessions[session_id] { name: session_name or f会话_{len(self.active_sessions) 1}, created_at: cl.user_session.get(current_time), message_count: 0 } if self.current_session_id is None: self.current_session_id session_id return session_id else: raise Exception(f创建会话失败: {response.status}) async def switch_session(self, session_id: str): 切换到指定会话 if session_id in self.active_sessions: self.current_session_id session_id # 加载该会话的历史记录 await self.load_session_history(session_id) return True return False async def load_session_history(self, session_id: str): 加载会话历史记录 async with aiohttp.ClientSession() as session: async with session.get(f{API_URL}/history/{session_id}) as response: if response.status 200: data await response.json() history data.get(history, []) # 清空当前聊天界面 await cl.Message(content).send() # 重新显示历史消息 for msg in history: if msg[role] user: await cl.Message( contentmsg[content], author用户 ).send() else: await cl.Message( contentmsg[content], author助手 ).send() async def delete_session(self, session_id: str): 删除会话 async with aiohttp.ClientSession() as session: async with session.delete(f{API_URL}/session/{session_id}) as response: if response.status 200: if session_id in self.active_sessions: del self.active_sessions[session_id] if self.current_session_id session_id: self.current_session_id None if self.active_sessions: self.current_session_id next(iter(self.active_sessions.keys())) return True return False def get_session_list(self) - List[Dict]: 获取会话列表 sessions [] for session_id, session_data in self.active_sessions.items(): sessions.append({ id: session_id, name: session_data[name], is_current: session_id self.current_session_id, message_count: session_data[message_count] }) return sessions # 全局会话管理器 session_manager MultiSessionManager() cl.on_chat_start async def on_chat_start(): 聊天开始时的初始化 # 设置用户会话数据 cl.user_session.set(current_time, cl.Timestamp()) # 创建第一个会话 first_session_id await session_manager.create_session(主会话) # 显示欢迎消息 welcome_msg # Nanbeige4.1-3B 多会话管理系统 欢迎使用多会话AI助手你可以 ## 主要功能 1. **创建多个独立会话** - 不同话题互不干扰 2. **并行对话** - 同时处理多个任务 3. **上下文隔离** - 每个会话有自己的记忆 4. **历史回溯** - 随时查看之前的对话 ## 使用方法 - 点击左下角的新建会话按钮创建新窗口 - 使用会话列表切换不同对话 - 每个会话都是完全独立的 现在开始你的第一个对话吧 await cl.Message(contentwelcome_msg).send() # 显示会话管理侧边栏 await update_session_sidebar() cl.on_message async def on_message(message: cl.Message): 处理用户消息 if session_manager.current_session_id is None: await cl.Message(content请先创建一个会话).send() return # 显示用户消息 user_msg cl.Message(contentmessage.content, author用户) await user_msg.send() # 调用API获取回复 async with aiohttp.ClientSession() as session: try: async with session.post( f{API_URL}/chat, json{ message: message.content, session_id: session_manager.current_session_id, temperature: 0.7, max_tokens: 512 }, timeout30 ) as response: if response.status 200: data await response.json() response_text data[response] # 更新会话消息计数 if session_manager.current_session_id in session_manager.active_sessions: session_manager.active_sessions[session_manager.current_session_id][message_count] 1 # 显示助手回复 assistant_msg cl.Message(contentresponse_text, author助手) await assistant_msg.send() # 更新侧边栏 await update_session_sidebar() else: error_text await response.text() await cl.Message(contentf请求失败: {error_text}).send() except asyncio.TimeoutError: await cl.Message(content请求超时请稍后重试).send() except Exception as e: await cl.Message(contentf发生错误: {str(e)}).send() async def update_session_sidebar(): 更新会话管理侧边栏 sessions session_manager.get_session_list() # 创建会话管理界面 session_elements [] # 添加新建会话按钮 session_elements.append( cl.Button( namenew_session, label➕ 新建会话, variantprimary ) ) session_elements.append(cl.Divider()) # 添加会话列表 for session in sessions: session_label f{session[name]} if session[is_current]: session_label f {session_label} session_elements.append( cl.Button( namefswitch_{session[id]}, labelsession_label, variantsecondary if not session[is_current] else primary ) ) # 添加删除按钮非当前会话 if not session[is_current]: session_elements.append( cl.Button( namefdelete_{session[id]}, label️, sizesm, variantghost ) ) # 发送到侧边栏 await cl.Sidebar( title会话管理, contentsession_elements ).send() cl.on_button_click async def on_button_click(button: cl.Button): 处理按钮点击事件 if button.name new_session: # 创建新会话 session_id await session_manager.create_session() await session_manager.switch_session(session_id) await cl.Message(contentf已创建并切换到新会话).send() await update_session_sidebar() elif button.name.startswith(switch_): # 切换会话 session_id button.name.replace(switch_, ) if await session_manager.switch_session(session_id): await cl.Message(contentf已切换到会话: {session_manager.active_sessions[session_id][name]}).send() await update_session_sidebar() elif button.name.startswith(delete_): # 删除会话 session_id button.name.replace(delete_, ) session_name session_manager.active_sessions.get(session_id, {}).get(name, 未知会话) # 确认删除 confirm await cl.AskActionMessage( contentf确定要删除会话 {session_name} 吗, actions[ cl.Action(nameconfirm_delete, label确认删除, valuesession_id), cl.Action(namecancel_delete, label取消, valuecancel) ] ).send() if confirm and confirm.get(name) confirm_delete: if await session_manager.delete_session(session_id): await cl.Message(contentf已删除会话: {session_name}).send() await update_session_sidebar() if __name__ __main__: # 启动Chainlit应用 cl.run() EOF echo 5. 创建启动脚本... cat start_all.sh EOF #!/bin/bash echo 启动Nanbeige4.1-3B多会话管理系统... # 激活虚拟环境 source venv/bin/activate # 启动模型API服务后台运行 echo 启动模型API服务... python deploy_model.py model_api.log 21 MODEL_PID$! echo 等待模型服务启动... sleep 15 # 启动Chainlit前端 echo 启动Chainlit前端... chainlit run chainlit_app.py -w --port 7860 # 清理当Chainlit退出时也停止模型服务 kill $MODEL_PID 2/dev/null echo 服务已停止 EOF chmod x start_all.sh chmod x deploy_nanbeige.sh echo 部署完成 echo echo 使用方法 echo 1. 运行部署脚本./deploy_nanbeige.sh echo 2. 启动服务./start_all.sh echo 3. 打开浏览器访问http://localhost:7860 echo echo 注意首次运行需要下载模型可能需要一些时间请耐心等待。3.3 一键启动系统保存脚本后只需要几个简单的命令# 给脚本添加执行权限 chmod x deploy_nanbeige.sh # 运行部署脚本 ./deploy_nanbeige.sh # 启动所有服务 ./start_all.sh脚本会自动完成以下工作创建Python虚拟环境安装所有必要的依赖包下载并部署Nanbeige4.1-3B模型启动后端API服务启动Chainlit前端界面启动成功后打开浏览器访问http://localhost:7860就能看到多会话管理界面了。4. 多会话功能详解现在系统已经运行起来了让我们深入看看它的核心功能。4.1 创建和管理多个会话当你第一次打开界面时系统会自动创建一个主会话。但真正的强大之处在于你可以随时创建新的独立会话。创建新会话的几种方式点击新建会话按钮在侧边栏直接点击系统会自动创建一个新窗口使用快捷键我们可以在代码中添加快捷键支持比如CtrlN通过命令在聊天框输入/new_session 会话名称来创建每个新会话都会获得一个唯一的ID并且有自己独立的对话上下文历史记录界面状态4.2 并行对话实践让我们来看一个实际的使用场景。假设你同时需要处理三个任务任务1编程问题用户帮我写一个Python函数计算斐波那契数列 助手def fibonacci(n): if n 0: return [] elif n 1: return [0] elif n 2: return [0, 1] fib [0, 1] for i in range(2, n): fib.append(fib[i-1] fib[i-2]) return fib任务2翻译工作用户翻译这段英文Artificial Intelligence is transforming every industry 助手人工智能正在改变每一个行业任务3学习咨询用户机器学习中的过拟合是什么意思 助手过拟合是指模型在训练数据上表现很好但在新数据上表现差的现象...在传统单会话系统中这些对话会混在一起导致上下文混乱。但在我们的多会话系统中你可以在编程会话中讨论代码切换到翻译会话处理文档在学习会话中咨询概念问题随时切回任何一个会话继续之前的对话4.3 上下文隔离机制多会话的核心是上下文隔离。让我们看看这是如何实现的# 这是简化后的会话管理逻辑 class SessionManager: def __init__(self): # 每个会话有独立的上下文存储 self.session_contexts { session_1: [消息1, 消息2, 消息3], session_2: [消息A, 消息B], session_3: [消息X, 消息Y, 消息Z] } def get_context(self, session_id): # 只返回当前会话的上下文 return self.session_contexts.get(session_id, []) def add_message(self, session_id, message): # 消息只添加到指定会话 if session_id not in self.session_contexts: self.session_contexts[session_id] [] self.session_contexts[session_id].append(message)这种设计确保了会话A看不到会话B的对话内容每个会话的上下文长度独立管理可以针对不同会话设置不同的上下文长度限制4.4 历史回溯功能历史回溯让你可以随时查看之前的对话记录。系统会自动保存所有会话的完整历史# 历史记录存储结构 session_history { session_1: [ {role: user, content: 你好, timestamp: 2024-01-01 10:00:00}, {role: assistant, content: 你好有什么可以帮助你的, timestamp: 2024-01-01 10:00:01}, # ... 更多历史消息 ], session_2: [ # 另一个会话的独立历史 ] }你可以通过多种方式查看历史在侧边栏点击会话名称自动加载该会话的历史记录使用历史查看命令输入/history 会话ID查看指定历史导出历史记录支持导出为JSON或文本格式5. 实际应用场景展示多会话管理不仅仅是一个技术特性它在实际工作中能带来实实在在的效率提升。5.1 场景一多项目并行开发假设你同时在开发两个项目项目A电商网站会话1讨论商品推荐算法会话2处理订单系统的数据库设计会话3编写用户认证代码项目B数据分析工具会话4讨论数据可视化方案会话5优化数据处理性能会话6设计用户界面传统方式下你需要在不同话题间不断切换每次都要重新解释上下文。现在每个项目、每个子任务都有独立的会话切换时上下文自动恢复。5.2 场景二学习与研究在学习新技术或进行研究时基础概念会话记录所有基础概念和定义代码实践会话保存实验代码和运行结果问题解决会话记录遇到的问题和解决方案参考资料会话整理相关文档和链接这样组织信息查找和回顾都变得非常方便。5.3 场景三团队协作虽然当前是个人使用但我们可以扩展为团队协作模式# 扩展的团队会话管理 class TeamSessionManager: def __init__(self): self.team_sessions { team_project_a: { design: [设计讨论记录...], development: [开发进度...], testing: [测试结果...] }, team_project_b: { # 另一个项目的会话组 } }6. 高级功能与定制6.1 会话模板功能对于经常重复的任务可以创建会话模板# 会话模板配置 session_templates { code_review: { system_prompt: 你是一个代码审查助手请仔细检查代码并提出改进建议。, temperature: 0.3, context_length: 2000 }, creative_writing: { system_prompt: 你是一个创意写作助手请帮助我进行故事创作。, temperature: 0.8, context_length: 4000 }, technical_support: { system_prompt: 你是一个技术支持专家请帮助解决技术问题。, temperature: 0.5, context_length: 3000 } }使用模板创建会话# 在聊天框输入 /new_template code_review 代码审查会话6.2 会话合并与拆分有时候你可能需要合并相关的会话或者将一个复杂的会话拆分成多个合并会话将两个相关会话的历史记录合并拆分会话将会话中的不同话题分离到新的会话中6.3 会话导出与导入所有会话都可以导出为标准格式方便备份或分享# 导出会话数据 def export_session(session_id, formatjson): history session_manager.get_session_history(session_id) metadata session_manager.sessions[session_id] export_data { metadata: metadata, history: history, export_time: datetime.now(), model_version: Nanbeige4.1-3B } if format json: return json.dumps(export_data, indent2, defaultstr) elif format txt: # 转换为文本格式 text_output f会话: {metadata[name]}\n text_output f创建时间: {metadata[created_at]}\n\n for msg in history: text_output f{msg[role]}: {msg[content]}\n\n return text_output7. 性能优化建议7.1 内存管理多会话系统会占用更多内存这里有一些优化建议# 智能上下文管理 class SmartContextManager: def __init__(self, max_sessions10, max_context_per_session20): self.max_sessions max_sessions self.max_context_per_session max_context_per_session def cleanup_old_sessions(self): 清理不活跃的会话 if len(self.sessions) self.max_sessions: # 按最后活跃时间排序清理最旧的会话 sorted_sessions sorted( self.sessions.items(), keylambda x: x[1][last_active] ) for session_id, _ in sorted_sessions[:len(self.sessions) - self.max_sessions]: self.delete_session(session_id) def trim_context(self, session_id): 修剪过长的上下文 if session_id in self.session_contexts: context self.session_contexts[session_id] if len(context) self.max_context_per_session: # 保留最近的对话但保留重要的系统提示 self.session_contexts[session_id] context[-self.max_context_per_session:]7.2 响应速度优化为了确保多会话下的响应速度异步处理使用异步IO处理多个会话的请求缓存机制缓存常用的模型输出连接池复用数据库和API连接7.3 扩展性考虑如果需要在生产环境部署可以考虑数据库存储将会话数据存储到数据库而不是内存分布式部署支持多节点部署负载均衡用户认证添加用户系统支持多用户使用8. 总结通过将Nanbeige4.1-3B与Chainlit结合我们构建了一个真正实用的多会话AI对话系统。这个系统解决了传统单会话AI助手的核心痛点让AI对话变得更加高效和有条理。核心价值总结真正的并行处理同时处理多个任务互不干扰智能上下文管理每个会话保持独立的对话记忆完整的历史回溯随时查看和恢复之前的对话灵活的场景适配为不同用途创建专门的会话实际使用建议项目开发为每个项目创建独立会话保持代码讨论的连贯性学习研究按主题分类建立知识管理系统日常工作区分工作、学习、娱乐等不同场景团队协作虽然当前是个人版但架构支持扩展到团队使用下一步探索方向会话智能合并AI自动识别相关会话并建议合并跨会话知识共享在保护隐私的前提下允许会话间有限的知识传递移动端支持开发手机App随时随地管理会话插件系统支持第三方插件扩展功能这个多会话管理系统不仅提升了AI对话的实用性更为未来的AI助手发展提供了一个重要的方向——让AI更好地适应人类多任务、多场景的工作方式。获取更多AI镜像想探索更多AI镜像和应用场景访问 CSDN星图镜像广场提供丰富的预置镜像覆盖大模型推理、图像生成、视频生成、模型微调等多个领域支持一键部署。
