""" WebSocket处理器 - 新版本（STT+LLM+TTS流水线，支持流式LLM） """ import json import base64 import asyncio import time import re import logging import traceback from typing import Any, Dict, Optional from fastapi import WebSocket logger = logging.getLogger(__name__) # 预编译正则表达式（性能优化） EMOTION_TAG_PATTERN = re.compile(r'\[EMOTION:\w+\]') from models import ( AudioChunkMessage, AudioEndMessage, CancelMessage, TextResultMessage, EmotionMessage, AudioResponseMessage, ErrorMessage, PingMessage, PongMessage, EnergyStoneEventMessage, EnergyStoneSpawnMessage, UserLevelMessage, ) from ai_service import ai_service from request_logger import request_logger, PerformanceMetrics from config import settings from stt_service import stt_service, STTSession from services.user_service import user_service from services.energy_stone_service import energy_stone_service class WebSocketHandler: """WebSocket连接处理器""" def __init__(self, websocket: WebSocket, user_id: str = None, redis_cache = None): self.websocket = websocket self.user_id = user_id or "guest_user" # 用户ID（默认游客） self.cancelled_requests: set = set() # 已取消的请求ID self.stt_sessions: Dict[str, STTSession] = {} # requestId -> STT会话对象 self.event_loop: Optional[asyncio.AbstractEventLoop] = None # 保存事件循环引用 self.request_start_times: Dict[str, float] = {} # 请求开始时间 self.audio_end_times: Dict[str, float] = {} # requestId -> 录音结束时间 self.ai_responses: Dict[str, list] = {} # requestId -> AI响应句子列表（用于日志记录） self.performance_metrics: Dict[str, PerformanceMetrics] = {} # requestId -> 性能指标 self.energy_stone_service = None # 初始化服务（延迟加载，避免循环导入） self.message_service = None self.context_service = None self.redis_cache = redis_cache # 🔥 Redis缓存服务 # 🔥 预加载任务 self.preload_tasks: Dict[str, asyncio.Task] = {} # 🔥 异步持久化队列 self.pending_persistence = [] self.persistence_task: Optional[asyncio.Task] = None self.persistence_interval = 5.0 # 每5秒批量持久化 # 🔥 异步总结队列 self.pending_summaries = [] self.summary_task: Optional[asyncio.Task] = None async def _init_services(self): """初始化服务（延迟加载，避免重复初始化）""" start_time = time.time() # 🔥 如果服务已初始化，直接返回（避免重复初始化导致的阻塞） if self.message_service and self.context_service: logger.debug(f"[{self.user_id}] 🔍 服务已初始化，跳过重复初始化") return try: logger.info(f"[{self.user_id}] 🔧 开始初始化服务...") from database import db_manager from services import MessageService, ContextService # 使用延迟初始化的全局实例 from services.message_service import message_service from services.context_service import context_service # 🔥 只在服务未初始化时才设置（避免重复调用set_db_manager） if not message_service.db or not context_service.db: logger.debug(f"[{self.user_id}] 🔧 设置数据库管理器...") await message_service.set_db_manager(db_manager) await context_service.set_db_manager(db_manager) if not energy_stone_service.db: logger.debug(f"[{self.user_id}] 🔧 设置能量石服务数据库管理器...") await energy_stone_service.set_db_manager(db_manager) self.message_service = message_service self.context_service = context_service self.energy_stone_service = energy_stone_service elapsed = (time.time() - start_time) * 1000 logger.info(f"[{self.user_id}] ✅ 服务初始化完成，耗时: {elapsed:.1f}ms") except Exception as e: elapsed = (time.time() - start_time) * 1000 logger.error(f"[{self.user_id}] ❌ 服务初始化失败，耗时: {elapsed:.1f}ms, 错误: {e}") logger.error(traceback.format_exc()) async def handle_connection(self): """处理WebSocket连接""" await self.websocket.accept() # 初始化服务 await self._init_services() # 保存事件循环引用，供回调线程使用 self.event_loop = asyncio.get_running_loop() try: while True: # 接收消息 data = await self.websocket.receive_text() message_data = json.loads(data) # 处理消息 await self.handle_message(message_data) except Exception as e: pass finally: pass async def handle_message(self, message_data: dict): """处理接收到的消息""" message_type = message_data.get("type") request_id = message_data.get("requestId") # 检查是否为已取消的请求 if request_id in self.cancelled_requests: return if message_type == "audio_chunk": await self.handle_audio_chunk(AudioChunkMessage(**message_data)) elif message_type == "audio_end": await self.handle_audio_end(AudioEndMessage(**message_data)) elif message_type == "cancel": await self.handle_cancel(CancelMessage(**message_data)) elif message_type == "ping": await self.handle_ping(PingMessage(**message_data)) elif message_type == "energy_stone_event": await self.handle_energy_stone_event(EnergyStoneEventMessage(**message_data)) async def handle_audio_chunk(self, message: AudioChunkMessage): """处理音频数据块 - 使用新的STT服务""" request_id = message.requestId # 检查是否取消 if request_id in self.cancelled_requests: return # Base64解码音频数据 audio_data = base64.b64decode(message.data) chunk_size = len(audio_data) # 第一个chunk：初始化请求并启动STT会话 if request_id not in self.request_start_times: current_time = time.time() self.request_start_times[request_id] = current_time # 初始化性能指标 self.performance_metrics[request_id] = PerformanceMetrics() self.performance_metrics[request_id].user_speech_start = current_time self.performance_metrics[request_id].stt_first_audio_received = current_time # 创建STT会话 session = await self._create_stt_session(request_id) if not session or not session.is_connected: await self.send_error(request_id, "STT_ERROR", "语音识别服务启动失败") return self.stt_sessions[request_id] = session # 🔥 启动Redis预加载任务（与录音并行） if self.redis_cache: self.preload_tasks[request_id] = asyncio.create_task( self._ensure_redis_context(request_id) ) # 发送音频帧到STT服务 session = self.stt_sessions.get(request_id) if session: success = await session.send_audio(audio_data) if not success: await self.send_error(request_id, "STT_ERROR", "发送音频数据失败") # 📝 更新STT音频统计（用于日志） if request_id in self.performance_metrics: self.performance_metrics[request_id].stt_audio_chunks += 1 self.performance_metrics[request_id].stt_total_bytes += chunk_size async def handle_audio_end(self, message: AudioEndMessage): """处理音频结束 - 使用新的STT服务""" request_id = message.requestId # 记录录音结束时间 audio_end_time = time.time() self.audio_end_times[request_id] = audio_end_time # 更新性能指标 if request_id in self.performance_metrics: self.performance_metrics[request_id].user_speech_end = audio_end_time self.performance_metrics[request_id].stt_last_audio_received = audio_end_time speech_duration_ms = message.speechDuration or 0 total_duration_ms = None level_update = None # 🔥 更新用户累积说话时长（如果有提供，使用超时避免阻塞） if speech_duration_ms > 0: duration_start = time.time() try: logger.info(f"[{request_id}] ⏱️ 开始更新用户说话时长: {speech_duration_ms}ms（超时: 1.0s）...") # 使用超时，避免长时间阻塞 level_update = await asyncio.wait_for( user_service.update_speech_duration(self.user_id, speech_duration_ms), timeout=1.0 # 1秒超时 ) duration_elapsed = (time.time() - duration_start) * 1000 logger.info(f"[{request_id}] ✅ 更新用户说话时长成功，耗时: {duration_elapsed:.1f}ms") except asyncio.TimeoutError: duration_elapsed = (time.time() - duration_start) * 1000 logger.warning(f"[{request_id}] ⚠️ 更新用户说话时长超时（耗时: {duration_elapsed:.1f}ms）: user_id={self.user_id}, duration_ms={speech_duration_ms}") except Exception as e: duration_elapsed = (time.time() - duration_start) * 1000 logger.warning(f"[{request_id}] ⚠️ 更新用户说话时长失败（耗时: {duration_elapsed:.1f}ms）: user_id={self.user_id}, duration_ms={speech_duration_ms}, error={e}") # 初始化AI响应收集器 self.ai_responses[request_id] = [] if level_update and isinstance(level_update, dict): total_duration_ms = level_update.get("total_duration_ms") user_level = level_update.get("user_level") if total_duration_ms is not None: await self._maybe_spawn_conversation_stones(total_duration_ms, speech_duration_ms, request_id) # 🔥 即使等级没有变化，也发送 user_level 消息，确保前端能同步 total_duration_ms # 获取当前用户等级（如果更新时等级没变，user_level 可能为 None，需要查询） if user_level is not None: await self.send_user_level_update( request_id=request_id, level=user_level, total_duration_ms=total_duration_ms or 0, ) elif total_duration_ms is not None: # 等级没变，但总时长更新了，需要查询当前等级并发送消息 try: user_info = await user_service.get_user(self.user_id) if user_info and user_info.get("user_level") is not None: await self.send_user_level_update( request_id=request_id, level=user_info.get("user_level"), total_duration_ms=total_duration_ms, ) except Exception as e: logger.warning(f"[{request_id}] ⚠️ 查询用户等级失败，无法发送 user_level 消息: {e}") # 检查是否取消 if request_id in self.cancelled_requests: self.cancelled_requests.remove(request_id) await self.cleanup_request(request_id) return try: # 1. STT: 等待完整识别结果 stt_start = time.time() session = self.stt_sessions.get(request_id) if not session: await self.send_error(request_id, "STT_ERROR", "语音识别会话不存在") request_logger.log_error( user_id=self.user_id, request_id=request_id, error_code="STT_ERROR", error_message="语音识别会话不存在", stage="stt" ) return # 🔥 手动提交音频，触发识别（禁用VAD时必需） try: commit_success = await session.commit_audio() if not commit_success: logger.error( f"[STT异常] request_id={request_id}, " f"错误类型=提交识别请求失败, " f"用户ID={self.user_id}, " f"音频块数={message.totalChunks}, " f"会话状态=is_connected={session.is_connected}" ) await self.send_error(request_id, "STT_ERROR", "提交识别请求失败") request_logger.log_error( user_id=self.user_id, request_id=request_id, error_code="STT_ERROR", error_message="提交识别请求失败", stage="stt" ) await self.cleanup_request(request_id) return except Exception as commit_error: logger.error( f"[STT异常] request_id={request_id}, " f"错误类型=提交音频异常, " f"异常类型={type(commit_error).__name__}, " f"错误消息={str(commit_error)}, " f"用户ID={self.user_id}, " f"音频块数={message.totalChunks}, " f"堆栈跟踪:\n{traceback.format_exc()}" ) await self.send_error(request_id, "STT_ERROR", f"提交音频异常: {str(commit_error)}") request_logger.log_error( user_id=self.user_id, request_id=request_id, error_code="STT_ERROR", error_message=f"提交音频异常: {str(commit_error)}", stage="stt" ) await self.cleanup_request(request_id) return # 等待最终结果（根据音频长度动态调整超时） total_chunks = message.totalChunks if total_chunks > 80: timeout = 5.0 elif total_chunks > 50: timeout = 4.0 else: timeout = 3.0 logger.info(f"[{request_id}] ⏳ 等待STT最终识别结果（超时: {timeout}s）...") text = await session.wait_for_final_result(timeout=timeout) # 获取识别结果 if not text: # 检查语音检测状态，提供更详细的错误信息 callback = session.callback speech_detected = callback.speech_detected if callback else False speech_start_count = callback.speech_start_count if callback else 0 if not speech_detected and speech_start_count == 0: error_msg = "未检测到有效语音，请检查麦克风权限或重新录音" else: error_msg = f"语音识别超时，已检测到语音但未返回识别结果（speech_detected={speech_detected}, speech_start_count={speech_start_count}）" logger.error( f"[{request_id}] ❌ STT识别失败: {error_msg}, " f"音频块数={message.totalChunks}, " f"录音时长={message.speechDuration}ms" ) await self.send_error(request_id, "STT_ERROR", error_msg) request_logger.log_error( user_id=self.user_id, request_id=request_id, error_code="STT_ERROR", error_message=error_msg, stage="stt" ) await self.cleanup_request(request_id) return stt_duration = time.time() - stt_start stt_complete_time = time.time() # 记录STT完成时间（在获取到text之后） stt_complete_elapsed = (stt_complete_time - self.request_start_times.get(request_id, stt_complete_time)) * 1000 logger.info(f"[{request_id}] ✅ STT识别完成，识别文本: '{text[:50]}{'...' if len(text) > 50 else ''}'，耗时: {stt_complete_elapsed:.1f}ms") # 更新性能指标 if request_id in self.performance_metrics: self.performance_metrics[request_id].stt_completed = stt_complete_time self.performance_metrics[request_id].user_text = text # 获取STT首个文本生成时间 first_text_time = session.get_first_text_time() if first_text_time: self.performance_metrics[request_id].stt_first_text_generated = first_text_time # 发送最终识别结果 await self.send_text_result(request_id, text, True) if request_id in self.cancelled_requests: return # 🔥 LLM提示词准备开始 llm_prompt_start = time.time() if request_id in self.performance_metrics: self.performance_metrics[request_id].llm_prompt_prepare_start = llm_prompt_start # 🔥 从Redis获取上下文（零延迟） context = None current_round = 0 next_round = 0 if self.redis_cache: # 尝试从Redis获取上下文（使用超时，避免长时间阻塞） redis_start = time.time() try: logger.info(f"[{request_id}] 🔍 开始从Redis获取上下文（超时: 2.0s）...") context = await asyncio.wait_for( self._get_context_from_redis(request_id), timeout=2.0 # 2秒超时 ) redis_elapsed = (time.time() - redis_start) * 1000 logger.info(f"[{request_id}] ✅ Redis上下文获取完成，耗时: {redis_elapsed:.1f}ms") except asyncio.TimeoutError: redis_elapsed = (time.time() - redis_start) * 1000 logger.warning(f"[{request_id}] ⚠️ 从Redis获取上下文超时（耗时: {redis_elapsed:.1f}ms），降级到MySQL") context = None if context: current_round = context.get('current_round', 0) next_round = current_round + 1 logger.debug(f"[{request_id}] 📊 当前轮次: {current_round}，下一轮次: {next_round}") # 🔥 立即写入Redis（用户消息，使用超时） write_start = time.time() try: logger.info(f"[{request_id}] 💾 开始写入Redis（用户消息，超时: 1.0s）...") await asyncio.wait_for( self.redis_cache.append_message( user_id=self.user_id, role='user', content=text, round_number=next_round ), timeout=1.0 # 1秒超时 ) append_elapsed = (time.time() - write_start) * 1000 logger.debug(f"[{request_id}] ✅ Redis写入用户消息完成，耗时: {append_elapsed:.1f}ms") increment_start = time.time() await asyncio.wait_for( self.redis_cache.increment_round(self.user_id), timeout=1.0 # 1秒超时 ) increment_elapsed = (time.time() - increment_start) * 1000 write_total = (time.time() - write_start) * 1000 logger.info(f"[{request_id}] ✅ Redis写入完成（增量轮次耗时: {increment_elapsed:.1f}ms，总耗时: {write_total:.1f}ms）") except asyncio.TimeoutError: write_elapsed = (time.time() - write_start) * 1000 logger.warning(f"[{request_id}] ⚠️ Redis写入超时（耗时: {write_elapsed:.1f}ms），但继续处理") except Exception as e: write_elapsed = (time.time() - write_start) * 1000 logger.warning(f"[{request_id}] ⚠️ Redis写入失败（耗时: {write_elapsed:.1f}ms）: {e}") # 🔥 异步保存user消息到MySQL（不阻塞主流程） await self._init_services() if self.message_service: asyncio.create_task( self.message_service.save_message( user_id=self.user_id, role='user', content=text ) ) else: # Redis未命中，降级到MySQL mysql_start = time.time() logger.info(f"[{request_id}] ⚠️ Redis未命中，降级到MySQL获取上下文...") await self._init_services() if self.message_service and self.context_service: try: # 保存用户消息 save_start = time.time() logger.info(f"[{request_id}] 💾 开始保存用户消息到MySQL（超时: 5.0s）...") user_message = await asyncio.wait_for( self.message_service.save_message( user_id=self.user_id, role='user', content=text ), timeout=5.0 ) save_elapsed = (time.time() - save_start) * 1000 next_round = user_message['round_number'] current_round = next_round - 1 logger.info(f"[{request_id}] ✅ 用户消息保存成功，轮次: {next_round}，耗时: {save_elapsed:.1f}ms") # 📝 记录用户消息ID if request_id in self.performance_metrics: self.performance_metrics[request_id].user_message_id = user_message.get('message_id') # 获取上下文 context_start = time.time() logger.info(f"[{request_id}] 🔍 开始从MySQL获取上下文（超时: 3.0s）...") context = await asyncio.wait_for( self.context_service.get_context_for_round( user_id=self.user_id, current_round=current_round ), timeout=3.0 ) context_elapsed = (time.time() - context_start) * 1000 mysql_total_elapsed = (time.time() - mysql_start) * 1000 if context: messages_count = len(context.get('messages', [])) logger.info(f"[{request_id}] ✅ 从MySQL获取上下文成功，消息数: {messages_count}，获取耗时: {context_elapsed:.1f}ms，总耗时: {mysql_total_elapsed:.1f}ms") else: logger.warning(f"[{request_id}] ⚠️ 从MySQL获取上下文为空，获取耗时: {context_elapsed:.1f}ms，总耗时: {mysql_total_elapsed:.1f}ms") except asyncio.TimeoutError as e: elapsed = (time.time() - mysql_start) * 1000 logger.error(f"[{request_id}] ❌ MySQL操作超时，耗时: {elapsed:.1f}ms, 错误: {e}") except Exception as e: elapsed = (time.time() - mysql_start) * 1000 logger.error(f"[{request_id}] ❌ MySQL操作失败，耗时: {elapsed:.1f}ms, 错误: {e}") logger.error(traceback.format_exc()) else: # 没有Redis，直接使用MySQL await self._init_services() if self.message_service and self.context_service: try: user_message = await asyncio.wait_for( self.message_service.save_message( user_id=self.user_id, role='user', content=text ), timeout=5.0 ) next_round = user_message['round_number'] current_round = next_round - 1 # 📝 记录用户消息ID if request_id in self.performance_metrics: self.performance_metrics[request_id].user_message_id = user_message.get('message_id') context = await asyncio.wait_for( self.context_service.get_context_for_round( user_id=self.user_id, current_round=current_round ), timeout=3.0 ) except asyncio.TimeoutError: pass except Exception as e: pass # 🔥 LLM提示词准备结束 llm_prompt_end = time.time() if request_id in self.performance_metrics: self.performance_metrics[request_id].llm_prompt_prepare_end = llm_prompt_end llm_start = time.time() # 🔥 LLM调用开始 if request_id in self.performance_metrics: self.performance_metrics[request_id].llm_call_start = llm_start chunk_index = 0 first_audio_time = None final_emotion = "neutral" sentence_count = 0 llm_first_sentence_time = None llm_first_text_time = None # 🔥 LLM首个文本输出时间 emotion_sent = False # 标记是否已发送情感 # 构建Prompt（如果有上下文，使用上下文；否则使用原始文本） prompt_build_start = time.time() llm_input = self._build_llm_input(text, context) if context else text prompt_build_elapsed = (time.time() - prompt_build_start) * 1000 # 获取system_prompt用于日志记录 system_prompt = ai_service._build_system_prompt() full_prompt = f"System: {system_prompt}\n\nUser: {llm_input}" # 📝 更新LLM输入统计（用于日志） if request_id in self.performance_metrics: self.performance_metrics[request_id].llm_prompt_length = len(llm_input) self.performance_metrics[request_id].llm_context_round = next_round self.performance_metrics[request_id].llm_full_prompt = full_prompt logger.info(f"[{request_id}] 🤖 开始调用LLM，Prompt长度: {len(llm_input)}，构建耗时: {prompt_build_elapsed:.1f}ms，轮次: {next_round}") # 🔥 流式处理：LLM生成一句 → 立即TTS try: llm_first_chunk_time = None async for sentence, emotion in ai_service.generate_response_stream(llm_input): if llm_first_chunk_time is None: llm_first_chunk_time = time.time() llm_first_chunk_elapsed = (llm_first_chunk_time - llm_start) * 1000 logger.info(f"[{request_id}] ✅ LLM首个响应到达，耗时: {llm_first_chunk_elapsed:.1f}ms") if request_id in self.cancelled_requests: return # 🎯 优先处理情感：第一时间发送给前端 if not emotion_sent and emotion: await self.send_emotion(request_id, emotion) emotion_sent = True final_emotion = emotion # ⚡ 如果是空句子（仅传递情感），跳过 TTS 处理 if not sentence: continue sentence_count += 1 # 🧹 清理文本：移除情感标记和特殊字符（保留文字内容） clean_sentence = EMOTION_TAG_PATTERN.sub('', sentence) clean_sentence = clean_sentence.translate(str.maketrans('', '', '()[]')).strip() # 如果清理后为空，跳过 if not clean_sentence: continue # 收集AI响应（用于日志记录） self.ai_responses[request_id].append(clean_sentence) # 记录第一个句子生成时间 if llm_first_sentence_time is None: llm_first_sentence_time = time.time() # 🔥 记录LLM首个文本输出时间 if llm_first_text_time is None: llm_first_text_time = time.time() if request_id in self.performance_metrics: self.performance_metrics[request_id].llm_first_text_output = llm_first_text_time tts_sentence_start = time.time() # 🔥 记录TTS调用开始时间（第一个句子） if request_id in self.performance_metrics and self.performance_metrics[request_id].tts_call_start is None: self.performance_metrics[request_id].tts_call_start = tts_sentence_start # 立即TTS合成这个句子（使用清理后的文本 + 情感参数） try: async for audio_chunk in ai_service.text_to_speech_stream(clean_sentence, emotion): if request_id in self.cancelled_requests: return # Base64编码 audio_base64 = base64.b64encode(audio_chunk).decode() # 发送音频数据 await self.send_audio_response( request_id, audio_base64, chunk_index, False, ) # 记录第一个音频块发送时间 if chunk_index == 0: first_audio_time = time.time() time_to_first_audio = first_audio_time - audio_end_time # 🔥 记录TTS首个音频输出时间 if request_id in self.performance_metrics: self.performance_metrics[request_id].tts_first_audio_output = first_audio_time # 📝 更新TTS音频统计（用于日志） if request_id in self.performance_metrics: self.performance_metrics[request_id].tts_audio_chunks += 1 self.performance_metrics[request_id].tts_total_bytes += len(audio_chunk) chunk_index += 1 except Exception as tts_error: # TTS异常已在ai_service中详细记录，这里只记录上下文信息 logger.error( f"[WebSocket异常] request_id={request_id}, " f"错误类型=TTS合成异常, " f"异常类型={type(tts_error).__name__}, " f"错误消息={str(tts_error)}, " f"句子内容={clean_sentence[:100]}... (前100字符), " f"情感={emotion}, " f"句子序号={sentence_count}, " f"已发送chunk数={chunk_index}, " f"堆栈跟踪:\n{traceback.format_exc()}" ) # 重新抛出异常，让外层处理 raise # 更新当前情感（但不再发送，因为已在开头发送） if emotion: final_emotion = emotion except Exception as llm_error: # LLM异常已在ai_service中详细记录，这里只记录上下文信息 logger.error( f"[WebSocket异常] request_id={request_id}, " f"错误类型=LLM生成异常, " f"异常类型={type(llm_error).__name__}, " f"错误消息={str(llm_error)}, " f"用户输入={text[:100]}... (前100字符), " f"LLM输入={llm_input[:100]}... (前100字符), " f"上下文轮次={next_round}, " f"已生成句子数={sentence_count}, " f"堆栈跟踪:\n{traceback.format_exc()}" ) # 重新抛出异常，让外层处理 raise # 发送TTS结束标记 await self.send_audio_response(request_id, "", chunk_index, True) # 记录最后一个音频块发送时间 last_audio_time = time.time() time_to_last_audio = last_audio_time - audio_end_time # 🔥 更新性能指标：LLM和TTS完成时间 if request_id in self.performance_metrics: self.performance_metrics[request_id].llm_all_text_completed = last_audio_time self.performance_metrics[request_id].llm_call_end = last_audio_time self.performance_metrics[request_id].tts_all_audio_completed = last_audio_time self.performance_metrics[request_id].tts_call_end = last_audio_time # 📝 更新LLM输出统计 self.performance_metrics[request_id].llm_emotion = final_emotion self.performance_metrics[request_id].llm_sentences = sentence_count # 📝 更新TTS输入统计 self.performance_metrics[request_id].tts_voice = settings.tts_voice # 🔥 保存AI响应 ai_response_text = ' '.join(self.ai_responses.get(request_id, [])) llm_total_elapsed = (time.time() - llm_start) * 1000 logger.info(f"[{request_id}] ✅ LLM调用完成，总耗时: {llm_total_elapsed:.1f}ms，句子数: {sentence_count}，情感: {final_emotion}") if ai_response_text: # 🔥 立即写入Redis（不阻塞） if self.redis_cache: redis_save_start = time.time() try: logger.debug(f"[{request_id}] 💾 开始保存AI响应到Redis...") await self.redis_cache.append_message( user_id=self.user_id, role='assistant', content=ai_response_text, round_number=next_round ) redis_save_elapsed = (time.time() - redis_save_start) * 1000 logger.debug(f"[{request_id}] ✅ AI响应保存到Redis完成，耗时: {redis_save_elapsed:.1f}ms") except Exception as e: redis_save_elapsed = (time.time() - redis_save_start) * 1000 logger.warning(f"[{request_id}] ⚠️ AI响应保存到Redis失败，耗时: {redis_save_elapsed:.1f}ms, 错误: {e}") # 🔥 立即保存AI消息到MySQL（获取message_id） if self.message_service: mysql_save_start = time.time() try: logger.info(f"[{request_id}] 💾 开始保存AI消息到MySQL（超时: 5.0s）...") assistant_message = await asyncio.wait_for( self.message_service.save_message( user_id=self.user_id, role='assistant', content=ai_response_text, emotion=final_emotion ), timeout=5.0 ) mysql_save_elapsed = (time.time() - mysql_save_start) * 1000 logger.info(f"[{request_id}] ✅ AI消息保存到MySQL成功，耗时: {mysql_save_elapsed:.1f}ms") # 📝 记录AI消息ID if request_id in self.performance_metrics: self.performance_metrics[request_id].assistant_message_id = assistant_message.get('message_id') except asyncio.TimeoutError: mysql_save_elapsed = (time.time() - mysql_save_start) * 1000 logger.warning(f"[{request_id}] ⚠️ AI消息保存到MySQL超时，耗时: {mysql_save_elapsed:.1f}ms") except Exception as e: mysql_save_elapsed = (time.time() - mysql_save_start) * 1000 logger.warning(f"[{request_id}] ⚠️ AI消息保存到MySQL失败，耗时: {mysql_save_elapsed:.1f}ms, 错误: {e}") # 保存失败不影响主流程，继续异步持久化 # 🔥 加入异步持久化队列（批量写入MySQL，用于重试或降级场景） asyncio.create_task( self._persist_conversation_async( request_id=request_id, user_text=text, ai_text=ai_response_text, round_number=next_round, emotion=final_emotion ) ) if next_round == 1: await self._maybe_spawn_mission_stone(request_id) # 注意：总结触发逻辑已移至 message_service.py # 在保存 assistant 消息时，如果 current_round % 15 == 0，会自动触发总结 # 计算总耗时 total_end = time.time() total_duration = total_end - self.request_start_times.get(request_id, stt_start) llm_tts_total = total_end - llm_start # 📝 记录详细性能日志到文件 try: ai_response_log = ' '.join(self.ai_responses.get(request_id, [])) # 更新最终的AI响应文本 if request_id in self.performance_metrics: self.performance_metrics[request_id].ai_response_text = ai_response_log # 记录详细性能数据 request_logger.log_request( user_id=self.user_id, request_id=request_id, metrics=self.performance_metrics[request_id] ) except Exception as log_error: pass except Exception as e: # 详细记录异常信息 session = self.stt_sessions.get(request_id) user_text_log = session.get_current_text() if session else "" # 判断异常来源（STT/LLM/TTS） error_type = "UNKNOWN_ERROR" error_stage = "unknown" error_str = str(e).lower() if "stt" in error_str or "语音识别" in error_str or "transcription" in error_str: error_type = "STT_ERROR" error_stage = "stt" elif "llm" in error_str or "chat" in error_str or "completion" in error_str or "generate" in error_str: error_type = "LLM_ERROR" error_stage = "llm" elif "tts" in error_str or "speech" in error_str or "audio" in error_str: error_type = "TTS_ERROR" error_stage = "tts" else: error_type = "NETWORK_ERROR" error_stage = "unknown" logger.error( f"[WebSocket异常] request_id={request_id}, " f"user_id={self.user_id}, " f"错误类型={error_type}, " f"异常类型={type(e).__name__}, " f"错误消息={str(e)}, " f"阶段={error_stage}, " f"用户输入={user_text_log[:100]}... (前100字符), " f"堆栈跟踪:\n{traceback.format_exc()}" ) # 记录错误日志 try: request_logger.log_error( user_id=self.user_id, request_id=request_id, error_code=error_type, error_message=str(e), stage=error_stage, user_text=user_text_log ) except Exception as log_error: logger.error( f"[日志记录异常] request_id={request_id}, " f"错误类型=记录错误日志失败, " f"异常类型={type(log_error).__name__}, " f"错误消息={str(log_error)}" ) await self.send_error(request_id, error_type, str(e)) finally: # 清理资源 await self.cleanup_request(request_id) async def _create_stt_session(self, request_id: str) -> Optional[STTSession]: """创建STT会话并设置回调""" async def on_text_update(text: str, is_final: bool): """STT文本更新回调""" await self.send_text_result(request_id, text, is_final) async def on_error(error_msg: str): """STT错误回调""" await self.send_error(request_id, "STT_ERROR", error_msg) # 创建会话 session = await stt_service.create_session( request_id=request_id, on_text_update=on_text_update, on_error=on_error, event_loop=self.event_loop ) return session if session.is_connected else None async def cleanup_request(self, request_id: str): """清理请求相关资源""" # 清理STT会话 if request_id in self.stt_sessions: session = self.stt_sessions[request_id] await session.close() del self.stt_sessions[request_id] # 清理取消标记 if request_id in self.cancelled_requests: self.cancelled_requests.remove(request_id) # 清理时间记录 if request_id in self.request_start_times: del self.request_start_times[request_id] if request_id in self.audio_end_times: del self.audio_end_times[request_id] # 清理AI响应收集器 if request_id in self.ai_responses: del self.ai_responses[request_id] # 清理性能指标 if request_id in self.performance_metrics: del self.performance_metrics[request_id] async def handle_cancel(self, message: CancelMessage): """处理取消请求""" request_id = message.requestId self.cancelled_requests.add(request_id) # 清理资源（包括关闭STT会话） await self.cleanup_request(request_id) async def handle_ping(self, message: PingMessage): """处理Ping""" await self.send_pong(message.requestId) async def handle_energy_stone_event(self, message: EnergyStoneEventMessage): """处理前端能量石事件""" if not self.energy_stone_service: return metadata = { "client_ts": message.clientTs, "position": message.position, "request_id": message.requestId, } try: await self.energy_stone_service.mark_event( stone_id=message.stoneId, event=message.event, metadata={k: v for k, v in metadata.items() if v is not None}, ) except Exception as exc: logger.warning( "[EnergyStone] 记录事件失败: stone_id=%s event=%s error=%s", message.stoneId, message.event, exc, ) async def _maybe_spawn_mission_stone(self, request_id: str): """首次对话奖励""" if not self.energy_stone_service or not self.user_id: return try: payload = await self.energy_stone_service.create_mission_stone_if_needed( user_id=self.user_id, request_id=request_id, ) if payload: await self.send_energy_stone_spawn(payload) except Exception as exc: logger.warning( "[EnergyStone] 任务能量石生成失败: user_id=%s error=%s", self.user_id, exc, ) async def _maybe_spawn_conversation_stones( self, total_duration_ms: int, increment_ms: int, request_id: str, ): """按累计说话时长发放""" if ( not self.energy_stone_service or not self.user_id or total_duration_ms <= 0 or increment_ms <= 0 ): return try: stones = await self.energy_stone_service.create_conversation_stones_if_needed( user_id=self.user_id, request_id=request_id, total_duration_ms=total_duration_ms, duration_increment_ms=increment_ms, ) for stone in stones: await self.send_energy_stone_spawn(stone) except Exception as exc: logger.warning( "[EnergyStone] 对话能量石生成失败: user_id=%s error=%s", self.user_id, exc, ) async def send_user_level_update(self, request_id: str, level: int, total_duration_ms: int): """发送用户等级更新消息""" message = UserLevelMessage( type="user_level", requestId=request_id, timestamp=self._get_timestamp(), level=level, totalDurationMs=total_duration_ms, ) await self.websocket.send_text(message.model_dump_json()) async def send_text_result(self, request_id: str, text: str, is_final: bool): """发送文本结果""" message = TextResultMessage( type="text_result", requestId=request_id, timestamp=self._get_timestamp(), text=text, isFinal=is_final, ) await self.websocket.send_text(message.model_dump_json()) async def send_energy_stone_spawn(self, payload: Dict[str, Any]): """发送能量石生成消息""" message = EnergyStoneSpawnMessage( type="energy_stone_spawn", requestId=payload.get("stone_id"), timestamp=self._get_timestamp(), stoneId=payload.get("stone_id"), stoneType=payload.get("stone_type"), ttlSeconds=payload.get("ttl_seconds", settings.ENERGY_STONE_TTL_SECONDS), appearAt=payload.get("appear_at", self._get_timestamp()), metadata=payload.get("metadata"), ) await self.websocket.send_text(message.model_dump_json()) async def send_emotion(self, request_id: str, emotion: str): """发送情感标签""" message = EmotionMessage( type="emotion", requestId=request_id, timestamp=self._get_timestamp(), emotion=emotion, ) await self.websocket.send_text(message.model_dump_json()) async def send_audio_response( self, request_id: str, data: str, chunk_index: int, is_last: bool, format: str = "pcm", sample_rate: int = 24000 ): """发送音频响应""" message = AudioResponseMessage( type="audio_response", requestId=request_id, timestamp=self._get_timestamp(), data=data, sampleRate=sample_rate, format=format, chunkIndex=chunk_index, isLast=is_last, ) await self.websocket.send_text(message.model_dump_json()) async def send_error(self, request_id: str, code: str, error_message: str): """发送错误消息""" message = ErrorMessage( type="error", requestId=request_id, timestamp=self._get_timestamp(), code=code, message=error_message, ) await self.websocket.send_text(message.model_dump_json()) async def send_pong(self, request_id: str): """发送Pong""" message = PongMessage( type="pong", requestId=request_id, timestamp=self._get_timestamp(), ) await self.websocket.send_text(message.model_dump_json()) def _get_timestamp(self) -> int: """获取当前时间戳（毫秒）""" return int(time.time() * 1000) def _build_llm_input(self, user_input: str, context: dict) -> str: """ 构建包含上下文的LLM输入 Args: user_input: 用户当前输入 context: 上下文信息（包含summary和recent_messages） Returns: 完整的LLM输入文本 """ parts = [] # 1. 最新总结（如果有，已包含所有历史信息） summary = context.get('summary') if summary: parts.append("【历史对话总结】") parts.append(summary) parts.append("") # 2. 最近对话（如果有） recent_messages = context.get('recent_messages', []) if recent_messages: parts.append("【最近对话】") for msg in recent_messages: role_label = "User" if msg['role'] == 'user' else "Assistant" parts.append(f"{role_label}: {msg['content']}") parts.append("") # 3. 当前用户输入 parts.append(f"User: {user_input}") return "\n".join(parts) async def _ensure_redis_context(self, request_id: str): """ 确保Redis中有最新上下文（预加载） 策略： 1. 检查Redis是否有缓存 2. 如果没有，从MySQL加载并更新Redis 3. 如果有，直接使用（零等待） """ try: start = time.time() # 检查Redis缓存 current_round = await self.redis_cache.get_current_round(self.user_id) cached_context = await self.redis_cache.get_context(self.user_id) if cached_context: # ✅ Redis已有缓存，无需操作 return # ❌ Redis缓存未命中，从MySQL加载 # 确保服务已初始化 await self._init_services() if not self.context_service: return # 从MySQL获取上下文 context = await self.context_service.get_context_for_round( user_id=self.user_id, current_round=current_round ) # 更新Redis缓存 await self.redis_cache.set_context(self.user_id, context) # 同时缓存消息列表 for msg in context.get('recent_messages', []): await self.redis_cache.append_message( user_id=self.user_id, role=msg['role'], content=msg['content'], round_number=msg.get('round_number', 0) ) # 同步轮次 if current_round > 0: await self.redis_cache.set_current_round(self.user_id, current_round) except asyncio.CancelledError: # 🔥 任务被取消时，确保清理资源 logger.debug(f"Redis上下文预加载任务被取消: request_id={request_id}") raise # 重新抛出 CancelledError except Exception as e: import traceback traceback.print_exc() async def _get_context_from_redis(self, request_id: str) -> Optional[dict]: """ 从Redis获取上下文（零延迟） Returns: 上下文字典 或 None（需降级到MySQL） """ try: if not self.redis_cache: return None start = time.time() # 获取当前轮次 current_round = await self.redis_cache.get_current_round(self.user_id) # 从Redis构建上下文（需要传入db_manager以获取所有总结） db_manager = None if self.context_service: db_manager = self.context_service.db context = await self.redis_cache.build_context_from_cache( user_id=self.user_id, current_round=current_round, db_manager=db_manager ) return context except Exception as e: return None async def _persist_conversation_async( self, request_id: str, user_text: str, ai_text: str, round_number: int, emotion: str ): """ 异步持久化对话到MySQL（不阻塞主流程） 策略：加入队列，批量写入 """ self.pending_persistence.append({ 'request_id': request_id, 'user_id': self.user_id, 'user_text': user_text, 'ai_text': ai_text, 'round_number': round_number, 'emotion': emotion, 'timestamp': time.time() }) # 启动持久化任务 if not self.persistence_task or self.persistence_task.done(): self.persistence_task = asyncio.create_task( self._batch_persistence_loop() ) async def _batch_persistence_loop(self): """批量持久化循环（后台任务）""" while True: try: await asyncio.sleep(self.persistence_interval) if not self.pending_persistence: continue # 批量提取 batch = self.pending_persistence.copy() self.pending_persistence.clear() start = time.time() # 确保服务已初始化 await self._init_services() if not self.message_service: # 重新加入队列 self.pending_persistence.extend(batch) continue # 批量写入MySQL # ⚠️ 注意：用户消息和AI消息都已在主流程中立即保存 # 这里不再重复保存，避免数据库中出现重复记录 # 如果立即保存失败，应该通过错误日志和监控告警处理，而不是在这里重试 for item in batch: # 记录日志，但不保存（避免重复） logger.debug( f"跳过批量持久化（已立即保存）: " f"user={item['user_id']}, round={item['round_number']}" ) except asyncio.CancelledError: # 🔥 任务被取消时，确保清理资源 logger.info("批量持久化任务被取消，正在清理...") raise # 重新抛出 CancelledError except Exception as e: import traceback traceback.print_exc() async def _generate_summary_async( self, request_id: str, user_id: str, start_round: int, end_round: int ): """ 异步生成对话总结（完全不阻塞） 触发时机： - 每15轮触发一次（15, 30, 45...） """ self.pending_summaries.append({ 'request_id': request_id, 'user_id': user_id, 'start_round': start_round, 'end_round': end_round, 'timestamp': time.time() }) # 启动总结生成任务 if not self.summary_task or self.summary_task.done(): self.summary_task = asyncio.create_task( self._summary_generation_loop() ) async def _summary_generation_loop(self): """总结生成循环（后台任务）""" while True: try: # 等待任务或间隔 await asyncio.sleep(1.0) if not self.pending_summaries: continue # 逐个处理总结任务 task = self.pending_summaries.pop(0) start = time.time() try: # 确保服务已初始化 await self._init_services() if not self.context_service or not self.redis_cache: continue # 调用总结服务（耗时2-3秒） # 注意：这里假设有summary_service，如果没有需要创建 from services.summary_service import SummaryService summary_service = SummaryService(self.context_service.db) start_round = task['start_round'] end_round = task['end_round'] # generate_summary 会自动从数据库获取消息并保存总结 # 该方法会自动获取之前的总结并进行累积总结 await summary_service.generate_summary( user_id=task['user_id'], start_round=start_round, end_round=end_round ) # 从数据库获取生成的总结文本（使用 ORM） from database import db_manager from models_db import ConversationSummary from sqlalchemy import select # 🔥 async with 会自动管理 session 生命周期，包括 CancelledError 情况 async with db_manager.get_session() as session: result = await session.execute( select(ConversationSummary) .where( ConversationSummary.user_id == task['user_id'], ConversationSummary.start_round == start_round, ConversationSummary.end_round == end_round ) .order_by(ConversationSummary.created_at.desc()) .limit(1) ) summary_obj = result.scalar_one_or_none() summary = summary_obj.summary_text if summary_obj else "" # 🔥 保存到Redis（立即生效） await self.redis_cache.set_summary( user_id=task['user_id'], end_round=end_round, summary=summary ) # 🔥 更新Redis上下文缓存（包含新总结） await self.redis_cache.invalidate_context(task['user_id']) except Exception as e: import traceback traceback.print_exc() except asyncio.CancelledError: # 🔥 任务被取消时，确保清理资源 logger.info("总结生成任务被取消，正在清理...") raise # 重新抛出 CancelledError except Exception as e: import traceback traceback.print_exc()