""" 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 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 -> Recognition对象 self.recognition_results: Dict[str, str] = {} # requestId -> 识别结果文本 self.recognition_sentences: Dict[str, list] = {} # 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服务 recognition = self.recognition_sessions.get(request_id) if recognition: try: # 在线程池中发送音频帧 send_start = time.time() await asyncio.to_thread(recognition.send_audio_frame, audio_data) 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}] 🔍 ⚠️ 等待期间没有新的识别结果更新") # 🔧 关键修复：如果没有收到最终结果事件，再主动等待一小段时间让 recognition_results 更新 if not got_final_event: print(f"[{request_id}] 🔄 未收到最终事件，再等待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: print(f"[{request_id}] ✅ 最终等待成功！识别结果已更新") else: print(f"[{request_id}] ⚠️ 最终等待后仍无更新，使用现有结果") # 获取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识别，返回是否成功""" try: stt_init_start = time.time() print(f"[{request_id}] 🔊 启动流式STT识别...") # 发送"正在听..."的中间状态 await self.send_text_result(request_id, "Listening...", False) from dashscope.audio.asr import Recognition, RecognitionCallback, RecognitionResult # 定义回调处理类 handler = self req_id = request_id class StreamCallback(RecognitionCallback): def on_open(self): print(f"[{req_id}] 🔗 STT连接已打开") def on_close(self): print(f"[{req_id}] 🔌 STT连接已关闭") # 标记为失败，停止后续音频帧发送 handler.recognition_failed.add(req_id) def on_error(self, error): print(f"[{req_id}] ❌ STT错误: {error}") # 标记为失败 handler.recognition_failed.add(req_id) def on_event(self, result: RecognitionResult): # 获取识别的句子 sentence = result.get_sentence() if sentence and 'text' in sentence: text = sentence['text'] begin_time = sentence.get('begin_time', 0) end_time = sentence.get('end_time') is_final = end_time is not None and end_time > 0 status = "✅最终" if is_final else "⏳中间" print(f"[{req_id}] 💬 STT识别({status}): {text} [时间: {begin_time}-{end_time}]") # 初始化句子列表 if req_id not in handler.recognition_sentences: handler.recognition_sentences[req_id] = [] sentence_list = handler.recognition_sentences[req_id] # 查找是否是已有句子的更新（根据begin_time匹配） updated = False for i, sent in enumerate(sentence_list): if abs(sent['begin_time'] - begin_time) < 100: # 100ms容差 # 更新现有句子 sentence_list[i] = {'text': text, 'begin_time': begin_time, 'end_time': end_time} updated = True break if not updated: # 新句子，追加 sentence_list.append({'text': text, 'begin_time': begin_time, 'end_time': end_time}) # 按时间排序并合并所有句子 sentence_list.sort(key=lambda x: x['begin_time']) accumulated_text = ' '.join([s['text'] for s in sentence_list]) # 🔍 记录识别结果更新时间 handler.last_recognition_update_time[req_id] = time.time() handler.recognition_results[req_id] = accumulated_text print(f"[{req_id}] 📝 累积文本: {accumulated_text} [更新时间: {time.strftime('%H:%M:%S.%f')[:-3]}]") # 🔥 如果是最终结果，触发事件 if is_final and 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, accumulated_text, False), handler.event_loop ) except Exception as e: print(f"[{req_id}] ❌ 发送中间结果失败: {e}") # 创建识别对象 def create_recognition(): return Recognition( model='paraformer-realtime-v2', format='pcm', sample_rate=16000, callback=StreamCallback() ) recognition = await asyncio.to_thread(create_recognition) # 启动识别 await asyncio.to_thread(recognition.start) # 保存识别会话 self.recognition_sessions[request_id] = recognition # 创建最终结果事件 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)") # 🔥 优化：移除不必要的100ms延迟，直接开始 # await asyncio.sleep(0.1) # 已移除 # 检查是否已经失败 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识别（异步非阻塞）""" try: recognition = self.recognition_sessions.get(request_id) if recognition: print(f"[{request_id}] 🛑 触发STT异步停止...") # 🔥 优化：异步停止STT，不等待完成 # 这样可以立即进入LLM处理，节省200-340ms async def _async_stop(): try: stop_start = time.time() await asyncio.to_thread(recognition.stop) 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.recognition_sentences: del self.recognition_sentences[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)