#!/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()