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