Hass/xiaozhi-esp32-server
1
0
Fork 0
mirror of https://github.com/xinnan-tech/xiaozhi-esp32-server.git synced 2026-07-29 04:53:55 +08:00
Code Issues Packages Projects Releases Wiki Activity

修复：讯飞asr最后一个字丢失的bug

Browse Source
This commit is contained in:
hrz
2025-10-23 17:04:33 +08:00
parent dbea7ae11e
commit a035ed525a
1 changed files with 156 additions and 75 deletions
Download Patch File Download Diff File Expand all files Collapse all files
main/xiaozhi-server/core/providers/asr/xunfei_stream.py
+153 -72
View File
@@ -22,6 +22,7 @@ STATUS_FIRST_FRAME = 0 # 第一帧的标识
STATUS_CONTINUE_FRAME = 1 # 中间帧标识 STATUS_CONTINUE_FRAME = 1 # 中间帧标识
STATUS_LAST_FRAME = 2 # 最后一帧的标识 STATUS_LAST_FRAME = 2 # 最后一帧的标识
class ASRProvider(ASRProviderBase): class ASRProvider(ASRProviderBase):
def __init__(self, config, delete_audio_file): def __init__(self, config, delete_audio_file):
super().__init__() super().__init__()
@@ -35,6 +36,7 @@ class ASRProvider(ASRProviderBase):
self.server_ready = False self.server_ready = False
self.last_frame_sent = False # 标记是否已发送最终帧 self.last_frame_sent = False # 标记是否已发送最终帧
self.best_text = "" # 保存最佳识别结果 self.best_text = "" # 保存最佳识别结果
self.has_final_result = False # 标记是否收到最终识别结果
# 讯飞配置 # 讯飞配置
self.app_id = config.get("app_id") self.app_id = config.get("app_id")
@@ -50,11 +52,7 @@ class ASRProvider(ASRProviderBase):
"language": config.get("language", "zh_cn"), "language": config.get("language", "zh_cn"),
"accent": config.get("accent", "mandarin"), "accent": config.get("accent", "mandarin"),
"dwa": config.get("dwa", "wpgs"), "dwa": config.get("dwa", "wpgs"),
"result": { "result": {"encoding": "utf8", "compress": "raw", "format": "plain"},
"encoding": "utf8",
"compress": "raw",
"format": "plain"
}
} }
self.output_dir = config.get("output_dir", "tmp/") self.output_dir = config.get("output_dir", "tmp/")
@@ -62,7 +60,7 @@ class ASRProvider(ASRProviderBase):
def create_url(self) -> str: def create_url(self) -> str:
"""生成认证URL""" """生成认证URL"""
url = 'ws://iat.cn-huabei-1.xf-yun.com/v1' url = "ws://iat.cn-huabei-1.xf-yun.com/v1"
# 生成RFC1123格式的时间戳 # 生成RFC1123格式的时间戳
now = datetime.now() now = datetime.now()
date = format_date_time(mktime(now.timetuple())) date = format_date_time(mktime(now.timetuple()))
@@ -73,23 +71,30 @@ class ASRProvider(ASRProviderBase):
signature_origin += "GET " + "/v1 " + "HTTP/1.1" signature_origin += "GET " + "/v1 " + "HTTP/1.1"
# 进行hmac-sha256进行加密 # 进行hmac-sha256进行加密
signature_sha = hmac.new(self.api_secret.encode('utf-8'), signature_origin.encode('utf-8'), signature_sha = hmac.new(
digestmod=hashlib.sha256).digest() self.api_secret.encode("utf-8"),
signature_sha = base64.b64encode(signature_sha).decode(encoding='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\"" % ( authorization_origin = (
self.api_key, "hmac-sha256", "host date request-line", signature_sha) 'api_key="%s", algorithm="%s", headers="%s", signature="%s"'
authorization = base64.b64encode(authorization_origin.encode('utf-8')).decode(encoding='utf-8') % (self.api_key, "hmac-sha256", "host date request-line", signature_sha)
)
authorization = base64.b64encode(authorization_origin.encode("utf-8")).decode(
encoding="utf-8"
)
# 将请求的鉴权参数组合为字典 # 将请求的鉴权参数组合为字典
v = { v = {
"authorization": authorization, "authorization": authorization,
"date": date, "date": date,
"host": "iat.cn-huabei-1.xf-yun.com" "host": "iat.cn-huabei-1.xf-yun.com",
} }
# 拼接鉴权参数，生成url # 拼接鉴权参数，生成url
url = url + '?' + urlencode(v) url = url + "?" + urlencode(v)
return url return url
async def open_audio_channels(self, conn): async def open_audio_channels(self, conn):
@@ -100,7 +105,7 @@ class ASRProvider(ASRProviderBase):
await super().receive_audio(conn, audio, audio_have_voice) 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 = []
conn.asr_audio_for_voiceprint.append(audio) conn.asr_audio_for_voiceprint.append(audio)
@@ -146,8 +151,10 @@ class ASRProvider(ASRProviderBase):
# 发送首帧音频 # 发送首帧音频
if conn.asr_audio and len(conn.asr_audio) > 0: if conn.asr_audio and len(conn.asr_audio) > 0:
first_audio = conn.asr_audio[-1] if conn.asr_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'' pcm_frame = (
self.decoder.decode(first_audio, 960) if first_audio else b""
)
await self._send_audio_frame(pcm_frame, STATUS_FIRST_FRAME) await self._send_audio_frame(pcm_frame, STATUS_FIRST_FRAME)
self.server_ready = True self.server_ready = True
logger.bind(tag=TAG).info("已发送首帧，开始识别") logger.bind(tag=TAG).info("已发送首帧，开始识别")
@@ -176,23 +183,14 @@ class ASRProvider(ASRProviderBase):
if not self.asr_ws: if not self.asr_ws:
return return
audio_b64 = base64.b64encode(audio_data).decode('utf-8') audio_b64 = base64.b64encode(audio_data).decode("utf-8")
frame_data = { frame_data = {
"header": { "header": {"status": status, "app_id": self.app_id},
"status": status, "parameter": {"iat": self.iat_params},
"app_id": self.app_id
},
"parameter": {
"iat": self.iat_params
},
"payload": { "payload": {
"audio": { "audio": {"audio": audio_b64, "sample_rate": 16000, "encoding": "raw"}
"audio": audio_b64, },
"sample_rate": 16000,
"encoding": "raw"
}
}
} }
await self.asr_ws.send(json.dumps(frame_data, ensure_ascii=False)) await self.asr_ws.send(json.dumps(frame_data, ensure_ascii=False))
@@ -207,11 +205,13 @@ class ASRProvider(ASRProviderBase):
try: try:
while self.asr_ws and not conn.stop_event.is_set(): 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: try:
# 如果已发送最终帧，增加超时时间等待完整结果 # 如果已发送最终帧，增加超时时间等待完整结果
timeout = 3.0 if self.last_frame_sent else 30.0 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) result = json.loads(response)
logger.bind(tag=TAG).debug(f"收到ASR结果: {result}") logger.bind(tag=TAG).debug(f"收到ASR结果: {result}")
@@ -221,7 +221,9 @@ class ASRProvider(ASRProviderBase):
status = header.get("status", 0) status = header.get("status", 0)
if code != 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]: # 连接问题 if code in [10114, 10160]: # 连接问题
break break
continue continue
@@ -231,52 +233,124 @@ class ASRProvider(ASRProviderBase):
text_data = payload["result"]["text"] text_data = payload["result"]["text"]
if text_data: if text_data:
# 解码base64文本 # 解码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_json = json.loads(decoded_text)
# 提取文本内容 # 提取文本内容
text_ws = text_json.get('ws', []) text_ws = text_json.get("ws", [])
result_text = '' result_text = ""
for i in text_ws: for i in text_ws:
for j in i.get("cw", []): for j in i.get("cw", []):
w = j.get("w", "") w = j.get("w", "")
result_text += w result_text += w
# 更新识别文本 - 实时更新策略 # 更新识别文本 - 实时更新策略
if result_text and result_text.strip() not in ['', '。', '.', ',', '，']: # 只检查是否为空字符串，不再过滤任何标点符号
# 这样可以确保所有识别到的内容，包括标点符号都能被实时更新
if result_text and result_text.strip():
# 实时更新：正常情况下都更新，提高响应速度 # 实时更新：正常情况下都更新，提高响应速度
should_update = True should_update = True
# 保存最佳文本（最长的有意义文本） # 保存最佳文本
if (len(result_text) > len(self.best_text) and # 1. 如果是识别完成状态或最终帧后收到的结果，优先保存
result_text.strip() not in ['？', '?', '。', '.']): # 2. 否则保存最长的有意义文本
self.best_text = result_text # 取消对标点符号的过滤，只检查是否为空
logger.bind(tag=TAG).debug(f"保存最佳文本: {self.best_text}") # 这样可以保留所有识别到的内容，包括各种标点符号
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 self.last_frame_sent:
# 最终帧后拒绝问号等明显错误的结果 # 只拒绝完全空的结果
if result_text.strip() in ['？', '?', '。', '.']: if not result_text.strip():
should_update = False should_update = False
logger.bind(tag=TAG).warning(f"最终帧后拒绝无效文本: {result_text}") logger.bind(tag=TAG).warning(
f"最终帧后拒绝空文本"
)
if should_update: 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: 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: if not self.last_frame_sent:
logger.bind(tag=TAG).info("识别完成但最终帧未发送，继续等待...") logger.bind(tag=TAG).info(
"识别完成但最终帧未发送，继续等待..."
)
continue continue
# 已发送最终帧且收到完成状态，使用最佳文本作为最终结果 # 已发送最终帧且收到完成状态，使用最佳策略选择最终结果
if self.best_text and len(self.best_text) > len(self.text): # 优先使用识别完成状态下的最新结果，而不是仅仅基于长度
logger.bind(tag=TAG).info(f"使用最佳文本作为最终结果: {self.text} -> {self.best_text}") if self.best_text:
self.text = 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}") logger.bind(tag=TAG).info(f"获取到最终完整文本: {self.text}")
conn.reset_vad_states() conn.reset_vad_states()
@@ -289,9 +363,13 @@ class ASRProvider(ASRProviderBase):
if self.last_frame_sent: if self.last_frame_sent:
# 超时时也使用最佳文本 # 超时时也使用最佳文本
if self.best_text and len(self.best_text) > len(self.text): 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 self.text = self.best_text
logger.bind(tag=TAG).info(f"最终帧后超时，使用结果: {self.text}") logger.bind(tag=TAG).info(
f"最终帧后超时，使用结果: {self.text}"
)
break break
# 如果还没发送最终帧，继续等待 # 如果还没发送最终帧，继续等待
continue continue
@@ -316,11 +394,11 @@ class ASRProvider(ASRProviderBase):
self.asr_ws = None self.asr_ws = None
self.is_processing = False self.is_processing = False
if conn: if conn:
if hasattr(conn, 'asr_audio_for_voiceprint'): if hasattr(conn, "asr_audio_for_voiceprint"):
conn.asr_audio_for_voiceprint = [] conn.asr_audio_for_voiceprint = []
if hasattr(conn, 'asr_audio'): if hasattr(conn, "asr_audio"):
conn.asr_audio = [] conn.asr_audio = []
if hasattr(conn, 'has_valid_voice'): if hasattr(conn, "has_valid_voice"):
conn.has_valid_voice = False conn.has_valid_voice = False
async def handle_voice_stop(self, conn, asr_audio_task: List[bytes]): 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: if self.asr_ws and self.is_processing:
try: try:
# 取最后一个有效的音频帧作为最后一帧数据 # 取最后一个有效的音频帧作为最后一帧数据
last_frame = b'' last_frame = b""
if asr_audio_task: if asr_audio_task:
last_audio = asr_audio_task[-1] last_audio = asr_audio_task[-1]
last_frame = self.decoder.decode(last_audio, 960) last_frame = self.decoder.decode(last_audio, 960)
@@ -338,8 +416,7 @@ class ASRProvider(ASRProviderBase):
logger.bind(tag=TAG).info("已发送最后一帧") logger.bind(tag=TAG).info("已发送最后一帧")
# 发送最终帧后，给_forward_results适当时间处理最终结果 # 发送最终帧后，给_forward_results适当时间处理最终结果
# 减少等待时间，提高响应速度 await asyncio.sleep(0.25)
await asyncio.sleep(0.1) # 从3秒减少到1秒
logger.bind(tag=TAG).info(f"准备处理最终识别结果: {self.text}") logger.bind(tag=TAG).info(f"准备处理最终识别结果: {self.text}")
except Exception as e: except Exception as e:
@@ -350,6 +427,7 @@ class ASRProvider(ASRProviderBase):
except Exception as e: except Exception as e:
logger.bind(tag=TAG).error(f"处理语音停止失败: {e}") logger.bind(tag=TAG).error(f"处理语音停止失败: {e}")
import traceback import traceback
logger.bind(tag=TAG).debug(f"异常详情: {traceback.format_exc()}") logger.bind(tag=TAG).debug(f"异常详情: {traceback.format_exc()}")
def stop_ws_connection(self): def stop_ws_connection(self):
@@ -360,12 +438,14 @@ class ASRProvider(ASRProviderBase):
async def _cleanup(self, conn): 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: if self.asr_ws and self.is_processing:
try: try:
await self._send_audio_frame(b'', STATUS_LAST_FRAME) await self._send_audio_frame(b"", STATUS_LAST_FRAME)
await asyncio.sleep(0.1) await asyncio.sleep(0.1)
logger.bind(tag=TAG).info("已发送最后一帧") logger.bind(tag=TAG).info("已发送最后一帧")
except Exception as e: except Exception as e:
@@ -376,6 +456,7 @@ class ASRProvider(ASRProviderBase):
self.server_ready = False self.server_ready = False
self.last_frame_sent = False self.last_frame_sent = False
self.best_text = "" self.best_text = ""
self.has_final_result = False
logger.bind(tag=TAG).info("ASR状态已重置") logger.bind(tag=TAG).info("ASR状态已重置")
# 清理任务 # 清理任务
@@ -403,11 +484,11 @@ class ASRProvider(ASRProviderBase):
# 清理连接的音频缓存 # 清理连接的音频缓存
if conn: if conn:
if hasattr(conn, 'asr_audio_for_voiceprint'): if hasattr(conn, "asr_audio_for_voiceprint"):
conn.asr_audio_for_voiceprint = [] conn.asr_audio_for_voiceprint = []
if hasattr(conn, 'asr_audio'): if hasattr(conn, "asr_audio"):
conn.asr_audio = [] conn.asr_audio = []
if hasattr(conn, 'has_valid_voice'): if hasattr(conn, "has_valid_voice"):
conn.has_valid_voice = False conn.has_valid_voice = False
logger.bind(tag=TAG).info("ASR会话清理完成") logger.bind(tag=TAG).info("ASR会话清理完成")
@@ -432,11 +513,11 @@ class ASRProvider(ASRProviderBase):
self.forward_task = None self.forward_task = None
self.is_processing = False self.is_processing = False
# 清理所有连接的音频缓冲区 # 清理所有连接的音频缓冲区
if hasattr(self, '_connections'): if hasattr(self, "_connections"):
for conn in self._connections.values(): 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 = [] conn.asr_audio_for_voiceprint = []
if hasattr(conn, 'asr_audio'): if hasattr(conn, "asr_audio"):
conn.asr_audio = [] conn.asr_audio = []
if hasattr(conn, 'has_valid_voice'): if hasattr(conn, "has_valid_voice"):
conn.has_valid_voice = False conn.has_valid_voice = False