Tolan/backend/websocket_handler.py

"""
WebSocket处理器 - 新版本（STT+LLM+TTS流水线，支持流式LLM）
"""
import json
import base64
import asyncio
import time
import re
from typing import Dict, Optional
from fastapi import WebSocket

# 预编译正则表达式（性能优化）
EMOTION_TAG_PATTERN = re.compile(r'\[EMOTION:\w+\]')
from models import (
    AudioChunkMessage,
    AudioEndMessage,
    CancelMessage,
    TextResultMessage,
    EmotionMessage,
    AudioResponseMessage,
    ErrorMessage,
    PingMessage,
    PongMessage,
)
from ai_service import ai_service
from request_logger import request_logger
from config import settings, get_dashscope_ws_url


class WebSocketHandler:
    """WebSocket连接处理器"""
    
    def __init__(self, websocket: WebSocket):
        self.websocket = websocket
        self.audio_buffers: Dict[str, bytearray] = {}  # requestId -> audio_buffer
        self.cancelled_requests: set = set()  # 已取消的请求ID
        self.recognition_sessions: Dict[str, any] = {}  # requestId -> OmniRealtimeConversation对象
        self.recognition_results: Dict[str, str] = {}  # requestId -> 识别结果文本
        self.recognition_failed: set = set()  # STT识别失败的请求ID
        self.event_loop: Optional[asyncio.AbstractEventLoop] = None  # 保存事件循环引用
        self.request_start_times: Dict[str, float] = {}  # 请求开始时间
        self.last_audio_frame_time: Dict[str, float] = {}  # 最后一帧音频到达时间
        self.last_recognition_update_time: Dict[str, float] = {}  # 最后一次识别结果更新时间
        self.stt_final_events: Dict[str, asyncio.Event] = {}  # STT最终结果事件(用于事件驱动)
        self.ai_responses: Dict[str, list] = {}  # requestId -> AI响应句子列表（用于日志记录）
        self.audio_end_times: Dict[str, float] = {}  # requestId -> 录音结束时间
        
    async def handle_connection(self):
        """处理WebSocket连接"""
        await self.websocket.accept()
        print("客户端已连接")
        
        # 保存事件循环引用，供回调线程使用
        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:
            print(f"WebSocket错误: {e}")
        finally:
            print("客户端已断开")
    
    async def handle_message(self, message_data: dict):
        """处理接收到的消息"""
        message_type = message_data.get("type")
        request_id = message_data.get("requestId")
        
        # 检查是否为已取消的请求
        if request_id in self.cancelled_requests:
            return
        
        if message_type == "audio_chunk":
            await self.handle_audio_chunk(AudioChunkMessage(**message_data))
            
        elif message_type == "audio_end":
            await self.handle_audio_end(AudioEndMessage(**message_data))
            
        elif message_type == "cancel":
            await self.handle_cancel(CancelMessage(**message_data))
            
        elif message_type == "ping":
            await self.handle_ping(PingMessage(**message_data))
    
    async def handle_audio_chunk(self, message: AudioChunkMessage):
        """处理音频数据块 - 流式STT"""
        request_id = message.requestId
        
        # 检查是否取消或STT已失败
        if request_id in self.cancelled_requests or request_id in self.recognition_failed:
            return
        
        # Base64解码音频数据
        audio_data = base64.b64decode(message.data)
        chunk_size = len(audio_data)
        
        # 初始化buffer（用于备份，以防STT失败）
        if request_id not in self.audio_buffers:
            self.audio_buffers[request_id] = bytearray()
            self.request_start_times[request_id] = time.time()
            print(f"\n[{request_id}] 🎤 开始接收音频流 [时间: {time.strftime('%H:%M:%S.%f')[:-3]}]")
            print(f"[{request_id}] 音频参数: {message.sampleRate}Hz, {message.channels}ch, {message.bitDepth}bit")
        
        self.audio_buffers[request_id].extend(audio_data)
        total_size = len(self.audio_buffers[request_id])
        
        # 计算已录制时长（16000 sample_rate, 16bit = 2 bytes per sample, mono）
        duration_seconds = total_size / (16000 * 2)
        
        # 打印音频块信息
        print(f"[{request_id}] 📦 接收chunk#{message.chunkIndex}: "
              f"大小={chunk_size}字节, 累计={total_size}字节 ({duration_seconds:.2f}秒)")
        
        # 第一个chunk：启动流式STT识别
        if request_id not in self.recognition_sessions:
            print(f"[{request_id}] 🚀 首个chunk，启动STT流式识别...")
            success = await self.start_streaming_stt(request_id)
            if not success:
                # STT启动失败，标记为失败状态
                print(f"[{request_id}] ❌ STT启动失败")
                self.recognition_failed.add(request_id)
                return
        
        # 🔍 记录最后一帧音频到达时间
        self.last_audio_frame_time[request_id] = time.time()
        
        # 发送音频帧到STT服务（OmniRealtimeConversation使用base64编码）
        conversation = self.recognition_sessions.get(request_id)
        if conversation:
            try:
                # Base64编码音频数据
                audio_b64 = base64.b64encode(audio_data).decode('ascii')
                
                # 在线程池中发送音频帧
                send_start = time.time()
                await asyncio.to_thread(conversation.append_audio, audio_b64)
                send_duration = (time.time() - send_start) * 1000
                print(f"[{request_id}] ➡️  发送至STT: {chunk_size}字节 (耗时: {send_duration:.1f}ms)")
            except Exception as e:
                # 发送失败，标记为失败状态，避免后续继续发送
                if request_id not in self.recognition_failed:
                    print(f"[{request_id}] ❌ 发送音频帧错误，停止发送: {e}")
                    self.recognition_failed.add(request_id)
    
    async def handle_audio_end(self, message: AudioEndMessage):
        """处理音频结束 - 使用流式LLM优化"""
        request_id = message.requestId
        
        # 记录录音结束时间
        audio_end_time = time.time()
        self.audio_end_times[request_id] = audio_end_time
        
        # 初始化AI响应收集器
        self.ai_responses[request_id] = []
        
        # 统计音频流信息
        total_size = len(self.audio_buffers.get(request_id, bytearray()))
        duration_seconds = total_size / (16000 * 2)
        total_chunks = message.totalChunks
        
        print(f"\n[{request_id}] 🎤 音频录制结束 [时间: {time.strftime('%H:%M:%S.%f')[:-3]}]")
        print(f"[{request_id}] 📊 音频流统计: 总大小={total_size}字节, 时长={duration_seconds:.2f}秒, 共{total_chunks}个chunk")
        
        # 🔍 时间分析：最后一帧音频到达时间
        last_frame_time = self.last_audio_frame_time.get(request_id, 0)
        if last_frame_time:
            frame_delay = (audio_end_time - last_frame_time) * 1000
            print(f"[{request_id}] 🔍 最后音频帧到达 → audio_end消息: {frame_delay:.1f}ms")
        
        # 检查是否取消
        if request_id in self.cancelled_requests:
            print(f"[{request_id}] 请求已取消")
            self.cancelled_requests.remove(request_id)
            await self.cleanup_request(request_id)
            return
        
        try:
            # 1. STT: 等待完整识别结果（保证准确性）
            stt_start = time.time()
            
            # 🔥 触发STT异步停止
            stop_trigger_time = time.time()
            await self.stop_streaming_stt(request_id)
            stop_return_time = time.time()
            stop_blocking = (stop_return_time - stop_trigger_time) * 1000
            print(f"[{request_id}] ⚡ STT stop()返回 (阻塞时间: {stop_blocking:.1f}ms) [目标: <10ms]")
            
            # 🔥 优化：并行预热LLM连接（在等待STT期间）
            async def warmup_llm_connection():
                """预热LLM HTTP连接，减少首次请求延迟"""
                try:
                    print(f"[{request_id}] 🔥 预热LLM连接...")
                    # 发送极短请求以建立连接（不等待完整响应）
                    await asyncio.wait_for(
                        asyncio.to_thread(
                            lambda: ai_service.openai_client.chat.completions.create(
                                model=settings.llm_model,
                                messages=[{"role": "user", "content": "."}],
                                max_tokens=1,
                                stream=False,
                            )
                        ),
                        timeout=0.5  # 最多等待500ms
                    )
                    print(f"[{request_id}] ✅ LLM连接预热成功")
                except Exception as e:
                    # 预热失败不影响主流程
                    print(f"[{request_id}] ⚠️  LLM预热失败（忽略）: {e}")
            
            # 🔥 优化：动态超时等待 - 根据最后更新时间智能调整
            print(f"[{request_id}] ⏳ 等待STT最终识别结果（动态超时）+ 预热LLM...")
            wait_start = time.time()
            
            # 🔍 检查最后一次识别结果更新时间，动态调整超时
            last_update = self.last_recognition_update_time.get(request_id, 0)
            if last_update:
                time_since_update = (wait_start - last_update) * 1000
                print(f"[{request_id}] 🔍 当前时刻距最后一次识别更新: {time_since_update:.1f}ms")
                
                # 🎯 动态超时策略：根据最近更新时间调整
                if time_since_update < 50:
                    # 刚更新过，使用短超时
                    timeout = 0.1
                    print(f"[{request_id}] 📊 动态超时: 100ms（最近刚更新）")
                elif time_since_update < 150:
                    # 中等时间，使用中等超时
                    timeout = 0.2
                    print(f"[{request_id}] 📊 动态超时: 200ms（中等间隔）")
                else:
                    # 较长时间，使用标准超时
                    timeout = 0.3
                    print(f"[{request_id}] 📊 动态超时: 300ms（较长间隔）")
            else:
                # 没有更新记录，使用保守超时
                timeout = 0.3
                print(f"[{request_id}] 📊 动态超时: 300ms（无更新记录）")
            
            # 🔥 并行执行：等待STT + 预热LLM（节省时间）
            final_event = self.stt_final_events.get(request_id)
            
            # 创建等待任务
            if final_event:
                stt_wait_task = asyncio.create_task(
                    asyncio.wait_for(final_event.wait(), timeout=timeout)
                )
            else:
                # 降级到固定等待
                stt_wait_task = asyncio.create_task(asyncio.sleep(timeout))
            
            # 创建预热任务
            warmup_task = asyncio.create_task(warmup_llm_connection())
            
            # 并行执行（谁先完成算谁，但都不阻塞）
            got_final_event = False
            try:
                # 等待STT完成，同时预热在后台进行
                await stt_wait_task
                print(f"[{request_id}] ⚡ 收到STT最终结果事件!")
                got_final_event = True
            except asyncio.TimeoutError:
                print(f"[{request_id}] ⏰ 等待超时({int(timeout*1000)}ms)，再等待50ms给最终结果一个机会...")
                # 🔧 关键修复：超时后再等待一小段时间，给STT最终结果到达的机会
                # 因为STT的最终结果可能在事件触发失败的情况下延迟到达
                try:
                    if final_event:
                        await asyncio.wait_for(final_event.wait(), timeout=0.05)  # 额外等待50ms
                        print(f"[{request_id}] ✅ 额外等待期间收到最终结果!")
                        got_final_event = True
                except asyncio.TimeoutError:
                    print(f"[{request_id}] ⏰ 额外等待也超时，使用当前结果")
            
            # 确保预热任务也完成（或超时）
            try:
                await asyncio.wait_for(warmup_task, timeout=0.1)
            except:
                pass  # 预热失败不影响
            
            wait_actual = (time.time() - wait_start) * 1000
            print(f"[{request_id}] ✓ STT等待完成 (实际: {wait_actual:.1f}ms)")
            
            # 🔍 再次检查识别结果更新时间
            last_update_after = self.last_recognition_update_time.get(request_id, 0)
            if last_update_after and last_update:
                if last_update_after > last_update:
                    new_update_delay = (last_update_after - wait_start) * 1000
                    print(f"[{request_id}] 🔍 ✅ 等待期间有新的识别结果更新！延迟: {new_update_delay:.1f}ms")
                else:
                    print(f"[{request_id}] 🔍 ⚠️  等待期间没有新的识别结果更新")
            
            # 🔧 关键修复：如果没有收到最终结果事件，再主动等待让 STT 最终结果到达
            if not got_final_event:
                print(f"[{request_id}] 🔄 未收到最终事件，再等待500ms让识别结果更新...")
                # 🎯 增加等待时间：从 100ms → 500ms
                # 原因：commit() 后的最终结果需要更长时间（特别是在长时间没有中间结果时）
                for i in range(5):  # 分5次检查，每次100ms
                    await asyncio.sleep(0.1)
                    # 检查是否有更新
                    last_update_final = self.last_recognition_update_time.get(request_id, 0)
                    if last_update_final > last_update_after:
                        wait_time = (i + 1) * 100
                        print(f"[{request_id}] ✅ 最终等待成功！识别结果已更新 (耗时: {wait_time}ms)")
                        got_final_event = True
                        break
                
                if not got_final_event:
                    print(f"[{request_id}] ⚠️  最终等待500ms后仍无更新，使用现有结果")
            
            # 获取STT识别结果
            text = self.recognition_results.get(request_id, "")
            text_length = len(text) if text else 0
            
            if not text:
                print(f"[{request_id}] ⚠️  STT未识别到文本，使用默认")
                text = "Hello, how can I help you?"
            else:
                print(f"[{request_id}] ✓ STT识别完成: \"{text}\" (长度: {text_length}字)")
            
            # 发送最终识别结果
            await self.send_text_result(request_id, text, True)
            
            stt_end = time.time()
            stt_duration = stt_end - self.request_start_times.get(request_id, stt_start)
            print(f"[{request_id}] ⏱️  STT总耗时: {stt_duration:.2f}秒")
            
            if request_id in self.cancelled_requests:
                return
            
            # 2+3. LLM流式 + TTS并行（优化：边生成边播放）
            print(f"[{request_id}] 步骤 2+3: LLM流式 + TTS并行")
            llm_start = time.time()
            
            chunk_index = 0
            first_audio_time = None
            final_emotion = "neutral"
            sentence_count = 0
            llm_first_sentence_time = None
            emotion_sent = False  # 标记是否已发送情感
            
            # 🔥 流式处理：LLM生成一句 → 立即TTS
            async for sentence, emotion in ai_service.generate_response_stream(text):
                if request_id in self.cancelled_requests:
                    return
                
                # 🎯 优先处理情感：第一时间发送给前端
                if not emotion_sent and emotion:
                    await self.send_emotion(request_id, emotion)
                    print(f"[{request_id}] 🎭 立即发送情感给前端: {emotion}")
                    emotion_sent = True
                    final_emotion = emotion
                
                # ⚡ 如果是空句子（仅传递情感），跳过 TTS 处理
                if not sentence:
                    print(f"[{request_id}] ⚡ 收到仅情感标记（无句子内容），继续等待实际文本")
                    continue
                
                sentence_count += 1
                
                # 🧹 清理文本：移除情感标记和特殊字符（保留文字内容）
                clean_sentence = EMOTION_TAG_PATTERN.sub('', sentence)
                # 移除括号符号本身，但保留括号内的文字（避免TTS发音异常）
                # 使用 str.translate() 比多次 replace() 更快
                clean_sentence = clean_sentence.translate(str.maketrans('', '', '()[]')).strip()
                
                # 如果清理后为空，跳过
                if not clean_sentence:
                    print(f"[{request_id}] ⚠️  跳过空句子（仅含情感标记）")
                    continue
                
                # 收集AI响应（用于日志记录）
                self.ai_responses[request_id].append(clean_sentence)
                
                # 记录第一个句子生成时间
                if llm_first_sentence_time is None:
                    llm_first_sentence_time = time.time()
                    llm_first_sentence_duration = llm_first_sentence_time - llm_start
                    print(f"[{request_id}] 💬 LLM第{sentence_count}句: {clean_sentence} (耗时: {llm_first_sentence_duration:.2f}秒)")
                else:
                    print(f"[{request_id}] 💬 LLM第{sentence_count}句: {clean_sentence}")
                
                print(f"[{request_id}] 🎵 立即启动TTS合成第{sentence_count}句...")
                tts_sentence_start = time.time()
                
                # 立即TTS合成这个句子（使用清理后的文本 + 情感参数）
                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
                        print(f"[{request_id}] 🎵 第一个音频块已发送 [耗时: {time_to_first_audio:.2f}秒 = 录音结束→第一个音频块] ⚡")
                    
                    chunk_index += 1
                
                tts_sentence_duration = time.time() - tts_sentence_start
                print(f"[{request_id}] ✓ 第{sentence_count}句TTS完成 (耗时: {tts_sentence_duration:.2f}秒)")
                
                # 更新当前情感（但不再发送，因为已在开头发送）
                if emotion:
                    final_emotion = emotion
            
            # 发送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
            print(f"[{request_id}] 🎵 最后一个音频块已发送 [耗时: {time_to_last_audio:.2f}秒 = 录音结束→最后一个音频块]")
            
            # 计算总耗时
            total_end = time.time()
            total_duration = total_end - self.request_start_times.get(request_id, stt_start)
            llm_tts_total = total_end - llm_start
            
            # 性能汇总日志
            print(f"\n{'='*60}")
            print(f"[{request_id}] 📊 性能汇总报告（流式优化）")
            print(f"{'='*60}")
            print(f"  STT 耗时:              {stt_duration:.2f}秒")
            print(f"  LLM+TTS 流式总耗时:    {llm_tts_total:.2f}秒")
            print(f"    ├─ LLM第1句:         {llm_first_sentence_duration:.2f}秒" if llm_first_sentence_time else "")
            print(f"    └─ 共{sentence_count}个句子")
            if first_audio_time:
                print(f"  🎯 录音结束→首个音频:  {time_to_first_audio:.2f}秒 ⭐ [流式优化]")
                print(f"  🎯 录音结束→全部音频:  {time_to_last_audio:.2f}秒")
            print(f"  总耗时:                {total_duration:.2f}秒")
            print(f"{'='*60}")
            print(f"  📝 优化验证指标:")
            print(f"     - STT等待实际耗时: {wait_actual:.0f}ms (期望: ~300ms, 前端100ms发送间隔)")
            print(f"     - STT stop阻塞时间: {stop_blocking:.0f}ms (期望: <10ms)")
            print(f"     - 识别文本长度: {text_length}字 (检查完整性)")
            if first_audio_time:
                baseline_time = 1.57  # 串行基准：LLM(0.77s) + TTS(0.8s)
                improvement = baseline_time - time_to_first_audio
                improvement_pct = (improvement / baseline_time) * 100
                print(f"     - 流式优化效果: {improvement:+.2f}秒 ({improvement_pct:+.1f}%) [基准: {baseline_time}s]")
                print(f"     - 用户感知延迟: {time_to_first_audio:.2f}秒 [目标: <0.5s]")
            print(f"{'='*60}")
            print(f"[{request_id}] ✅ 请求处理完成\n")
            
            # 📝 记录对话日志到文件
            try:
                user_text_log = text
                ai_response_log = ''.join(self.ai_responses.get(request_id, []))
                first_audio_latency_ms = (first_audio_time - audio_end_time) * 1000 if first_audio_time else None
                
                request_logger.log_request(
                    user_text=user_text_log,
                    ai_response=ai_response_log,
                    first_audio_latency=first_audio_latency_ms,
                    request_id=request_id
                )
            except Exception as log_error:
                print(f"[{request_id}] ⚠️  记录日志失败: {log_error}")
            
        except Exception as e:
            print(f"[{request_id}] 处理错误: {e}")
            import traceback
            traceback.print_exc()
            
            # 记录错误日志
            try:
                user_text_log = self.recognition_results.get(request_id, "")
                request_logger.log_error(
                    user_text=user_text_log,
                    error_message=str(e),
                    request_id=request_id
                )
            except:
                pass
            
            await self.send_error(request_id, "NETWORK_ERROR", str(e))
        finally:
            # 清理资源
            await self.cleanup_request(request_id)
    
    async def start_streaming_stt(self, request_id: str) -> bool:
        """启动流式STT识别（使用OmniRealtimeConversation API），返回是否成功"""
        try:
            stt_init_start = time.time()
            print(f"[{request_id}] 🔊 启动流式STT识别（OmniRealtimeConversation API）...")
            
            # 发送"正在听..."的中间状态
            await self.send_text_result(request_id, "Listening...", False)
            
            from dashscope.audio.qwen_omni import OmniRealtimeConversation, OmniRealtimeCallback
            from dashscope.audio.qwen_omni.omni_realtime import TranscriptionParams, MultiModality
            import dashscope
            
            # 设置API Key（根据国际版/国内版自动选择）
            dashscope.api_key = settings.active_dashscope_api_key
            
            # 定义回调处理类
            handler = self
            req_id = request_id
            
            class StreamCallback(OmniRealtimeCallback):
                def __init__(self, conversation):
                    self.conversation = conversation
                    self.handlers = {
                        'session.created': self._handle_session_created,
                        'conversation.item.input_audio_transcription.completed': self._handle_final_text,
                        'conversation.item.input_audio_transcription.text': self._handle_stash_text,
                        'input_audio_buffer.speech_started': lambda r: print(f'[{req_id}] ======Speech Start======'),
                        'input_audio_buffer.speech_stopped': lambda r: print(f'[{req_id}] ======Speech Stop======'),
                        'error': self._handle_error
                    }
                
                def on_open(self):
                    print(f"[{req_id}] 🔗 STT连接已打开")
                
                def on_close(self, code, msg):
                    print(f"[{req_id}] 🔌 STT连接已关闭 (code: {code}, msg: {msg})")
                    # 标记为失败，停止后续音频帧发送
                    if code != 1000:  # 1000 是正常关闭
                        handler.recognition_failed.add(req_id)
                
                def on_event(self, response):
                    try:
                        event_type = response.get('type')
                        event_handler = self.handlers.get(event_type)
                        
                        # 🔍 记录所有事件（用于诊断）
                        if event_type not in ['input_audio_buffer.speech_started', 'input_audio_buffer.speech_stopped']:
                            print(f"[{req_id}] 📡 STT事件: {event_type}")
                        
                        if event_handler:
                            event_handler(response)
                        else:
                            # 🔍 记录未处理的事件
                            if event_type not in ['session.created', 'session.updated', 'input_audio_buffer.committed']:
                                print(f"[{req_id}] ⚠️  未处理的STT事件: {event_type}")
                    except Exception as e:
                        print(f"[{req_id}] ❌ 事件处理错误: {e}")
                
                def _handle_session_created(self, response):
                    print(f"[{req_id}] ✅ 会话创建: {response['session']['id']}")
                
                def _handle_final_text(self, response):
                    text = response.get('transcript', '')
                    # 🔍 增强日志：显示文本长度和完整响应
                    print(f"[{req_id}] 💬 STT识别(✅最终): {text} (长度: {len(text)}字)")
                    print(f"[{req_id}] 🔍 最终结果完整响应: {response}")
                    
                    # 更新识别结果
                    handler.recognition_results[req_id] = text
                    handler.last_recognition_update_time[req_id] = time.time()
                    
                    # 触发最终结果事件
                    if req_id in handler.stt_final_events:
                        try:
                            handler.event_loop.call_soon_threadsafe(handler.stt_final_events[req_id].set)
                            print(f"[{req_id}] ⚡ 触发STT最终结果事件")
                        except Exception as e:
                            print(f"[{req_id}] ❌ 触发事件失败: {e}")
                    
                    # 发送最终结果到前端
                    if handler.event_loop:
                        try:
                            asyncio.run_coroutine_threadsafe(
                                handler.send_text_result(req_id, text, False),
                                handler.event_loop
                            )
                        except Exception as e:
                            print(f"[{req_id}] ❌ 发送最终结果失败: {e}")
                
                def _handle_stash_text(self, response):
                    # ✅ 修复：合并 text（已确认部分）和 stash（正在识别部分）
                    confirmed_text = response.get('text', '')  # 已确认的文本
                    stash_text = response.get('stash', '')     # 正在识别的文本
                    
                    # 合并完整文本
                    full_text = confirmed_text + stash_text
                    
                    # 🔍 增强日志：显示完整文本
                    print(f"[{req_id}] 💬 STT识别(⏳中间): {full_text} (长度: {len(full_text)}字)")
                    print(f"[{req_id}]    ├─ 已确认: {confirmed_text} ({len(confirmed_text)}字)")
                    print(f"[{req_id}]    └─ 缓冲区: {stash_text} ({len(stash_text)}字)")
                    
                    # 🔍 诊断：打印原始响应（只在文本变化时）
                    if full_text and full_text != handler.recognition_results.get(req_id, ""):
                        print(f"[{req_id}] 🔍 中间结果完整响应: {response}")
                    
                    # 更新中间识别结果
                    if full_text:
                        handler.recognition_results[req_id] = full_text
                        handler.last_recognition_update_time[req_id] = time.time()
                        
                        # 发送中间结果到前端
                        if handler.event_loop:
                            try:
                                asyncio.run_coroutine_threadsafe(
                                    handler.send_text_result(req_id, full_text, False),
                                    handler.event_loop
                                )
                            except Exception as e:
                                print(f"[{req_id}] ❌ 发送中间结果失败: {e}")
                
                def _handle_error(self, response):
                    error_msg = response.get('error', {}).get('message', 'Unknown error')
                    print(f"[{req_id}] ❌ STT错误: {error_msg}")
                    handler.recognition_failed.add(req_id)
            
            # 创建会话对象
            def create_conversation():
                ws_url = get_dashscope_ws_url()
                conversation = OmniRealtimeConversation(
                    model='qwen3-asr-flash-realtime',
                    url=ws_url,
                    callback=StreamCallback(conversation=None)  # 先传None
                )
                # 注入自身引用
                conversation.callback.conversation = conversation
                return conversation
            
            conversation = await asyncio.to_thread(create_conversation)
            
            # 连接
            await asyncio.to_thread(conversation.connect)
            
            # 配置转录参数
            transcription_params = TranscriptionParams(
                language='zh',  # 中文识别
                sample_rate=16000,
                input_audio_format="pcm"
            )
            
            # 更新会话配置
            await asyncio.to_thread(
                conversation.update_session,
                output_modalities=[MultiModality.TEXT],
                enable_input_audio_transcription=True,
                transcription_params=transcription_params,
                enable_turn_detection=False  # 禁用VAD，手动提交
            )
            
            # 保存会话对象
            self.recognition_sessions[request_id] = conversation
            
            # 创建最终结果事件
            self.stt_final_events[request_id] = asyncio.Event()
            
            stt_init_duration = (time.time() - stt_init_start) * 1000
            print(f"[{request_id}] ✅ STT识别会话已启动 (耗时: {stt_init_duration:.1f}ms)")
            
            # 检查是否已经失败
            if request_id in self.recognition_failed:
                print(f"[{request_id}] ❌ STT连接启动后立即失败")
                return False
            
            return True
            
        except Exception as e:
            print(f"[{request_id}] 启动STT识别失败: {e}")
            import traceback
            traceback.print_exc()
            return False
    
    async def stop_streaming_stt(self, request_id: str):
        """停止流式STT识别（异步非阻塞）- OmniRealtimeConversation版本"""
        try:
            conversation = self.recognition_sessions.get(request_id)
            if conversation:
                print(f"[{request_id}] 🛑 提交音频并触发STT异步停止...")
                
                # 🔥 优化：异步提交并关闭，不等待完成
                async def _async_stop():
                    try:
                        stop_start = time.time()
                        # 先提交音频（触发最终识别）
                        await asyncio.to_thread(conversation.commit)
                        print(f"[{request_id}] ✅ 音频已提交")
                        
                        # 🎯 增加等待时间：从 50ms → 100ms
                        # 让 commit() 后的最终结果有更多时间到达
                        await asyncio.sleep(0.1)
                        
                        # 关闭连接
                        await asyncio.to_thread(conversation.close)
                        stop_duration = (time.time() - stop_start) * 1000
                        print(f"[{request_id}] ✅ STT连接已关闭 (耗时: {stop_duration:.1f}ms)")
                    except Exception as e:
                        print(f"[{request_id}] ⚠️  STT停止异常: {e}")
                
                # 启动异步任务但不等待
                asyncio.create_task(_async_stop())
                print(f"[{request_id}] ⚡ STT异步停止已触发，继续处理...")
                
        except Exception as e:
            print(f"[{request_id}] ❌ 停止STT识别错误: {e}")
    
    async def cleanup_request(self, request_id: str):
        """清理请求相关资源"""
        # 清理audio buffer
        if request_id in self.audio_buffers:
            del self.audio_buffers[request_id]
        
        # 清理识别会话
        if request_id in self.recognition_sessions:
            del self.recognition_sessions[request_id]
        
        # 清理识别结果
        if request_id in self.recognition_results:
            del self.recognition_results[request_id]
        
        # 清理取消标记
        if request_id in self.cancelled_requests:
            self.cancelled_requests.remove(request_id)
        
        # 清理失败标记
        if request_id in self.recognition_failed:
            self.recognition_failed.remove(request_id)
        
        # 清理时间记录
        if request_id in self.request_start_times:
            del self.request_start_times[request_id]
        
        if request_id in self.last_audio_frame_time:
            del self.last_audio_frame_time[request_id]
        
        if request_id in self.last_recognition_update_time:
            del self.last_recognition_update_time[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.stt_final_events:
            del self.stt_final_events[request_id]
        
        print(f"[{request_id}] 资源已清理")
    
    async def handle_cancel(self, message: CancelMessage):
        """处理取消请求"""
        request_id = message.requestId
        print(f"[{request_id}] 收到取消请求")
        
        self.cancelled_requests.add(request_id)
        
        # 停止STT识别
        await self.stop_streaming_stt(request_id)
        
        # 清理资源
        await self.cleanup_request(request_id)
    
    async def handle_ping(self, message: PingMessage):
        """处理Ping"""
        await self.send_pong(message.requestId)
    
    async def send_text_result(self, request_id: str, text: str, is_final: bool):
        """发送文本结果"""
        message = TextResultMessage(
            type="text_result",
            requestId=request_id,
            timestamp=self._get_timestamp(),
            text=text,
            isFinal=is_final,
        )
        await self.websocket.send_text(message.model_dump_json())
    
    async def send_emotion(self, request_id: str, emotion: str):
        """发送情感标签"""
        message = EmotionMessage(
            type="emotion",
            requestId=request_id,
            timestamp=self._get_timestamp(),
            emotion=emotion,
        )
        await self.websocket.send_text(message.model_dump_json())
    
    async def send_audio_response(
        self, request_id: str, data: str, chunk_index: int, is_last: bool,
        format: str = "pcm", sample_rate: int = 24000
    ):
        """发送音频响应"""
        message = AudioResponseMessage(
            type="audio_response",
            requestId=request_id,
            timestamp=self._get_timestamp(),
            data=data,
            sampleRate=sample_rate,
            format=format,
            chunkIndex=chunk_index,
            isLast=is_last,
        )
        await self.websocket.send_text(message.model_dump_json())
    
    async def send_error(self, request_id: str, code: str, error_message: str):
        """发送错误消息"""
        message = ErrorMessage(
            type="error",
            requestId=request_id,
            timestamp=self._get_timestamp(),
            code=code,
            message=error_message,
        )
        await self.websocket.send_text(message.model_dump_json())
    
    async def send_pong(self, request_id: str):
        """发送Pong"""
        message = PongMessage(
            type="pong",
            requestId=request_id,
            timestamp=self._get_timestamp(),
        )
        await self.websocket.send_text(message.model_dump_json())
    
    def _get_timestamp(self) -> int:
        """获取当前时间戳（毫秒）"""
        return int(time.time() * 1000)