Phi-3-mini-128k-instruct Chainlit插件开发：添加思维链可视化、Token用量统计面板

优质文章学习记录

05 Apr 2026 — 27 min read

Phi-3-mini-128k-instruct Chainlit插件开发：添加思维链可视化、Token用量统计面板

你是不是也遇到过这种情况？用Chainlit调用大模型，模型回答得挺好，但你心里总有个问号：它到底是怎么“想”出这个答案的？每次对话消耗了多少计算资源？如果能让模型的“思考过程”和“资源消耗”变得一目了然，那该多好。

今天，我们就来解决这个问题。我将手把手教你为基于vLLM部署的Phi-3-mini-128k-instruct模型，开发两个实用的Chainlit插件：思维链可视化面板和Token用量统计面板。通过这两个插件，你不仅能“看见”模型的推理路径，还能精确掌握每次对话的成本，让AI对话从“黑盒”走向“透明”。

1. 项目准备与环境确认

在开始插件开发之前，我们先确保基础环境已经就绪。你已经按照官方说明，成功部署了Phi-3-mini-128k-instruct模型，并且能够通过Chainlit前端正常调用。

1.1 验证模型服务状态

打开终端，使用Webshell检查模型服务日志，确认部署成功。

cat /root/workspace/llm.log

如果看到类似下面的输出，说明模型服务正在正常运行：

INFO 04-10 14:30:15 llm_engine.py:73] Initializing an LLM engine with config: model="/root/workspace/models/Phi-3-mini-128k-instruct", tokenizer="/root/workspace/models/Phi-3-mini-128k-instruct", ... INFO 04-10 14:30:20 llm_engine.py:210] # GPU blocks: 496, # CPU blocks: 512 INFO 04-10 14:30:22 model_runner.py:96] Loading model weights took 4.85 GB INFO 04-10 14:30:25 llm_engine.py:287] LLM engine is ready.

1.2 测试基础Chainlit功能

打开Chainlit前端界面，进行简单的提问测试，确保基础调用流程畅通无阻。

输入一个问题，比如“请用中文解释什么是机器学习”，你应该能正常收到模型的回复。这个基础功能是我们后续插件开发的基石。

2. 理解插件开发的核心思路

在动手写代码之前，我们先搞清楚两个关键问题：思维链可视化和Token统计到底要做什么，以及Chainlit插件是怎么工作的。

2.1 思维链可视化：让AI“思考”可见

大语言模型在生成回答时，内部其实有一个复杂的推理过程。虽然我们无法直接看到神经元如何激活，但可以通过一些技术手段，让模型的“思考步骤”展现出来。

我们的实现思路是：

拦截模型输出：在模型生成文本的过程中，不仅获取最终答案，还要捕获中间的关键推理步骤。
解析思维结构：将模型的“内心独白”解析成清晰的步骤，比如“第一步：理解问题”、“第二步：回忆相关知识”、“第三步：组织答案”。
可视化呈现：在Chainlit界面中，以流程图、步骤列表或时间线的形式，将这些思考过程展示给用户。

2.2 Token用量统计：量化对话成本

每次与AI对话都会消耗计算资源，而Token是衡量这种消耗的基本单位。了解Token用量，对于成本控制、性能优化都至关重要。

我们需要统计哪些数据：

输入Token数：你提问用了多少Token
输出Token数：模型回答用了多少Token
总Token数：本次对话的总消耗
预估成本：如果知道每个Token的价格，还能估算出本次对话的费用

2.3 Chainlit插件工作机制

Chainlit提供了灵活的插件系统，允许我们在不修改核心代码的情况下，扩展界面功能。插件主要通过以下方式工作：

自定义消息元素：在对话流中插入自定义的UI组件
侧边栏面板：在界面侧边添加新的信息面板
回调函数：在特定事件发生时执行自定义逻辑
状态管理：在会话中保存和读取自定义数据

理解了这些基础概念，我们就可以开始动手开发了。

3. 开发思维链可视化插件

这个插件会在模型生成回答时，自动分析并展示其推理过程。我们分步骤来实现。

3.1 创建插件基础结构

首先，在你的Chainlit项目目录中创建插件文件夹和文件。

mkdir -p plugins/thought_chain touch plugins/thought_chain/__init__.py touch plugins/thought_chain/visualizer.py touch plugins/thought_chain/utils.py

3.2 实现思维链解析器

编辑plugins/thought_chain/utils.py，添加思维链解析的核心逻辑。

import re from typing import List, Dict, Any import json class ThoughtChainParser: """思维链解析器，用于从模型输出中提取推理步骤""" def __init__(self): # 定义常见的思维链提示模式 self.patterns = { 'step_by_step': r'(?:步骤|Step)\s*\d+[:：]\s*(.*?)(?=(?:步骤|Step)\s*\d+[:：]|$)', 'reasoning': r'(?:思考|推理|Reasoning)[:：]\s*(.*?)(?=(?:因此|所以|Thus|Therefore)|$)', 'bullet_points': r'[•\-*]\s*(.*?)(?=\n[•\-*]|\n\n|$)', 'numbered_list': r'\d+\.\s*(.*?)(?=\n\d+\.|\n\n|$)' } def parse(self, text: str) -> List[Dict[str, Any]]: """ 从文本中解析思维链 Args: text: 模型生成的文本 Returns: 思维步骤列表，每个步骤包含内容和类型 """ steps = [] # 尝试不同的解析模式 for pattern_name, pattern in self.patterns.items(): matches = re.findall(pattern, text, re.DOTALL) if matches: for i, match in enumerate(matches): step_text = match.strip() if step_text and len(step_text) > 10: # 过滤掉太短的内容 steps.append({ 'step': len(steps) + 1, 'content': step_text, 'type': pattern_name, 'confidence': 0.8 # 置信度评分 }) # 如果没有找到结构化步骤，尝试按句子分割 if not steps: sentences = re.split(r'[。！？.!?]\s*', text) for i, sentence in enumerate(sentences): sentence = sentence.strip() if sentence and len(sentence) > 15: # 过滤掉太短的句子 steps.append({ 'step': i + 1, 'content': sentence, 'type': 'sentence', 'confidence': 0.5 }) return steps def extract_final_answer(self, text: str, steps: List[Dict]) -> str: """ 从文本中提取最终答案 Args: text: 完整文本 steps: 解析出的思维步骤 Returns: 最终答案文本 """ if not steps: return text # 找到最后一个步骤的位置 last_step_content = steps[-1]['content'] last_step_pos = text.rfind(last_step_content) if last_step_pos != -1: # 提取最后一个步骤之后的内容作为最终答案 answer_start = last_step_pos + len(last_step_content) final_answer = text[answer_start:].strip() # 清理答案中的常见前缀 prefixes = ['因此', '所以', '综上所述', '答案是', 'Answer:', 'Thus', 'Therefore'] for prefix in prefixes: if final_answer.startswith(prefix): final_answer = final_answer[len(prefix):].strip() return final_answer if final_answer else "（答案已包含在推理过程中）" return text

3.3 创建可视化组件

编辑plugins/thought_chain/visualizer.py，实现Chainlit的可视化组件。

import chainlit as cl from typing import List, Dict, Any import json from .utils import ThoughtChainParser class ThoughtChainVisualizer: """思维链可视化组件""" def __init__(self): self.parser = ThoughtChainParser() async def visualize(self, message: cl.Message): """ 可视化消息中的思维链 Args: message: Chainlit消息对象 """ text = message.content # 解析思维链 steps = self.parser.parse(text) final_answer = self.parser.extract_final_answer(text, steps) if steps: # 创建思维链可视化面板 await self._create_thought_chain_panel(steps, final_answer) else: # 如果没有检测到明显的思维链，显示简单分析 await self._create_simple_analysis(text) async def _create_thought_chain_panel(self, steps: List[Dict], final_answer: str): """创建思维链可视化面板""" # 创建步骤时间线 elements = [] # 添加标题 elements.append( cl.Text( name="thought_chain_title", content="## 🤔 模型思考过程", display="side" ) ) # 添加步骤列表 for step in steps: step_content = f"**步骤 {step['step']}** ({step['type']})\n\n{step['content']}" elements.append( cl.Text( name=f"step_{step['step']}", content=step_content, display="side" ) ) # 添加最终答案 if final_answer: elements.append( cl.Text( name="final_answer", content=f"## ✅ 最终答案\n\n{final_answer}", display="side" ) ) # 添加分析摘要 analysis = self._create_analysis_summary(steps) elements.append( cl.Text( name="analysis_summary", content=f"## 📊 分析摘要\n\n{analysis}", display="side" ) ) # 发送所有元素 for element in elements: await element.send() async def _create_simple_analysis(self, text: str): """创建简单分析面板""" analysis = self._analyze_text_structure(text) await cl.Text( name="simple_analysis", content=f"## 📝 回答结构分析\n\n{analysis}", display="side" ).send() def _create_analysis_summary(self, steps: List[Dict]) -> str: """创建分析摘要""" total_steps = len(steps) step_types = {} for step in steps: step_type = step['type'] step_types[step_type] = step_types.get(step_type, 0) + 1 summary = f"- **总推理步骤**: {total_steps} 步\n" for step_type, count in step_types.items(): type_name = { 'step_by_step': '分步推理', 'reasoning': '逻辑推理', 'bullet_points': '要点列举', 'numbered_list': '编号列表', 'sentence': '句子分析' }.get(step_type, step_type) summary += f"- **{type_name}**: {count} 步\n" # 计算平均置信度 avg_confidence = sum(step['confidence'] for step in steps) / total_steps summary += f"- **推理清晰度**: {avg_confidence:.1%}\n" return summary def _analyze_text_structure(self, text: str) -> str: """分析文本结构""" sentences = [s.strip() for s in re.split(r'[。！？.!?]\s*', text) if s.strip()] words = len(''.join(text.split())) analysis = f"- **总句子数**: {len(sentences)}\n" analysis += f"- **总字数**: {words} 字\n" analysis += f"- **平均句长**: {words/len(sentences):.1f} 字/句\n" # 检测文本类型 if any(keyword in text.lower() for keyword in ['首先', '其次', '然后', '最后']): analysis += "- **文本类型**: 顺序说明\n" elif any(keyword in text.lower() for keyword in ['因为', '所以', '因此', '由于']): analysis += "- **文本类型**: 因果论证\n" elif text.count('\n') > 3 or '•' in text or '-' in text: analysis += "- **文本类型**: 列表说明\n" else: analysis += "- **文本类型**: 一般叙述\n" return analysis

3.4 集成到Chainlit应用

现在我们需要将思维链可视化插件集成到主应用中。编辑你的app.py或主应用文件。

import chainlit as cl from chainlit import run_sync from plugins.thought_chain.visualizer import ThoughtChainVisualizer import asyncio # 初始化可视化器 thought_visualizer = ThoughtChainVisualizer() @cl.on_message async def main(message: cl.Message): """处理用户消息""" # 显示用户消息 await cl.Message( content=f"用户提问: {message.content}", author="User" ).send() # 这里应该是调用Phi-3模型的代码 # 为了示例，我们模拟一个包含思维链的回复"让我思考一下这个问题。 首先，我需要理解什么是机器学习。机器学习是人工智能的一个分支，它使计算机能够从数据中学习并做出决策，而无需显式编程。 接下来，我可以从几个关键方面来解释： 1. 核心思想：让机器从经验中学习 2. 主要类型：监督学习、无监督学习、强化学习 3. 常见应用：图像识别、语音识别、推荐系统 具体来说，监督学习就像有老师指导，我们给机器带有标签的数据进行训练；无监督学习则是让机器自己发现数据中的模式；强化学习则是通过试错来学习最佳策略。 因此，简单来说，机器学习就是让计算机像人类一样从经验中学习，并不断改进其性能的技术。""" # 发送模型回复 msg = cl.Message(content="") await msg.send() # 模拟流式输出 for chunk in simulated_response.split(): await msg.stream_token(chunk + " ") await asyncio.sleep(0.05) # 更新消息内容 msg.content = simulated_response await msg.update() # 可视化思维链 await thought_visualizer.visualize(msg) @cl.on_chat_start async def start_chat(): """聊天开始时的初始化""" # 欢迎消息 welcome_msg = cl.Message( content="欢迎使用Phi-3-mini智能助手！我已启用思维链可视化功能，您可以看到模型的推理过程。", author="Assistant" ) await welcome_msg.send() # 初始化侧边栏 await cl.Sidebar( name="thought_chain_sidebar", content="## 思维链可视化面板\n\n模型的思考过程将在这里显示。", size="large" ).send() if __name__ == "__main__": # 运行Chainlit应用 cl.run()

4. 开发Token用量统计插件

接下来，我们开发第二个插件：Token用量统计面板。这个插件会实时统计并展示每次对话的Token消耗。

4.1 创建Token统计插件结构

mkdir -p plugins/token_stats touch plugins/token_stats/__init__.py touch plugins/token_stats/counter.py touch plugins/token_stats/panel.py

4.2 实现Token计数器

编辑plugins/token_stats/counter.py，实现Token统计的核心逻辑。

import tiktoken from typing import Dict, Any, Optional import time from dataclasses import dataclass from datetime import datetime @dataclass class TokenStats: """Token统计数据结构""" input_tokens: int = 0 output_tokens: int = 0 total_tokens: int = 0 start_time: Optional[float] = None end_time: Optional[float] = None cost_estimate: float = 0.0 @property def duration(self) -> float: """计算处理时长（秒）""" if self.start_time and self.end_time: return self.end_time - self.start_time return 0.0 @property def tokens_per_second(self) -> float: """计算Token生成速度""" if self.duration > 0 and self.output_tokens > 0: return self.output_tokens / self.duration return 0.0 class TokenCounter: """Token计数器""" def __init__(self, model_name: str = "gpt-3.5-turbo"): """ 初始化Token计数器 Args: model_name: 模型名称，用于选择编码器 """ try: # 尝试使用tiktoken获取编码器 self.encoder = tiktoken.encoding_for_model(model_name) except: # 如果失败，使用cl100k_base（GPT-3.5/GPT-4使用的编码） self.encoder = tiktoken.get_encoding("cl100k_base") # Phi-3的定价信息（示例值，实际请参考官方定价） self.pricing = { "input": 0.0015 / 1000, # 每千Token输入价格 "output": 0.0020 / 1000, # 每千Token输出价格 } def count_tokens(self, text: str) -> int: """ 计算文本的Token数量 Args: text: 输入文本 Returns: Token数量 """ if not text: return 0 try: tokens = self.encoder.encode(text) return len(tokens) except: # 如果编码失败，使用简单估算（英文约4字符=1Token，中文约2字符=1Token） chinese_chars = sum(1 for c in text if '\u4e00' <= c <= '\u9fff') english_chars = len(text) - chinese_chars return int(chinese_chars / 2 + english_chars / 4) def calculate_cost(self, input_tokens: int, output_tokens: int) -> float: """ 计算预估成本 Args: input_tokens: 输入Token数 output_tokens: 输出Token数 Returns: 预估成本（美元） """ input_cost = input_tokens * self.pricing["input"] output_cost = output_tokens * self.pricing["output"] return input_cost + output_cost def create_stats(self, input_text: str, output_text: str, start_time: Optional[float] = None, end_time: Optional[float] = None) -> TokenStats: """ 创建完整的Token统计 Args: input_text: 输入文本 output_text: 输出文本 start_time: 开始时间戳 end_time: 结束时间戳 Returns: TokenStats对象 """ input_tokens = self.count_tokens(input_text) output_tokens = self.count_tokens(output_text) total_tokens = input_tokens + output_tokens stats = TokenStats( input_tokens=input_tokens, output_tokens=output_tokens, total_tokens=total_tokens, start_time=start_time, end_time=end_time or time.time(), cost_estimate=self.calculate_cost(input_tokens, output_tokens) ) return stats def format_stats(self, stats: TokenStats) -> Dict[str, Any]: """ 格式化统计信息 Args: stats: TokenStats对象 Returns: 格式化后的统计信息字典 """ return { "输入Token数": f"{stats.input_tokens:,}", "输出Token数": f"{stats.output_tokens:,}", "总Token数": f"{stats.total_tokens:,}", "处理时间": f"{stats.duration:.2f}秒", "生成速度": f"{stats.tokens_per_second:.1f} Token/秒", "预估成本": f"${stats.cost_estimate:.6f}", "时间戳": datetime.now().strftime("%Y-%m-%d %H:%M:%S") }

4.3 创建统计面板组件

编辑plugins/token_stats/panel.py，实现Chainlit的统计面板。

import chainlit as cl from typing import Dict, Any, List import json from datetime import datetime from .counter import TokenCounter, TokenStats class TokenStatsPanel: """Token统计面板""" def __init__(self): self.counter = TokenCounter() self.session_stats: List[Dict[str, Any]] = [] async def update_panel(self, input_text: str, output_text: str, start_time: float, end_time: float): """ 更新统计面板 Args: input_text: 用户输入 output_text: 模型输出 start_time: 开始时间戳 end_time: 结束时间戳 """ # 计算统计信息 stats = self.counter.create_stats( input_text=input_text, output_text=output_text, start_time=start_time, end_time=end_time ) # 格式化显示 formatted_stats = self.counter.format_stats(stats) # 保存到会话历史 self.session_stats.append({ "timestamp": datetime.now().isoformat(), "stats": formatted_stats, "input_preview": input_text[:100] + ("..." if len(input_text) > 100 else ""), "output_preview": output_text[:100] + ("..." if len(output_text) > 100 else "") }) # 更新面板显示 await self._render_panel(formatted_stats) # 更新历史记录 await self._update_history_panel() async def _render_panel(self, stats: Dict[str, Any]): """渲染统计面板""" # 创建统计卡片 stats_content = "## 📊 本次对话统计\n\n" for key, value in stats.items(): if key != "时间戳": stats_content += f"**{key}**: {value}\n\n" # 添加进度条可视化 total_tokens = int(stats["总Token数"].replace(",", "")) max_tokens = 128000 # Phi-3-mini-128k的上下文长度 if total_tokens > 0: percentage = min(100, (total_tokens / max_tokens) * 100) # 创建简单的文本进度条 bar_length = 20 filled = int(bar_length * percentage / 100) bar = "█" * filled + "░" * (bar_length - filled) stats_content += f"**上下文使用率**: {percentage:.1f}%\n" stats_content += f"`[{bar}]`\n\n" stats_content += f"*统计时间: {stats.get('时间戳', 'N/A')}*" # 发送到侧边栏 await cl.Sidebar( name="token_stats", content=stats_content, size="medium" ).send() async def _update_history_panel(self): """更新历史记录面板""" if len(self.session_stats) <= 1: return history_content = "## 📈 历史统计\n\n" # 只显示最近5次记录 recent_stats = self.session_stats[-5:] for i, record in enumerate(recent_stats): history_content += f"### 对话 {len(self.session_stats) - len(recent_stats) + i + 1}\n" history_content += f"**时间**: {record['timestamp'][11:19]}\n" history_content += f"**输入**: {record['input_preview']}\n" history_content += f"**输出**: {record['output_preview']}\n" history_content += f"**总Token**: {record['stats']['总Token数']}\n" history_content += "---\n\n" # 计算累计统计 total_input = sum(int(s['stats']['输入Token数'].replace(",", "")) for s in self.session_stats) total_output = sum(int(s['stats']['输出Token数'].replace(",", "")) for s in self.session_stats) total_cost = sum(float(s['stats']['预估成本'].replace("$", "")) for s in self.session_stats) history_content += "### 累计统计\n" history_content += f"**总对话数**: {len(self.session_stats)}\n" history_content += f"**累计输入Token**: {total_input:,}\n" history_content += f"**累计输出Token**: {total_output:,}\n" history_content += f"**累计预估成本**: ${total_cost:.6f}\n" # 发送历史面板 await cl.Sidebar( name="token_history", content=history_content, size="large" ).send() async def show_initial_panel(self): """显示初始面板"""" ## 📊 Token用量统计面板 本面板将实时显示： - ✅ 每次对话的Token消耗 - ✅ 处理时间和生成速度 - ✅ 上下文使用率 - ✅ 预估成本统计 - 📈 历史记录追踪 等待第一次对话开始... """ await cl.Sidebar( name="token_stats", content=initial_content, size="medium" ).send()

4.4 集成Token统计到主应用

现在更新主应用文件，集成Token统计功能。

import chainlit as cl from chainlit import run_sync from plugins.thought_chain.visualizer import ThoughtChainVisualizer from plugins.token_stats.panel import TokenStatsPanel import asyncio import time # 初始化插件 thought_visualizer = ThoughtChainVisualizer() token_panel = TokenStatsPanel() @cl.on_message async def main(message: cl.Message): """处理用户消息""" # 记录开始时间 start_time = time.time() # 显示用户消息 user_msg = cl.Message( content=f"用户提问: {message.content}", author="User" ) await user_msg.send() # 这里应该是调用Phi-3模型的真实代码 # 为了示例，我们使用模拟回复"让我思考一下这个问题。 首先，我需要理解什么是机器学习。机器学习是人工智能的一个分支，它使计算机能够从数据中学习并做出决策，而无需显式编程。 接下来，我可以从几个关键方面来解释： 1. 核心思想：让机器从经验中学习 2. 主要类型：监督学习、无监督学习、强化学习 3. 常见应用：图像识别、语音识别、推荐系统 具体来说，监督学习就像有老师指导，我们给机器带有标签的数据进行训练；无监督学习则是让机器自己发现数据中的模式；强化学习则是通过试错来学习最佳策略。 因此，简单来说，机器学习就是让计算机像人类一样从经验中学习，并不断改进其性能的技术。""" # 模拟处理延迟 await asyncio.sleep(0.5) # 发送模型回复 msg = cl.Message(content="") await msg.send() # 模拟流式输出 for chunk in simulated_response.split(): await msg.stream_token(chunk + " ") await asyncio.sleep(0.05) # 更新消息内容 msg.content = simulated_response await msg.update() # 记录结束时间 end_time = time.time() # 更新Token统计面板 await token_panel.update_panel( input_text=message.content, output_text=simulated_response, start_time=start_time, end_time=end_time ) # 可视化思维链 await thought_visualizer.visualize(msg) @cl.on_chat_start async def start_chat(): """聊天开始时的初始化""" # 欢迎消息 welcome_msg = cl.Message("欢迎使用Phi-3-mini智能助手！ 我已启用以下增强功能： 🔍 **思维链可视化** - 查看模型的推理过程 📊 **Token用量统计** - 实时监控资源消耗 请开始提问吧！""", author="Assistant" ) await welcome_msg.send() # 初始化侧边栏 await cl.Sidebar( name="thought_chain_sidebar", content="## 思维链可视化面板\n\n模型的思考过程将在这里显示。", size="large" ).send() # 显示初始Token统计面板 await token_panel.show_initial_panel() if __name__ == "__main__": # 运行Chainlit应用 cl.run()

5. 插件配置与高级功能

现在我们已经完成了两个核心插件的开发，接下来让我们添加一些配置选项和高级功能，让插件更加实用。

5.1 创建配置文件

创建config.py来管理插件配置。

# config.py import os from dataclasses import dataclass from typing import Optional @dataclass class PluginConfig: """插件配置类""" # 思维链可视化配置 thought_chain_enabled: bool = True thought_chain_min_confidence: float = 0.3 # 最小置信度阈值 thought_chain_max_steps: int = 10 # 最大显示步骤数 # Token统计配置 token_stats_enabled: bool = True token_cost_per_input: float = 0.0015 / 1000 # 输入Token单价 token_cost_per_output: float = 0.0020 / 1000 # 输出Token单价 show_cost_estimation: bool = True # 是否显示成本估算 # 显示配置 sidebar_position: str = "right" # 侧边栏位置 auto_collapse: bool = False # 是否自动折叠 @classmethod def from_env(cls): """从环境变量加载配置""" return cls( thought_chain_enabled=os.getenv("THOUGHT_CHAIN_ENABLED", "true").lower() == "true", thought_chain_min_confidence=float(os.getenv("THOUGHT_CHAIN_MIN_CONFIDENCE", "0.3")), thought_chain_max_steps=int(os.getenv("THOUGHT_CHAIN_MAX_STEPS", "10")), token_stats_enabled=os.getenv("TOKEN_STATS_ENABLED", "true").lower() == "true", token_cost_per_input=float(os.getenv("TOKEN_COST_INPUT", "0.0000015")), token_cost_per_output=float(os.getenv("TOKEN_COST_OUTPUT", "0.0000020")), show_cost_estimation=os.getenv("SHOW_COST_ESTIMATION", "true").lower() == "true", sidebar_position=os.getenv("SIDEBAR_POSITION", "right"), auto_collapse=os.getenv("AUTO_COLLAPSE", "false").lower() == "true" ) # 全局配置实例 config = PluginConfig.from_env()

5.2 添加设置界面

让我们添加一个简单的设置界面，允许用户动态调整插件配置。

# plugins/settings/panel.py import chainlit as cl from typing import Dict, Any import json class SettingsPanel: """设置面板""" def __init__(self, config): self.config = config async def show_settings(self): """显示设置面板"""" ## ⚙️ 插件设置 ### 思维链可视化 - **启用/禁用**: 控制是否显示思维链 - **置信度阈值**: 过滤低置信度的推理步骤 - **最大步骤数**: 限制显示的步骤数量 ### Token统计 - **启用/禁用**: 控制是否显示Token统计 - **成本估算**: 显示/隐藏成本估算 - **单价设置**: 自定义Token单价 ### 显示设置 - **侧边栏位置**: 左右切换 - **自动折叠**: 自动折叠非活动面板 """ # 创建设置表单 settings_elements = [ cl.Checkbox(, label="启用思维链可视化", initial=self.config.thought_chain_enabled ), cl.Slider(, label="置信度阈值", initial=self.config.thought_chain_min_confidence, min=0, max=1, step=0.1, marks={0: "0", 0.5: "0.5", 1: "1"} ), cl.NumberInput(, label="最大显示步骤", initial=self.config.thought_chain_max_steps, min=1, max=20 ), cl.Divider(), cl.Checkbox(, label="启用Token统计", initial=self.config.token_stats_enabled ), cl.Checkbox(, label="显示成本估算", initial=self.config.show_cost_estimation ), cl.Button( name="save_settings", label="保存设置", variant="primary" ) ] # 发送设置面板 await cl.Sidebar( name="settings_panel", content=settings_content, elements=settings_elements, size="medium" ).send() async def handle_settings_change(self, settings: Dict[str, Any]): """处理设置变更""" # 更新配置 if "thought_chain_enabled" in settings: self.config.thought_chain_enabled = settings["thought_chain_enabled"] if "thought_chain_confidence" in settings: self.config.thought_chain_min_confidence = settings["thought_chain_confidence"] if "thought_chain_max_steps" in settings: self.config.thought_chain_max_steps = settings["thought_chain_max_steps"] if "token_stats_enabled" in settings: self.config.token_stats_enabled = settings["token_stats_enabled"] if "show_cost_estimation" in settings: self.config.show_cost_estimation = settings["show_cost_estimation"] # 发送确认消息 await cl.Message( content="✅ 设置已更新！", author="System" ).send()

5.3 更新主应用集成所有功能

最后，更新主应用文件，集成所有插件和设置功能。

# app.py - 完整版本 import chainlit as cl import asyncio import time from config import config from plugins.thought_chain.visualizer import ThoughtChainVisualizer from plugins.token_stats.panel import TokenStatsPanel from plugins.settings.panel import SettingsPanel # 初始化所有插件 thought_visualizer = ThoughtChainVisualizer() token_panel = TokenStatsPanel() settings_panel = SettingsPanel(config) @cl.on_message async def main(message: cl.Message): """处理用户消息""" # 记录开始时间 start_time = time.time() # 显示用户消息 user_msg = cl.Message( content=f"用户提问: {message.content}", author="User" ) await user_msg.send() # 调用Phi-3模型（这里需要替换为实际的模型调用代码） # 注意：实际部署时，这里应该调用vLLM部署的Phi-3模型 try: # 模拟调用过程 response = await call_phi3_model(message.content) except Exception as e: response = f"抱歉，调用模型时出现错误: {str(e)}" # 发送模型回复 msg = cl.Message(content="") await msg.send() # 模拟流式输出（实际部署时应该是真正的流式输出） for chunk in response.split(): await msg.stream_token(chunk + " ") await asyncio.sleep(0.03) # 稍微加快速度 # 更新消息内容 msg.content = response await msg.update() # 记录结束时间 end_time = time.time() # 根据配置决定是否显示插件内容 if config.token_stats_enabled: await token_panel.update_panel( input_text=message.content, output_text=response, start_time=start_time, end_time=end_time ) if config.thought_chain_enabled: await thought_visualizer.visualize(msg) async def call_phi3_model(prompt: str) -> str: """ 调用Phi-3模型的函数 实际部署时需要替换为真实的模型调用代码 """ # 这里应该是调用vLLM API的代码 # 示例：使用requests调用本地vLLM服务 """ import requests import json url = "http://localhost:8000/v1/completions" headers = {"Content-Type": "application/json"} data = { "model": "Phi-3-mini-128k-instruct", "prompt": prompt, "max_tokens": 500, "temperature": 0.7 } response = requests.post(url, headers=headers, json=data) result = response.json() return result["choices"][0]["text"] """ # 模拟返回 return f"""这是一个模拟的Phi-3模型回复。 对于您的问题"{prompt}"，我的思考过程如下： 1. 首先，我需要理解问题的核心要点。您询问的是关于{prompt.split()[0] if prompt.split() else "某个主题"}的内容。 2. 接下来，我会回忆相关的知识体系。根据我的训练数据，这个话题涉及多个方面。 3. 然后，我会组织回答的结构。一个好的回答应该包括：定义、关键特点、实际应用和注意事项。 4. 最后，我会用清晰易懂的语言呈现答案。 基于以上思考，我的回答是：这是一个非常重要且有趣的话题。在实际应用中，需要考虑多个因素，包括技术实现、成本效益和用户体验。建议进一步深入研究具体案例以获得更深入的理解。""" @cl.on_chat_start async def start_chat(): """聊天开始时的初始化""" # 欢迎消息 welcome_msg = cl.Message("🤖 欢迎使用Phi-3-mini智能助手增强版！ 已加载功能： 🔍 思维链可视化 - 查看模型推理过程 📊 Token用量统计 - 实时监控资源消耗 ⚙️ 可配置设置 - 个性化您的体验 输入 /settings 打开设置面板 输入 /help 查看帮助信息""", author="Assistant" ) await welcome_msg.send() # 初始化侧边栏 await cl.Sidebar( name="main_sidebar", content="## 功能面板\n\n选择要查看的面板：", elements=[ cl.Button(name="show_thought", label="思维链可视化", variant="primary"), cl.Button(name="show_stats", label="Token统计", variant="primary"), cl.Button(name="show_settings", label="设置", variant="secondary") ], size="medium" ).send() @cl.on_action("show_thought") async def on_show_thought(): """显示思维链面板""" await cl.Sidebar( name="thought_panel", content="## 思维链可视化\n\n模型的思考过程将在这里显示。", size="large" ).send() @cl.on_action("show_stats") async def on_show_stats(): """显示统计面板""" await token_panel.show_initial_panel() @cl.on_action("show_settings") async def on_show_settings(): """显示设置面板""" await settings_panel.show_settings() @cl.on_chat_resume async def on_chat_resume(): """聊天恢复时的处理""" await cl.Message( content="会话已恢复，所有插件功能已重新加载。", author="System" ).send() if __name__ == "__main__": # 运行Chainlit应用 cl.run()

6. 部署与使用指南

现在我们已经完成了所有插件的开发，接下来看看如何部署和使用这个增强版的Chainlit应用。

6.1 项目结构整理

确保你的项目结构如下：

phi3-chainlit-enhanced/ ├── app.py # 主应用文件 ├── config.py # 配置文件 ├── requirements.txt # 依赖文件 ├── plugins/ │ ├── __init__.py │ ├── thought_chain/ │ │ ├── __init__.py │ │ ├── visualizer.py │ │ └── utils.py │ ├── token_stats/ │ │ ├── __init__.py │ │ ├── counter.py │ │ └── panel.py │ └── settings/ │ ├── __init__.py │ └── panel.py └── README.md # 说明文档

6.2 安装依赖

创建requirements.txt文件：

chainlit>=1.0.0 tiktoken>=0.5.0 requests>=2.31.0

安装依赖：

pip install -r requirements.txt

6.3 环境配置

你可以通过环境变量配置插件行为：

# 启用/禁用功能 export THOUGHT_CHAIN_ENABLED=true export TOKEN_STATS_ENABLED=true # 配置参数 export THOUGHT_CHAIN_MIN_CONFIDENCE=0.3 export THOUGHT_CHAIN_MAX_STEPS=10 # Token成本（根据实际定价调整） export TOKEN_COST_INPUT=0.0000015 export TOKEN_COST_OUTPUT=0.0000020 # 显示设置 export SIDEBAR_POSITION=right export AUTO_COLLAPSE=false

6.4 运行应用

启动Chainlit应用：

chainlit run app.py

或者指定端口：

chainlit run app.py --port 7860

6.5 使用说明

访问应用：打开浏览器，访问 http://localhost:7860
开始对话：在输入框中提问，模型会生成回答
查看思维链：在右侧侧边栏查看模型的推理过程
监控Token使用：实时查看Token消耗和成本估算
调整设置：点击设置按钮调整插件参数

6.6 实际集成Phi-3模型

在实际部署中，你需要将模拟的call_phi3_model函数替换为真实的vLLM API调用：

import requests import json async def call_phi3_model_real(prompt: str) -> str: """实际调用vLLM部署的Phi-3模型""" # vLLM API端点 url = "http://localhost:8000/v1/completions" # 请求头 headers = { "Content-Type": "application/json" } # 请求数据 data = { "model": "Phi-3-mini-128k-instruct", "prompt": prompt, "max_tokens": 1024, "temperature": 0.7, "top_p": 0.9, "stream": True # 启用流式输出 } try: # 发送请求 response = requests.post(url, headers=headers, json=data, stream=True) # 处理流式响应 for line in response.iter_lines(): if line: line = line.decode('utf-8') if line.startswith('data: '): json_str = line[6:] # 去掉'data: '前缀 if json_str != '[DONE]': try: chunk = json.loads(json_str) token = chunk['choices'][0]['text'] full_response += token # 这里可以添加流式输出的逻辑 except json.JSONDecodeError: continue return full_response except Exception as e: return f"调用模型时出错: {str(e)}"

7. 总结

通过本文的教程，我们成功为Phi-3-mini-128k-instruct模型开发了两个实用的Chainlit插件：思维链可视化面板和Token用量统计面板。这两个插件极大地增强了AI对话的透明度和可控性。

7.1 主要成果回顾

思维链可视化插件让我们能够：

实时查看模型的推理步骤和思考过程
理解模型是如何从问题推导出答案的
分析回答的结构和质量
提高对话的可解释性和可信度

Token用量统计插件帮助我们：

精确监控每次对话的资源消耗
了解输入和输出的Token分布
估算对话成本，便于预算管理
追踪历史使用情况，优化使用策略

7.2 核心价值

这两个插件的核心价值在于：

提升透明度：让AI的"黑盒"决策过程变得可见
优化成本：帮助用户更好地管理和控制使用成本
改善体验：提供更丰富、更有价值的信息展示
促进理解：帮助用户更好地理解模型的能力和局限

7.3 扩展可能性

基于这个基础框架，你还可以进一步扩展：

更多可视化类型：添加图表、流程图等更丰富的可视化形式
性能监控：添加响应时间、内存使用等性能指标
质量评估：自动评估回答的相关性、准确性和完整性
批量处理：支持批量提问和批量分析
导出功能：支持将对话记录和统计数据导出为报告

7.4 实际应用建议

在实际使用中，建议：

根据需求调整：根据具体应用场景调整插件的显示内容和详细程度
性能考虑：在资源受限的环境中，可以关闭部分高开销功能
用户教育：向最终用户解释这些可视化信息的含义和价值
持续优化：根据用户反馈不断改进插件功能和用户体验

通过这两个插件，我们不仅增强了Phi-3模型的使用体验，也为其他大语言模型的Chainlit集成提供了可复用的模式。希望这个教程能帮助你更好地理解和利用大语言模型，让AI对话变得更加透明、可控和高效。

获取更多AI镜像

想探索更多AI镜像和应用场景？访问 ZEEKLOG星图镜像广场，提供丰富的预置镜像，覆盖大模型推理、图像生成、视频生成、模型微调等多个领域，支持一键部署。