""" WebSocket处理器 - 新版本（STT+LLM+TTS流水线，支持流式LLM） """ import json import base64 import asyncio import time import re import logging import traceback from typing import 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, ) 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 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.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): """初始化服务（延迟加载）""" try: from database import db_manager from services import MessageService, ContextService # 使用延迟初始化的全局实例 from services.message_service import message_service from services.context_service import context_service await message_service.set_db_manager(db_manager) await context_service.set_db_manager(db_manager) self.message_service = message_service self.context_service = context_service except Exception as e: pass 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)) 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 # 🔥 更新用户累积说话时长（如果有提供） if message.speechDuration and message.speechDuration > 0: try: await user_service.update_speech_duration(self.user_id, message.speechDuration) except Exception as e: logger.warning(f"⚠️ 更新用户说话时长失败: user_id={self.user_id}, duration_ms={message.speechDuration}, error={e}") # 初始化AI响应收集器 self.ai_responses[request_id] = [] # 检查是否取消 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 text = await session.wait_for_final_result(timeout=timeout) # 获取识别结果 if not text: 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 stt_duration = time.time() - stt_start stt_complete_time = time.time() # 更新性能指标 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获取上下文 context = await self._get_context_from_redis(request_id) if context: current_round = context.get('current_round', 0) next_round = current_round + 1 # 🔥 立即写入Redis（用户消息） await self.redis_cache.append_message( user_id=self.user_id, role='user', content=text, round_number=next_round ) await self.redis_cache.increment_round(self.user_id) # 🔥 异步保存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 await self._init_services() if self.message_service and self.context_service: try: # 保存用户消息 logger.info("🔥 [WebSocket] 准备保存用户消息") 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 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（如果有上下文，使用上下文；否则使用原始文本） llm_input = self._build_llm_input(text, context) if context else text # 获取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 # 🔥 流式处理：LLM生成一句 → 立即TTS try: async for sentence, emotion in ai_service.generate_response_stream(llm_input): 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, [])) if ai_response_text: # 🔥 立即写入Redis（不阻塞） if self.redis_cache: await self.redis_cache.append_message( user_id=self.user_id, role='assistant', content=ai_response_text, round_number=next_round ) # 🔥 立即保存AI消息到MySQL（获取message_id） if self.message_service: try: 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 ) # 📝 记录AI消息ID if request_id in self.performance_metrics: self.performance_metrics[request_id].assistant_message_id = assistant_message.get('message_id') except Exception as e: # 保存失败不影响主流程，继续异步持久化 pass # 🔥 加入异步持久化队列（批量写入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 ) ) # 注意：总结触发逻辑已移至 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 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_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 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 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 ) # 从数据库获取生成的总结文本 summary_result = await self.context_service.db.execute_one( """ SELECT summary_text FROM conversation_summaries WHERE user_id = %(user_id)s AND start_round = %(start_round)s AND end_round = %(end_round)s ORDER BY created_at DESC LIMIT 1 """, { "user_id": task['user_id'], "start_round": start_round, "end_round": end_round } ) summary = summary_result['summary_text'] if summary_result 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 Exception as e: import traceback traceback.print_exc()