Tolan/backend/websocket_handler.py

"""
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,
)
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

        # 🔥 更新用户累积说话时长（如果有提供，使用超时避免阻塞）
        if speech_duration_ms > 0:
            duration_start = time.time()
            try:
                logger.info(f"[{request_id}] ⏱️ 开始更新用户说话时长: {speech_duration_ms}ms（超时: 1.0s）...")
                # 使用超时，避免长时间阻塞
                total_duration_ms = 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 total_duration_ms:
            await self._maybe_spawn_conversation_stones(total_duration_ms, speech_duration_ms, 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
            
            # 发送最终识别结果
            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_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 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
                    )
                    
                    # 从数据库获取生成的总结文本（使用 ORM）
                    from database import db_manager
                    from models_db import ConversationSummary
                    from sqlalchemy import select
                    
                    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 Exception as e:
                import traceback
                traceback.print_exc()