Qwen3-0.6B-FP8代码实例：自定义Chainlit前端样式、添加历史会话与流式响应支持

Qwen3-0.6B-FP8代码实例自定义Chainlit前端样式、添加历史会话与流式响应支持你是不是已经用vLLM部署好了Qwen3-0.6B-FP8模型也通过Chainlit的默认界面和它聊过天了但总觉得那个界面太简单功能也不够用比如每次对话都是孤立的想回顾一下刚才聊了什么得自己翻记录模型生成回复时只能干等着看不到一个字一个字蹦出来的效果界面样式千篇一律想换个颜色、改个布局也不知道从何下手。如果你有这些想法那今天这篇文章就是为你准备的。我将带你一步步改造Chainlit前端实现三个实用功能自定义界面样式换个主题色调整布局让前端更符合你的审美或品牌需求。添加历史会话管理保存和加载之前的对话让AI记住“上下文”。支持流式响应让模型回复像真人打字一样逐字显示体验更流畅。我们会从最基础的Chainlit调用代码开始逐步添加这些功能。即使你之前没怎么接触过Chainlit跟着做也能搞定。1. 环境准备与基础代码回顾在开始魔改之前我们先确保环境是通的并回顾一下最基础的调用方式。1.1 确认模型服务状态按照你提供的说明模型应该已经用vLLM部署好了。我们可以通过查看日志来确认# 在你的部署环境中执行 cat /root/workspace/llm.log如果看到模型加载成功的日志信息通常包含模型名称、加载的GPU信息等就说明服务端准备好了。我们的Chainlit应用将通过HTTP请求与这个服务端通信。1.2 最基础的Chainlit调用脚本我们先创建一个最简单的app.py文件这是Chainlit应用的入口。这个脚本只做一件事向vLLM服务发送请求并获取回复。# app.py - 基础版本 import chainlit as cl import aiohttp import json import asyncio # 配置你的vLLM服务地址和端口 VLLM_SERVER_URL http://localhost:8000/v1/completions # 请根据你的实际部署地址修改 cl.on_message async def main(message: cl.Message): 处理用户消息的核心函数 # 1. 准备发送给vLLM的请求数据 payload { model: Qwen3-0.6B-FP8, # 模型名称需与vLLM加载的模型名一致 prompt: message.content, # 用户输入的问题 max_tokens: 512, # 生成的最大token数 temperature: 0.7, # 温度参数控制随机性 stream: False # 非流式响应 } # 2. 显示“正在思考”的提示 msg cl.Message(content) await msg.send() try: # 3. 发送HTTP请求到vLLM服务 async with aiohttp.ClientSession() as session: async with session.post( VLLM_SERVER_URL, jsonpayload, headers{Content-Type: application/json} ) as response: if response.status 200: # 4. 解析响应并显示结果 result await response.json() answer result[choices][0][text] # 5. 更新消息内容 msg.content answer await msg.update() else: error_text await response.text() msg.content f请求失败状态码{response.status}\n错误信息{error_text} await msg.update() except Exception as e: # 6. 异常处理 msg.content f请求过程中发生错误{str(e)} await msg.update() # Chainlit应用配置 if __name__ __main__: # 运行Chainlit应用 # 在终端执行: chainlit run app.py pass这个基础版本已经能工作了。在终端进入脚本所在目录执行chainlit run app.py就会在默认的8000端口如果没被占用启动一个Web服务。打开浏览器访问提示的地址通常是http://localhost:8000就能看到Chainlit的聊天界面输入问题就能得到Qwen3模型的回复。但正如开头所说这个界面太基础了。接下来我们开始给它“升级”。2. 自定义Chainlit前端样式Chainlit允许我们通过CSS来自定义界面样式。我们可以修改主题色、字体、布局等让前端看起来更专业或更有个性。2.1 创建自定义CSS文件首先在项目根目录创建一个名为chainlit.md的文件。这个文件是Chainlit的配置文件我们可以在其中指定自定义的CSS。# chainlit.md # 在这里可以配置应用的一些元信息比如名称、描述等 # 但最重要的是下面这个CSS配置 # 加载自定义CSS文件 CSS_PATH assets/custom.css接着创建assets文件夹并在里面创建custom.css文件。这个文件将包含我们所有的样式代码。2.2 编写自定义CSS代码打开assets/custom.css我们可以开始编写样式了。这里提供几个常见的自定义示例/* assets/custom.css */ /* 1. 修改整体主题色将默认的蓝色改为深紫色 */ :root { --primary: #6d28d9; /* 主色调 - 深紫色 */ --primary-dark: #5b21b6; /* 主色调深色 */ --primary-light: #8b5cf6; /* 主色调浅色 */ } /* 应用主色调到各种元素 */ .cl-button-primary { background-color: var(--primary) !important; border-color: var(--primary) !important; } .cl-button-primary:hover { background-color: var(--primary-dark) !important; } /* 2. 修改聊天消息气泡样式 */ /* 用户消息气泡 */ .cl-message-user { background-color: #f0f9ff; /* 更柔和的用户消息背景 */ border-color: #bae6fd; border-radius: 18px 18px 4px 18px; /* 圆角调整 */ } /* AI消息气泡 */ .cl-message-assistant { background-color: #fefce8; /* 温暖的AI消息背景 */ border-color: #fde68a; border-radius: 18px 18px 18px 4px; } /* 3. 修改输入框样式 */ .cl-input-container { border: 2px solid #e5e7eb; border-radius: 12px; transition: border-color 0.2s; } .cl-input-container:focus-within { border-color: var(--primary-light); box-shadow: 0 0 0 3px rgba(109, 40, 217, 0.1); } /* 4. 修改侧边栏样式 */ .cl-sidebar { background: linear-gradient(180deg, #f8fafc 0%, #f1f5f9 100%); border-right: 1px solid #e2e8f0; } /* 5. 添加自定义字体如果需要 */ import url(https://fonts.googleapis.com/css2?familyInter:wght300;400;500;600displayswap); body { font-family: Inter, -apple-system, BlinkMacSystemFont, sans-serif; } /* 6. 调整消息内容的字体和行高提升可读性 */ .cl-message-content { line-height: 1.6; font-size: 15px; } /* 7. 为代码块添加特殊样式 */ .cl-message-content pre { background-color: #1e293b; border-radius: 8px; padding: 16px; } .cl-message-content code { background-color: #f1f5f9; padding: 2px 6px; border-radius: 4px; font-size: 14px; }保存CSS文件后重新启动Chainlit应用chainlit run app.py刷新浏览器就能看到样式已经生效了。你可以根据自己的喜好调整颜色、间距、圆角等所有CSS属性。3. 添加历史会话管理功能现在我们的聊天是“健忘”的每次刷新页面之前的对话就没了。对于实际应用来说保存对话历史非常重要。Chainlit本身不提供持久化的会话管理但我们可以自己实现。3.1 设计会话管理方案我们将实现一个简单的基于文件的会话管理系统每个会话有一个唯一ID会话数据以JSON格式保存到本地文件在侧边栏显示历史会话列表可以创建新会话、加载旧会话、删除会话3.2 修改app.py实现会话管理我们需要大幅修改app.py添加会话管理相关的逻辑# app.py - 添加会话管理功能 import chainlit as cl import aiohttp import json import asyncio import uuid import os from datetime import datetime from typing import Dict, List, Optional # 配置 VLLM_SERVER_URL http://localhost:8000/v1/completions SESSIONS_DIR sessions # 会话保存目录 # 确保会话目录存在 os.makedirs(SESSIONS_DIR, exist_okTrue) class SessionManager: 会话管理器 def __init__(self): self.current_session_id None self.sessions self.load_all_sessions() def load_all_sessions(self) - Dict: 加载所有会话 sessions {} if os.path.exists(SESSIONS_DIR): for filename in os.listdir(SESSIONS_DIR): if filename.endswith(.json): session_id filename[:-5] # 去掉.json后缀 try: with open(os.path.join(SESSIONS_DIR, filename), r, encodingutf-8) as f: sessions[session_id] json.load(f) except: pass return sessions def create_new_session(self, title: str 新对话) - str: 创建新会话 session_id str(uuid.uuid4()) session_data { id: session_id, title: title, created_at: datetime.now().isoformat(), updated_at: datetime.now().isoformat(), messages: [] # 存储消息历史 } self.sessions[session_id] session_data self.current_session_id session_id self.save_session(session_id) return session_id def save_session(self, session_id: str): 保存会话到文件 if session_id in self.sessions: filepath os.path.join(SESSIONS_DIR, f{session_id}.json) with open(filepath, w, encodingutf-8) as f: json.dump(self.sessions[session_id], f, ensure_asciiFalse, indent2) def add_message_to_session(self, session_id: str, role: str, content: str): 向会话添加消息 if session_id in self.sessions: self.sessions[session_id][messages].append({ role: role, content: content, timestamp: datetime.now().isoformat() }) self.sessions[session_id][updated_at] datetime.now().isoformat() self.save_session(session_id) def get_session_messages(self, session_id: str) - List[Dict]: 获取会话的所有消息 if session_id in self.sessions: return self.sessions[session_id][messages] return [] def delete_session(self, session_id: str): 删除会话 if session_id in self.sessions: # 删除文件 filepath os.path.join(SESSIONS_DIR, f{session_id}.json) if os.path.exists(filepath): os.remove(filepath) # 从内存中删除 del self.sessions[session_id] # 如果删除的是当前会话清空当前会话ID if self.current_session_id session_id: self.current_session_id None # 全局会话管理器实例 session_manager SessionManager() cl.on_chat_start async def on_chat_start(): 聊天开始时执行设置侧边栏 # 创建默认会话 if not session_manager.current_session_id: session_manager.create_new_session(初始对话) # 在侧边栏添加会话管理控件 await setup_sidebar() async def setup_sidebar(): 设置侧边栏内容 # 创建新会话按钮 new_session_btn cl.Button( namenew_session, label 新对话, variantprimary ) # 会话列表 sessions_list [] for session_id, session_data in session_manager.sessions.items(): # 截断过长的标题 title session_data[title] if len(title) 20: title title[:17] ... # 格式化时间 updated_at datetime.fromisoformat(session_data[updated_at]) time_str updated_at.strftime(%m-%d %H:%M) sessions_list.append( cl.Button( namefload_session_{session_id}, labelf{title} ({time_str}), variantsecondary ) ) # 将控件添加到侧边栏 await cl.Sidebar( cl.Title(会话管理), new_session_btn, cl.Divider(), cl.Title(历史会话), *sessions_list ).send() cl.on_button_click async def on_button_click(button: cl.Button): 处理按钮点击事件 if button.name new_session: # 创建新会话 session_id session_manager.create_new_session(新对话) # 清空当前聊天界面 await cl.Message( content已创建新会话开始聊天吧, author系统 ).send() # 刷新侧边栏 await setup_sidebar() elif button.name.startswith(load_session_): # 加载历史会话 session_id button.name.replace(load_session_, ) if session_id in session_manager.sessions: session_manager.current_session_id session_id # 清空当前界面 await cl.Message( contentf已加载会话{session_manager.sessions[session_id][title]}, author系统 ).send() # 加载历史消息 messages session_manager.get_session_messages(session_id) for msg in messages: if msg[role] user: await cl.Message( contentmsg[content], author用户 ).send() else: await cl.Message( contentmsg[content], author助手 ).send() # 刷新侧边栏 await setup_sidebar() cl.on_message async def main(message: cl.Message): 处理用户消息 # 确保有当前会话 if not session_manager.current_session_id: session_manager.create_new_session(自动创建对话) current_session_id session_manager.current_session_id # 保存用户消息到会话 session_manager.add_message_to_session( current_session_id, user, message.content ) # 更新会话标题如果这是第一条消息 if len(session_manager.get_session_messages(current_session_id)) 1: # 用第一条消息的前20个字符作为标题 title message.content[:20] (... if len(message.content) 20 else ) session_manager.sessions[current_session_id][title] title session_manager.save_session(current_session_id) await setup_sidebar() # 刷新侧边栏显示新标题 # 显示正在思考提示 msg cl.Message(content) await msg.send() try: # 准备请求数据 payload { model: Qwen3-0.6B-FP8, prompt: message.content, max_tokens: 512, temperature: 0.7, stream: False } # 发送请求 async with aiohttp.ClientSession() as session: async with session.post( VLLM_SERVER_URL, jsonpayload, headers{Content-Type: application/json} ) as response: if response.status 200: result await response.json() answer result[choices][0][text] # 更新消息 msg.content answer await msg.update() # 保存AI回复到会话 session_manager.add_message_to_session( current_session_id, assistant, answer ) else: error_text await response.text() msg.content f请求失败{response.status}\n{error_text} await msg.update() except Exception as e: msg.content f错误{str(e)} await msg.update() if __name__ __main__: # 运行应用 pass现在重启Chainlit应用你会看到侧边栏多了会话管理功能。你可以创建新对话、加载历史对话所有对话都会自动保存到sessions文件夹中。4. 实现流式响应支持流式响应能让用户体验大幅提升看着文字一个个出现而不是等待全部生成完。vLLM支持流式响应我们只需要修改Chainlit端的处理方式。4.1 修改请求为流式模式关键改动是将stream参数设为True然后逐块处理响应数据。4.2 更新app.py支持流式响应我们修改main函数让它支持流式响应# app.py - 添加流式响应功能修改cl.on_message部分 cl.on_message async def main(message: cl.Message): 处理用户消息支持流式响应 # 确保有当前会话 if not session_manager.current_session_id: session_manager.create_new_session(自动创建对话) current_session_id session_manager.current_session_id # 保存用户消息到会话 session_manager.add_message_to_session( current_session_id, user, message.content ) # 更新会话标题如果这是第一条消息 if len(session_manager.get_session_messages(current_session_id)) 1: title message.content[:20] (... if len(message.content) 20 else ) session_manager.sessions[current_session_id][title] title session_manager.save_session(current_session_id) await setup_sidebar() # 创建消息对象用于流式更新 msg cl.Message(content) await msg.send() # 收集完整的回复内容用于保存到历史 full_response try: # 准备请求数据 - 注意stream设为True payload { model: Qwen3-0.6B-FP8, prompt: message.content, max_tokens: 512, temperature: 0.7, stream: True # 启用流式响应 } # 发送流式请求 async with aiohttp.ClientSession() as session: async with session.post( VLLM_SERVER_URL.replace(/completions, /completions), # vLLM流式端点 jsonpayload, headers{Content-Type: application/json} ) as response: if response.status 200: # 逐块读取流式响应 buffer async for chunk in response.content: if chunk: chunk_str chunk.decode(utf-8) buffer chunk_str # 处理可能的多行数据 lines buffer.split(\n) for line in lines[:-1]: # 处理完整的行 line line.strip() if line.startswith(data: ): data_str line[6:] # 去掉data: 前缀 if data_str [DONE]: continue try: data json.loads(data_str) if choices in data and len(data[choices]) 0: token data[choices][0][text] if token: full_response token # 流式更新消息内容 await msg.stream_token(token) except json.JSONDecodeError: continue buffer lines[-1] # 保留不完整的行 # 流式传输完成 await msg.update() # 保存完整的AI回复到会话历史 session_manager.add_message_to_session( current_session_id, assistant, full_response ) else: error_text await response.text() msg.content f请求失败{response.status}\n{error_text} await msg.update() except Exception as e: msg.content f错误{str(e)} await msg.update()现在重启应用再次提问时你会看到回复是一个字一个字出现的就像真人打字一样。这种体验比等待全部生成完再显示要好得多。5. 完整代码与使用建议5.1 项目文件结构完成所有功能后你的项目目录应该类似这样your_project/ ├── app.py # 主应用文件 ├── chainlit.md # Chainlit配置文件 ├── assets/ │ └── custom.css # 自定义CSS样式 ├── sessions/ # 自动创建的会话存储目录 │ ├── session1.json │ └── session2.json └── requirements.txt # 依赖文件可选5.2 requirements.txt 内容如果你需要分享项目或部署到其他环境可以创建requirements.txtchainlit1.0.0 aiohttp3.9.0安装依赖pip install -r requirements.txt5.3 运行应用在终端中执行chainlit run app.py然后打开浏览器访问显示的地址通常是http://localhost:8000。5.4 使用建议与注意事项vLLM服务地址确保VLLM_SERVER_URL变量指向正确的vLLM服务地址和端口。会话存储会话数据保存在本地sessions目录的JSON文件中定期清理不需要的会话文件可以释放磁盘空间。样式调整CSS样式可以根据需要随时修改修改后刷新页面即可生效。性能考虑如果会话历史非常长加载时可能会有延迟。可以考虑添加分页或限制显示数量。错误处理当前代码包含了基本的错误处理但在生产环境中可能需要更完善的异常处理机制。6. 总结通过这篇文章我们为基于Qwen3-0.6B-FP8和Chainlit的AI对话应用添加了三个实用功能自定义界面样式通过CSS文件我们可以轻松调整前端的外观让它更符合个性化需求或品牌风格。历史会话管理实现了基于文件的会话存储和加载功能让对话有了“记忆”用户体验更完整。流式响应支持将响应方式从“等待-显示”改为“边生成边显示”大幅提升了交互的流畅感和实时性。这三个功能都是从实际使用角度出发的改进它们让一个基础的AI对话Demo变得更像是一个可用的产品。你可以在此基础上继续扩展比如添加对话导出功能导出为Markdown或PDF实现会话搜索功能添加多模型切换支持集成更多Chainlit的UI组件最重要的是你现在有了一个功能相对完整的AI对话前端可以直接用于各种实际场景或者作为进一步开发的基础。获取更多AI镜像想探索更多AI镜像和应用场景访问 CSDN星图镜像广场提供丰富的预置镜像覆盖大模型推理、图像生成、视频生成、模型微调等多个领域支持一键部署。