|
|
@@ -86,21 +86,33 @@ class WebSocketHandler:
|
|
|
|
|
|
# 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()
|
|
|
import time
|
|
|
self.request_start_times[request_id] = time.time()
|
|
|
- print(f"\n[{request_id}] 开始接收音频数据 [时间: {time.strftime('%H:%M:%S')}]")
|
|
|
+ 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
|
|
|
|
|
|
@@ -109,11 +121,15 @@ class WebSocketHandler:
|
|
|
if recognition:
|
|
|
try:
|
|
|
# 在线程池中发送音频帧
|
|
|
+ import time
|
|
|
+ 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}")
|
|
|
+ print(f"[{request_id}] ❌ 发送音频帧错误,停止发送: {e}")
|
|
|
self.recognition_failed.add(request_id)
|
|
|
|
|
|
async def handle_audio_end(self, message: AudioEndMessage):
|
|
|
@@ -121,7 +137,13 @@ class WebSocketHandler:
|
|
|
request_id = message.requestId
|
|
|
import time
|
|
|
|
|
|
- print(f"\n[{request_id}] 音频录制结束 [时间: {time.strftime('%H:%M:%S')}]")
|
|
|
+ # 统计音频流信息
|
|
|
+ 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")
|
|
|
|
|
|
# 检查是否取消
|
|
|
if request_id in self.cancelled_requests:
|
|
|
@@ -153,69 +175,56 @@ class WebSocketHandler:
|
|
|
if request_id in self.cancelled_requests:
|
|
|
return
|
|
|
|
|
|
- # 2. LLM: 流式生成回复(核心优化点)
|
|
|
- print(f"[{request_id}] 步骤 2/3: LLM 流式生成回复")
|
|
|
+ # 2. LLM: 生成回复(使用非流式,保证音频连贯性)
|
|
|
+ print(f"[{request_id}] 步骤 2/3: LLM 生成回复")
|
|
|
llm_start = time.time()
|
|
|
|
|
|
- sentence_index = 0
|
|
|
- emotion = "neutral"
|
|
|
- first_sentence_time = 0
|
|
|
- tts_tasks = []
|
|
|
-
|
|
|
- # 流式处理LLM输出 - 边生成边TTS
|
|
|
- async for sentence, sent_emotion in ai_service.generate_response_stream(text):
|
|
|
- if request_id in self.cancelled_requests:
|
|
|
- # 取消所有TTS任务
|
|
|
- for task in tts_tasks:
|
|
|
- task.cancel()
|
|
|
- return
|
|
|
-
|
|
|
- # 记录第一句时间(用户感知延迟)
|
|
|
- if sentence_index == 0:
|
|
|
- first_sentence_time = time.time()
|
|
|
- user_perceived_latency = first_sentence_time - self.request_start_times.get(request_id, llm_start)
|
|
|
- print(f"[{request_id}] ⏱️ 首句生成耗时: {first_sentence_time - llm_start:.2f}秒")
|
|
|
- print(f"[{request_id}] ⏱️ **用户感知延迟**: {user_perceived_latency:.2f}秒 (从说话结束到听到回复)")
|
|
|
-
|
|
|
- # 保存情感(使用最后一句的情感)
|
|
|
- emotion = sent_emotion
|
|
|
-
|
|
|
- # 立即TTS这个句子(异步,不阻塞下一句生成)
|
|
|
- print(f"[{request_id}] TTS句子{sentence_index + 1}: {sentence}")
|
|
|
- task = asyncio.create_task(
|
|
|
- self.process_tts_sentence(request_id, sentence, emotion, sentence_index)
|
|
|
- )
|
|
|
- tts_tasks.append(task)
|
|
|
-
|
|
|
- sentence_index += 1
|
|
|
+ # 调用非流式 LLM,获取完整回复和情感
|
|
|
+ response_text, emotion = await ai_service.generate_response(text)
|
|
|
|
|
|
llm_end = time.time()
|
|
|
llm_duration = llm_end - llm_start
|
|
|
+ print(f"[{request_id}] ⏱️ LLM耗时: {llm_duration:.2f}秒")
|
|
|
|
|
|
- print(f"[{request_id}] ⏱️ LLM总耗时: {llm_duration:.2f}秒 (生成{sentence_index}个句子)")
|
|
|
+ if request_id in self.cancelled_requests:
|
|
|
+ return
|
|
|
|
|
|
# 发送情感标签
|
|
|
await self.send_emotion(request_id, emotion)
|
|
|
|
|
|
- # 等待所有TTS任务完成
|
|
|
- if tts_tasks:
|
|
|
- await asyncio.gather(*tts_tasks, return_exceptions=True)
|
|
|
+ # 3. TTS: 一次性生成完整音频(保证音色一致、无停顿)
|
|
|
+ print(f"[{request_id}] 步骤 3/3: TTS 生成语音")
|
|
|
+ tts_start = time.time()
|
|
|
+
|
|
|
+ chunk_index = 0
|
|
|
+ async for audio_chunk in ai_service.text_to_speech_stream(response_text, 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,
|
|
|
+ )
|
|
|
+ chunk_index += 1
|
|
|
+
|
|
|
+ tts_end = time.time()
|
|
|
+ tts_duration = tts_end - tts_start
|
|
|
+ print(f"[{request_id}] ⏱️ TTS耗时: {tts_duration:.2f}秒 (共{chunk_index}个数据块)")
|
|
|
|
|
|
# 发送TTS结束标记
|
|
|
- await self.send_audio_response(request_id, "", sentence_index * 100, True)
|
|
|
+ await self.send_audio_response(request_id, "", chunk_index, True)
|
|
|
|
|
|
# 计算总耗时
|
|
|
total_end = time.time()
|
|
|
- total_duration = total_end - self.request_start_times.get(request_id, llm_start)
|
|
|
- print(f"[{request_id}] ⏱️ 总耗时: {total_duration:.2f}秒")
|
|
|
-
|
|
|
- # 性能对比输出
|
|
|
- if first_sentence_time > 0:
|
|
|
- user_perceived_latency = first_sentence_time - self.request_start_times.get(request_id, llm_start)
|
|
|
- improvement = ((stt_duration + llm_duration) - user_perceived_latency) / (stt_duration + llm_duration) * 100
|
|
|
- print(f"[{request_id}] 📊 性能提升: 用户感知延迟减少 {improvement:.1f}% (传统: {stt_duration + llm_duration:.2f}秒 vs 流式: {user_perceived_latency:.2f}秒)")
|
|
|
-
|
|
|
- print(f"[{request_id}] 请求处理完成\n")
|
|
|
+ total_duration = total_end - self.request_start_times.get(request_id, stt_start)
|
|
|
+ print(f"[{request_id}] ⏱️ 总耗时: {total_duration:.2f}秒 (STT: {stt_duration:.2f}s + LLM: {llm_duration:.2f}s + TTS: {tts_duration:.2f}s)")
|
|
|
+ print(f"[{request_id}] ✅ 请求处理完成\n")
|
|
|
|
|
|
except Exception as e:
|
|
|
print(f"[{request_id}] 处理错误: {e}")
|
|
|
@@ -229,7 +238,9 @@ class WebSocketHandler:
|
|
|
async def start_streaming_stt(self, request_id: str) -> bool:
|
|
|
"""启动流式STT识别,返回是否成功"""
|
|
|
try:
|
|
|
- print(f"[{request_id}] 启动流式STT识别")
|
|
|
+ import time
|
|
|
+ stt_init_start = time.time()
|
|
|
+ print(f"[{request_id}] 🔊 启动流式STT识别...")
|
|
|
|
|
|
# 发送"正在听..."的中间状态
|
|
|
await self.send_text_result(request_id, "Listening...", False)
|
|
|
@@ -242,15 +253,15 @@ class WebSocketHandler:
|
|
|
|
|
|
class StreamCallback(RecognitionCallback):
|
|
|
def on_open(self):
|
|
|
- print(f"[{req_id}] STT连接已打开")
|
|
|
+ print(f"[{req_id}] 🔗 STT连接已打开")
|
|
|
|
|
|
def on_close(self):
|
|
|
- print(f"[{req_id}] STT连接已关闭")
|
|
|
+ print(f"[{req_id}] 🔌 STT连接已关闭")
|
|
|
# 标记为失败,停止后续音频帧发送
|
|
|
handler.recognition_failed.add(req_id)
|
|
|
|
|
|
def on_error(self, error):
|
|
|
- print(f"[{req_id}] STT错误: {error}")
|
|
|
+ print(f"[{req_id}] ❌ STT错误: {error}")
|
|
|
# 标记为失败
|
|
|
handler.recognition_failed.add(req_id)
|
|
|
|
|
|
@@ -259,7 +270,11 @@ class WebSocketHandler:
|
|
|
sentence = result.get_sentence()
|
|
|
if sentence and 'text' in sentence:
|
|
|
text = sentence['text']
|
|
|
- print(f"[{req_id}] STT实时结果: {text}")
|
|
|
+ # 判断是否为最终结果(end_time 可能为 None)
|
|
|
+ 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}")
|
|
|
# 保存最新的识别结果
|
|
|
handler.recognition_results[req_id] = text
|
|
|
# 从回调线程安全地发送到事件循环
|
|
|
@@ -270,7 +285,7 @@ class WebSocketHandler:
|
|
|
handler.event_loop
|
|
|
)
|
|
|
except Exception as e:
|
|
|
- print(f"[{req_id}] 发送中间结果失败: {e}")
|
|
|
+ print(f"[{req_id}] ❌ 发送中间结果失败: {e}")
|
|
|
|
|
|
# 创建识别对象
|
|
|
def create_recognition():
|
|
|
@@ -288,14 +303,16 @@ class WebSocketHandler:
|
|
|
|
|
|
# 保存识别会话
|
|
|
self.recognition_sessions[request_id] = recognition
|
|
|
- print(f"[{request_id}] STT识别会话已启动")
|
|
|
+
|
|
|
+ stt_init_duration = (time.time() - stt_init_start) * 1000
|
|
|
+ print(f"[{request_id}] ✅ STT识别会话已启动 (耗时: {stt_init_duration:.1f}ms)")
|
|
|
|
|
|
# 短暂延迟,确保连接稳定
|
|
|
await asyncio.sleep(0.1)
|
|
|
|
|
|
# 检查是否已经失败
|
|
|
if request_id in self.recognition_failed:
|
|
|
- print(f"[{request_id}] STT连接启动后立即失败")
|
|
|
+ print(f"[{request_id}] ❌ STT连接启动后立即失败")
|
|
|
return False
|
|
|
|
|
|
return True
|
|
|
@@ -311,36 +328,14 @@ class WebSocketHandler:
|
|
|
try:
|
|
|
recognition = self.recognition_sessions.get(request_id)
|
|
|
if recognition:
|
|
|
- print(f"[{request_id}] 停止STT识别")
|
|
|
+ print(f"[{request_id}] 🛑 停止STT识别...")
|
|
|
+ import time
|
|
|
+ stop_start = time.time()
|
|
|
await asyncio.to_thread(recognition.stop)
|
|
|
- print(f"[{request_id}] STT识别已停止")
|
|
|
- except Exception as e:
|
|
|
- print(f"[{request_id}] 停止STT识别错误: {e}")
|
|
|
-
|
|
|
- async def process_tts_sentence(self, request_id: str, sentence: str, emotion: str, sentence_index: int):
|
|
|
- """处理单个句子的TTS(异步,支持并行)"""
|
|
|
- try:
|
|
|
- chunk_index = sentence_index * 100 # 每个句子的chunk从不同起点开始
|
|
|
-
|
|
|
- # 调用AI服务生成语音
|
|
|
- async for audio_chunk in ai_service.text_to_speech_stream(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,
|
|
|
- )
|
|
|
- chunk_index += 1
|
|
|
-
|
|
|
+ stop_duration = (time.time() - stop_start) * 1000
|
|
|
+ print(f"[{request_id}] ✅ STT识别已停止 (耗时: {stop_duration:.1f}ms)")
|
|
|
except Exception as e:
|
|
|
- print(f"[{request_id}] TTS句子{sentence_index}错误: {e}")
|
|
|
+ print(f"[{request_id}] ❌ 停止STT识别错误: {e}")
|
|
|
|
|
|
async def cleanup_request(self, request_id: str):
|
|
|
"""清理请求相关资源"""
|