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

Merge branch 'main' of https://github.com/xinnan-tech/xiaozhi-esp32-server

Browse Source
This commit is contained in:
Chingfeng Li
2025-10-24 09:02:09 +08:00
parent 7035a57cf1 4e3701e62c
commit 7ce1672d98
2 changed files with 194 additions and 109 deletions
Download Patch File Download Diff File Expand all files Collapse all files
main/xiaozhi-server/core/providers/asr/xunfei_stream.py
+156 -75
View File
@@ -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
if hasattr(conn, "has_valid_voice"):
conn.has_valid_voice = False
main/xiaozhi-server/core/providers/tts/huoshan_double_stream.py
+38 -34
View File
@@ -149,6 +149,7 @@ class TTSProvider(TTSProviderBase):
self.access_token = config.get("access_token")
self.cluster = config.get("cluster")
self.resource_id = config.get("resource_id")
self.activate_session = False
if config.get("private_voice"):
self.voice = config.get("private_voice")
else:
@@ -180,7 +181,7 @@ class TTSProvider(TTSProviderBase):
raise
async def _ensure_connection(self):
"""建立新的WebSocket连接"""
"""建立新的WebSocket连接，并启动监听任务（仅第一次）"""
try:
if self.ws:
logger.bind(tag=TAG).info(f"使用已有链接...")
@@ -196,6 +197,12 @@ class TTSProvider(TTSProviderBase):
self.ws_url, additional_headers=ws_header, max_size=1000000000
)
logger.bind(tag=TAG).info("WebSocket连接建立成功")
# 连接建立成功后，启动监听任务
if self._monitor_task is None or self._monitor_task.done():
logger.bind(tag=TAG).info("启动监听任务...")
self._monitor_task = asyncio.create_task(self._start_monitor_tts_response())
return self.ws
except Exception as e:
logger.bind(tag=TAG).error(f"建立连接失败: {str(e)}")
@@ -314,22 +321,24 @@ class TTSProvider(TTSProviderBase):
async def start_session(self, session_id):
logger.bind(tag=TAG).info(f"开始会话～～{session_id}")
try:
# 会话开始时检测上个会话的监听状态
if (
self._monitor_task is not None
and isinstance(self._monitor_task, Task)
and not self._monitor_task.done()
):
logger.bind(tag=TAG).info("检测到未完成的上个会话，关闭监听任务和连接...")
try:
# 等待上一个会话结束，最多等待3次
for _ in range(3):
if not self.activate_session:
break
logger.bind(tag=TAG).debug(f"等待上一个会话结束...")
await asyncio.sleep(0.1)
else:
# 等待超时，强制清除连接状态
logger.bind(tag=TAG).debug("等待上一个会话超时，清除连接状态...")
await self.close()
# 建立新连接
# 设置会话激活标志
self.activate_session = True
# 确保连接建立
await self._ensure_connection()
# 启动监听任务
self._monitor_task = asyncio.create_task(self._start_monitor_tts_response())
header = Header(
message_type=FULL_CLIENT_REQUEST,
message_type_specific_flags=MsgTypeFlagWithEvent,
@@ -365,17 +374,6 @@ class TTSProvider(TTSProviderBase):
await self.send_event(self.ws, header, optional, payload)
logger.bind(tag=TAG).info("会话结束请求已发送")
# 等待监听任务完成
if self._monitor_task:
try:
await self._monitor_task
except Exception as e:
logger.bind(tag=TAG).error(
f"等待监听任务完成时发生错误: {str(e)}"
)
finally:
self._monitor_task = None
except Exception as e:
logger.bind(tag=TAG).error(f"关闭会话失败: {str(e)}")
# 确保清理资源
@@ -405,6 +403,7 @@ class TTSProvider(TTSProviderBase):
async def close(self):
"""资源清理方法"""
self.activate_session = False
# 取消监听任务
if self._monitor_task:
try:
@@ -424,9 +423,8 @@ class TTSProvider(TTSProviderBase):
self.ws = None
async def _start_monitor_tts_response(self):
"""监听TTS响应"""
"""监听TTS响应 - 长期运行"""
try:
session_finished = False # 标记会话是否正常结束
while not self.conn.stop_event.is_set():
try:
# 确保 `recv()` 运行在同一个 event loop
@@ -434,10 +432,17 @@ class TTSProvider(TTSProviderBase):
res = self.parser_response(msg)
self.print_response(res, "send_text res:")
# 只处理当前活跃会话的响应
if res.optional.sessionId and self.conn.sentence_id != res.optional.sessionId:
# 如果是会话结束相关事件，即使会话ID不匹配也要重置状态
if res.optional.event in [EVENT_SessionCanceled, EVENT_SessionFailed, EVENT_SessionFinished]:
logger.bind(tag=TAG).debug(f"收到残余下行结束响应重置会话状态～～")
self.activate_session = False
continue
if res.optional.event == EVENT_SessionCanceled:
logger.bind(tag=TAG).debug(f"释放服务端资源成功～～")
session_finished = True
break
self.activate_session = False
elif res.optional.event == EVENT_TTSSentenceStart:
json_data = json.loads(res.payload.decode("utf-8"))
self.tts_text = json_data.get("text", "")
@@ -454,9 +459,8 @@ class TTSProvider(TTSProviderBase):
logger.bind(tag=TAG).info(f"句子语音生成成功：{self.tts_text}")
elif res.optional.event == EVENT_SessionFinished:
logger.bind(tag=TAG).debug(f"会话结束～～")
self.activate_session = False
self._process_before_stop_play_files()
session_finished = True
break
except websockets.ConnectionClosed:
logger.bind(tag=TAG).warning("WebSocket连接已关闭")
break
@@ -466,8 +470,8 @@ class TTSProvider(TTSProviderBase):
)
traceback.print_exc()
break
# 仅在连接异常时才关闭
if not session_finished and self.ws:
# 连接异常时关闭WebSocket
if self.ws:
try:
await self.ws.close()
except:
@@ -513,7 +517,7 @@ class TTSProvider(TTSProviderBase):
def read_res_content(self, res: bytes, offset: int):
content_size = int.from_bytes(res[offset : offset + 4], "big", signed=True)
offset += 4
content = str(res[offset : offset + content_size])
content = res[offset : offset + content_size].decode('utf-8')
offset += content_size
return content, offset