From 60521b0a7ef6fc2d9ec7cdfd96a4d6688fb0f582 Mon Sep 17 00:00:00 2001 From: FAN-yeB <1442100690@qq.com> Date: Tue, 9 Dec 2025 14:42:50 +0800 Subject: [PATCH 1/8] =?UTF-8?q?update:=E9=95=BF=E6=8C=89=E8=AF=B4=E8=AF=9D?= =?UTF-8?q?=E4=B8=8D=E8=B5=B0VAD=E7=9B=B4=E6=8E=A5=E8=A7=A6=E5=8F=91ASR?= =?UTF-8?q?=E8=AF=86=E5=88=AB?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../textHandler/listenMessageHandler.py | 8 ++++++- .../xiaozhi-server/core/providers/asr/base.py | 23 ++++++++++++------- .../core/providers/vad/silero.py | 4 ++++ 3 files changed, 26 insertions(+), 9 deletions(-) diff --git a/main/xiaozhi-server/core/handle/textHandler/listenMessageHandler.py b/main/xiaozhi-server/core/handle/textHandler/listenMessageHandler.py index 97286dfe..65b442ad 100644 --- a/main/xiaozhi-server/core/handle/textHandler/listenMessageHandler.py +++ b/main/xiaozhi-server/core/handle/textHandler/listenMessageHandler.py @@ -30,7 +30,13 @@ class ListenTextMessageHandler(TextMessageHandler): conn.client_have_voice = True conn.client_voice_stop = True if len(conn.asr_audio) > 0: - await handleAudioMessage(conn, b"") + # 手动模式下直接触发ASR识别，不需要再调用handleAudioMessage + asr_audio_task = conn.asr_audio.copy() + conn.asr_audio.clear() + conn.reset_vad_states() + + if len(asr_audio_task) > 0: + await conn.asr.handle_voice_stop(conn, asr_audio_task) elif msg_json["state"] == "detect": conn.client_have_voice = False conn.asr_audio.clear() diff --git a/main/xiaozhi-server/core/providers/asr/base.py b/main/xiaozhi-server/core/providers/asr/base.py index 7e36de5c..6c3c2d25 100644 --- a/main/xiaozhi-server/core/providers/asr/base.py +++ b/main/xiaozhi-server/core/providers/asr/base.py @@ -56,21 +56,28 @@ class ASRProviderBase(ABC): async def receive_audio(self, conn, audio, audio_have_voice): if conn.client_listen_mode == "auto" or conn.client_listen_mode == "realtime": have_voice = audio_have_voice + + conn.asr_audio.append(audio) + if not have_voice and not conn.client_have_voice: + conn.asr_audio = conn.asr_audio[-10:] + return else: - have_voice = conn.client_have_voice - - conn.asr_audio.append(audio) - if not have_voice and not conn.client_have_voice: - conn.asr_audio = conn.asr_audio[-10:] - return + # 手动模式：总是缓存音频，忽略VAD检测结果 + conn.asr_audio.append(audio) if conn.client_voice_stop: asr_audio_task = conn.asr_audio.copy() conn.asr_audio.clear() conn.reset_vad_states() - if len(asr_audio_task) > 15: - await self.handle_voice_stop(conn, asr_audio_task) + # 手动模式下允许短语音识别，自动模式保持原有限制 + if conn.client_listen_mode == "auto" or conn.client_listen_mode == "realtime": + if len(asr_audio_task) > 15: + await self.handle_voice_stop(conn, asr_audio_task) + else: + # 手动模式：只要有音频就进行识别 + if len(asr_audio_task) > 0: + await self.handle_voice_stop(conn, asr_audio_task) # 处理语音停止 async def handle_voice_stop(self, conn, asr_audio_task: List[bytes]): diff --git a/main/xiaozhi-server/core/providers/vad/silero.py b/main/xiaozhi-server/core/providers/vad/silero.py index ae51b88f..28a3e61e 100644 --- a/main/xiaozhi-server/core/providers/vad/silero.py +++ b/main/xiaozhi-server/core/providers/vad/silero.py @@ -45,6 +45,10 @@ class VADProvider(VADProviderBase): pass def is_vad(self, conn, opus_packet): + # 手动模式：直接返回True，不进行实时VAD检测，所有音频都缓存 + if conn.client_listen_mode not in ["auto", "realtime"]: + return True + try: pcm_frame = self.decoder.decode(opus_packet, 960) conn.client_audio_buffer.extend(pcm_frame) # 将新数据加入缓冲区 From 48094f7e376324d893773f667500b5eb14293404 Mon Sep 17 00:00:00 2001 From: Sakura-RanChen <1908198662@qq.com> Date: Tue, 9 Dec 2025 18:06:56 +0800 Subject: [PATCH 2/8] =?UTF-8?q?update:=20=E9=9F=B3=E9=A2=91=E5=90=8E?= =?UTF-8?q?=E5=8F=B0=E9=98=9F=E5=88=97=E7=A8=B3=E5=AE=9A=E5=8F=91=E9=80=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- main/xiaozhi-server/core/connection.py | 5 + .../core/handle/sendAudioHandle.py | 199 ++++++++++-------- .../core/utils/audioRateController.py | 84 +++++--- .../core/utils/opus_encoder_utils.py | 3 + 4 files changed, 166 insertions(+), 125 deletions(-) diff --git a/main/xiaozhi-server/core/connection.py b/main/xiaozhi-server/core/connection.py index 6b34e32f..b37745c7 100644 --- a/main/xiaozhi-server/core/connection.py +++ b/main/xiaozhi-server/core/connection.py @@ -1171,6 +1171,11 @@ class ConnectionHandler: except queue.Empty: break + # 重置音频流控器（取消后台任务并清空队列） + if hasattr(self, "audio_rate_controller") and self.audio_rate_controller: + self.audio_rate_controller.reset() + self.logger.bind(tag=TAG).debug("已重置音频流控器") + self.logger.bind(tag=TAG).debug( f"清理结束: TTS队列大小={self.tts.tts_text_queue.qsize()}, 音频队列大小={self.tts.tts_audio_queue.qsize()}" ) diff --git a/main/xiaozhi-server/core/handle/sendAudioHandle.py b/main/xiaozhi-server/core/handle/sendAudioHandle.py index ea952fbe..f662b6c6 100644 --- a/main/xiaozhi-server/core/handle/sendAudioHandle.py +++ b/main/xiaozhi-server/core/handle/sendAudioHandle.py @@ -16,7 +16,14 @@ async def sendAudioMessage(conn, sentenceType, audios, text): await send_tts_message(conn, "start", None) if sentenceType == SentenceType.FIRST: - await send_tts_message(conn, "sentence_start", text) + # 同一句子的后续消息加入流控队列，其他情况立即发送 + if hasattr(conn, "audio_rate_controller") and conn.audio_rate_controller and getattr(conn, "audio_flow_control", {}).get("sentence_id") == conn.sentence_id: + conn.audio_rate_controller.add_message( + lambda: send_tts_message(conn, "sentence_start", text) + ) + else: + # 新句子或流控器未初始化，立即发送 + await send_tts_message(conn, "sentence_start", text) await sendAudio(conn, audios) # 发送句子开始消息 @@ -31,6 +38,22 @@ async def sendAudioMessage(conn, sentenceType, audios, text): await conn.close() +async def _wait_for_audio_completion(conn): + """ + 等待音频队列清空 + + Args: + conn: 连接对象 + """ + if hasattr(conn, "audio_rate_controller") and conn.audio_rate_controller: + rate_controller = conn.audio_rate_controller + conn.logger.bind(tag=TAG).debug( + f"等待音频发送完成，队列中还有 {len(rate_controller.queue)} 个包" + ) + await rate_controller.queue_empty_event.wait() + conn.logger.bind(tag=TAG).debug("音频发送完成") + + async def _send_to_mqtt_gateway(conn, opus_packet, timestamp, sequence): """ 发送带16字节头部的opus数据包给mqtt_gateway @@ -53,7 +76,6 @@ async def _send_to_mqtt_gateway(conn, opus_packet, timestamp, sequence): await conn.websocket.send(complete_packet) -# 播放音频 - 使用 AudioRateController 进行精确流控 async def sendAudio(conn, audios, frame_duration=60): """ 发送音频包，使用 AudioRateController 进行精确的流量控制 @@ -62,130 +84,121 @@ async def sendAudio(conn, audios, frame_duration=60): conn: 连接对象 audios: 单个opus包(bytes) 或 opus包列表 frame_duration: 帧时长（毫秒），默认60ms - - 改进点： - 1. 使用单一时间基准，避免累积误差 - 2. 每次检查队列时重新计算 elapsed_ms，更精准 - 3. 支持高并发而不产生时间偏差 """ if audios is None or len(audios) == 0: return - # 获取发送延迟配置 send_delay = conn.config.get("tts_audio_send_delay", -1) / 1000.0 + is_single_packet = isinstance(audios, bytes) - if isinstance(audios, bytes): - # 单个 opus 包处理 - await _sendAudio_single(conn, audios, send_delay, frame_duration) - else: - # 音频列表处理（如文件型音频） - await _sendAudio_list(conn, audios, send_delay, frame_duration) + # 初始化或获取 RateController + rate_controller, flow_control = _get_or_create_rate_controller( + conn, frame_duration, is_single_packet + ) + + # 统一转换为列表处理 + audio_list = [audios] if is_single_packet else audios + + # 发送音频包 + await _send_audio_with_rate_control( + conn, audio_list, rate_controller, flow_control, send_delay + ) -async def _sendAudio_single(conn, opus_packet, send_delay, frame_duration=60): +def _get_or_create_rate_controller(conn, frame_duration, is_single_packet): """ - 发送单个 opus 包 - 使用 AudioRateController 进行流控 + 获取或创建 RateController 和 flow_control + + Args: + conn: 连接对象 + frame_duration: 帧时长 + is_single_packet: 是否单包模式（True: TTS流式单包, False: 批量包） + + Returns: + (rate_controller, flow_control) """ - # 重置流控状态，第一次读取和会话发生转变时 - if not hasattr(conn, "audio_rate_controller") or conn.audio_flow_control.get("sentence_id") != conn.sentence_id: - if hasattr(conn, "audio_rate_controller"): - conn.audio_rate_controller.reset() - else: + # 判断是否需要重置：单包模式且 sentence_id 变化，或者控制器不存在 + need_reset = ( + is_single_packet + and getattr(conn, "audio_flow_control", {}).get("sentence_id") != conn.sentence_id + ) or not hasattr(conn, "audio_rate_controller") + + if need_reset: + # 创建或获取 rate_controller + if not hasattr(conn, "audio_rate_controller"): conn.audio_rate_controller = AudioRateController(frame_duration) + else: conn.audio_rate_controller.reset() + # 初始化 flow_control conn.audio_flow_control = { "packet_count": 0, "sequence": 0, "sentence_id": conn.sentence_id, } - if conn.client_abort: - return + # 启动后台发送循环 + _start_background_sender(conn, conn.audio_rate_controller, conn.audio_flow_control) - conn.last_activity_time = time.time() * 1000 + return conn.audio_rate_controller, conn.audio_flow_control - rate_controller = conn.audio_rate_controller - flow_control = conn.audio_flow_control - packet_count = flow_control["packet_count"] - # 预缓冲：前5个包直接发送，不做延迟 +def _start_background_sender(conn, rate_controller, flow_control): + """ + 启动后台发送循环任务 + + Args: + conn: 连接对象 + rate_controller: 速率控制器 + flow_control: 流控状态 + """ + async def send_callback(packet): + # 检查是否应该中止 + if conn.client_abort: + raise asyncio.CancelledError("客户端已中止") + + conn.last_activity_time = time.time() * 1000 + await _do_send_audio(conn, packet, flow_control) + conn.client_is_speaking = True + + # 使用 start_sending 启动后台循环 + rate_controller.start_sending(send_callback) + + +async def _send_audio_with_rate_control(conn, audio_list, rate_controller, flow_control, send_delay): + """ + 使用 rate_controller 发送音频包 + + Args: + conn: 连接对象 + audio_list: 音频包列表 + rate_controller: 速率控制器 + flow_control: 流控状态 + send_delay: 固定延迟（秒），-1表示使用动态流控 + """ pre_buffer_count = 5 - if packet_count < pre_buffer_count or send_delay > 0: - # 预缓冲阶段或固定延迟模式，直接发送 - await _do_send_audio(conn, opus_packet, flow_control, frame_duration) - conn.client_is_speaking = True - - if send_delay > 0 and packet_count >= pre_buffer_count: - await asyncio.sleep(send_delay) - else: - # 使用流控器进行精确的速率控制 - rate_controller.add_audio(opus_packet) - - async def send_callback(packet): - await _do_send_audio(conn, packet, flow_control, frame_duration) - - await rate_controller.check_queue(send_callback) - conn.client_is_speaking = True - - # 更新流控状态 - flow_control["packet_count"] += 1 - flow_control["sequence"] += 1 - - -async def _sendAudio_list(conn, audios, send_delay, frame_duration=60): - """ - 发送音频列表（如文件型音频） - """ - if not audios: - return - - rate_controller = AudioRateController(frame_duration) - rate_controller.reset() - - flow_control = { - "packet_count": 0, - "sequence": 0, - } - - # 预缓冲：前5个包直接发送 - pre_buffer_frames = min(5, len(audios)) - for i in range(pre_buffer_frames): + for packet in audio_list: if conn.client_abort: return - await _do_send_audio(conn, audios[i], flow_control, frame_duration) - conn.client_is_speaking = True - - remaining_audios = audios[pre_buffer_frames:] - - # 处理剩余音频帧 - for i, opus_packet in enumerate(remaining_audios): - if conn.client_abort: - break conn.last_activity_time = time.time() * 1000 - if send_delay > 0: + # 预缓冲：前5个包直接发送 + if flow_control["packet_count"] < pre_buffer_count: + await _do_send_audio(conn, packet, flow_control) + conn.client_is_speaking = True + elif send_delay > 0: # 固定延迟模式 await asyncio.sleep(send_delay) - else: - # 使用流控器进行精确延迟 - rate_controller.add_audio(opus_packet) - - async def send_callback(packet): - await _do_send_audio(conn, packet, flow_control, frame_duration) - - await rate_controller.check_queue(send_callback) + await _do_send_audio(conn, packet, flow_control) conn.client_is_speaking = True - continue - - await _do_send_audio(conn, opus_packet, flow_control, frame_duration) - conn.client_is_speaking = True + else: + # 动态流控模式：仅添加到队列，由后台循环负责发送 + rate_controller.add_audio(packet) -async def _do_send_audio(conn, opus_packet, flow_control, frame_duration=60): +async def _do_send_audio(conn, opus_packet, flow_control): """ 执行实际的音频发送 """ @@ -224,6 +237,8 @@ async def send_tts_message(conn, state, text=None): ) audios = audio_to_data(stop_tts_notify_voice, is_opus=True) await sendAudio(conn, audios) + # 等待所有音频包发送完成 + await _wait_for_audio_completion(conn) # 清除服务端讲话状态 conn.clearSpeakStatus() diff --git a/main/xiaozhi-server/core/utils/audioRateController.py b/main/xiaozhi-server/core/utils/audioRateController.py index b404e6f1..e6d883f1 100644 --- a/main/xiaozhi-server/core/utils/audioRateController.py +++ b/main/xiaozhi-server/core/utils/audioRateController.py @@ -10,12 +10,6 @@ class AudioRateController: """ 音频速率控制器 - 按照60ms帧时长精确控制音频发送 解决高并发下的时间累积误差问题 - - 关键改进： - 1. 单一时间基准（start_timestamp 只初始化一次） - 2. 每次检查队列时重新计算 elapsed_ms，避免累积误差 - 3. 分离"检查时间"和"发送"两个操作 - 4. 支持高并发而不产生延迟 """ def __init__(self, frame_duration=60): @@ -29,24 +23,34 @@ class AudioRateController: self.start_timestamp = None # 开始时间戳（只读，不修改） self.pending_send_task = None self.logger = logger + self.queue_empty_event = asyncio.Event() # 队列清空事件 + self.queue_empty_event.set() # 初始为空状态 def reset(self): """重置控制器状态""" if self.pending_send_task and not self.pending_send_task.done(): self.pending_send_task.cancel() - try: - # 等待任务取消完成 - asyncio.get_event_loop().run_until_complete(self.pending_send_task) - except asyncio.CancelledError: - pass + # 取消任务后，任务会在下次事件循环时清理，无需阻塞等待 self.queue.clear() self.play_position = 0 self.start_timestamp = time.time() + self.queue_empty_event.set() # 队列已清空 def add_audio(self, opus_packet): """添加音频包到队列""" self.queue.append(("audio", opus_packet)) + self.queue_empty_event.clear() # 队列非空，清除事件 + + def add_message(self, message_callback): + """ + 添加消息到队列（立即发送，不占用播放时间） + + Args: + message_callback: 消息发送回调函数 async def() + """ + self.queue.append(("message", message_callback)) + self.queue_empty_event.clear() # 队列非空，清除事件 def _get_elapsed_ms(self): """获取已经过的时间（毫秒）""" @@ -62,34 +66,47 @@ class AudioRateController: send_audio_callback: 发送音频的回调函数 async def(opus_packet) """ if self.start_timestamp is None: - self.reset() + self.start_timestamp = time.time() while self.queue: item = self.queue[0] item_type = item[0] - if item_type == "audio": + if item_type == "message": + # 消息类型：立即发送，不占用播放时间 + _, message_callback = item + self.queue.pop(0) + try: + await message_callback() + except Exception as e: + self.logger.bind(tag=TAG).error(f"发送消息失败: {e}") + raise + + elif item_type == "audio": _, opus_packet = item - # 计算时间差 - elapsed_ms = self._get_elapsed_ms() - output_ms = self.play_position + # 循环等待直到时间到达 + while True: + # 计算时间差 + elapsed_ms = self._get_elapsed_ms() + output_ms = self.play_position - if elapsed_ms < output_ms: - # 还不到发送时间，计算等待时长 - wait_ms = output_ms - elapsed_ms + if elapsed_ms < output_ms: + # 还不到发送时间，计算等待时长 + wait_ms = output_ms - elapsed_ms - # 等待后继续检查（允许被中断） - try: - await asyncio.sleep(wait_ms / 1000) - except asyncio.CancelledError: - self.logger.bind(tag=TAG).debug("音频发送任务被取消") - raise + # 等待后继续检查（允许被中断） + try: + await asyncio.sleep(wait_ms / 1000) + except asyncio.CancelledError: + self.logger.bind(tag=TAG).debug("音频发送任务被取消") + raise + # 等待结束后重新检查时间（循环回到 while True） + else: + # 时间已到，跳出等待循环 + break - # 继续循环检查（时间可能已到） - continue - - # 时间已到，发送音频 + # 时间已到，从队列移除并发送 self.queue.pop(0) self.play_position += self.frame_duration @@ -99,14 +116,15 @@ class AudioRateController: self.logger.bind(tag=TAG).error(f"发送音频失败: {e}") raise + self.queue_empty_event.set() - async def start_sending(self, send_audio_callback, send_message_callback=None): + + def start_sending(self, send_audio_callback): """ 启动异步发送任务 Args: send_audio_callback: 发送音频的回调函数 - send_message_callback: 发送消息的回调函数 Returns: asyncio.Task: 发送任务 @@ -114,11 +132,11 @@ class AudioRateController: async def _send_loop(): try: while True: - await self.check_queue(send_audio_callback, send_message_callback) + await self.check_queue(send_audio_callback) # 如果队列空了，短暂等待后再检查（避免 busy loop） await asyncio.sleep(0.01) except asyncio.CancelledError: - self.logger.bind(tag=TAG).info("音频发送循环已停止") + self.logger.bind(tag=TAG).debug("音频发送循环已停止") except Exception as e: self.logger.bind(tag=TAG).error(f"音频发送循环异常: {e}") diff --git a/main/xiaozhi-server/core/utils/opus_encoder_utils.py b/main/xiaozhi-server/core/utils/opus_encoder_utils.py index 300b2422..f2fcc266 100644 --- a/main/xiaozhi-server/core/utils/opus_encoder_utils.py +++ b/main/xiaozhi-server/core/utils/opus_encoder_utils.py @@ -103,6 +103,9 @@ class OpusEncoderUtils: def _encode(self, frame: np.ndarray) -> Optional[bytes]: """编码一帧音频数据""" try: + # 编码器已释放，跳过编码 + if not hasattr(self, 'encoder') or self.encoder is None: + return None # 将numpy数组转换为bytes frame_bytes = frame.tobytes() # opuslib要求输入字节数必须是channels*2的倍数 From 3a74b30a0e5669e7e1d8e69bdd4ba0a892c54497 Mon Sep 17 00:00:00 2001 From: FAN-yeB <1442100690@qq.com> Date: Wed, 10 Dec 2025 17:57:26 +0800 Subject: [PATCH 3/8] =?UTF-8?q?=E6=9B=B4=E6=96=B0=20qwen3=5Fasr=5Fflash.py?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../core/providers/asr/qwen3_asr_flash.py | 23 ++----------------- 1 file changed, 2 insertions(+), 21 deletions(-) diff --git a/main/xiaozhi-server/core/providers/asr/qwen3_asr_flash.py b/main/xiaozhi-server/core/providers/asr/qwen3_asr_flash.py index 84fe1979..7101448d 100644 --- a/main/xiaozhi-server/core/providers/asr/qwen3_asr_flash.py +++ b/main/xiaozhi-server/core/providers/asr/qwen3_asr_flash.py @@ -1,8 +1,5 @@ import os -import json -import asyncio import tempfile -import difflib from typing import Optional, Tuple, List import dashscope from config.logger import setup_logging @@ -130,27 +127,11 @@ class ASRProvider(ASRProviderBase): # 处理流式响应 full_text = "" - last_text = "" # 用于存储上一个文本片段 for chunk in response: try: text = chunk["output"]["choices"][0]["message"].content[0]["text"] - # 标准化文本片段（去除首尾空格） - normalized_text = text.strip() - # 只有当新文本片段与上一个不同时才处理 - if normalized_text != last_text: - # 提取新增的文本部分 - # 通过比较当前文本和上一个文本，找到新增的部分 - if normalized_text.startswith(last_text): - # 如果当前文本以最后一个文本开头，则新增部分是两者的差集 - new_part = normalized_text[len(last_text):] - else: - # 如果不以最后一个文本开头，说明识别结果发生了较大变化，直接使用当前文本 - new_part = normalized_text - - # 将新增部分添加到完整文本中 - full_text += new_part - last_text = normalized_text - # 这里可以实时处理文本片段，例如通过回调函数 + # 更新为最新的完整文本 + full_text = text.strip() except: pass From fbfc408e94427e70075724ae389fc46e7c6e0cd4 Mon Sep 17 00:00:00 2001 From: Sakura-RanChen <1908198662@qq.com> Date: Thu, 11 Dec 2025 17:12:56 +0800 Subject: [PATCH 4/8] =?UTF-8?q?update:=20=E9=95=BF=E6=8C=89=E8=AE=BE?= =?UTF-8?q?=E5=A4=87ASR=E9=80=82=E9=85=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- main/xiaozhi-server/config.yaml | 1 - .../core/handle/reportHandle.py | 1 - .../textHandler/listenMessageHandler.py | 24 +- .../core/providers/asr/aliyun_stream.py | 141 +++++----- .../xiaozhi-server/core/providers/asr/base.py | 27 +- .../core/providers/asr/doubao_stream.py | 66 +++-- .../core/providers/asr/qwen3_asr_flash.py | 3 +- .../core/providers/asr/xunfei_stream.py | 250 +++--------------- .../core/providers/vad/silero.py | 3 +- .../core/utils/opus_encoder_utils.py | 1 - main/xiaozhi-server/core/utils/util.py | 1 - 11 files changed, 195 insertions(+), 323 deletions(-) diff --git a/main/xiaozhi-server/config.yaml b/main/xiaozhi-server/config.yaml index 2c005356..e7b61dcd 100644 --- a/main/xiaozhi-server/config.yaml +++ b/main/xiaozhi-server/config.yaml @@ -471,7 +471,6 @@ ASR: domain: slm # 识别领域，iat:日常用语，medical:医疗，finance:金融等 language: zh_cn # 语言，zh_cn:中文，en_us:英文 accent: mandarin # 方言，mandarin:普通话 - dwa: wpgs # 动态修正，wpgs:实时返回中间结果 # 调整音频处理参数以提高长语音识别质量 output_dir: tmp/ diff --git a/main/xiaozhi-server/core/handle/reportHandle.py b/main/xiaozhi-server/core/handle/reportHandle.py index 973ebb38..053e8f2e 100644 --- a/main/xiaozhi-server/core/handle/reportHandle.py +++ b/main/xiaozhi-server/core/handle/reportHandle.py @@ -10,7 +10,6 @@ TTS上报功能已集成到ConnectionHandler类中。 """ import time -import gc import opuslib_next from config.manage_api_client import report as manage_report diff --git a/main/xiaozhi-server/core/handle/textHandler/listenMessageHandler.py b/main/xiaozhi-server/core/handle/textHandler/listenMessageHandler.py index 65b442ad..a2c96836 100644 --- a/main/xiaozhi-server/core/handle/textHandler/listenMessageHandler.py +++ b/main/xiaozhi-server/core/handle/textHandler/listenMessageHandler.py @@ -1,12 +1,14 @@ import time +import asyncio from typing import Dict, Any -from core.handle.receiveAudioHandle import handleAudioMessage, startToChat +from core.handle.receiveAudioHandle import startToChat from core.handle.reportHandle import enqueue_asr_report from core.handle.sendAudioHandle import send_stt_message, send_tts_message from core.handle.textMessageHandler import TextMessageHandler from core.handle.textMessageType import TextMessageType from core.utils.util import remove_punctuation_and_length +from core.providers.asr.dto.dto import InterfaceType TAG = __name__ @@ -29,14 +31,18 @@ class ListenTextMessageHandler(TextMessageHandler): elif msg_json["state"] == "stop": conn.client_have_voice = True conn.client_voice_stop = True - if len(conn.asr_audio) > 0: - # 手动模式下直接触发ASR识别，不需要再调用handleAudioMessage - asr_audio_task = conn.asr_audio.copy() - conn.asr_audio.clear() - conn.reset_vad_states() - - if len(asr_audio_task) > 0: - await conn.asr.handle_voice_stop(conn, asr_audio_task) + if conn.asr.interface_type == InterfaceType.STREAM: + # 流式模式下，发送结束请求 + asyncio.create_task(conn.asr._send_stop_request()) + else: + # 非流式模式：直接触发ASR识别 + if len(conn.asr_audio) > 0: + asr_audio_task = conn.asr_audio.copy() + conn.asr_audio.clear() + conn.reset_vad_states() + + if len(asr_audio_task) > 0: + await conn.asr.handle_voice_stop(conn, asr_audio_task) elif msg_json["state"] == "detect": conn.client_have_voice = False conn.asr_audio.clear() diff --git a/main/xiaozhi-server/core/providers/asr/aliyun_stream.py b/main/xiaozhi-server/core/providers/asr/aliyun_stream.py index 179685bd..4ce588a7 100644 --- a/main/xiaozhi-server/core/providers/asr/aliyun_stream.py +++ b/main/xiaozhi-server/core/providers/asr/aliyun_stream.py @@ -5,12 +5,9 @@ import hmac import base64 import hashlib import asyncio -import gc import requests import websockets import opuslib_next -import random -from typing import Optional, Tuple, List from urllib import parse from datetime import datetime from config.logger import setup_logging @@ -140,13 +137,13 @@ class ASRProvider(ASRProviderBase): conn.asr_audio.append(audio) conn.asr_audio = conn.asr_audio[-10:] - # 只在有声音且没有连接时建立连接 - if audio_have_voice and not self.is_processing: + # 只在有声音且没有连接时建立连接（排除正在停止的情况） + if audio_have_voice and not self.is_processing and not self.asr_ws: try: await self._start_recognition(conn) except Exception as e: logger.bind(tag=TAG).error(f"开始识别失败: {str(e)}") - await self._cleanup(conn) + await self._cleanup() return if self.asr_ws and self.is_processing and self.server_ready: @@ -186,10 +183,8 @@ class ASRProvider(ASRProviderBase): "header": { "namespace": "SpeechTranscriber", "name": "StartTranscription", - "status": 20000000, "message_id": uuid.uuid4().hex, "task_id": self.task_id, - "status_text": "Gateway:SUCCESS:Success.", "appkey": self.appkey }, "payload": { @@ -208,18 +203,21 @@ class ASRProvider(ASRProviderBase): async def _forward_results(self, conn): """转发识别结果""" try: - while self.asr_ws and not conn.stop_event.is_set(): + while not conn.stop_event.is_set(): try: response = await asyncio.wait_for(self.asr_ws.recv(), timeout=1.0) result = json.loads(response) - + header = result.get("header", {}) payload = result.get("payload", {}) message_name = header.get("name", "") status = header.get("status", 0) - + if status != 20000000: - if status in [40000004, 40010004]: # 连接超时或客户端断开 + if status == 40010004: + logger.bind(tag=TAG).warning(f"请在服务端响应完成后再关闭链接，状态码: {status}") + break + if status in [40000004, 40010003]: # 连接超时或客户端断开 logger.bind(tag=TAG).warning(f"连接问题，状态码: {status}") break elif status in [40270002, 40270003]: # 音频问题 @@ -228,12 +226,12 @@ class ASRProvider(ASRProviderBase): else: logger.bind(tag=TAG).error(f"识别错误，状态码: {status}, 消息: {header.get('status_text', '')}") continue - + # 收到TranscriptionStarted表示服务器准备好接收音频数据 if message_name == "TranscriptionStarted": self.server_ready = True logger.bind(tag=TAG).debug("服务器已准备，开始发送缓存音频...") - + # 发送缓存音频 if conn.asr_audio: for cached_audio in conn.asr_audio[-10:]: @@ -244,89 +242,89 @@ class ASRProvider(ASRProviderBase): logger.bind(tag=TAG).warning(f"发送缓存音频失败: {e}") break continue - - if message_name == "TranscriptionResultChanged": - # 中间结果 - text = payload.get("result", "") - if text: - self.text = text elif message_name == "SentenceEnd": - # 最终结果 + # 句子结束（每个句子都会触发） text = payload.get("result", "") if text: - self.text = text - conn.reset_vad_states() - # 传递缓存的音频数据 - audio_data = getattr(conn, 'asr_audio_for_voiceprint', []) - await self.handle_voice_stop(conn, audio_data) - # 清空缓存 - conn.asr_audio_for_voiceprint = [] - break - elif message_name == "TranscriptionCompleted": - # 识别完成 - self.is_processing = False - break - + logger.bind(tag=TAG).info(f"识别到文本: {text}") + + # 手动模式下累积识别结果 + if conn.client_listen_mode == "manual": + if self.text: + self.text += text + else: + self.text = text + + # 手动模式下，只有在收到stop信号后才触发处理（仅处理一次） + if conn.client_voice_stop: + audio_data = getattr(conn, 'asr_audio_for_voiceprint', []) + if len(audio_data) > 0: + logger.bind(tag=TAG).debug("收到最终识别结果，触发处理") + await self.handle_voice_stop(conn, audio_data) + # 清理音频缓存 + conn.asr_audio.clear() + conn.reset_vad_states() + break + else: + # 自动模式下直接覆盖 + self.text = text + conn.reset_vad_states() + audio_data = getattr(conn, 'asr_audio_for_voiceprint', []) + await self.handle_voice_stop(conn, audio_data) + break + except asyncio.TimeoutError: - continue - except websockets.exceptions.ConnectionClosed: + logger.bind(tag=TAG).error("接收结果超时") + break + except websockets.ConnectionClosed: + logger.bind(tag=TAG).info("ASR服务连接已关闭") + self.is_processing = False break except Exception as e: logger.bind(tag=TAG).error(f"处理结果失败: {str(e)}") break - + except Exception as e: logger.bind(tag=TAG).error(f"结果转发失败: {str(e)}") finally: - await self._cleanup(conn) + # 清理连接的音频缓存 + await self._cleanup() + if conn: + if hasattr(conn, 'asr_audio_for_voiceprint'): + conn.asr_audio_for_voiceprint = [] + if hasattr(conn, 'asr_audio'): + conn.asr_audio = [] - async def _cleanup(self, conn): - """清理资源""" - logger.bind(tag=TAG).debug(f"开始ASR会话清理 | 当前状态: processing={self.is_processing}, server_ready={self.server_ready}") - - # 清理连接的音频缓存 - if conn and hasattr(conn, 'asr_audio_for_voiceprint'): - conn.asr_audio_for_voiceprint = [] - - # 判断是否需要发送终止请求 - should_stop = self.is_processing or self.server_ready - - # 发送停止识别请求 - if self.asr_ws and should_stop: + async def _send_stop_request(self): + """发送停止识别请求（不关闭连接）""" + if self.asr_ws: try: + # 先停止音频发送 + self.is_processing = False + stop_msg = { "header": { "namespace": "SpeechTranscriber", "name": "StopTranscription", - "status": 20000000, "message_id": uuid.uuid4().hex, "task_id": self.task_id, - "status_text": "Client:Stop", "appkey": self.appkey } } - logger.bind(tag=TAG).debug("正在发送ASR终止请求") + logger.bind(tag=TAG).debug("停止识别请求已发送") await self.asr_ws.send(json.dumps(stop_msg, ensure_ascii=False)) - await asyncio.sleep(0.1) - logger.bind(tag=TAG).debug("ASR终止请求已发送") except Exception as e: - logger.bind(tag=TAG).error(f"ASR终止请求发送失败: {e}") - - # 状态重置（在终止请求发送后） + logger.bind(tag=TAG).error(f"发送停止识别请求失败: {e}") + + async def _cleanup(self): + """清理资源（关闭连接）""" + logger.bind(tag=TAG).debug(f"开始ASR会话清理 | 当前状态: processing={self.is_processing}, server_ready={self.server_ready}") + + # 状态重置 self.is_processing = False self.server_ready = False logger.bind(tag=TAG).debug("ASR状态已重置") - # 清理任务 - if self.forward_task and not self.forward_task.done(): - self.forward_task.cancel() - try: - await asyncio.wait_for(self.forward_task, timeout=1.0) - except Exception as e: - logger.bind(tag=TAG).debug(f"forward_task取消异常: {e}") - finally: - self.forward_task = None - # 关闭连接 if self.asr_ws: try: @@ -337,7 +335,10 @@ class ASRProvider(ASRProviderBase): logger.bind(tag=TAG).error(f"关闭WebSocket连接失败: {e}") finally: self.asr_ws = None - + + # 清理任务引用 + self.forward_task = None + logger.bind(tag=TAG).debug("ASR会话清理完成") async def speech_to_text(self, opus_data, session_id, audio_format): diff --git a/main/xiaozhi-server/core/providers/asr/base.py b/main/xiaozhi-server/core/providers/asr/base.py index 6c3c2d25..671484e9 100644 --- a/main/xiaozhi-server/core/providers/asr/base.py +++ b/main/xiaozhi-server/core/providers/asr/base.py @@ -10,7 +10,6 @@ import traceback import threading import opuslib_next import concurrent.futures -import gc from abc import ABC, abstractmethod from config.logger import setup_logging from typing import Optional, Tuple, List @@ -54,30 +53,26 @@ class ASRProviderBase(ABC): # 接收音频 async def receive_audio(self, conn, audio, audio_have_voice): - if conn.client_listen_mode == "auto" or conn.client_listen_mode == "realtime": + if conn.client_listen_mode == "manual": + # 手动模式：缓存音频用于ASR识别 + conn.asr_audio.append(audio) + else: + # 自动/实时模式：使用VAD检测 have_voice = audio_have_voice - + conn.asr_audio.append(audio) if not have_voice and not conn.client_have_voice: conn.asr_audio = conn.asr_audio[-10:] return - else: - # 手动模式：总是缓存音频，忽略VAD检测结果 - conn.asr_audio.append(audio) - if conn.client_voice_stop: - asr_audio_task = conn.asr_audio.copy() - conn.asr_audio.clear() - conn.reset_vad_states() + # 自动模式下通过VAD检测到语音停止时触发识别 + if conn.client_voice_stop: + asr_audio_task = conn.asr_audio.copy() + conn.asr_audio.clear() + conn.reset_vad_states() - # 手动模式下允许短语音识别，自动模式保持原有限制 - if conn.client_listen_mode == "auto" or conn.client_listen_mode == "realtime": if len(asr_audio_task) > 15: await self.handle_voice_stop(conn, asr_audio_task) - else: - # 手动模式：只要有音频就进行识别 - if len(asr_audio_task) > 0: - await self.handle_voice_stop(conn, asr_audio_task) # 处理语音停止 async def handle_voice_stop(self, conn, asr_audio_task: List[bytes]): diff --git a/main/xiaozhi-server/core/providers/asr/doubao_stream.py b/main/xiaozhi-server/core/providers/asr/doubao_stream.py index 2d179b86..e141e25d 100644 --- a/main/xiaozhi-server/core/providers/asr/doubao_stream.py +++ b/main/xiaozhi-server/core/providers/asr/doubao_stream.py @@ -4,7 +4,6 @@ import uuid import asyncio import websockets import opuslib_next -import gc from core.providers.asr.base import ASRProviderBase from config.logger import setup_logging from core.providers.asr.dto.dto import InterfaceType @@ -19,8 +18,6 @@ class ASRProvider(ASRProviderBase): self.interface_type = InterfaceType.STREAM self.config = config self.text = "" - self.max_retries = 3 - self.retry_delay = 2 self.decoder = opuslib_next.Decoder(16000, 1) self.asr_ws = None self.forward_task = None @@ -57,14 +54,13 @@ class ASRProvider(ASRProviderBase): async def receive_audio(self, conn, audio, audio_have_voice): conn.asr_audio.append(audio) conn.asr_audio = conn.asr_audio[-10:] - # 存储音频数据 if not hasattr(conn, 'asr_audio_for_voiceprint'): conn.asr_audio_for_voiceprint = [] conn.asr_audio_for_voiceprint.append(audio) - + # 当没有音频数据时处理完整语音片段 - if not audio and len(conn.asr_audio_for_voiceprint) > 0: + if conn.client_listen_mode != "manual" and not audio and len(conn.asr_audio_for_voiceprint) > 0: await self.handle_voice_stop(conn, conn.asr_audio_for_voiceprint) conn.asr_audio_for_voiceprint = [] @@ -180,6 +176,7 @@ class ASRProvider(ASRProviderBase): payload.get("audio_info", {}).get("duration", 0) > 2000 and not utterances and not payload["result"].get("text") + and conn.client_listen_mode != "manual" ): logger.bind(tag=TAG).error(f"识别文本：空") self.text = "" @@ -188,15 +185,44 @@ class ASRProvider(ASRProviderBase): await self.handle_voice_stop(conn, audio_data) break + # 专门处理没有文本的识别结果（手动模式下可能已经识别完成但是没松按键） + elif not payload["result"].get("text") and not utterances: + if conn.client_voice_stop and len(audio_data) > 0: + logger.bind(tag=TAG).debug("消息结束收到停止信号，触发处理") + await self.handle_voice_stop(conn, audio_data) + # 清理音频缓存 + conn.asr_audio.clear() + conn.reset_vad_states() + break + for utterance in utterances: if utterance.get("definite", False): - self.text = utterance["text"] + current_text = utterance["text"] logger.bind(tag=TAG).info( - f"识别到文本: {self.text}" + f"识别到文本: {current_text}" ) - conn.reset_vad_states() - if len(audio_data) > 15: # 确保有足够音频数据 - await self.handle_voice_stop(conn, audio_data) + + # 手动模式下累积识别结果 + if conn.client_listen_mode == "manual": + if self.text: + self.text += current_text + else: + self.text = current_text + + # 在接收消息中途时收到停止信号 + if conn.client_voice_stop and len(audio_data) > 0: + logger.bind(tag=TAG).debug("消息中途收到停止信号，触发处理") + await self.handle_voice_stop(conn, audio_data) + # 清理音频缓存 + conn.asr_audio.clear() + conn.reset_vad_states() + break + else: + # 自动模式下直接覆盖 + self.text = current_text + conn.reset_vad_states() + if len(audio_data) > 15: # 确保有足够音频数据 + await self.handle_voice_stop(conn, audio_data) break elif "error" in payload: error_msg = payload.get("error", "未知错误") @@ -228,8 +254,6 @@ class ASRProvider(ASRProviderBase): conn.asr_audio_for_voiceprint = [] if hasattr(conn, 'asr_audio'): conn.asr_audio = [] - if hasattr(conn, 'has_valid_voice'): - conn.has_valid_voice = False def stop_ws_connection(self): if self.asr_ws: @@ -237,6 +261,20 @@ class ASRProvider(ASRProviderBase): self.asr_ws = None self.is_processing = False + async def _send_stop_request(self): + """发送最后一个音频帧以通知服务器结束""" + if self.asr_ws: + try: + # 发送结束标记的音频帧（gzip压缩的空数据） + empty_payload = gzip.compress(b"") + last_audio_request = bytearray(self.generate_last_audio_default_header()) + last_audio_request.extend(len(empty_payload).to_bytes(4, "big")) + last_audio_request.extend(empty_payload) + await self.asr_ws.send(last_audio_request) + logger.bind(tag=TAG).debug("已发送结束音频帧") + except Exception as e: + logger.bind(tag=TAG).debug(f"发送结束音频帧时出错: {e}") + def construct_request(self, reqid): req = { "app": { @@ -388,5 +426,3 @@ class ASRProvider(ASRProviderBase): conn.asr_audio_for_voiceprint = [] if hasattr(conn, 'asr_audio'): conn.asr_audio = [] - if hasattr(conn, 'has_valid_voice'): - conn.has_valid_voice = False diff --git a/main/xiaozhi-server/core/providers/asr/qwen3_asr_flash.py b/main/xiaozhi-server/core/providers/asr/qwen3_asr_flash.py index 7101448d..51c84a37 100644 --- a/main/xiaozhi-server/core/providers/asr/qwen3_asr_flash.py +++ b/main/xiaozhi-server/core/providers/asr/qwen3_asr_flash.py @@ -13,7 +13,8 @@ logger = setup_logging() class ASRProvider(ASRProviderBase): def __init__(self, config: dict, delete_audio_file: bool): super().__init__() - self.interface_type = InterfaceType.STREAM + # 音频文件上传类型，流式文本识别输出 + self.interface_type = InterfaceType.NON_STREAM """Qwen3-ASR-Flash ASR初始化""" # 配置参数 diff --git a/main/xiaozhi-server/core/providers/asr/xunfei_stream.py b/main/xiaozhi-server/core/providers/asr/xunfei_stream.py index 6a895ebc..b7e7886e 100644 --- a/main/xiaozhi-server/core/providers/asr/xunfei_stream.py +++ b/main/xiaozhi-server/core/providers/asr/xunfei_stream.py @@ -35,9 +35,6 @@ class ASRProvider(ASRProviderBase): self.forward_task = None self.is_processing = False self.server_ready = False - self.last_frame_sent = False # 标记是否已发送最终帧 - self.best_text = "" # 保存最佳识别结果 - self.has_final_result = False # 标记是否收到最终识别结果 # 讯飞配置 self.app_id = config.get("app_id") @@ -52,7 +49,6 @@ class ASRProvider(ASRProviderBase): "domain": config.get("domain", "slm"), "language": config.get("language", "zh_cn"), "accent": config.get("accent", "mandarin"), - "dwa": config.get("dwa", "wpgs"), "result": {"encoding": "utf8", "compress": "raw", "format": "plain"}, } @@ -116,7 +112,7 @@ class ASRProvider(ASRProviderBase): await self._start_recognition(conn) except Exception as e: logger.bind(tag=TAG).error(f"建立ASR连接失败: {str(e)}") - await self._cleanup(conn) + await self._cleanup() return # 发送当前音频数据 @@ -126,7 +122,7 @@ class ASRProvider(ASRProviderBase): await self._send_audio_frame(pcm_frame, STATUS_CONTINUE_FRAME) except Exception as e: logger.bind(tag=TAG).warning(f"发送音频数据时发生错误: {e}") - await self._cleanup(conn) + await self._cleanup() async def _start_recognition(self, conn): """开始识别会话""" @@ -136,6 +132,10 @@ class ASRProvider(ASRProviderBase): ws_url = self.create_url() logger.bind(tag=TAG).info(f"正在连接ASR服务: {ws_url[:50]}...") + # 如果为手动模式,设置超时时长为一分钟 + if conn.client_listen_mode == "manual": + self.iat_params["eos"] = 60000 + self.asr_ws = await websockets.connect( ws_url, max_size=1000000000, @@ -146,8 +146,6 @@ class ASRProvider(ASRProviderBase): logger.bind(tag=TAG).info("ASR WebSocket连接已建立") self.server_ready = False - self.last_frame_sent = False - self.best_text = "" self.forward_task = asyncio.create_task(self._forward_results(conn)) # 发送首帧音频 @@ -196,23 +194,12 @@ class ASRProvider(ASRProviderBase): await self.asr_ws.send(json.dumps(frame_data, ensure_ascii=False)) - # 标记是否发送了最终帧 - if status == STATUS_LAST_FRAME: - self.last_frame_sent = True - logger.bind(tag=TAG).info("标记最终帧已发送") - async def _forward_results(self, conn): """转发识别结果""" try: - while self.asr_ws and not conn.stop_event.is_set(): - # 获取当前连接的音频数据 - audio_data = getattr(conn, "asr_audio_for_voiceprint", []) + while not conn.stop_event.is_set(): try: - # 如果已发送最终帧，增加超时时间等待完整结果 - timeout = 3.0 if self.last_frame_sent else 30.0 - response = await asyncio.wait_for( - self.asr_ws.recv(), timeout=timeout - ) + response = await asyncio.wait_for(self.asr_ws.recv(), timeout=60) result = json.loads(response) logger.bind(tag=TAG).debug(f"收到ASR结果: {result}") @@ -236,144 +223,27 @@ class ASRProvider(ASRProviderBase): # 解码base64文本 decoded_text = base64.b64decode(text_data).decode("utf-8") text_json = json.loads(decoded_text) - # 提取文本内容 text_ws = text_json.get("ws", []) - result_text = "" for i in text_ws: for j in i.get("cw", []): w = j.get("w", "") - result_text += w + self.text += w - # 更新识别文本 - 实时更新策略 - # 只检查是否为空字符串，不再过滤任何标点符号 - # 这样可以确保所有识别到的内容，包括标点符号都能被实时更新 - if result_text and result_text.strip(): - # 实时更新：正常情况下都更新，提高响应速度 - should_update = True - - # 保存最佳文本 - # 1. 如果是识别完成状态或最终帧后收到的结果，优先保存 - # 2. 否则保存最长的有意义文本 - # 取消对标点符号的过滤，只检查是否为空 - # 这样可以保留所有识别到的内容，包括各种标点符号 - is_valid_text = len(result_text.strip()) > 0 - - if ( - self.last_frame_sent or status == 2 - ) and is_valid_text: - self.best_text = result_text - self.has_final_result = True # 标记已收到最终结果 - logger.bind(tag=TAG).debug( - f"保存最终识别结果: {self.best_text}" - ) - elif ( - len(result_text) > len(self.best_text) - and is_valid_text - and not self.has_final_result - ): - self.best_text = result_text - logger.bind(tag=TAG).debug( - f"保存中间最佳文本: {self.best_text}" - ) - - # 如果已发送最终帧，只过滤空文本 - if self.last_frame_sent: - # 只拒绝完全空的结果 - if not result_text.strip(): - should_update = False - logger.bind(tag=TAG).warning( - f"最终帧后拒绝空文本" - ) - - if should_update: - # 处理流式识别结果，避免简单替换导致内容丢失 - # 1. 如果是中间状态（非最终帧后），可能需要替换为更完整的识别 - # 2. 如果是最终帧后收到的结果，可能是对前面文本的补充 - if self.last_frame_sent: - # 最终帧后收到的结果可能是标点符号等补充内容 - # 检查是否需要合并文本而不是替换 - # 如果当前文本是纯标点而前面已有内容，应该追加而不是替换 - if len( - self.text - ) > 0 and result_text.strip() in [ - "。", - ".", - "?", - "？", - "!", - "！", - ",", - "，", - ";", - "；", - ]: - # 对于标点符号，追加到现有文本后 - self.text = ( - self.text.rstrip().rstrip("。.") - + result_text - ) - else: - # 其他情况保持替换逻辑 - self.text = result_text - else: - # 中间状态替换为新的识别结果 - self.text = result_text - - logger.bind(tag=TAG).info( - f"实时更新识别文本: {self.text} (最终帧已发送: {self.last_frame_sent})" - ) - - # 识别完成，但如果还没发送最终帧，继续等待 if status == 2: - logger.bind(tag=TAG).info( - f"识别完成状态已到达，当前识别文本: {self.text}" - ) - - # 如果还没发送最终帧，继续等待 - if not self.last_frame_sent: - logger.bind(tag=TAG).info( - "识别完成但最终帧未发送，继续等待..." - ) - continue - - # 已发送最终帧且收到完成状态，使用最佳策略选择最终结果 - # 优先使用识别完成状态下的最新结果，而不是仅仅基于长度 - if self.best_text: - # 如果当前文本是在最终帧发送后或识别完成状态下收到的，优先使用 - if ( - self.last_frame_sent or status == 2 - ) and self.text.strip(): - logger.bind(tag=TAG).info( - f"使用完成状态下的最新识别结果: {self.text}" - ) - elif len(self.best_text) > len(self.text): - logger.bind(tag=TAG).info( - f"使用更长的最佳文本作为最终结果: {self.text} -> {self.best_text}" - ) - self.text = self.best_text - - logger.bind(tag=TAG).info(f"获取到最终完整文本: {self.text}") + if conn.client_listen_mode == "manual": + audio_data = getattr(conn, 'asr_audio_for_voiceprint', []) + if len(audio_data) > 0: + logger.bind(tag=TAG).debug("收到最终识别结果，触发处理") + await self.handle_voice_stop(conn, audio_data) + # 清理音频缓存 + conn.asr_audio.clear() conn.reset_vad_states() - if len(audio_data) > 15: # 确保有足够音频数据 - # 准备处理结果 - pass break except asyncio.TimeoutError: - if self.last_frame_sent: - # 超时时也使用最佳文本 - if self.best_text and len(self.best_text) > len(self.text): - logger.bind(tag=TAG).info( - f"超时，使用最佳文本: {self.text} -> {self.best_text}" - ) - self.text = self.best_text - logger.bind(tag=TAG).info( - f"最终帧后超时，使用结果: {self.text}" - ) - break - # 如果还没发送最终帧，继续等待 - continue + logger.bind(tag=TAG).error("接收结果超时") + break except websockets.ConnectionClosed: logger.bind(tag=TAG).info("ASR服务连接已关闭") self.is_processing = False @@ -390,17 +260,15 @@ class ASRProvider(ASRProviderBase): if hasattr(e, "__cause__") and e.__cause__: logger.bind(tag=TAG).error(f"错误原因: {str(e.__cause__)}") finally: - if self.asr_ws: - await self.asr_ws.close() - self.asr_ws = None - self.is_processing = False + # 清理连接资源 + await self._cleanup() + + # 清理连接的音频缓存 if conn: if hasattr(conn, "asr_audio_for_voiceprint"): conn.asr_audio_for_voiceprint = [] if hasattr(conn, "asr_audio"): conn.asr_audio = [] - if hasattr(conn, "has_valid_voice"): - conn.has_valid_voice = False async def handle_voice_stop(self, conn, asr_audio_task: List[bytes]): """处理语音停止，发送最后一帧并处理识别结果""" @@ -408,22 +276,13 @@ class ASRProvider(ASRProviderBase): # 先发送最后一帧表示音频结束 if self.asr_ws and self.is_processing: try: - # 取最后一个有效的音频帧作为最后一帧数据 - last_frame = b"" - if asr_audio_task: - last_audio = asr_audio_task[-1] - last_frame = self.decoder.decode(last_audio, 960) - await self._send_audio_frame(last_frame, STATUS_LAST_FRAME) - logger.bind(tag=TAG).info("已发送最后一帧") + await self._send_audio_frame(b"", STATUS_LAST_FRAME) + logger.bind(tag=TAG).debug(f"已发送停止请求") - # 发送最终帧后，给_forward_results适当时间处理最终结果 await asyncio.sleep(0.25) - - logger.bind(tag=TAG).info(f"准备处理最终识别结果: {self.text}") except Exception as e: - logger.bind(tag=TAG).error(f"发送最后一帧失败: {e}") + logger.bind(tag=TAG).error(f"发送停止请求失败: {e}") - # 调用父类的handle_voice_stop方法处理识别结果 await super().handle_voice_stop(conn, asr_audio_task) except Exception as e: logger.bind(tag=TAG).error(f"处理语音停止失败: {e}") @@ -437,40 +296,27 @@ class ASRProvider(ASRProviderBase): self.asr_ws = None self.is_processing = False - async def _cleanup(self, conn): - """清理资源""" - logger.bind(tag=TAG).info( + async def _send_stop_request(self): + """发送停止识别请求（不关闭连接）""" + if self.asr_ws: + try: + # 先停止音频发送 + self.is_processing = False + await self._send_audio_frame(b"", STATUS_LAST_FRAME) + logger.bind(tag=TAG).debug("已发送停止请求") + except Exception as e: + logger.bind(tag=TAG).error(f"发送停止请求失败: {e}") + + async def _cleanup(self): + """清理资源（关闭连接）""" + logger.bind(tag=TAG).debug( f"开始ASR会话清理 | 当前状态: processing={self.is_processing}, server_ready={self.server_ready}" ) - # 发送最后一帧 - if self.asr_ws and self.is_processing: - try: - await self._send_audio_frame(b"", STATUS_LAST_FRAME) - await asyncio.sleep(0.1) - logger.bind(tag=TAG).info("已发送最后一帧") - except Exception as e: - logger.bind(tag=TAG).error(f"发送最后一帧失败: {e}") - # 状态重置 self.is_processing = False self.server_ready = False - self.last_frame_sent = False - self.best_text = "" - self.has_final_result = False - logger.bind(tag=TAG).info("ASR状态已重置") - - # 清理任务 - if self.forward_task and not self.forward_task.done(): - self.forward_task.cancel() - try: - await asyncio.wait_for(self.forward_task, timeout=1.0) - except asyncio.CancelledError: - pass - except Exception as e: - logger.bind(tag=TAG).debug(f"forward_task取消异常: {e}") - finally: - self.forward_task = None + logger.bind(tag=TAG).debug("ASR状态已重置") # 关闭连接 if self.asr_ws: @@ -483,16 +329,10 @@ class ASRProvider(ASRProviderBase): finally: self.asr_ws = None - # 清理连接的音频缓存 - if conn: - if hasattr(conn, "asr_audio_for_voiceprint"): - conn.asr_audio_for_voiceprint = [] - if hasattr(conn, "asr_audio"): - conn.asr_audio = [] - if hasattr(conn, "has_valid_voice"): - conn.has_valid_voice = False + # 清理任务引用 + self.forward_task = None - logger.bind(tag=TAG).info("ASR会话清理完成") + logger.bind(tag=TAG).debug("ASR会话清理完成") async def speech_to_text(self, opus_data, session_id, audio_format): """获取识别结果""" @@ -513,7 +353,7 @@ class ASRProvider(ASRProviderBase): pass self.forward_task = None self.is_processing = False - + # 显式释放decoder资源 if hasattr(self, 'decoder') and self.decoder is not None: try: @@ -530,5 +370,3 @@ class ASRProvider(ASRProviderBase): conn.asr_audio_for_voiceprint = [] if hasattr(conn, "asr_audio"): conn.asr_audio = [] - if hasattr(conn, "has_valid_voice"): - conn.has_valid_voice = False diff --git a/main/xiaozhi-server/core/providers/vad/silero.py b/main/xiaozhi-server/core/providers/vad/silero.py index 28a3e61e..81215681 100644 --- a/main/xiaozhi-server/core/providers/vad/silero.py +++ b/main/xiaozhi-server/core/providers/vad/silero.py @@ -2,7 +2,6 @@ import time import numpy as np import torch import opuslib_next -import gc from config.logger import setup_logging from core.providers.vad.base import VADProviderBase @@ -46,7 +45,7 @@ class VADProvider(VADProviderBase): def is_vad(self, conn, opus_packet): # 手动模式：直接返回True，不进行实时VAD检测，所有音频都缓存 - if conn.client_listen_mode not in ["auto", "realtime"]: + if conn.client_listen_mode == "manual": return True try: diff --git a/main/xiaozhi-server/core/utils/opus_encoder_utils.py b/main/xiaozhi-server/core/utils/opus_encoder_utils.py index f2fcc266..8d603e22 100644 --- a/main/xiaozhi-server/core/utils/opus_encoder_utils.py +++ b/main/xiaozhi-server/core/utils/opus_encoder_utils.py @@ -6,7 +6,6 @@ Opus编码工具类 import logging import traceback import numpy as np -import gc from opuslib_next import Encoder from opuslib_next import constants from typing import Optional, Callable, Any diff --git a/main/xiaozhi-server/core/utils/util.py b/main/xiaozhi-server/core/utils/util.py index 37388be8..f72dda2c 100644 --- a/main/xiaozhi-server/core/utils/util.py +++ b/main/xiaozhi-server/core/utils/util.py @@ -8,7 +8,6 @@ import requests import subprocess import numpy as np import opuslib_next -import gc from io import BytesIO from core.utils import p3 from pydub import AudioSegment From f88abd7638acb6724531fe4b9f09f4afc53194aa Mon Sep 17 00:00:00 2001 From: Sakura-RanChen <1908198662@qq.com> Date: Fri, 12 Dec 2025 09:46:15 +0800 Subject: [PATCH 5/8] =?UTF-8?q?fix:=20=E8=81=86=E5=90=AC=E8=AE=BE=E5=A4=87?= =?UTF-8?q?=E8=AF=AF=E8=A7=A6=E5=8F=91?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- main/xiaozhi-server/core/providers/asr/doubao_stream.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/main/xiaozhi-server/core/providers/asr/doubao_stream.py b/main/xiaozhi-server/core/providers/asr/doubao_stream.py index e141e25d..0c1cf727 100644 --- a/main/xiaozhi-server/core/providers/asr/doubao_stream.py +++ b/main/xiaozhi-server/core/providers/asr/doubao_stream.py @@ -187,7 +187,7 @@ class ASRProvider(ASRProviderBase): # 专门处理没有文本的识别结果（手动模式下可能已经识别完成但是没松按键） elif not payload["result"].get("text") and not utterances: - if conn.client_voice_stop and len(audio_data) > 0: + if conn.client_listen_mode == "manual" and conn.client_voice_stop and len(audio_data) > 0: logger.bind(tag=TAG).debug("消息结束收到停止信号，触发处理") await self.handle_voice_stop(conn, audio_data) # 清理音频缓存 From c8d2d2255db5ceb16c8256f16d00d62c41926a09 Mon Sep 17 00:00:00 2001 From: Sakura-RanChen <1908198662@qq.com> Date: Fri, 12 Dec 2025 15:08:40 +0800 Subject: [PATCH 6/8] =?UTF-8?q?fix:=20=E9=9F=B3=E9=A2=91=E5=BD=B1=E5=93=8D?= =?UTF-8?q?=E7=BA=BF=E7=A8=8B=E9=97=AE=E9=A2=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../xiaozhi-server/core/providers/asr/base.py | 102 ++++++------------ 1 file changed, 33 insertions(+), 69 deletions(-) diff --git a/main/xiaozhi-server/core/providers/asr/base.py b/main/xiaozhi-server/core/providers/asr/base.py index 671484e9..6c6969fd 100644 --- a/main/xiaozhi-server/core/providers/asr/base.py +++ b/main/xiaozhi-server/core/providers/asr/base.py @@ -9,7 +9,6 @@ import asyncio import traceback import threading import opuslib_next -import concurrent.futures from abc import ABC, abstractmethod from config.logger import setup_logging from typing import Optional, Tuple, List @@ -79,98 +78,63 @@ class ASRProviderBase(ABC): """并行处理ASR和声纹识别""" try: total_start_time = time.monotonic() - + # 准备音频数据 if conn.audio_format == "pcm": pcm_data = asr_audio_task else: pcm_data = self.decode_opus(asr_audio_task) - + combined_pcm_data = b"".join(pcm_data) - + # 预先准备WAV数据 wav_data = None if conn.voiceprint_provider and combined_pcm_data: wav_data = self._pcm_to_wav(combined_pcm_data) - + # 定义ASR任务 - def run_asr(): - start_time = time.monotonic() - try: - loop = asyncio.new_event_loop() - asyncio.set_event_loop(loop) - try: - result = loop.run_until_complete( - self.speech_to_text(asr_audio_task, conn.session_id, conn.audio_format) - ) - end_time = time.monotonic() - logger.bind(tag=TAG).debug(f"ASR耗时: {end_time - start_time:.3f}s") - return result - finally: - loop.close() - except Exception as e: - end_time = time.monotonic() - logger.bind(tag=TAG).error(f"ASR失败: {e}") - return ("", None) - - # 定义声纹识别任务 - def run_voiceprint(): - if not wav_data: - return None - try: - loop = asyncio.new_event_loop() - asyncio.set_event_loop(loop) - try: - # 使用连接的声纹识别提供者 - result = loop.run_until_complete( - conn.voiceprint_provider.identify_speaker(wav_data, conn.session_id) - ) - return result - finally: - loop.close() - except Exception as e: - logger.bind(tag=TAG).error(f"声纹识别失败: {e}") - return None - - # 使用线程池执行器并行运行 - with concurrent.futures.ThreadPoolExecutor(max_workers=2) as thread_executor: - asr_future = thread_executor.submit(run_asr) - - if conn.voiceprint_provider and wav_data: - voiceprint_future = thread_executor.submit(run_voiceprint) - - # 等待两个线程都完成 - asr_result = asr_future.result(timeout=15) - voiceprint_result = voiceprint_future.result(timeout=15) - - results = {"asr": asr_result, "voiceprint": voiceprint_result} - else: - asr_result = asr_future.result(timeout=15) - results = {"asr": asr_result, "voiceprint": None} - - - # 处理结果 - raw_text, _ = results.get("asr", ("", None)) - speaker_name = results.get("voiceprint", None) - - # 记录识别结果 + asr_task = self.speech_to_text(asr_audio_task, conn.session_id, conn.audio_format) + + if conn.voiceprint_provider and wav_data: + voiceprint_task = conn.voiceprint_provider.identify_speaker(wav_data, conn.session_id) + # 并发等待两个结果 + asr_result, voiceprint_result = await asyncio.gather( + asr_task, voiceprint_task, return_exceptions=True + ) + else: + asr_result = await asr_task + voiceprint_result = None + + # 记录识别结果 - 检查是否为异常 + if isinstance(asr_result, Exception): + logger.bind(tag=TAG).error(f"ASR识别失败: {asr_result}") + raw_text = "" + else: + raw_text, _ = asr_result + + if isinstance(voiceprint_result, Exception): + logger.bind(tag=TAG).error(f"声纹识别失败: {voiceprint_result}") + speaker_name = "" + else: + speaker_name = voiceprint_result + if raw_text: logger.bind(tag=TAG).info(f"识别文本: {raw_text}") if speaker_name: logger.bind(tag=TAG).info(f"识别说话人: {speaker_name}") - + # 性能监控 total_time = time.monotonic() - total_start_time logger.bind(tag=TAG).debug(f"总处理耗时: {total_time:.3f}s") - + # 检查文本长度 text_len, _ = remove_punctuation_and_length(raw_text) self.stop_ws_connection() - + if text_len > 0: # 构建包含说话人信息的JSON字符串 enhanced_text = self._build_enhanced_text(raw_text, speaker_name) - + # 使用自定义模块进行上报 await startToChat(conn, enhanced_text) enqueue_asr_report(conn, enhanced_text, asr_audio_task) From 20f601e6071641aae574da0ea3d08ea1d93881ff Mon Sep 17 00:00:00 2001 From: Sakura-RanChen <1908198662@qq.com> Date: Fri, 12 Dec 2025 15:39:30 +0800 Subject: [PATCH 7/8] =?UTF-8?q?fix:=20=E5=90=8C=E6=AD=A5=E6=96=B9=E6=B3=95?= =?UTF-8?q?=E4=BD=BF=E7=94=A8=E7=BA=BF=E7=A8=8B=E6=B1=A0=E9=81=BF=E5=85=8D?= =?UTF-8?q?=E9=98=BB=E5=A1=9E?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../core/providers/asr/fun_local.py | 17 ++++++++++------- 1 file changed, 10 insertions(+), 7 deletions(-) diff --git a/main/xiaozhi-server/core/providers/asr/fun_local.py b/main/xiaozhi-server/core/providers/asr/fun_local.py index 217f17ff..2fd49c36 100644 --- a/main/xiaozhi-server/core/providers/asr/fun_local.py +++ b/main/xiaozhi-server/core/providers/asr/fun_local.py @@ -1,14 +1,16 @@ -import time import os -import sys import io +import sys +import time +import shutil import psutil +import asyncio + from config.logger import setup_logging from typing import Optional, Tuple, List -from core.providers.asr.base import ASRProviderBase from funasr import AutoModel from funasr.utils.postprocess_utils import rich_transcription_postprocess -import shutil +from core.providers.asr.base import ASRProviderBase from core.providers.asr.dto.dto import InterfaceType TAG = __name__ @@ -90,16 +92,17 @@ class ASRProvider(ASRProviderBase): else: file_path = self.save_audio_to_file(pcm_data, session_id) - # 语音识别 + # 语音识别 - 使用线程池避免阻塞事件循环 start_time = time.time() - result = self.model.generate( + result = await asyncio.to_thread( + self.model.generate, input=combined_pcm_data, cache={}, language="auto", use_itn=True, batch_size_s=60, ) - text = rich_transcription_postprocess(result[0]["text"]) + text = await asyncio.to_thread(rich_transcription_postprocess, result[0]["text"]) logger.bind(tag=TAG).debug( f"语音识别耗时: {time.time() - start_time:.3f}s | 结果: {text}" ) From 5dbf796fa931efe15ec217a7ab170fdda9fde676 Mon Sep 17 00:00:00 2001 From: hrz <1710360675@qq.com> Date: Fri, 12 Dec 2025 16:33:50 +0800 Subject: [PATCH 8/8] =?UTF-8?q?update:=E6=9B=B4=E6=96=B0=E7=89=88=E6=9C=AC?= =?UTF-8?q?=E5=8F=B7?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../src/main/java/xiaozhi/common/constant/Constant.java | 2 +- main/manager-mobile/src/pages/settings/index.vue | 2 +- main/xiaozhi-server/config/logger.py | 2 +- 3 files changed, 3 insertions(+), 3 deletions(-) diff --git a/main/manager-api/src/main/java/xiaozhi/common/constant/Constant.java b/main/manager-api/src/main/java/xiaozhi/common/constant/Constant.java index 1ea19355..917f1e50 100644 --- a/main/manager-api/src/main/java/xiaozhi/common/constant/Constant.java +++ b/main/manager-api/src/main/java/xiaozhi/common/constant/Constant.java @@ -299,7 +299,7 @@ public interface Constant { /** * 版本号 */ - public static final String VERSION = "0.8.9"; + public static final String VERSION = "0.8.10"; /** * 无效固件URL diff --git a/main/manager-mobile/src/pages/settings/index.vue b/main/manager-mobile/src/pages/settings/index.vue index 2f7be6b4..b1fa487b 100644 --- a/main/manager-mobile/src/pages/settings/index.vue +++ b/main/manager-mobile/src/pages/settings/index.vue @@ -235,7 +235,7 @@ function showAbout() { title: t('settings.aboutApp', { appName: import.meta.env.VITE_APP_TITLE }), content: t('settings.aboutContent', { appName: import.meta.env.VITE_APP_TITLE, - version: '0.8.9' + version: '0.8.10' }), showCancel: false, confirmText: t('common.confirm'), diff --git a/main/xiaozhi-server/config/logger.py b/main/xiaozhi-server/config/logger.py index a0b64dd7..bf64893c 100644 --- a/main/xiaozhi-server/config/logger.py +++ b/main/xiaozhi-server/config/logger.py @@ -5,7 +5,7 @@ from config.config_loader import load_config from config.settings import check_config_file from datetime import datetime -SERVER_VERSION = "0.8.9" +SERVER_VERSION = "0.8.10" _logger_initialized = False