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