Tolan/backend/analyze_performance.py

#!/usr/bin/env python3
"""
性能日志分析工具
分析后端日志，统计STT、LLM、TTS各阶段耗时
"""
import re
import sys
from typing import Dict, List
from dataclasses import dataclass, field
from statistics import mean, median


@dataclass
class RequestStats:
    """单个请求的统计数据"""
    request_id: str
    stt_duration: float = 0.0
    llm_duration: float = 0.0
    tts_duration: float = 0.0
    total_duration: float = 0.0
    user_input: str = ""
    response_text: str = ""
    emotion: str = ""
    
    def is_complete(self) -> bool:
        """检查是否有完整的统计数据"""
        return self.total_duration > 0


@dataclass
class PerformanceStats:
    """性能统计汇总"""
    total_requests: int = 0
    completed_requests: int = 0
    
    stt_times: List[float] = field(default_factory=list)
    llm_times: List[float] = field(default_factory=list)
    tts_times: List[float] = field(default_factory=list)
    total_times: List[float] = field(default_factory=list)
    
    def add_request(self, req: RequestStats):
        """添加一个请求的统计数据"""
        self.total_requests += 1
        if req.is_complete():
            self.completed_requests += 1
            self.stt_times.append(req.stt_duration)
            self.llm_times.append(req.llm_duration)
            self.tts_times.append(req.tts_duration)
            self.total_times.append(req.total_duration)
    
    def print_summary(self):
        """打印统计摘要"""
        print("\n" + "=" * 70)
        print("性能统计分析报告".center(70))
        print("=" * 70)
        
        print(f"\n总请求数: {self.total_requests}")
        print(f"完成请求数: {self.completed_requests}")
        
        if not self.completed_requests:
            print("\n⚠️  没有找到完整的请求数据")
            return
        
        # STT统计
        print("\n" + "-" * 70)
        print("🎤 STT (语音识别) 性能".center(70))
        print("-" * 70)
        self._print_stats("STT", self.stt_times)
        
        # LLM统计
        print("\n" + "-" * 70)
        print("🤖 LLM (大语言模型) 性能".center(70))
        print("-" * 70)
        self._print_stats("LLM", self.llm_times)
        
        # TTS统计
        print("\n" + "-" * 70)
        print("🔊 TTS (语音合成) 性能".center(70))
        print("-" * 70)
        self._print_stats("TTS", self.tts_times)
        
        # 总计统计
        print("\n" + "-" * 70)
        print("⏱️  总体性能".center(70))
        print("-" * 70)
        self._print_stats("总计", self.total_times)
        
        # 耗时占比
        print("\n" + "-" * 70)
        print("📊 耗时占比分析".center(70))
        print("-" * 70)
        avg_total = mean(self.total_times)
        avg_stt = mean(self.stt_times)
        avg_llm = mean(self.llm_times)
        avg_tts = mean(self.tts_times)
        
        stt_pct = (avg_stt / avg_total * 100) if avg_total > 0 else 0
        llm_pct = (avg_llm / avg_total * 100) if avg_total > 0 else 0
        tts_pct = (avg_tts / avg_total * 100) if avg_total > 0 else 0
        
        print(f"STT 占比: {stt_pct:5.1f}% | {'█' * int(stt_pct / 2)}")
        print(f"LLM 占比: {llm_pct:5.1f}% | {'█' * int(llm_pct / 2)}")
        print(f"TTS 占比: {tts_pct:5.1f}% | {'█' * int(tts_pct / 2)}")
        
        # 优化建议
        print("\n" + "-" * 70)
        print("💡 优化建议".center(70))
        print("-" * 70)
        
        max_stage = max([
            ("STT", avg_stt, stt_pct),
            ("LLM", avg_llm, llm_pct),
            ("TTS", avg_tts, tts_pct)
        ], key=lambda x: x[1])
        
        print(f"1. {max_stage[0]} 耗时最长 ({max_stage[1]:.2f}秒, {max_stage[2]:.1f}%)，建议优先优化")
        
        if avg_stt > 2.0:
            print("2. STT 响应较慢，建议检查网络连接或考虑本地STT模型")
        
        if avg_llm > 3.0:
            print("3. LLM 响应较慢，可考虑使用更快的模型或流式响应")
        
        if avg_tts > 2.0:
            print("4. TTS 合成较慢，可考虑音频缓存或预生成常用回复")
        
        print("\n" + "=" * 70 + "\n")
    
    def _print_stats(self, name: str, times: List[float]):
        """打印单个阶段的统计信息"""
        if not times:
            print(f"{name}: 无数据")
            return
        
        avg = mean(times)
        med = median(times)
        min_time = min(times)
        max_time = max(times)
        
        print(f"平均耗时: {avg:6.2f}秒")
        print(f"中位耗时: {med:6.2f}秒")
        print(f"最小耗时: {min_time:6.2f}秒")
        print(f"最大耗时: {max_time:6.2f}秒")
        print(f"样本数量: {len(times):6d}次")


def analyze_log(log_content: str) -> PerformanceStats:
    """分析日志内容"""
    stats = PerformanceStats()
    requests: Dict[str, RequestStats] = {}
    
    lines = log_content.split('\n')
    
    for line in lines:
        # 提取 request_id
        request_id_match = re.search(r'\[([a-f0-9\-]{36})\]', line)
        if not request_id_match:
            continue
        
        request_id = request_id_match.group(1)
        
        # 确保有这个请求的记录
        if request_id not in requests:
            requests[request_id] = RequestStats(request_id=request_id)
        
        req = requests[request_id]
        
        # 提取STT耗时
        stt_match = re.search(r'⏱️\s+STT耗时:\s+([\d.]+)秒', line)
        if stt_match:
            req.stt_duration = float(stt_match.group(1))
        
        # 提取LLM耗时
        llm_match = re.search(r'⏱️\s+LLM耗时:\s+([\d.]+)秒', line)
        if llm_match:
            req.llm_duration = float(llm_match.group(1))
        
        # 提取TTS耗时
        tts_match = re.search(r'⏱️\s+TTS耗时:\s+([\d.]+)秒', line)
        if tts_match:
            req.tts_duration = float(tts_match.group(1))
        
        # 提取总耗时
        total_match = re.search(r'⏱️\s+总耗时:\s+([\d.]+)秒', line)
        if total_match:
            req.total_duration = float(total_match.group(1))
        
        # 提取用户输入
        if 'STT识别结果:' in line:
            text_match = re.search(r'STT识别结果:\s*(.+)$', line)
            if text_match:
                req.user_input = text_match.group(1).strip()
        
        # 提取回复文本
        if '回复文本:' in line:
            text_match = re.search(r'回复文本:\s*(.+)$', line)
            if text_match:
                req.response_text = text_match.group(1).strip()
        
        # 提取情感
        if re.search(r'^情感:', line):
            emotion_match = re.search(r'情感:\s*(\w+)', line)
            if emotion_match:
                req.emotion = emotion_match.group(1)
    
    # 添加所有请求到统计
    for req in requests.values():
        stats.add_request(req)
    
    # 打印详细信息（可选）
    print("\n" + "=" * 70)
    print("详细请求列表".center(70))
    print("=" * 70)
    
    for i, req in enumerate(requests.values(), 1):
        if req.is_complete():
            print(f"\n请求 #{i} [{req.request_id[:8]}...]")
            print(f"  用户输入: {req.user_input[:50]}..." if len(req.user_input) > 50 else f"  用户输入: {req.user_input}")
            print(f"  情感: {req.emotion}")
            print(f"  STT: {req.stt_duration:.2f}秒 | LLM: {req.llm_duration:.2f}秒 | TTS: {req.tts_duration:.2f}秒 | 总计: {req.total_duration:.2f}秒")
    
    return stats


def main():
    """主函数"""
    if len(sys.argv) > 1:
        # 从文件读取
        log_file = sys.argv[1]
        print(f"正在分析日志文件: {log_file}")
        try:
            with open(log_file, 'r', encoding='utf-8') as f:
                log_content = f.read()
        except Exception as e:
            print(f"读取日志文件失败: {e}")
            sys.exit(1)
    else:
        # 从标准输入读取
        print("从标准输入读取日志内容 (Ctrl+D 结束输入)")
        log_content = sys.stdin.read()
    
    # 分析日志
    stats = analyze_log(log_content)
    
    # 打印统计
    stats.print_summary()


if __name__ == "__main__":
    main()