""" 请求日志记录器 - 记录每个语音对话请求的详细性能数据 使用JSON Lines格式，按天轮转日志文件 """ import json import time import queue import threading from datetime import datetime from pathlib import Path from typing import Optional, Dict, Any from dataclasses import dataclass from config import settings @dataclass class PerformanceMetrics: """性能指标数据类""" # 用户说话 user_speech_start: Optional[float] = None user_speech_end: Optional[float] = None user_text: str = "" # STT相关 stt_first_audio_received: Optional[float] = None stt_first_text_generated: Optional[float] = None stt_last_audio_received: Optional[float] = None stt_completed: Optional[float] = None stt_audio_chunks: int = 0 # 音频块数量 stt_total_bytes: int = 0 # 总字节数 # LLM提示词准备 llm_prompt_prepare_start: Optional[float] = None llm_prompt_prepare_end: Optional[float] = None llm_prompt_length: int = 0 # 提示词长度 llm_context_round: int = 0 # 上下文轮次 llm_full_prompt: str = "" # 完整prompt（system_prompt + user_input） # LLM生成 llm_call_start: Optional[float] = None llm_first_text_output: Optional[float] = None llm_all_text_completed: Optional[float] = None llm_call_end: Optional[float] = None llm_emotion: str = "neutral" # 情感 llm_sentences: int = 0 # 句子数 # TTS生成 tts_call_start: Optional[float] = None tts_first_audio_output: Optional[float] = None tts_all_audio_completed: Optional[float] = None tts_call_end: Optional[float] = None tts_voice: str = "Cherry" # 音色 tts_audio_chunks: int = 0 # 音频块数量 tts_total_bytes: int = 0 # 总字节数 # AI响应 ai_response_text: str = "" # 消息ID（数据库关联） user_message_id: Optional[str] = None # 用户消息的 message_id assistant_message_id: Optional[str] = None # AI消息的 message_id def calculate_user_speech_duration_ms(self) -> Optional[float]: """计算用户说话时长(ms)""" if self.user_speech_start and self.user_speech_end: return (self.user_speech_end - self.user_speech_start) * 1000 return None def calculate_stt_latency_ms(self) -> Optional[float]: """计算STT延迟(ms): STT完成时间 - 收到最后一个音频时间""" if self.stt_completed and self.stt_last_audio_received: return (self.stt_completed - self.stt_last_audio_received) * 1000 return None def calculate_llm_prompt_latency_ms(self) -> Optional[float]: """计算LLM提示词准备延迟(ms)""" if self.llm_prompt_prepare_end and self.llm_prompt_prepare_start: return (self.llm_prompt_prepare_end - self.llm_prompt_prepare_start) * 1000 return None def calculate_llm_generation_latency_ms(self) -> Optional[float]: """计算LLM生成延迟(ms)""" if self.llm_first_text_output and self.llm_call_start: return (self.llm_first_text_output - self.llm_call_start) * 1000 return None def calculate_tts_latency_ms(self) -> Optional[float]: """计算TTS延迟(ms)""" if self.tts_first_audio_output and self.tts_call_start: return (self.tts_first_audio_output - self.tts_call_start) * 1000 return None def calculate_total_latency_ms(self) -> Optional[float]: """计算总延迟(ms): 开始输出第一个音频时间 - 用户结束说话时间""" if self.tts_first_audio_output and self.user_speech_end: return (self.tts_first_audio_output - self.user_speech_end) * 1000 return None class RequestLogger: """请求日志记录器（线程安全，异步写入）""" def __init__(self, log_dir: str = "logs"): """ 初始化日志记录器 Args: log_dir: 日志目录 """ self.log_dir = Path(log_dir) self.log_dir.mkdir(parents=True, exist_ok=True) # 当前日志文件路径 self.current_log_file: Optional[Path] = None self.current_date: Optional[str] = None # 异步写入队列 self.log_queue: queue.Queue = queue.Queue() self.writer_thread: Optional[threading.Thread] = None self.running = True # 启动后台写入线程 self._start_writer_thread() def _get_log_file_path(self, date_str: str) -> Path: """获取指定日期的日志文件路径""" return self.log_dir / f"request_{date_str}.log" def _get_current_log_file(self) -> Path: """获取当前日期的日志文件路径（自动轮转）""" today = datetime.now().strftime("%Y-%m-%d") # 如果日期变化，切换日志文件 if self.current_date != today: self.current_date = today self.current_log_file = self._get_log_file_path(today) return self.current_log_file def _start_writer_thread(self): """启动后台写入线程""" def writer_loop(): """后台写入循环""" while self.running: try: # 从队列获取日志条目（超时1秒） log_entry = self.log_queue.get(timeout=1.0) # 获取当前日志文件 log_file = self._get_current_log_file() # 写入JSON格式日志文件（追加模式） with open(log_file, 'a', encoding='utf-8') as f: f.write(json.dumps(log_entry, ensure_ascii=False) + '\n') # 如果是请求日志，同时写入conversation_log.txt if log_entry.get("类型") == "请求": self._write_conversation_log(log_entry) self.log_queue.task_done() except queue.Empty: continue except Exception as e: # 写入失败，记录到stderr（避免循环） import sys print(f"[RequestLogger] 写入日志失败: {e}", file=sys.stderr) self.writer_thread = threading.Thread(target=writer_loop, daemon=True) self.writer_thread.start() def _format_timestamp(self, timestamp: Optional[float]) -> Optional[str]: """格式化时间戳为ISO 8601格式""" if timestamp is None: return None dt = datetime.fromtimestamp(timestamp) return dt.strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "Z" def _format_iso_time_to_conversation_log(self, iso_time: Optional[str]) -> str: """将ISO时间戳格式化为conversation_log.txt格式 (HH:MM:SS.mmm)""" if not iso_time: return "N/A" try: # 从ISO格式解析: 2025-11-11T22:42:46.123Z dt = datetime.fromisoformat(iso_time.replace('Z', '+00:00')) return dt.strftime("%H:%M:%S.%f")[:-3] except: return "N/A" def _write_conversation_log(self, log_entry: Dict[str, Any]): """写入conversation_log.txt格式的日志""" try: conversation_log_path = Path("conversation_log.txt") stages = log_entry.get("阶段", {}) summary = log_entry.get("总结", {}) request_id = log_entry.get("请求ID", "unknown") timestamp = log_entry.get("时间戳", "") # 解析时间戳获取日期时间 if timestamp: try: # 从ISO格式解析: 2025-11-11T22:42:46.123Z dt = datetime.fromisoformat(timestamp.replace('Z', '+00:00')) date_str = dt.strftime("%Y-%m-%d %H:%M:%S") except: date_str = datetime.now().strftime("%Y-%m-%d %H:%M:%S") else: date_str = datetime.now().strftime("%Y-%m-%d %H:%M:%S") lines = [] lines.append(f"[{date_str}] [{request_id}]") # 用户说话 user_speech = stages.get("用户说话", {}) user_text = "" if "语音识别" in stages and "输出" in stages["语音识别"]: user_text = stages["语音识别"]["输出"].get("文本", "") user_start = user_speech.get("开始时间") user_end = user_speech.get("结束时间") user_duration = user_speech.get("时长_毫秒") user_start_str = self._format_iso_time_to_conversation_log(user_start) user_end_str = self._format_iso_time_to_conversation_log(user_end) user_duration_str = f"{int(user_duration)}ms" if user_duration else "N/A" lines.append(f"用户说话: {user_start_str}, {user_text}, {user_end_str}, {user_duration_str}") # STT stt = stages.get("语音识别", {}) stt_first_audio = stt.get("首次收到音频") stt_first_text = stt.get("首次生成文本") stt_last_audio = stt.get("最后收到音频") stt_completed = stt.get("完成时间") stt_latency = stt.get("延迟_毫秒") stt_first_audio_str = self._format_iso_time_to_conversation_log(stt_first_audio) stt_first_text_str = self._format_iso_time_to_conversation_log(stt_first_text) stt_last_audio_str = self._format_iso_time_to_conversation_log(stt_last_audio) stt_completed_str = self._format_iso_time_to_conversation_log(stt_completed) stt_latency_str = f"{int(stt_latency)}ms" if stt_latency else "N/A" lines.append(f"STT: {stt_first_audio_str}, {stt_first_text_str}, {stt_last_audio_str}, {stt_completed_str}, {stt_latency_str}") # LLM提示词 llm_prompt = stages.get("LLM提示词准备", {}) llm_prompt_start = llm_prompt.get("开始时间") llm_prompt_end = llm_prompt.get("结束时间") llm_prompt_duration = llm_prompt.get("时长_毫秒") llm_prompt_start_str = self._format_iso_time_to_conversation_log(llm_prompt_start) llm_prompt_end_str = self._format_iso_time_to_conversation_log(llm_prompt_end) llm_prompt_duration_str = f"{int(llm_prompt_duration)}ms" if llm_prompt_duration else "N/A" lines.append(f"LLM提示词: {llm_prompt_start_str}, {llm_prompt_end_str}, {llm_prompt_duration_str}") # LLM生成 llm = stages.get("LLM生成", {}) llm_call_start = llm.get("调用开始") llm_first_text = llm.get("首次文本输出") llm_all_completed = llm.get("全部文本完成") llm_call_end = llm.get("调用结束") llm_latency = llm.get("延迟_毫秒") llm_call_start_str = self._format_iso_time_to_conversation_log(llm_call_start) llm_first_text_str = self._format_iso_time_to_conversation_log(llm_first_text) llm_all_completed_str = self._format_iso_time_to_conversation_log(llm_all_completed) llm_call_end_str = self._format_iso_time_to_conversation_log(llm_call_end) llm_latency_str = f"{int(llm_latency)}ms" if llm_latency else "N/A" lines.append(f"LLM生成: {llm_call_start_str}, {llm_first_text_str}, {llm_all_completed_str}, {llm_call_end_str}, {llm_latency_str}") # TTS生成 tts = stages.get("TTS生成", {}) tts_call_start = tts.get("调用开始") tts_first_audio = tts.get("首次音频输出") tts_all_completed = tts.get("全部音频完成") tts_call_end = tts.get("调用结束") tts_latency = tts.get("延迟_毫秒") tts_call_start_str = self._format_iso_time_to_conversation_log(tts_call_start) tts_first_audio_str = self._format_iso_time_to_conversation_log(tts_first_audio) tts_all_completed_str = self._format_iso_time_to_conversation_log(tts_all_completed) tts_call_end_str = self._format_iso_time_to_conversation_log(tts_call_end) tts_latency_str = f"{int(tts_latency)}ms" if tts_latency else "N/A" lines.append(f"TTS生成: {tts_call_start_str}, {tts_first_audio_str}, {tts_all_completed_str}, {tts_call_end_str}, {tts_latency_str}") # 总结 ai_response_text = "" if "LLM生成" in stages and "输出" in stages["LLM生成"]: ai_response_text = stages["LLM生成"]["输出"].get("文本", "") total_latency = summary.get("首次音频延迟_毫秒") total_latency_str = f"{int(total_latency)}ms" if total_latency else "N/A" lines.append(f"总结: {ai_response_text}, {total_latency_str}") lines.append("-" * 120) lines.append("") # 写入文件 with open(conversation_log_path, 'a', encoding='utf-8') as f: f.write('\n'.join(lines) + '\n') except Exception as e: # 写入失败不影响主流程 import sys print(f"[RequestLogger] 写入conversation_log.txt失败: {e}", file=sys.stderr) def _build_request_log( self, request_id: str, user_id: str, metrics: PerformanceMetrics ) -> Dict[str, Any]: """构建请求日志JSON对象""" # 构建各阶段数据 stages = {} # 用户说话阶段 if metrics.user_speech_start and metrics.user_speech_end: stages["用户说话"] = { "开始时间": self._format_timestamp(metrics.user_speech_start), "结束时间": self._format_timestamp(metrics.user_speech_end), "时长_毫秒": metrics.calculate_user_speech_duration_ms() } # STT阶段 if metrics.stt_first_audio_received: stt_stage = { "首次收到音频": self._format_timestamp(metrics.stt_first_audio_received), "最后收到音频": self._format_timestamp(metrics.stt_last_audio_received), "完成时间": self._format_timestamp(metrics.stt_completed), "输入": { "音频块数": metrics.stt_audio_chunks, "总字节数": metrics.stt_total_bytes }, "输出": { "文本": metrics.user_text, "是否最终": True } } if metrics.stt_first_text_generated: stt_stage["首次生成文本"] = self._format_timestamp(metrics.stt_first_text_generated) if metrics.stt_completed and metrics.stt_last_audio_received: stt_stage["延迟_毫秒"] = metrics.calculate_stt_latency_ms() stages["语音识别"] = stt_stage # LLM提示词准备阶段 if metrics.llm_prompt_prepare_start and metrics.llm_prompt_prepare_end: stages["LLM提示词准备"] = { "开始时间": self._format_timestamp(metrics.llm_prompt_prepare_start), "结束时间": self._format_timestamp(metrics.llm_prompt_prepare_end), "时长_毫秒": metrics.calculate_llm_prompt_latency_ms() } # LLM生成阶段 if metrics.llm_call_start: llm_stage = { "调用开始": self._format_timestamp(metrics.llm_call_start), "调用结束": self._format_timestamp(metrics.llm_call_end), "输入": { "提示词长度": metrics.llm_prompt_length, "上下文轮次": metrics.llm_context_round, "完整提示词": metrics.llm_full_prompt if metrics.llm_full_prompt else "" }, "输出": { "文本": metrics.ai_response_text, "情感": metrics.llm_emotion, "句子数": metrics.llm_sentences } } if metrics.llm_first_text_output: llm_stage["首次文本输出"] = self._format_timestamp(metrics.llm_first_text_output) if metrics.llm_all_text_completed: llm_stage["全部文本完成"] = self._format_timestamp(metrics.llm_all_text_completed) if metrics.llm_first_text_output and metrics.llm_call_start: llm_stage["延迟_毫秒"] = metrics.calculate_llm_generation_latency_ms() stages["LLM生成"] = llm_stage # TTS阶段 if metrics.tts_call_start: tts_stage = { "调用开始": self._format_timestamp(metrics.tts_call_start), "调用结束": self._format_timestamp(metrics.tts_call_end), "输入": { "文本": metrics.ai_response_text, "情感": metrics.llm_emotion, "音色": metrics.tts_voice }, "输出": { "音频块数": metrics.tts_audio_chunks, "总字节数": metrics.tts_total_bytes } } if metrics.tts_first_audio_output: tts_stage["首次音频输出"] = self._format_timestamp(metrics.tts_first_audio_output) if metrics.tts_all_audio_completed: tts_stage["全部音频完成"] = self._format_timestamp(metrics.tts_all_audio_completed) if metrics.tts_first_audio_output and metrics.tts_call_start: tts_stage["延迟_毫秒"] = metrics.calculate_tts_latency_ms() stages["TTS生成"] = tts_stage # 构建总结 summary = {} if metrics.tts_first_audio_output and metrics.user_speech_end: summary["首次音频延迟_毫秒"] = metrics.calculate_total_latency_ms() if metrics.tts_all_audio_completed and metrics.user_speech_start: total_duration = (metrics.tts_all_audio_completed - metrics.user_speech_start) * 1000 summary["总延迟_毫秒"] = total_duration # 构建完整日志对象 log_entry = { "类型": "请求", "时间戳": self._format_timestamp(time.time()), "请求ID": request_id, "用户ID": user_id, "阶段": stages } if summary: log_entry["总结"] = summary # 添加消息ID（数据库关联） if metrics.user_message_id: log_entry["用户消息ID"] = metrics.user_message_id if metrics.assistant_message_id: log_entry["助手消息ID"] = metrics.assistant_message_id return log_entry def log_request( self, user_id: str, request_id: str, metrics: PerformanceMetrics ): """ 记录一条对话请求 Args: user_id: 用户ID request_id: 请求ID metrics: 性能指标对象 """ try: log_entry = self._build_request_log(request_id, user_id, metrics) self.log_queue.put(log_entry) except Exception as e: # 记录失败不影响主流程 import sys print(f"[RequestLogger] 记录请求日志失败: {e}", file=sys.stderr) def log_error( self, user_id: str, request_id: str, error_code: str, error_message: str, stage: Optional[str] = None, user_text: Optional[str] = None ): """ 记录错误请求 Args: user_id: 用户ID request_id: 请求ID error_code: 错误代码 error_message: 错误信息 stage: 错误发生的阶段（可选） user_text: 用户输入文本（可选） """ try: log_entry = { "类型": "错误", "时间戳": self._format_timestamp(time.time()), "请求ID": request_id, "用户ID": user_id, "错误": { "代码": error_code, "消息": error_message } } if stage: log_entry["错误"]["阶段"] = stage if user_text: log_entry["上下文"] = { "用户文本": user_text, "阶段": stage or "unknown" } self.log_queue.put(log_entry) except Exception as e: import sys print(f"[RequestLogger] 记录错误日志失败: {e}", file=sys.stderr) def shutdown(self): """关闭日志记录器（等待队列清空）""" self.running = False if self.writer_thread: self.log_queue.join() # 等待队列清空 self.writer_thread.join(timeout=5.0) # 全局单例 request_logger = RequestLogger(log_dir=settings.LOG_DIR)