""" WebSocket处理器 - 新版本（STT+LLM+TTS流水线，支持流式LLM） """ import json import base64 import asyncio import time import re import logging import traceback from typing import Any, Dict, Literal, Optional from fastapi import WebSocket from starlette.websockets import WebSocketDisconnect, WebSocketState logger = logging.getLogger(__name__) from emotion_text_utils import ( clean_text_for_display, strip_emotion_from_text, ) # 表情符号正则表达式（用于TTS前清理） EMOJI_PATTERN = re.compile( "[" "\U0001F600-\U0001F64F" # emoticons "\U0001F300-\U0001F5FF" # symbols & pictographs "\U0001F680-\U0001F6FF" # transport & map symbols "\U0001F1E0-\U0001F1FF" # flags (iOS) "\U00002702-\U000027B0" "\U000024C2-\U0001F251" "]+", flags=re.UNICODE ) 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 ab_experiment_media_config import resolve_llm_model, resolve_tts_params from stt_service import stt_service, STTSession from services.user_service import user_service from services.energy_stone_service import energy_stone_service from services.scenario_topic_service import scenario_topic_service from services.scenario_gift_service import grant_scene_gift_if_needed from services.llm_task_service import llm_task_service from scenario_config import ( SCENE_DEFAULT, SCENE_WINDOW, SCENE_READING, SCENE_EATING, SCENE_WEIGHTS, SCENARIO_TOPICS, get_special_first_line, build_topics_payload, is_special_ab_experiment_user, ) import random import uuid from sqlalchemy import text class WebSocketHandler: """WebSocket连接处理器""" def __init__( self, websocket: WebSocket, user_id: str = None, token: str = None, redis_cache = None, session_store = None, ): self.websocket = websocket self.user_id = user_id or "guest_user" # 用户ID（默认游客） self.token = token # 会话 token（单活校验） 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.session_store = session_store # 会话存储（单活 token） # 🔥 预加载任务 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 # 情景对话：首条语音携带的 scene/topic，在进入 LLM 前落库 self._pending_scenario_log: Dict[str, Dict[str, str]] = {} 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) if not scenario_topic_service.db: await scenario_topic_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 _close_with_session_invalid(self, reason: str = "session_invalidated"): """发送会话失效错误并关闭连接""" try: await self.send_error( request_id="session", code="SESSION_INVALIDATED", error_message=reason, ) except Exception: pass try: await self.websocket.close(code=4401, reason=reason) except Exception: pass async def _ensure_valid_session(self) -> bool: """校验会话 token，确保单用户单活""" if not self.session_store: return True if not self.user_id or not self.token: await self._close_with_session_invalid("missing_session_token") return False if not await self.session_store.validate(self.user_id, self.token, refresh=True): await self._close_with_session_invalid("session_invalidated") return False await self.session_store.bind_connection(self.user_id, self.token, self.websocket) return True async def handle_connection(self): """处理WebSocket连接""" await self.websocket.accept() # 初始化服务 await self._init_services() # 校验会话 token if not await self._ensure_valid_session(): return # 🔥 恢复未过期的能量石（页面切换后重新显示） await self._restore_active_stones() # 保存事件循环引用，供回调线程使用 self.event_loop = asyncio.get_running_loop() try: while True: if not self._is_ws_connected(): break try: data = await self.websocket.receive_text() except RuntimeError as exc: # 同用户重连时 bind_connection 会主动关闭旧连接，Starlette 抛 RuntimeError 而非 WebSocketDisconnect msg = str(exc).lower() if "websocket is not connected" in msg or "need to call \"accept\" first" in msg: logger.info( f"[{self.user_id}] ℹ️ WebSocket 已关闭（可能被新连接替换），停止接收消息" ) break raise message_data = json.loads(data) await self.handle_message(message_data) except WebSocketDisconnect as e: close_code = getattr(e, "code", None) reason = getattr(e, "reason", None) logger.warning(f"[{self.user_id}] ⚠️ WebSocket断开: code={close_code}, reason={reason}") except Exception as e: logger.error(f"[{self.user_id}] ❌ WebSocket异常: {e.__class__.__name__}: {e}") logger.error(traceback.format_exc()) finally: await self._cleanup_on_disconnect() close_code = getattr(self.websocket, "close_code", None) logger.info(f"[{self.user_id}] ℹ️ WebSocket连接结束，close_code={close_code}") def _is_ws_connected(self) -> bool: try: return self.websocket.client_state == WebSocketState.CONNECTED except Exception: return False async def _cleanup_on_disconnect(self) -> None: """连接关闭时释放 STT 会话等资源。""" if self.session_store: try: await self.session_store.unbind_connection(self.user_id, self.websocket) except Exception: pass for request_id in list(self.stt_sessions.keys()): try: await self.cleanup_request(request_id) except Exception as exc: logger.warning( f"[{self.user_id}] ⚠️ 断开连接时清理 request={request_id} 失败: {exc}" ) for task in self.preload_tasks.values(): if not task.done(): task.cancel() self.preload_tasks.clear() async def handle_message(self, message_data: dict): """处理接收到的消息""" message_type = message_data.get("type") request_id = message_data.get("requestId") incoming_token = message_data.get("token") if incoming_token: self.token = incoming_token if not await self._ensure_valid_session(): return # 检查是否为已取消的请求 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)) elif message_type == "start_greeting": # 由前端显式触发首句 greeting；支持 data.forceScene 测试用 data = message_data.get("data") or {} force_scene = data.get("forceScene") if isinstance(data, dict) else None await self._send_greeting_message(force_scene=force_scene) elif message_type == "play_topic_first_line": # 情景对话：用户切换话题后请求播放该话题首句 TTS data = message_data.get("data") or {} if isinstance(data, dict) and data.get("topicKey"): await self._play_topic_first_line( request_id=message_data.get("requestId") or f"topic_{uuid.uuid4().hex[:8]}", topic_key=data.get("topicKey"), ) 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 # 情景对话：首条语音携带的 scene/topic，进入 LLM 前落库 if getattr(message, "scene", None) and getattr(message, "topic", None): self._pending_scenario_log[request_id] = {"scene": message.scene, "topic": message.topic} # 初始化性能指标 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 # 情景对话：若首 chunk 未带 scene/topic，可在此补带（仅首条语音） if getattr(message, "scene", None) and getattr(message, "topic", None) and request_id not in self._pending_scenario_log: self._pending_scenario_log[request_id] = {"scene": message.scene, "topic": message.topic} # 记录录音结束时间 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 # 说话时长写库改异步，不阻塞 STT → LLM → TTS 主链路（全用户生效） if speech_duration_ms > 0: asyncio.create_task( self._update_speech_duration_async(request_id, speech_duration_ms) ) # 初始化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 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 # 发送最终识别结果（source=user，前端不用于字幕） await self.send_text_result(request_id, text, True, source="user") 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 # 标记是否已发送情感 # 情景对话：首条语音若携带 scene/topic，在此落库（仅记录一次） pending = self._pending_scenario_log.pop(request_id, None) if pending: await scenario_topic_service.record_conversation( self.user_id, pending["scene"], pending["topic"] ) # 构建Prompt（如果有上下文，使用上下文；否则使用原始文本） prompt_build_start = time.time() # 获取用户信息（用于运行时上下文） user_info = await user_service.get_user(self.user_id) is_new_user = False if user_info: # 判断是否为新用户（total_rounds == 0） is_new_user = (user_info.get("total_rounds", 0) == 0) ab_task = None if is_special_ab_experiment_user(user_info): ab_task = await llm_task_service.get_latest_task_cached() if not ab_task: logger.warning(f"[{self.user_id}] ⚠️ A/B 用户未找到任务表 prompt，将回退默认 base prompt") ab_task_prompt = ab_task.get("task_prompt") if ab_task else None ab_task_name = ab_task.get("task_name") if ab_task else None # 获取system_prompt和runtime_context用于日志记录 system_prompt = ai_service._build_system_prompt( user_info, is_new_user, ab_task_prompt=ab_task_prompt, ) runtime_context = ai_service._build_runtime_context(user_info, ab_task_name=ab_task_name) # 构建对话历史（用于上下文）- v0.3: 近20轮对话 chat_history = [] if context and context.get("recent_messages"): for msg in context.get("recent_messages", []): role = "user" if msg.get("role") == "user" else "assistant" chat_history.append({"role": role, "content": msg.get("content", "")}) # 📝 构建完整的 messages 数组（用于保存 LLM 提示词记录和日志） llm_messages = [ {"role": "system", "content": system_prompt}, {"role": "system", "content": runtime_context}, ] if chat_history: llm_messages.extend(chat_history) llm_messages.append({"role": "user", "content": text}) # ✅ v0.3: 使用原始用户输入 # 构建完整提示词用于日志（文本格式） full_prompt_parts = [ f"System (Base): {system_prompt}", f"System (Runtime Context): {runtime_context}", ] if chat_history: full_prompt_parts.append(f"Chat History ({len(chat_history)} messages)") full_prompt_parts.append(f"User: {text}") full_prompt = "\n\n".join(full_prompt_parts) prompt_build_elapsed = (time.time() - prompt_build_start) * 1000 # 📝 更新LLM输入统计（用于日志） if request_id in self.performance_metrics: self.performance_metrics[request_id].llm_prompt_length = len(text) 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，用户输入长度: {len(text)}，构建耗时: {prompt_build_elapsed:.1f}ms，轮次: {next_round}") # 🔥 流式处理：LLM生成一句 → 立即TTS # ✅ v0.3: 传入原始用户输入，generate_response_stream 内部会构建完整的 messages 数组 try: llm_first_chunk_time = None async for sentence, emotion in ai_service.generate_response_stream( text, # ✅ v0.3: 原始用户输入，generate_response_stream 内部会构建完整的 messages 数组 user_info=user_info, is_new_user=is_new_user, chat_history=chat_history if chat_history else None, ab_task=ab_task, ): 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: logger.info(f"[{request_id}] 🎭 发送情感标签给前端: {emotion}, 句子内容: {sentence[:100] if sentence else '(空句子)'}") await self.send_emotion(request_id, emotion) emotion_sent = True final_emotion = emotion # ⚡ 如果是空句子（仅传递情感），跳过 TTS 处理 if not sentence: continue sentence_count += 1 # 字幕展示文本：剥离 [EMOTION:xxx] 及 LLM 偶发的裸 emotion 词 display_text = clean_text_for_display(sentence) # 🧹 TTS 文本清理（括号/emoji 仅影响语音，不影响字幕） clean_sentence = strip_emotion_from_text(sentence) clean_sentence = clean_sentence.translate(str.maketrans('', '', '()[]')).strip() clean_sentence = EMOJI_PATTERN.sub('', clean_sentence) # 如果清理后为空，跳过TTS（emotion 已通过 send_emotion 下发，无需空字幕） if not clean_sentence: continue # 发送文本给前端（source=assistant 用于字幕拼接） if display_text: await self.send_text_result( request_id, display_text, is_final=False, source="assistant" ) # 收集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, user_info=user_info ): 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 结束标记：整轮 AI 回复（可能多句话/多段 TTS）全部发完后只发一次 isLast=True，前端据此判定「对话结束」并在全部播完后才切 idle 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 = resolve_tts_params(user_info)["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') # 🔥 保存LLM提示词记录（异步，不阻塞主流程） db_message_id = assistant_message.get('id') if db_message_id and self.message_service: try: # 获取LLM调用参数 llm_call_metadata = {} if request_id in self.performance_metrics: metrics = self.performance_metrics[request_id] llm_call_metadata = { "llm_call_start": getattr(metrics, 'llm_call_start', None), "llm_first_chunk_time": getattr(metrics, 'llm_first_chunk_time', None), "llm_first_text_output": getattr(metrics, 'llm_first_text_output', None), "llm_prompt_length": getattr(metrics, 'llm_prompt_length', None), } # 异步保存LLM提示词记录（不阻塞主流程） asyncio.create_task( self.message_service.save_llm_prompt( message_id=db_message_id, user_id=self.user_id, round_number=next_round, system_prompt=system_prompt, runtime_context=runtime_context, messages=llm_messages, model_name=resolve_llm_model(user_info), temperature="0.7", max_tokens=500, stream=True, metadata=llm_call_metadata ) ) logger.debug(f"[{request_id}] 📝 LLM提示词记录保存任务已创建: message_id={db_message_id}") except Exception as e: logger.warning(f"[{request_id}] ⚠️ 创建LLM提示词记录保存任务失败: {e}") 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 ) ) # 首次对话“水晶石”赠送已关闭：不再生成首句奖励（MissionEnergyStone） # 也不再更新 `users.crystal_stone_total`。 # 注意：总结触发逻辑已移至 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文本更新回调（source=user，前端不用于字幕）""" await self.send_text_result(request_id, text, is_final, source="user") 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 _update_speech_duration_async( self, request_id: str, speech_duration_ms: int ) -> None: """后台更新累积说话时长，并同步等级/能量石（不阻塞对话主链路）。""" duration_start = time.time() try: logger.info( f"[{request_id}] ⏱️ 异步更新用户说话时长: {speech_duration_ms}ms..." ) level_update = await asyncio.wait_for( user_service.update_speech_duration(self.user_id, speech_duration_ms), timeout=5.0, ) 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）: " f"user_id={self.user_id}, duration_ms={speech_duration_ms}" ) return except Exception as e: duration_elapsed = (time.time() - duration_start) * 1000 logger.warning( f"[{request_id}] ⚠️ 异步更新用户说话时长失败（耗时: {duration_elapsed:.1f}ms）: " f"user_id={self.user_id}, duration_ms={speech_duration_ms}, error={e}" ) return if not level_update or not isinstance(level_update, dict): return 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 ) 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}" ) 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_greeting_message(self, force_scene: Optional[str] = None): """ 发送首句（AI先说话） - 若 force_scene 有效则强制该场景；否则按权重随机场景（特殊场景需排除已聊完的）。 - default：新用户固定首句 / 回归用户动态登录问候。 - 特殊场景：未完成话题等概率选一，发 greeting_context 后发固定首句+TTS。 """ request_id = f"greeting_{uuid.uuid4().hex[:8]}" try: user_info = await user_service.get_user(self.user_id) if not user_info: logger.warning(f"[{self.user_id}] ⚠️ 无法获取用户信息，跳过首句发送") return user_name = (user_info.get("username") or "").strip() or "Neo" is_new_user = (user_info.get("total_rounds", 0) == 0) ab_experiment_user = is_special_ab_experiment_user(user_info) # 1) 决定本次场景 chosen_scene = SCENE_DEFAULT scene_anchor: Optional[Dict[str, Any]] = None valid_force = force_scene in (SCENE_DEFAULT, SCENE_WINDOW, SCENE_READING, SCENE_EATING) if ab_experiment_user: chosen_scene = SCENE_DEFAULT if valid_force and force_scene != SCENE_DEFAULT: logger.info( f"[{self.user_id}] A/B 实验用户忽略强制特殊场景 force_scene={force_scene}，使用 default" ) else: logger.info(f"[{self.user_id}] A/B 实验用户，跳过特殊情景场景随机") elif valid_force: if force_scene == SCENE_DEFAULT: chosen_scene = force_scene logger.info(f"[{self.user_id}] 🎯 首句强制场景: {chosen_scene}") else: forced_anchor = await self._pick_scene_anchor_for_user(force_scene) if forced_anchor: chosen_scene = force_scene scene_anchor = forced_anchor logger.info(f"[{self.user_id}] 🎯 首句强制场景: {chosen_scene}, anchor_layout_item_id={scene_anchor.get('anchorLayoutItemId')}") else: chosen_scene = SCENE_DEFAULT logger.info(f"[{self.user_id}] ℹ️ 强制场景 {force_scene} 无可用锚点，回退 default") else: await self._init_services() completed = await scenario_topic_service.get_completed_count_per_scenario(self.user_id) candidates = [] for scene_key, weight in SCENE_WEIGHTS: if scene_key == SCENE_DEFAULT: candidates.append((scene_key, weight)) else: if completed.get(scene_key, 0) < 3 and await self._scene_has_available_anchor(scene_key): candidates.append((scene_key, weight)) if not candidates: candidates = [(SCENE_DEFAULT, 70)] total = sum(w for _, w in candidates) r = random.uniform(0, total) acc = 0 for sk, w in candidates: acc += w if r < acc: chosen_scene = sk break # 2) 默认场景：沿用原逻辑 if chosen_scene == SCENE_DEFAULT: lumi_name = (user_info.get("bot_name") or "").strip() or "Lumi" if is_special_ab_experiment_user(user_info): ab_task = await llm_task_service.get_latest_task_cached() lumi_first_line = ab_task.get("lumi_first_line") if ab_task else None greeting_text = ai_service.ab_group_greeting_line( user_name, lumi_name, is_new_user, lumi_first_line=lumi_first_line, ) logger.info(f"[{self.user_id}] 🎯 A/B 实验用户首句: {greeting_text}") elif is_new_user: greeting_text = ai_service.first_meeting_line(user_name, lumi_name) logger.info(f"[{self.user_id}] 🎯 新用户首句: {greeting_text}") else: await self._init_services() chat_history = [] if self.context_service: try: current_round = await self.redis_cache.get_current_round(self.user_id) if self.redis_cache else 1 context = await self.context_service.get_context_for_round( user_id=self.user_id, current_round=current_round ) if context and context.get("recent_messages"): for msg in context.get("recent_messages", []): role = "user" if msg.get("role") == "user" else "assistant" chat_history.append({"role": role, "content": msg.get("content", "")}) except Exception as e: logger.warning(f"[{self.user_id}] ⚠️ 获取对话历史失败: {e}") greeting_text = await ai_service.generate_login_greeting( user_info=user_info, chat_history=chat_history if chat_history else None ) logger.info(f"[{self.user_id}] 🎯 回归用户登录问候: {greeting_text}") emotion = "friendly" if is_new_user else "neutral" clean_text = greeting_text if ( not is_special_ab_experiment_user(user_info) and not is_new_user and "[EMOTION:" in greeting_text ): emotion_start = greeting_text.find("[EMOTION:") emotion_end = greeting_text.find("]", emotion_start) if emotion_end > emotion_start: emotion_str = greeting_text[emotion_start + 9:emotion_end].strip().lower() valid_emotions = ["neutral", "happy", "sad", "excited", "angry", "surprised", "thoughtful", "encouraging", "confused", "worried", "satisfied", "friendly", "sorry"] if emotion_str in valid_emotions: emotion = emotion_str clean_text = greeting_text[emotion_end + 1:].strip() clean_text = clean_text.translate(str.maketrans('', '', '()[]')).strip() clean_text = EMOJI_PATTERN.sub('', clean_text) if not clean_text or not clean_text.strip(): clean_text = "Welcome back! How are you today?" await self.send_emotion(request_id, emotion) await self.send_text_result(request_id, clean_text, is_final=True, source="assistant") chunk_index = 0 async for audio_chunk in ai_service.text_to_speech_stream( clean_text, emotion, user_info=user_info ): audio_base64 = base64.b64encode(audio_chunk).decode('utf-8') await self.send_audio_response(request_id=request_id, data=audio_base64, chunk_index=chunk_index, is_last=False) chunk_index += 1 await self.send_audio_response(request_id=request_id, data="", chunk_index=chunk_index, is_last=True) logger.info(f"[{self.user_id}] ✅ 首句发送完成: request_id={request_id}") return # 3) 特殊场景：选话题、发 greeting_context、再发首句+TTS if not scene_anchor: scene_anchor = await self._pick_scene_anchor_for_user(chosen_scene) if not scene_anchor: logger.warning(f"[{self.user_id}] ⚠️ 特殊场景 {chosen_scene} 无可用锚点，回退默认首句") await self._send_greeting_message(force_scene=SCENE_DEFAULT) return done_topics = await scenario_topic_service.get_done_topic_keys(self.user_id, chosen_scene) all_topics = SCENARIO_TOPICS.get(chosen_scene, []) available = [t for t in all_topics if t not in done_topics] if not available: logger.warning(f"[{self.user_id}] ⚠️ 特殊场景 {chosen_scene} 无可用话题，回退默认首句") await self._send_greeting_message(force_scene=SCENE_DEFAULT) return chosen_topic = random.choice(available) first_line = get_special_first_line(chosen_topic, user_name) topics_payload = build_topics_payload(available) await self._send_greeting_context( request_id=request_id, scene=chosen_scene, topic=chosen_topic, topics=topics_payload, first_line=first_line, scene_anchor=scene_anchor, ) emotion = "friendly" await self.send_emotion(request_id, emotion) await self.send_text_result(request_id, first_line, is_final=True, source="assistant") chunk_index = 0 async for audio_chunk in ai_service.text_to_speech_stream( first_line, emotion, user_info=user_info ): audio_base64 = base64.b64encode(audio_chunk).decode('utf-8') await self.send_audio_response(request_id=request_id, data=audio_base64, chunk_index=chunk_index, is_last=False) chunk_index += 1 await self.send_audio_response(request_id=request_id, data="", chunk_index=chunk_index, is_last=True) await grant_scene_gift_if_needed(self.user_id, chosen_scene) logger.info(f"[{self.user_id}] ✅ 特殊场景首句发送完成: scene={chosen_scene}, topic={chosen_topic}, request_id={request_id}") except Exception as e: logger.error(f"[{self.user_id}] ❌ 发送首句失败: {e}") logger.error(traceback.format_exc()) async def _send_greeting_context( self, request_id: str, scene: str, topic: str, topics: list, first_line: str, scene_anchor: Optional[Dict[str, Any]] = None, ): """发送 greeting_context 消息（仅特殊场景）""" payload = { "type": "greeting_context", "requestId": request_id, "timestamp": self._get_timestamp(), "scene": scene, "topic": topic, "topics": topics, "firstLine": first_line, } if scene_anchor: payload.update(scene_anchor) await self.websocket.send_text(json.dumps(payload)) async def _scene_has_available_anchor(self, scene: str) -> bool: if scene == SCENE_DEFAULT: return True anchor = await self._pick_scene_anchor_for_user(scene) return anchor is not None async def _pick_scene_anchor_for_user(self, scene: str) -> Optional[Dict[str, Any]]: """为特殊场景在当前布局中随机选取一个锚点（优先新槽位模型）。""" anchor_type: Optional[str] = None if scene in (SCENE_READING, SCENE_EATING): anchor_type = "seat" elif scene == SCENE_WINDOW: anchor_type = "window" if not anchor_type: return None try: from database import db_manager async with db_manager.get_session() as session: layout_user_type_res = await session.execute( text( """ SELECT DATA_TYPE FROM information_schema.columns WHERE table_schema = :db AND table_name = 'chatbot_layouts' AND column_name = 'user_id' """ ), {"db": settings.MYSQL_DATABASE}, ) layout_user_type = (layout_user_type_res.mappings().first() or {}).get("DATA_TYPE") or "" layout_user_id_where_expr = ( "(SELECT id FROM users WHERE user_id = :user_id LIMIT 1)" if "bigint" in str(layout_user_type).lower() else ":user_id" ) # 1) 新模型：基于 user_chatbot_slot_placements + chatbot_backdrop_slots 选锚点 backdrop_user_type_res = await session.execute( text( """ SELECT DATA_TYPE FROM information_schema.columns WHERE table_schema = :db AND table_name = 'user_chatbot_backdrops' AND column_name = 'user_id' """ ), {"db": settings.MYSQL_DATABASE}, ) backdrop_user_type = (backdrop_user_type_res.mappings().first() or {}).get("DATA_TYPE") or "" backdrop_user_id_where_expr = ( "(SELECT id FROM users WHERE user_id = :user_id LIMIT 1)" if "bigint" in str(backdrop_user_type).lower() else ":user_id" ) backdrop_row_res = await session.execute( text( f""" SELECT backdrop_item_id FROM user_chatbot_backdrops WHERE user_id = {backdrop_user_id_where_expr} ORDER BY updated_at DESC, id DESC LIMIT 1 """ ), {"user_id": self.user_id}, ) backdrop_row = backdrop_row_res.mappings().first() backdrop_item_id = ( int(backdrop_row["backdrop_item_id"]) if backdrop_row and backdrop_row.get("backdrop_item_id") is not None else None ) candidates: list[dict[str, Any]] = [] if backdrop_item_id is not None: slot_candidates_res = await session.execute( text( f""" SELECT usp.id, s.start_row, s.start_col, s.size_rows, s.size_cols, COALESCE(usp.z_index, s.z_index, si.default_z_index, 0) AS z_index FROM user_chatbot_slot_placements usp JOIN chatbot_backdrop_slots s ON s.backdrop_item_id = usp.backdrop_item_id AND s.slot_code = usp.slot_code AND s.is_active = 1 LEFT JOIN shop_items si ON si.id = usp.item_id WHERE usp.user_id = {layout_user_id_where_expr} AND usp.backdrop_item_id = :backdrop_item_id AND s.anchor_type = :anchor_type ORDER BY usp.id ASC """ ), { "user_id": self.user_id, "backdrop_item_id": backdrop_item_id, "anchor_type": anchor_type, }, ) candidates = [dict(r) for r in slot_candidates_res.mappings().all()] # 2) 兼容旧模型：新模型未命中时回退 layout_items + scene_anchor_type if not candidates: layout_row_res = await session.execute( text( f""" SELECT id FROM chatbot_layouts l WHERE l.user_id = {layout_user_id_where_expr} ORDER BY l.updated_at DESC, l.id DESC LIMIT 1 """ ), {"user_id": self.user_id}, ) layout_row = layout_row_res.mappings().first() if not layout_row or layout_row.get("id") is None: return None items_res = await session.execute( text( """ SELECT li.id, li.start_row, li.start_col, li.size_rows, li.size_cols, li.z_index FROM chatbot_layout_items li JOIN shop_items si ON si.id = li.item_id WHERE li.layout_id = :layout_id AND si.scene_anchor_type = :anchor_type ORDER BY li.id ASC """ ), {"layout_id": layout_row["id"], "anchor_type": anchor_type}, ) candidates = [dict(r) for r in items_res.mappings().all()] if not candidates: return None picked = random.choice(candidates) anchor_z = int(picked.get("z_index") or 0) return { "anchorLayoutItemId": int(picked["id"]), "anchorPlacement": "window_left_bottom" if scene == SCENE_WINDOW else "seat_center_bottom", "anchorZIndex": anchor_z, "specialAvatarZIndex": max(anchor_z + 1, 3), } except Exception as e: logger.warning(f"[{self.user_id}] 获取场景锚点失败: scene={scene}, error={e}") return None async def _play_topic_first_line(self, request_id: str, topic_key: str): """情景对话：播放某话题的规范首句 TTS（用户切换话题后调用）""" try: user_info = await user_service.get_user(self.user_id) user_name = (user_info.get("username") or "").strip() or "Neo" if user_info else "Neo" first_line = get_special_first_line(topic_key, user_name) emotion = "friendly" await self.send_emotion(request_id, emotion) await self.send_text_result(request_id, first_line, is_final=True, source="assistant") chunk_index = 0 async for audio_chunk in ai_service.text_to_speech_stream( first_line, emotion, user_info=user_info ): audio_base64 = base64.b64encode(audio_chunk).decode("utf-8") await self.send_audio_response(request_id=request_id, data=audio_base64, chunk_index=chunk_index, is_last=False) chunk_index += 1 await self.send_audio_response(request_id=request_id, data="", chunk_index=chunk_index, is_last=True) except Exception as e: logger.warning(f"[{self.user_id}] play_topic_first_line failed: topic_key={topic_key}, e={e}") 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, source: Optional[Literal["user", "assistant"]] = None ): """发送文本结果。source=user 表示用户STT(不参与字幕)，source=assistant 表示AI回复(参与字幕拼接)。""" message = TextResultMessage( type="text_result", requestId=request_id, timestamp=self._get_timestamp(), text=text, isFinal=is_final, source=source, ) 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 _restore_active_stones(self): """恢复未过期的能量石（页面切换后重新显示）""" if not self.energy_stone_service or not self.user_id: return try: active_stones = await self.energy_stone_service.get_active_stones(self.user_id) if active_stones: logger.info( f"[{self.user_id}] 🔄 恢复 {len(active_stones)} 个未过期的能量石" ) for stone_payload in active_stones: await self.send_energy_stone_spawn(stone_payload) except Exception as exc: logger.warning( f"[{self.user_id}] ⚠️ 恢复能量石失败: {exc}" ) 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()