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):
        """打印统计摘要（已禁用）"""
        pass
    
    def _print_stats(self, name: str, times: List[float]):
        """打印单个阶段的统计信息（已禁用）"""
        pass


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)
    
    return stats


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


if __name__ == "__main__":
    main()