diff --git a/main/manager-api/src/main/resources/db/changelog/202512261524.sql b/main/manager-api/src/main/resources/db/changelog/202512261524.sql new file mode 100644 index 00000000..649fb240 --- /dev/null +++ b/main/manager-api/src/main/resources/db/changelog/202512261524.sql @@ -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'; \ No newline at end of file diff --git a/main/xiaozhi-server/config.yaml b/main/xiaozhi-server/config.yaml index 1139dbf4..11bdd87a 100644 --- a/main/xiaozhi-server/config.yaml +++ b/main/xiaozhi-server/config.yaml @@ -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 diff --git a/main/xiaozhi-server/core/providers/asr/aliyunbl_stream.py b/main/xiaozhi-server/core/providers/asr/aliyunbl_stream.py new file mode 100644 index 00000000..43241926 --- /dev/null +++ b/main/xiaozhi-server/core/providers/asr/aliyunbl_stream.py @@ -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}") \ No newline at end of file