基于 Chainlit 的 Phi-3 模型插件开发：思维链可视化与 Token 统计

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

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 模型的代码
    # 为了示例，我们模拟一个包含思维链的回复
    simulated_response = "让我思考一下这个问题。首先，我需要理解什么是机器学习。机器学习是人工智能的一个分支，它使计算机能够从数据中学习并做出决策，而无需显式编程。接下来，我可以从几个关键方面来解释：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):
        """显示初始面板"""
        initial_content = """## 📊 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 模型的真实代码
    # 为了示例，我们使用模拟回复
    simulated_response = "让我思考一下这个问题。首先，我需要理解什么是机器学习。机器学习是人工智能的一个分支，它使计算机能够从数据中学习并做出决策，而无需显式编程。接下来，我可以从几个关键方面来解释：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(
        content="欢迎使用 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):
        """显示设置面板"""
        initial_content = """## ⚙️ 插件设置
### 思维链可视化
- **启用/禁用**: 控制是否显示思维链
- **置信度阈值**: 过滤低置信度的推理步骤
- **最大步骤数**: 限制显示的步骤数量
### Token 统计
- **启用/禁用**: 控制是否显示 Token 统计
- **成本估算**: 显示/隐藏成本估算
- **单价设置**: 自定义 Token 单价
### 显示设置
- **侧边栏位置**: 左右切换
- **自动折叠**: 自动折叠非活动面板"""
        # 创建设置表单
        settings_elements = [
            cl.Checkbox(name="thought_chain_enabled", label="启用思维链可视化", initial=self.config.thought_chain_enabled),
            cl.Slider(name="thought_chain_confidence", 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(name="thought_chain_max_steps", label="最大显示步骤", initial=self.config.thought_chain_max_steps, min=1, max=20),
            cl.Divider(),
            cl.Checkbox(name="token_stats_enabled", label="启用 Token 统计", initial=self.config.token_stats_enabled),
            cl.Checkbox(name="show_cost_estimation", label="显示成本估算", initial=self.config.show_cost_estimation),
            cl.Button(name="save_settings", label="保存设置", variant="primary")
        ]
        # 发送设置面板
        await cl.Sidebar(
            name="settings_panel",
            content=initial_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(
        content="🤖 欢迎使用 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 使用说明

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

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)
        # 处理流式响应
        full_response = ""
        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 核心价值

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

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

7.3 扩展可能性

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

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

7.4 实际应用建议

在实际使用中，建议：

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

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