From a035ed525a3c72de0db71227ee73f334515a6c56 Mon Sep 17 00:00:00 2001 From: hrz <1710360675@qq.com> Date: Thu, 23 Oct 2025 17:04:33 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E5=A4=8D=EF=BC=9A=E8=AE=AF=E9=A3=9Eas?= =?UTF-8?q?r=E6=9C=80=E5=90=8E=E4=B8=80=E4=B8=AA=E5=AD=97=E4=B8=A2?= =?UTF-8?q?=E5=A4=B1=E7=9A=84bug?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../core/providers/asr/xunfei_stream.py | 231 ++++++++++++------ 1 file changed, 156 insertions(+), 75 deletions(-) diff --git a/main/xiaozhi-server/core/providers/asr/xunfei_stream.py b/main/xiaozhi-server/core/providers/asr/xunfei_stream.py index 7f696428..e91c8f09 100644 --- a/main/xiaozhi-server/core/providers/asr/xunfei_stream.py +++ b/main/xiaozhi-server/core/providers/asr/xunfei_stream.py @@ -22,6 +22,7 @@ STATUS_FIRST_FRAME = 0 # 第一帧的标识 STATUS_CONTINUE_FRAME = 1 # 中间帧标识 STATUS_LAST_FRAME = 2 # 最后一帧的标识 + class ASRProvider(ASRProviderBase): def __init__(self, config, delete_audio_file): super().__init__() @@ -35,6 +36,7 @@ class ASRProvider(ASRProviderBase): self.server_ready = False self.last_frame_sent = False # 标记是否已发送最终帧 self.best_text = "" # 保存最佳识别结果 + self.has_final_result = False # 标记是否收到最终识别结果 # 讯飞配置 self.app_id = config.get("app_id") @@ -50,19 +52,15 @@ class ASRProvider(ASRProviderBase): "language": config.get("language", "zh_cn"), "accent": config.get("accent", "mandarin"), "dwa": config.get("dwa", "wpgs"), - "result": { - "encoding": "utf8", - "compress": "raw", - "format": "plain" - } + "result": {"encoding": "utf8", "compress": "raw", "format": "plain"}, } self.output_dir = config.get("output_dir", "tmp/") self.delete_audio_file = delete_audio_file - + def create_url(self) -> str: """生成认证URL""" - url = 'ws://iat.cn-huabei-1.xf-yun.com/v1' + url = "ws://iat.cn-huabei-1.xf-yun.com/v1" # 生成RFC1123格式的时间戳 now = datetime.now() date = format_date_time(mktime(now.timetuple())) @@ -73,23 +71,30 @@ class ASRProvider(ASRProviderBase): signature_origin += "GET " + "/v1 " + "HTTP/1.1" # 进行hmac-sha256进行加密 - signature_sha = hmac.new(self.api_secret.encode('utf-8'), signature_origin.encode('utf-8'), - digestmod=hashlib.sha256).digest() - signature_sha = base64.b64encode(signature_sha).decode(encoding='utf-8') + signature_sha = hmac.new( + self.api_secret.encode("utf-8"), + signature_origin.encode("utf-8"), + digestmod=hashlib.sha256, + ).digest() + signature_sha = base64.b64encode(signature_sha).decode(encoding="utf-8") - authorization_origin = "api_key=\"%s\", algorithm=\"%s\", headers=\"%s\", signature=\"%s\"" % ( - self.api_key, "hmac-sha256", "host date request-line", signature_sha) - authorization = base64.b64encode(authorization_origin.encode('utf-8')).decode(encoding='utf-8') + authorization_origin = ( + 'api_key="%s", algorithm="%s", headers="%s", signature="%s"' + % (self.api_key, "hmac-sha256", "host date request-line", signature_sha) + ) + authorization = base64.b64encode(authorization_origin.encode("utf-8")).decode( + encoding="utf-8" + ) # 将请求的鉴权参数组合为字典 v = { "authorization": authorization, "date": date, - "host": "iat.cn-huabei-1.xf-yun.com" + "host": "iat.cn-huabei-1.xf-yun.com", } # 拼接鉴权参数，生成url - url = url + '?' + urlencode(v) + url = url + "?" + urlencode(v) return url async def open_audio_channels(self, conn): @@ -100,7 +105,7 @@ class ASRProvider(ASRProviderBase): await super().receive_audio(conn, audio, audio_have_voice) # 存储音频数据用于声纹识别 - if not hasattr(conn, 'asr_audio_for_voiceprint'): + if not hasattr(conn, "asr_audio_for_voiceprint"): conn.asr_audio_for_voiceprint = [] conn.asr_audio_for_voiceprint.append(audio) @@ -146,8 +151,10 @@ class ASRProvider(ASRProviderBase): # 发送首帧音频 if conn.asr_audio and len(conn.asr_audio) > 0: - first_audio = conn.asr_audio[-1] if conn.asr_audio else b'' - pcm_frame = self.decoder.decode(first_audio, 960) if first_audio else b'' + first_audio = conn.asr_audio[-1] if conn.asr_audio else b"" + pcm_frame = ( + self.decoder.decode(first_audio, 960) if first_audio else b"" + ) await self._send_audio_frame(pcm_frame, STATUS_FIRST_FRAME) self.server_ready = True logger.bind(tag=TAG).info("已发送首帧，开始识别") @@ -176,23 +183,14 @@ class ASRProvider(ASRProviderBase): if not self.asr_ws: return - audio_b64 = base64.b64encode(audio_data).decode('utf-8') + audio_b64 = base64.b64encode(audio_data).decode("utf-8") frame_data = { - "header": { - "status": status, - "app_id": self.app_id - }, - "parameter": { - "iat": self.iat_params - }, + "header": {"status": status, "app_id": self.app_id}, + "parameter": {"iat": self.iat_params}, "payload": { - "audio": { - "audio": audio_b64, - "sample_rate": 16000, - "encoding": "raw" - } - } + "audio": {"audio": audio_b64, "sample_rate": 16000, "encoding": "raw"} + }, } await self.asr_ws.send(json.dumps(frame_data, ensure_ascii=False)) @@ -207,11 +205,13 @@ class ASRProvider(ASRProviderBase): try: while self.asr_ws and not conn.stop_event.is_set(): # 获取当前连接的音频数据 - audio_data = getattr(conn, 'asr_audio_for_voiceprint', []) + audio_data = getattr(conn, "asr_audio_for_voiceprint", []) 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=timeout + ) result = json.loads(response) logger.bind(tag=TAG).debug(f"收到ASR结果: {result}") @@ -221,7 +221,9 @@ class ASRProvider(ASRProviderBase): status = header.get("status", 0) if code != 0: - logger.bind(tag=TAG).error(f"识别错误，错误码: {code}, 消息: {header.get('message', '')}") + logger.bind(tag=TAG).error( + f"识别错误，错误码: {code}, 消息: {header.get('message', '')}" + ) if code in [10114, 10160]: # 连接问题 break continue @@ -231,52 +233,124 @@ class ASRProvider(ASRProviderBase): text_data = payload["result"]["text"] if text_data: # 解码base64文本 - decoded_text = base64.b64decode(text_data).decode('utf-8') + decoded_text = base64.b64decode(text_data).decode("utf-8") text_json = json.loads(decoded_text) # 提取文本内容 - text_ws = text_json.get('ws', []) - result_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 # 更新识别文本 - 实时更新策略 - if result_text and result_text.strip() not in ['', '。', '.', ',', '，']: + # 只检查是否为空字符串，不再过滤任何标点符号 + # 这样可以确保所有识别到的内容，包括标点符号都能被实时更新 + if result_text and result_text.strip(): # 实时更新：正常情况下都更新，提高响应速度 should_update = True - # 保存最佳文本（最长的有意义文本） - if (len(result_text) > len(self.best_text) and - result_text.strip() not in ['？', '?', '。', '.']): - self.best_text = result_text - logger.bind(tag=TAG).debug(f"保存最佳文本: {self.best_text}") + # 保存最佳文本 + # 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 result_text.strip() in ['？', '?', '。', '.']: + # 只拒绝完全空的结果 + if not result_text.strip(): should_update = False - logger.bind(tag=TAG).warning(f"最终帧后拒绝无效文本: {result_text}") + logger.bind(tag=TAG).warning( + f"最终帧后拒绝空文本" + ) if should_update: - self.text = result_text - logger.bind(tag=TAG).info(f"实时更新识别文本: {self.text} (最终帧已发送: {self.last_frame_sent})") + # 处理流式识别结果，避免简单替换导致内容丢失 + # 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}") + logger.bind(tag=TAG).info( + f"识别完成状态已到达，当前识别文本: {self.text}" + ) # 如果还没发送最终帧，继续等待 if not self.last_frame_sent: - logger.bind(tag=TAG).info("识别完成但最终帧未发送，继续等待...") + logger.bind(tag=TAG).info( + "识别完成但最终帧未发送，继续等待..." + ) continue - # 已发送最终帧且收到完成状态，使用最佳文本作为最终结果 - 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 + # 已发送最终帧且收到完成状态，使用最佳策略选择最终结果 + # 优先使用识别完成状态下的最新结果，而不是仅仅基于长度 + 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}") conn.reset_vad_states() @@ -289,9 +363,13 @@ class ASRProvider(ASRProviderBase): 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}") + logger.bind(tag=TAG).info( + f"超时，使用最佳文本: {self.text} -> {self.best_text}" + ) self.text = self.best_text - logger.bind(tag=TAG).info(f"最终帧后超时，使用结果: {self.text}") + logger.bind(tag=TAG).info( + f"最终帧后超时，使用结果: {self.text}" + ) break # 如果还没发送最终帧，继续等待 continue @@ -316,11 +394,11 @@ class ASRProvider(ASRProviderBase): self.asr_ws = None self.is_processing = False if conn: - if hasattr(conn, 'asr_audio_for_voiceprint'): + if hasattr(conn, "asr_audio_for_voiceprint"): conn.asr_audio_for_voiceprint = [] - if hasattr(conn, 'asr_audio'): + if hasattr(conn, "asr_audio"): conn.asr_audio = [] - if hasattr(conn, 'has_valid_voice'): + if hasattr(conn, "has_valid_voice"): conn.has_valid_voice = False async def handle_voice_stop(self, conn, asr_audio_task: List[bytes]): @@ -330,7 +408,7 @@ class ASRProvider(ASRProviderBase): if self.asr_ws and self.is_processing: try: # 取最后一个有效的音频帧作为最后一帧数据 - last_frame = b'' + last_frame = b"" if asr_audio_task: last_audio = asr_audio_task[-1] last_frame = self.decoder.decode(last_audio, 960) @@ -338,8 +416,7 @@ class ASRProvider(ASRProviderBase): logger.bind(tag=TAG).info("已发送最后一帧") # 发送最终帧后，给_forward_results适当时间处理最终结果 - # 减少等待时间，提高响应速度 - await asyncio.sleep(0.1) # 从3秒减少到1秒 + await asyncio.sleep(0.25) logger.bind(tag=TAG).info(f"准备处理最终识别结果: {self.text}") except Exception as e: @@ -350,8 +427,9 @@ class ASRProvider(ASRProviderBase): except Exception as e: logger.bind(tag=TAG).error(f"处理语音停止失败: {e}") import traceback + logger.bind(tag=TAG).debug(f"异常详情: {traceback.format_exc()}") - + def stop_ws_connection(self): if self.asr_ws: asyncio.create_task(self.asr_ws.close()) @@ -360,12 +438,14 @@ class ASRProvider(ASRProviderBase): async def _cleanup(self, conn): """清理资源""" - logger.bind(tag=TAG).info(f"开始ASR会话清理 | 当前状态: processing={self.is_processing}, server_ready={self.server_ready}") + logger.bind(tag=TAG).info( + 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 self._send_audio_frame(b"", STATUS_LAST_FRAME) await asyncio.sleep(0.1) logger.bind(tag=TAG).info("已发送最后一帧") except Exception as e: @@ -376,6 +456,7 @@ class ASRProvider(ASRProviderBase): self.server_ready = False self.last_frame_sent = False self.best_text = "" + self.has_final_result = False logger.bind(tag=TAG).info("ASR状态已重置") # 清理任务 @@ -403,11 +484,11 @@ class ASRProvider(ASRProviderBase): # 清理连接的音频缓存 if conn: - if hasattr(conn, 'asr_audio_for_voiceprint'): + if hasattr(conn, "asr_audio_for_voiceprint"): conn.asr_audio_for_voiceprint = [] - if hasattr(conn, 'asr_audio'): + if hasattr(conn, "asr_audio"): conn.asr_audio = [] - if hasattr(conn, 'has_valid_voice'): + if hasattr(conn, "has_valid_voice"): conn.has_valid_voice = False logger.bind(tag=TAG).info("ASR会话清理完成") @@ -432,11 +513,11 @@ class ASRProvider(ASRProviderBase): self.forward_task = None self.is_processing = False # 清理所有连接的音频缓冲区 - if hasattr(self, '_connections'): + if hasattr(self, "_connections"): for conn in self._connections.values(): - if hasattr(conn, 'asr_audio_for_voiceprint'): + if hasattr(conn, "asr_audio_for_voiceprint"): conn.asr_audio_for_voiceprint = [] - if hasattr(conn, 'asr_audio'): + if hasattr(conn, "asr_audio"): conn.asr_audio = [] - if hasattr(conn, 'has_valid_voice'): - conn.has_valid_voice = False \ No newline at end of file + if hasattr(conn, "has_valid_voice"): + conn.has_valid_voice = False