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:
FAN-yeB
2025-12-26 16:13:14 +08:00
parent 2a42d86db0
commit 46abb1009a
3 changed files with 399 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
main/manager-api/src/main/resources/db/changelog/202512261524.sql
+26
View File
@@ -0,0 +1,26 @@
-- 添加阿里百炼Paraformer实时语音识别服务配置
delete from `ai_model_provider` where id = 'SYSTEM_ASR_AliyunBLStream';
INSERT INTO `ai_model_provider` (`id`, `model_type`, `provider_code`, `name`, `fields`, `sort`, `creator`, `create_date`, `updater`, `update_date`) VALUES
('SYSTEM_ASR_AliyunBLStream', 'ASR', 'aliyunbl_stream', '阿里百炼Paraformer实时语音识别', '[{"key":"api_key","label":"API密钥","type":"password"},{"key":"model","label":"模型名称","type":"string"},{"key":"format","label":"音频格式","type":"string"},{"key":"sample_rate","label":"采样率","type":"number"},{"key":"output_dir","label":"输出目录","type":"string"}]', 18, 1, NOW(), 1, NOW());
delete from `ai_model_config` where id = 'ASR_AliyunBLStream';
INSERT INTO `ai_model_config` VALUES ('ASR_AliyunBLStream', 'ASR', 'AliyunBLStream', '阿里百炼Paraformer实时语音识别', 0, 1, '{"type": "aliyunbl_stream", "api_key": "", "model": "paraformer-realtime-v2", "format": "pcm", "sample_rate": 16000, "disfluency_removal_enabled": false, "semantic_punctuation_enabled": false, "max_sentence_silence": 800, "multi_threshold_mode_enabled": false, "punctuation_prediction_enabled": true, "heartbeat": false, "inverse_text_normalization_enabled": true, "output_dir": "tmp/"}', 'https://help.aliyun.com/zh/model-studio/websocket-for-paraformer-real-time-service', '支持多语言、热词定制、语义断句等高级功能', 21, NULL, NULL, NULL, NULL);
-- 更新阿里百炼Paraformer模型配置的说明文档
UPDATE `ai_model_config` SET
`doc_link` = 'https://help.aliyun.com/zh/model-studio/websocket-for-paraformer-real-time-service',
`remark` = '阿里百炼Paraformer实时语音识别配置说明：
1. 登录阿里云百炼平台 https://bailian.console.aliyun.com/
2. 创建API-KEY https://bailian.console.aliyun.com/#/api-key
3. 支持模型：paraformer-realtime-v2(推荐)、paraformer-realtime-8k-v2、paraformer-realtime-v1、paraformer-realtime-8k-v1
4. 功能特性：
- 多语言支持(中文含方言、英文、日语、韩语、德语、法语、俄语)
- 热词定制(vocabulary_id参数)
- 语义断句/VAD断句(semantic_punctuation_enabled参数)
- 自动标点符号、ITN、过滤语气词等
5. 参数说明：
- model: 模型名称，推荐paraformer-realtime-v2
- sample_rate: 采样率(Hz)，v2支持任意采样率，v1仅支持16000，8k版本仅支持8000
- semantic_punctuation_enabled: false为VAD断句(低延迟)，true为语义断句(高准确)
- max_sentence_silence: VAD断句静音时长阈值(200-6000ms)
' WHERE `id` = 'ASR_AliyunBLStream';
main/xiaozhi-server/config.yaml
+26 -1
View File
@@ -478,7 +478,32 @@ ASR:
accent: mandarin # 方言，mandarin:普通话
# 调整音频处理参数以提高长语音识别质量
output_dir: tmp/
AliyunBLStreamASR:
# 阿里百炼Paraformer实时语音识别服务
# WebSocket实时流式语音识别，支持多语言、热词定制、语义断句等高级功能
# 平台地址：https://bailian.console.aliyun.com/
# API Key地址：https://bailian.console.aliyun.com/#/api-key
# 文档地址：https://help.aliyun.com/zh/model-studio/websocket-for-paraformer-real-time-service
# 支持模型：paraformer-realtime-v2(推荐), paraformer-realtime-8k-v2, paraformer-realtime-v1, paraformer-realtime-8k-v1
type: aliyunbl_stream
# 必填参数
api_key: 你的阿里云百炼API密钥
# 模型选择，推荐使用v2版本
model: paraformer-realtime-v2
# 音频格式和采样率
format: pcm
sample_rate: 16000 # v2支持任意采样率，v1仅支持16000，8k版本仅支持8000
# 可选参数
disfluency_removal_enabled: false # 是否过滤语气词(如"嗯"、"啊"等)
semantic_punctuation_enabled: false # 语义断句(true:会议场景,准确;false:VAD断句,交互场景,低延迟)
max_sentence_silence: 800 # VAD断句静音时长阈值(ms)，范围200-6000，仅VAD断句时生效
multi_threshold_mode_enabled: false # 防止VAD断句切割过长，仅VAD断句时生效
punctuation_prediction_enabled: true # 是否自动添加标点符号
heartbeat: false # 是否开启长连接心跳(持续静音下保持连接)
inverse_text_normalization_enabled: true # 是否开启ITN(中文数字转阿拉伯数字)
# vocabulary_id: vocab-xxx-24ee19fa8cfb4d52902170a0xxxxxxxx # 热词ID(可选)
# language_hints: ["zh", "en"] # 指定语言(可选)，支持zh、en、ja、yue、ko、de、fr、ru
output_dir: tmp/
VAD:
SileroVAD:
type: silero
main/xiaozhi-server/core/providers/asr/aliyunbl_stream.py
+347
View File
@@ -0,0 +1,347 @@
import json
import uuid
import asyncio
import websockets
import opuslib_next
from typing import List
from config.logger import setup_logging
from core.providers.asr.base import ASRProviderBase
from core.providers.asr.dto.dto import InterfaceType
TAG = __name__
logger = setup_logging()
class ASRProvider(ASRProviderBase):
def __init__(self, config, delete_audio_file):
super().__init__()
self.interface_type = InterfaceType.STREAM
self.config = config
self.text = ""
self.decoder = opuslib_next.Decoder(16000, 1)
self.asr_ws = None
self.forward_task = None
self.is_processing = False
self.server_ready = False # 服务器准备状态
self.task_id = None # 当前任务ID
# 阿里百炼配置
self.api_key = config.get("api_key")
self.model = config.get("model", "paraformer-realtime-v2")
self.sample_rate = config.get("sample_rate", 16000)
self.format = config.get("format", "pcm")
# 可选参数
self.vocabulary_id = config.get("vocabulary_id") # 热词ID
self.disfluency_removal_enabled = config.get("disfluency_removal_enabled", False) # 过滤语气词
self.language_hints = config.get("language_hints") # 语言提示,如 ["zh", "en"]
self.semantic_punctuation_enabled = config.get("semantic_punctuation_enabled", False) # 语义断句
self.max_sentence_silence = config.get("max_sentence_silence", 800) # VAD断句静音时长(ms)
self.multi_threshold_mode_enabled = config.get("multi_threshold_mode_enabled", False) # 防止VAD断句切割过长
self.punctuation_prediction_enabled = config.get("punctuation_prediction_enabled", True) # 标点符号预测
self.heartbeat = config.get("heartbeat", False) # 长连接心跳
self.inverse_text_normalization_enabled = config.get("inverse_text_normalization_enabled", True) # ITN
# WebSocket URL
self.ws_url = "wss://dashscope.aliyuncs.com/api-ws/v1/inference"
self.output_dir = config.get("output_dir", "./audio_output")
self.delete_audio_file = delete_audio_file
async def open_audio_channels(self, conn):
await super().open_audio_channels(conn)
async def receive_audio(self, conn, audio, audio_have_voice):
# 初始化音频缓存
if not hasattr(conn, 'asr_audio_for_voiceprint'):
conn.asr_audio_for_voiceprint = []
# 存储音频数据
if audio:
conn.asr_audio_for_voiceprint.append(audio)
conn.asr_audio.append(audio)
conn.asr_audio = conn.asr_audio[-10:]
# 只在有声音且没有连接时建立连接
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()
return
# 发送音频数据
if self.asr_ws and self.is_processing and self.server_ready:
try:
pcm_frame = self.decoder.decode(audio, 960)
# 直接发送PCM音频数据(二进制)
await self.asr_ws.send(pcm_frame)
except Exception as e:
logger.bind(tag=TAG).warning(f"发送音频失败: {str(e)}")
await self._cleanup()
async def _start_recognition(self, conn):
"""开始识别会话"""
try:
self.is_processing = True
self.task_id = uuid.uuid4().hex
# 建立WebSocket连接
headers = {
"Authorization": f"Bearer {self.api_key}"
}
logger.bind(tag=TAG).debug(f"正在连接阿里百炼ASR服务, task_id: {self.task_id}")
self.asr_ws = await websockets.connect(
self.ws_url,
additional_headers=headers,
max_size=1000000000,
ping_interval=None,
ping_timeout=None,
close_timeout=5,
)
logger.bind(tag=TAG).debug("WebSocket连接建立成功")
self.server_ready = False
self.forward_task = asyncio.create_task(self._forward_results(conn))
# 发送run-task指令
run_task_msg = self._build_run_task_message()
await self.asr_ws.send(json.dumps(run_task_msg, ensure_ascii=False))
logger.bind(tag=TAG).debug("已发送run-task指令，等待服务器准备...")
except Exception as e:
logger.bind(tag=TAG).error(f"建立ASR连接失败: {str(e)}")
if self.asr_ws:
await self.asr_ws.close()
self.asr_ws = None
self.is_processing = False
raise
def _build_run_task_message(self) -> dict:
"""构建run-task指令"""
message = {
"header": {
"action": "run-task",
"task_id": self.task_id,
"streaming": "duplex"
},
"payload": {
"task_group": "audio",
"task": "asr",
"function": "recognition",
"model": self.model,
"parameters": {
"format": self.format,
"sample_rate": self.sample_rate,
"disfluency_removal_enabled": self.disfluency_removal_enabled,
"semantic_punctuation_enabled": self.semantic_punctuation_enabled,
"max_sentence_silence": self.max_sentence_silence,
"multi_threshold_mode_enabled": self.multi_threshold_mode_enabled,
"punctuation_prediction_enabled": self.punctuation_prediction_enabled,
"heartbeat": self.heartbeat,
"inverse_text_normalization_enabled": self.inverse_text_normalization_enabled,
},
"input": {}
}
}
# 添加可选参数
if self.vocabulary_id:
message["payload"]["parameters"]["vocabulary_id"] = self.vocabulary_id
if self.language_hints:
message["payload"]["parameters"]["language_hints"] = self.language_hints
return message
async def _forward_results(self, conn):
"""转发识别结果"""
try:
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", {})
event = header.get("event", "")
# 处理task-started事件
if event == "task-started":
self.server_ready = True
logger.bind(tag=TAG).debug("服务器已准备，开始发送缓存音频...")
# 发送缓存音频
if conn.asr_audio:
for cached_audio in conn.asr_audio[-10:]:
try:
pcm_frame = self.decoder.decode(cached_audio, 960)
await self.asr_ws.send(pcm_frame)
except Exception as e:
logger.bind(tag=TAG).warning(f"发送缓存音频失败: {e}")
break
continue
# 处理result-generated事件
elif event == "result-generated":
output = payload.get("output", {})
sentence = output.get("sentence", {})
if not sentence:
continue
# 跳过心跳消息
if sentence.get("heartbeat", False):
continue
text = sentence.get("text", "")
sentence_end = sentence.get("sentence_end", False)
end_time = sentence.get("end_time")
# 只处理有文本的结果
if not text:
continue
# 判断是否为最终结果(sentence_end为True且end_time不为null)
is_final = sentence_end and end_time is not None
if is_final:
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
# 处理task-finished事件
elif event == "task-finished":
logger.bind(tag=TAG).debug("任务已完成")
break
# 处理task-failed事件
elif event == "task-failed":
error_code = header.get("error_code", "UNKNOWN")
error_message = header.get("error_message", "未知错误")
logger.bind(tag=TAG).error(f"任务失败: {error_code} - {error_message}")
break
except asyncio.TimeoutError:
continue
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()
if conn:
if hasattr(conn, 'asr_audio_for_voiceprint'):
conn.asr_audio_for_voiceprint = []
if hasattr(conn, 'asr_audio'):
conn.asr_audio = []
async def _send_stop_request(self):
"""发送停止请求(用于手动模式停止录音)"""
if self.asr_ws and self.task_id and self.is_processing:
try:
logger.bind(tag=TAG).debug("收到停止请求，发送finish-task指令")
await self._send_finish_task()
except Exception as e:
logger.bind(tag=TAG).error(f"发送停止请求失败: {e}")
async def _send_finish_task(self):
"""发送finish-task指令"""
if self.asr_ws and self.task_id:
try:
finish_msg = {
"header": {
"action": "finish-task",
"task_id": self.task_id,
"streaming": "duplex"
},
"payload": {
"input": {}
}
}
await self.asr_ws.send(json.dumps(finish_msg, ensure_ascii=False))
logger.bind(tag=TAG).debug("已发送finish-task指令")
except Exception as e:
logger.bind(tag=TAG).error(f"发送finish-task指令失败: {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.asr_ws:
try:
# 先发送finish-task指令
await self._send_finish_task()
# 等待一小段时间让服务器处理
await asyncio.sleep(0.1)
logger.bind(tag=TAG).debug("正在关闭WebSocket连接")
await asyncio.wait_for(self.asr_ws.close(), timeout=2.0)
logger.bind(tag=TAG).debug("WebSocket连接已关闭")
except Exception as e:
logger.bind(tag=TAG).error(f"关闭WebSocket连接失败: {e}")
finally:
self.asr_ws = None
# 清理任务引用
self.forward_task = None
self.task_id = None
logger.bind(tag=TAG).debug("ASR会话清理完成")
async def speech_to_text(self, opus_data, session_id, audio_format):
"""获取识别结果"""
result = self.text
self.text = ""
return result, None
async def close(self):
"""关闭资源"""
await self._cleanup()
if hasattr(self, 'decoder') and self.decoder is not None:
try:
del self.decoder
self.decoder = None
logger.bind(tag=TAG).debug("Aliyun BL decoder resources released")
except Exception as e:
logger.bind(tag=TAG).debug(f"释放Aliyun BL decoder资源时出错: {e}")