SmallThinker-3B开源镜像实操：日志监控、推理耗时统计与Token效率分析

1. 环境准备与快速部署

SmallThinker-3B-Preview是一个基于Qwen2.5-3b-Instruct模型微调而来的轻量级语言模型，专门为边缘设备部署和高效推理场景设计。这个模型最大的特点是体积小巧但能力不俗，特别适合资源受限的环境使用。

首先需要确保你的系统环境满足基本要求：

• 操作系统：Linux Ubuntu 18.04+ 或 Windows 10+
• 内存：至少8GB RAM
• 存储：10GB可用空间
• Python版本：3.8或更高版本

安装过程非常简单，只需要几个命令：

# 克隆项目仓库
git clone https://github.com/sonhhxg/smallthinker-3b.git
cd smallthinker-3b

# 创建Python虚拟环境
python -m venv venv
source venv/bin/activate  # Linux/Mac
# 或者 venv\Scripts\activate  # Windows

# 安装依赖包
pip install -r requirements.txt

如果你想要更快速的部署，也可以使用Docker方式：

# 拉取预构建的Docker镜像
docker pull sonhhxg/smallthinker-3b:latest

# 运行容器
docker run -p 8000:8000 sonhhxg/smallthinker-3b

2. 基础功能快速上手

2.1 模型基本使用

SmallThinker-3B的使用方式非常直观，下面是一个最简单的示例：

from smallthinker import SmallThinker

# 初始化模型
model = SmallThinker("smallthinker:3b")

# 简单提问
response = model.generate("请解释一下人工智能的基本概念")
print(response)

模型会返回一个结构化的响应，包含生成的文本和相关的元数据信息。

2.2 通过Ollama界面使用

对于不喜欢命令行的用户，可以通过Ollama的图形界面来使用模型：

1. 打开Ollama模型显示入口
2. 在页面顶部的模型选择中，选择【smallthinker:3b】
3. 在下方输入框中直接提问即可

这种方式特别适合快速测试和演示，不需要任何编程知识就能体验模型的能力。

2.3 批量处理示例

如果你需要处理多个问题，可以使用批量生成功能：

questions = [
    "什么是机器学习？",
    "深度学习与机器学习有什么区别？",
    "请用简单的话解释神经网络"
]

results = model.generate_batch(questions)
for i, result in enumerate(results):
    print(f"问题 {i+1}: {result['text']}")
    print(f"生成耗时: {result['time_cost']}秒")
    print("---")

3. 日志监控实战指南

3.1 启用详细日志记录

要深入了解模型的运行情况，首先需要启用详细的日志记录：

import logging
import sys

# 配置日志系统
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('smallthinker.log'),
        logging.StreamHandler(sys.stdout)
    ]
)

# 获取模型日志器
model_logger = logging.getLogger('smallthinker')

3.2 监控关键指标

在实际使用中，建议监控以下几个关键指标：

class ModelMonitor:
    def __init__(self):
        self.total_requests = 0
        self.total_tokens = 0
        self.total_time = 0.0
        self.error_count = 0
    
    def record_request(self, tokens_generated, time_taken):
        self.total_requests += 1
        self.total_tokens += tokens_generated
        self.total_time += time_taken
        
        # 记录到日志
        model_logger.info(
            f"请求完成: 生成{tokens_generated}个token, "
            f"耗时{time_taken:.2f}秒, "
            f"速度{tokens_generated/time_taken:.1f} token/秒"
        )
    
    def get_stats(self):
        avg_time = self.total_time / self.total_requests if self.total_requests > 0 else 0
        avg_tokens = self.total_tokens / self.total_requests if self.total_requests > 0 else 0
        avg_speed = self.total_tokens / self.total_time if self.total_time > 0 else 0
        
        return {
            "总请求数": self.total_requests,
            "总生成token数": self.total_tokens,
            "总耗时": f"{self.total_time:.2f}秒",
            "平均生成时间": f"{avg_time:.2f}秒",
            "平均生成token数": f"{avg_tokens:.1f}",
            "平均生成速度": f"{avg_speed:.1f} token/秒",
            "错误次数": self.error_count
        }

# 使用监控器
monitor = ModelMonitor()

3.3 实时日志分析

你可以创建一个实时日志分析脚本来监控模型运行状态：

import re
import time
from collections import defaultdict

class LogAnalyzer:
    def __init__(self, log_file='smallthinker.log'):
        self.log_file = log_file
        self.key_metrics = defaultdict(list)
    
    def analyze_live(self):
        """实时分析日志文件"""
        with open(self.log_file, 'r') as f:
            # 移动到文件末尾
            f.seek(0, 2)
            
            while True:
                line = f.readline()
                if not line:
                    time.sleep(0.1)
                    continue
                
                # 分析日志行
                self._analyze_line(line)
    
    def _analyze_line(self, line):
        # 提取时间信息
        time_match = re.search(r'(\d+\.\d+)秒', line)
        if time_match:
            time_taken = float(time_match.group(1))
            self.key_metrics['inference_time'].append(time_taken)
        
        # 提取token数量
        token_match = re.search(r'生成(\d+)个token', line)
        if token_match:
            tokens = int(token_match.group(1))
            self.key_metrics['tokens_generated'].append(tokens)
        
        # 提取生成速度
        speed_match = re.search(r'速度(\d+\.\d+) token/秒', line)
        if speed_match:
            speed = float(speed_match.group(1))
            self.key_metrics['generation_speed'].append(speed)

4. 推理耗时统计与分析

4.1 基础耗时统计

了解模型的推理耗时对于优化使用体验非常重要：

import time
import statistics

class TimeProfiler:
    def __init__(self):
        self.timings = {
            'total': [],
            'preprocessing': [],
            'inference': [],
            'postprocessing': []
        }
    
    def time_function(self, func, *args, **kwargs):
        start_time = time.time()
        result = func(*args, **kwargs)
        end_time = time.time()
        
        return result, end_time - start_time
    
    def profile_generation(self, prompt, max_tokens=100):
        """完整生成过程性能分析"""
        # 预处理阶段
        preprocess_result, preprocess_time = self.time_function(
            self._preprocess, prompt
        )
        
        # 推理阶段
        inference_result, inference_time = self.time_function(
            self._inference, preprocess_result, max_tokens
        )
        
        # 后处理阶段
        final_result, postprocess_time = self.time_function(
            self._postprocess, inference_result
        )
        
        total_time = preprocess_time + inference_time + postprocess_time
        
        # 记录时间
        self.timings['preprocessing'].append(preprocess_time)
        self.timings['inference'].append(inference_time)
        self.timings['postprocessing'].append(postprocess_time)
        self.timings['total'].append(total_time)
        
        return {
            'result': final_result,
            'timings': {
                'preprocessing': preprocess_time,
                'inference': inference_time,
                'postprocessing': postprocess_time,
                'total': total_time
            }
        }
    
    def get_statistics(self):
        """获取统计信息"""
        stats = {}
        for stage, times in self.timings.items():
            if times:
                stats[stage] = {
                    'count': len(times),
                    'total': sum(times),
                    'average': statistics.mean(times),
                    'median': statistics.median(times),
                    'min': min(times),
                    'max': max(times),
                    'stdev': statistics.stdev(times) if len(times) > 1 else 0
                }
        return stats

4.2 不同输入长度的耗时对比

测试不同长度输入的推理耗时可以帮助我们了解模型的性能特征：

def test_different_lengths(model, lengths=[10, 50, 100, 200, 500]):
    """测试不同输入长度的性能"""
    results = {}
    
    for length in lengths:
        # 生成测试文本
        test_prompt = "请回答以下问题：" + "字" * length
        
        profiler = TimeProfiler()
        result = profiler.profile_generation(test_prompt)
        
        results[length] = {
            'total_time': result['timings']['total'],
            'inference_time': result['timings']['inference'],
            'output_length': len(result['result'])
        }
        
        print(f"输入长度 {length}: 总耗时 {result['timings']['total']:.2f}秒, "
              f"推理耗时 {result['timings']['inference']:.2f}秒")
    
    return results

4.3 性能优化建议

基于耗时统计结果，这里有一些优化建议：

1. 批量处理：如果需要处理多个请求，尽量使用批量处理功能
2. 输入长度控制：过长的输入会显著增加推理时间，尽量保持输入简洁
3. 输出长度限制：设置合理的max_tokens参数，避免生成过长内容
4. 硬件优化：确保有足够的内存和合适的CPU/GPU资源

5. Token效率深度分析

5.1 Token使用统计

Token效率直接影响使用成本和响应速度：

class TokenEfficiencyAnalyzer:
    def __init__(self):
        self.records = []
    
    def analyze_efficiency(self, input_text, output_text, inference_time):
        input_tokens = self._estimate_tokens(input_text)
        output_tokens = self._estimate_tokens(output_text)
        
        efficiency = {
            'input_tokens': input_tokens,
            'output_tokens': output_tokens,
            'total_tokens': input_tokens + output_tokens,
            'inference_time': inference_time,
            'tokens_per_second': output_tokens / inference_time if inference_time > 0 else 0,
            'efficiency_ratio': output_tokens / (input_tokens + 1)  # 避免除零
        }
        
        self.records.append(efficiency)
        return efficiency
    
    def _estimate_tokens(self, text):
        """粗略估计token数量（实际使用中应该用tokenizer）"""
        # 中文大致按字词计算，英文按单词计算
        chinese_chars = sum(1 for char in text if '\u4e00' <= char <= '\u9fff')
        english_words = len(text.split()) - chinese_chars / 2  # 粗略估计
        
        return int(chinese_chars + english_words * 1.3)
    
    def get_summary_stats(self):
        """获取汇总统计"""
        if not self.records:
            return {}
        
        tokens_per_second = [r['tokens_per_second'] for r in self.records]
        efficiency_ratios = [r['efficiency_ratio'] for r in self.records]
        
        return {
            'total_requests': len(self.records),
            'avg_tokens_per_second': statistics.mean(tokens_per_second),
            'avg_efficiency_ratio': statistics.mean(efficiency_ratios),
            'max_efficiency': max(efficiency_ratios),
            'min_efficiency': min(efficiency_ratios)
        }

5.2 不同任务类型的Token效率对比

分析不同任务类型的Token使用效率：

def compare_task_efficiency(model, tasks):
    """比较不同任务类型的效率"""
    analyzer = TokenEfficiencyAnalyzer()
    results = {}
    
    for task_name, prompts in tasks.items():
        task_results = []
        for prompt in prompts:
            start_time = time.time()
            output = model.generate(prompt)
            inference_time = time.time() - start_time
            
            efficiency = analyzer.analyze_efficiency(prompt, output, inference_time)
            task_results.append(efficiency)
        
        # 计算该任务类型的平均效率
        avg_tps = statistics.mean([r['tokens_per_second'] for r in task_results])
        avg_ratio = statistics.mean([r['efficiency_ratio'] for r in task_results])
        
        results[task_name] = {
            'avg_tokens_per_second': avg_tps,
            'avg_efficiency_ratio': avg_ratio,
            'sample_count': len(prompts)
        }
    
    return results

# 示例任务定义
example_tasks = {
    '问答': [
        "什么是人工智能？",
        "机器学习有哪些主要类型？",
        "请解释深度学习的基本概念"
    ],
    '摘要': [
        "请总结这篇关于气候变化的文章",
        "用三句话概括这篇技术文档的主要内容"
    ],
    '创意写作': [
        "写一个关于机器人朋友的短故事",
        "创作一首关于春天的诗歌"
    ]
}

5.3 效率优化策略

根据分析结果，可以采取以下策略提高Token效率：

1. 提示工程优化：设计更有效的提示词，减少不必要的输入token
2. 输出格式控制：明确指定输出格式和要求，避免模型生成冗余内容
3. 温度参数调整：适当调整temperature参数，平衡创造性和效率
4. 停止条件设置：使用stop sequences来精确控制输出长度

6. 完整监控解决方案

6.1 集成监控仪表板

创建一个完整的监控解决方案：

class SmallThinkerMonitor:
    def __init__(self):
        self.time_profiler = TimeProfiler()
        self.token_analyzer = TokenEfficiencyAnalyzer()
        self.model_monitor = ModelMonitor()
        self.log_analyzer = LogAnalyzer()
    
    def monitor_generation(self, prompt, max_tokens=100):
        """完整的监控生成过程"""
        # 时间分析
        start_time = time.time()
        result = model.generate(prompt, max_tokens=max_tokens)
        inference_time = time.time() - start_time
        
        # Token效率分析
        token_stats = self.token_analyzer.analyze_efficiency(
            prompt, result, inference_time
        )
        
        # 记录到监控器
        self.model_monitor.record_request(
            token_stats['output_tokens'], inference_time
        )
        
        return {
            'result': result,
            'metrics': {
                'inference_time': inference_time,
                'input_tokens': token_stats['input_tokens'],
                'output_tokens': token_stats['output_tokens'],
                'tokens_per_second': token_stats['tokens_per_second'],
                'efficiency_ratio': token_stats['efficiency_ratio']
            }
        }
    
    def generate_report(self):
        """生成监控报告"""
        time_stats = self.time_profiler.get_statistics()
        token_stats = self.token_analyzer.get_summary_stats()
        model_stats = self.model_monitor.get_stats()
        
        return {
            'time_statistics': time_stats,
            'token_efficiency': token_stats,
            'model_usage': model_stats,
            'timestamp': time.strftime('%Y-%m-%d %H:%M:%S')
        }

6.2 自动化监控脚本

创建一个可以定期运行的监控脚本：

def automated_monitoring(model, test_prompts, interval_minutes=60):
    """自动化监控任务"""
    monitor = SmallThinkerMonitor()
    
    while True:
        print(f"\n=== 开始监控周期 {time.strftime('%Y-%m-%d %H:%M:%S')} ===")
        
        # 运行测试提示
        for i, prompt in enumerate(test_prompts, 1):
            print(f"测试 {i}/{len(test_prompts)}: {prompt[:50]}...")
            result = monitor.monitor_generation(prompt)
            
            print(f"  耗时: {result['metrics']['inference_time']:.2f}秒")
            print(f"  生成: {result['metrics']['output_tokens']} tokens")
            print(f"  速度: {result['metrics']['tokens_per_second']:.1f} tokens/秒")
        
        # 生成报告
        report = monitor.generate_report()
        print("\n📊 本次监控报告:")
        print(f"总请求数: {report['model_usage']['总请求数']}")
        print(f"平均生成速度: {report['model_usage']['平均生成速度']}")
        print(f"平均Token效率: {report['token_efficiency']['avg_efficiency_ratio']:.2f}")
        
        # 等待下一个周期
        print(f"\n等待下一个监控周期（{interval_minutes}分钟后）...")
        time.sleep(interval_minutes * 60)

# 示例测试提示
test_prompts = [
    "请解释Transformer架构的核心思想",
    "写一个Python函数计算斐波那契数列",
    "用简单的语言说明机器学习的基本概念",
    "总结一下注意力机制在深度学习中的作用"
]

7. 总结

通过本文的实践指南，你应该已经掌握了SmallThinker-3B模型的日志监控、推理耗时统计和Token效率分析的关键技术。这些监控和分析能力对于在实际项目中优化模型使用、控制成本和提升用户体验都非常重要。

记住几个关键点：首先是要建立完整的监控体系，从日志记录到性能分析都要覆盖；其次是要定期分析模型性能，及时发现潜在问题；最后是要根据分析结果不断优化使用策略，提高整体效率。

SmallThinker-3B作为一个轻量级模型，在边缘设备和资源受限环境中有着独特的优势。通过有效的监控和优化，你可以充分发挥其潜力，在各种应用场景中获得良好的性能表现。

在实际使用中如果遇到任何问题，建议参考官方文档或者通过提供的联系方式寻求帮助。监控数据的长期积累和分析往往会带来意想不到的发现和优化机会。

---

获取更多AI镜像

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