Phi-3-mini-128k-instruct Chainlit 插件开发：思维链可视化与 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()

    simulated_response = "让我思考一下这个问题。首先，我需要理解什么是机器学习。机器学习是人工智能的一个分支..."
    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__":
    cl.run()

4. 开发 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。

import tiktoken
from typing import Dict, Any, Optional
import time
from dataclasses import dataclass
from datetime import datetime

@dataclass
class TokenStats:
    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:
        if self.duration > 0 and self.output_tokens > 0:
            return self.output_tokens / self.duration
        return 0.0

class TokenCounter:
    def __init__(self, model_name: str = "gpt-3.5-turbo"):
        try:
            self.encoder = tiktoken.encoding_for_model(model_name)
        except:
            self.encoder = tiktoken.get_encoding("cl100k_base")
        self.pricing = {
            "input": 0.0015 / 1000,
            "output": 0.0020 / 1000
        }

    def count_tokens(self, text: str) -> int:
        if not text:
            return 0
        try:
            tokens = self.encoder.encode(text)
            return len(tokens)
        except:
            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:
        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:
        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]:
        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。

import chainlit as cl
from typing import Dict, Any, List
import json
from datetime import datetime
from .counter import TokenCounter, TokenStats

class TokenStatsPanel:
    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):
        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
        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"
        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 用量统计面板\n本面板将实时显示 Token 消耗与成本统计。"
        await cl.Sidebar(
            name="token_stats",
            content=initial_content,
            size="medium"
        ).send()

4.4 集成 Token 统计到主应用

更新 app.py。

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()

    simulated_response = "这是一个模拟的 Phi-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()

    end_time = time.time()
    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 token_panel.show_initial_panel()

if __name__ == "__main__":
    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_stats_enabled: bool = True
    token_cost_per_input: float = 0.0015 / 1000
    token_cost_per_output: float = 0.0020 / 1000
    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。

# 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):
        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),
            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.Button(name="save_settings", label="保存设置", variant="primary")
        ]
        settings_content = "## ⚙️ 插件设置\n支持动态调整可视化与统计参数。"
        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"]
        await cl.Message(
            content="✅ 设置已更新！",
            author="System"
        ).send()

6. 部署与使用指南

6.1 项目结构整理

phi3-chainlit-enhanced/
├── app.py
├── config.py
├── requirements.txt
├── plugins/
│   ├── thought_chain/
│   │   ├── visualizer.py
│   │   └── utils.py
│   ├── token_stats/
│   │   ├── counter.py
│   │   └── panel.py
│   └── settings/
│       └── panel.py
└── README.md

6.2 安装依赖

pip install -r requirements.txt

6.3 运行应用

chainlit run app.py --port 7860

6.4 实际集成 Phi-3 模型

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

import requests
import json

async def call_phi3_model_real(prompt: str) -> str:
    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,
        "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:]
                    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 对话的可解释性与透明度，便于成本控制与性能优化。