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

update: 长按设备ASR适配

Browse Source
This commit is contained in:
Sakura-RanChen
2025-12-11 17:12:56 +08:00
parent 3a74b30a0e
commit fbfc408e94
11 changed files with 195 additions and 323 deletions
Download Patch File Download Diff File Expand all files Collapse all files
main/xiaozhi-server/config.yaml
-1
View File
@@ -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/
main/xiaozhi-server/core/handle/reportHandle.py
-1
View File
@@ -10,7 +10,6 @@ TTS上报功能已集成到ConnectionHandler类中。
"""
import time
import gc
import opuslib_next
from config.manage_api_client import report as manage_report
main/xiaozhi-server/core/handle/textHandler/listenMessageHandler.py
+15 -9
View File
@@ -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()
main/xiaozhi-server/core/providers/asr/aliyun_stream.py
+71 -70
View File
@@ -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):
main/xiaozhi-server/core/providers/asr/base.py
+11 -16
View File
@@ -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]):
main/xiaozhi-server/core/providers/asr/doubao_stream.py
+51 -15
View File
@@ -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
main/xiaozhi-server/core/providers/asr/qwen3_asr_flash.py
+2 -1
View File
@@ -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初始化"""
# 配置参数
main/xiaozhi-server/core/providers/asr/xunfei_stream.py
+44 -206
View File
@@ -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
main/xiaozhi-server/core/providers/vad/silero.py
+1 -2
View File
@@ -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:
main/xiaozhi-server/core/utils/opus_encoder_utils.py
-1
View File
@@ -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
main/xiaozhi-server/core/utils/util.py
-1
View File
@@ -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