diff --git a/main/xiaozhi-server/config.yaml b/main/xiaozhi-server/config.yaml index 71c13ec7..bd9f024b 100644 --- a/main/xiaozhi-server/config.yaml +++ b/main/xiaozhi-server/config.yaml @@ -298,9 +298,13 @@ ASR: output_dir: tmp/ AliyunASR: # 阿里云智能语音交互服务，需要先在阿里云平台开通服务，然后获取验证信息 + # HTTP POST请求，一次性处理完整音频 # 平台地址：https://nls-portal.console.aliyun.com/ # appkey地址：https://nls-portal.console.aliyun.com/applist # token地址：https://nls-portal.console.aliyun.com/overview + # AliyunASR和AliyunStreamASR的区别是：AliyunASR是批量处理场景，AliyunStreamASR是实时交互场景 + # 一般来说非流式ASR更便宜（0.004元/秒，¥0.24/分钟） + # 但是AliyunStreamASR实时性更好（0.005元/秒，¥0.3/分钟） # 定义ASR API类型 type: aliyun appkey: 你的阿里云智能语音交互服务项目Appkey @@ -308,6 +312,24 @@ ASR: access_key_id: 你的阿里云账号access_key_id access_key_secret: 你的阿里云账号access_key_secret output_dir: tmp/ + AliyunStreamASR: + # 阿里云智能语音交互服务 - 实时流式语音识别 + # WebSocket连接，实时处理音频流 + # 平台地址：https://nls-portal.console.aliyun.com/ + # appkey地址：https://nls-portal.console.aliyun.com/applist + # token地址：https://nls-portal.console.aliyun.com/overview + # AliyunASR和AliyunStreamASR的区别是：AliyunASR是批量处理场景，AliyunStreamASR是实时交互场景 + # 一般来说非流式ASR更便宜（0.004元/秒，¥0.24/分钟） + # 但是AliyunStreamASR实时性更好（0.005元/秒，¥0.3/分钟） + # 定义ASR API类型 + type: aliyun + appkey: 你的阿里云智能语音交互服务项目Appkey + token: 你的阿里云智能语音交互服务AccessToken，临时的24小时，要长期用下方的access_key_id，access_key_secret + access_key_id: 你的阿里云账号access_key_id + access_key_secret: 你的阿里云账号access_key_secret + # 服务器地域选择，如nls-gateway-cn-hangzhou.aliyuncs.com(杭州)等 + host: 根据自己的地理位置选择最近的服务器地域，不填默认为nls-gateway-cn-shanghai.aliyuncs.com(上海) + output_dir: tmp/ BaiduASR: # 获取AppID、API Key、Secret Key：https://console.bce.baidu.com/ai-engine/old/#/ai/speech/app/list # 查看资源额度：https://console.bce.baidu.com/ai-engine/old/#/ai/speech/overview/resource/list diff --git a/main/xiaozhi-server/core/providers/asr/aliyun_stream.py b/main/xiaozhi-server/core/providers/asr/aliyun_stream.py new file mode 100644 index 00000000..94727798 --- /dev/null +++ b/main/xiaozhi-server/core/providers/asr/aliyun_stream.py @@ -0,0 +1,286 @@ +import json +import time +import uuid +import hmac +import base64 +import hashlib +import asyncio +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 +from core.providers.asr.base import ASRProviderBase +from core.providers.asr.dto.dto import InterfaceType + +TAG = __name__ +logger = setup_logging() + + +class AccessToken: + @staticmethod + def _encode_text(text): + encoded_text = parse.quote_plus(text) + return encoded_text.replace("+", "%20").replace("*", "%2A").replace("%7E", "~") + + @staticmethod + def _encode_dict(dic): + keys = dic.keys() + dic_sorted = [(key, dic[key]) for key in sorted(keys)] + encoded_text = parse.urlencode(dic_sorted) + return encoded_text.replace("+", "%20").replace("*", "%2A").replace("%7E", "~") + + @staticmethod + def create_token(access_key_id, access_key_secret): + parameters = { + "AccessKeyId": access_key_id, + "Action": "CreateToken", + "Format": "JSON", + "RegionId": "cn-shanghai", + "SignatureMethod": "HMAC-SHA1", + "SignatureNonce": str(uuid.uuid1()), + "SignatureVersion": "1.0", + "Timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), + "Version": "2019-02-28", + } + query_string = AccessToken._encode_dict(parameters) + string_to_sign = ( + "GET" + "&" + AccessToken._encode_text("/") + "&" + AccessToken._encode_text(query_string) + ) + secreted_string = hmac.new( + bytes(access_key_secret + "&", encoding="utf-8"), + bytes(string_to_sign, encoding="utf-8"), + hashlib.sha1, + ).digest() + signature = base64.b64encode(secreted_string) + signature = AccessToken._encode_text(signature) + full_url = "http://nls-meta.cn-shanghai.aliyuncs.com/?Signature=%s&%s" % (signature, query_string) + response = requests.get(full_url) + if response.ok: + root_obj = response.json() + if "Token" in root_obj: + return root_obj["Token"]["Id"], root_obj["Token"]["ExpireTime"] + return None, None + + +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.access_key_id = config.get("access_key_id") + self.access_key_secret = config.get("access_key_secret") + self.appkey = config.get("appkey") + self.token = config.get("token") + self.host = config.get("host", "nls-gateway-cn-shanghai.aliyuncs.com") + self.ws_url = f"wss://{self.host}/ws/v1" + self.max_sentence_silence = config.get("max_sentence_silence") + self.output_dir = config.get("output_dir", "./audio_output") + self.delete_audio_file = delete_audio_file + self.expire_time = None + + # Token管理 + if self.access_key_id and self.access_key_secret: + self._refresh_token() + elif not self.token: + raise ValueError("必须提供access_key_id+access_key_secret或者直接提供token") + + def _refresh_token(self): + """刷新Token""" + self.token, expire_time_str = AccessToken.create_token(self.access_key_id, self.access_key_secret) + if not self.token: + raise ValueError("无法获取有效的访问Token") + + try: + expire_str = str(expire_time_str).strip() + if expire_str.isdigit(): + expire_time = datetime.fromtimestamp(int(expire_str)) + else: + expire_time = datetime.strptime(expire_str, "%Y-%m-%dT%H:%M:%SZ") + self.expire_time = expire_time.timestamp() - 60 + except: + self.expire_time = None + + def _is_token_expired(self): + """检查Token是否过期""" + return self.expire_time and time.time() > self.expire_time + + async def open_audio_channels(self, conn): + await super().open_audio_channels(conn) + + async def receive_audio(self, conn, audio, audio_have_voice): + conn.asr_audio.append(audio) + conn.asr_audio = conn.asr_audio[-10:] + + # 参考豆包ASR：只在有声音且没有连接时建立连接 + if audio_have_voice and not self.is_processing: + 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) + 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): + """开始识别会话""" + if self._is_token_expired(): + self._refresh_token() + + # 建立连接 + headers = {"X-NLS-Token": self.token} + self.asr_ws = await websockets.connect( + self.ws_url, + additional_headers=headers, + max_size=1000000000, + ping_interval=None, + ping_timeout=None, + close_timeout=5, + ) + + self.is_processing = True + self.server_ready = False # 重置服务器准备状态 + self.forward_task = asyncio.create_task(self._forward_results(conn)) + + # 发送开始请求 + start_request = { + "header": { + "namespace": "SpeechTranscriber", + "name": "StartTranscription", + "status": 20000000, + "message_id": ''.join(random.choices('0123456789abcdef', k=32)), + "task_id": ''.join(random.choices('0123456789abcdef', k=32)), + "status_text": "Gateway:SUCCESS:Success.", + "appkey": self.appkey + }, + "payload": { + "format": "pcm", + "sample_rate": 16000, + "enable_intermediate_result": True, + "enable_punctuation_prediction": True, + "enable_inverse_text_normalization": True, + "max_sentence_silence": self.max_sentence_silence, + "enable_voice_detection": False, + } + } + await self.asr_ws.send(json.dumps(start_request, ensure_ascii=False)) + logger.bind(tag=TAG).info("已发送开始请求，等待服务器准备...") + + async def _forward_results(self, conn): + """转发识别结果""" + try: + while self.asr_ws and 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]: # 连接超时或客户端断开 + logger.bind(tag=TAG).warning(f"连接问题，状态码: {status}") + break + elif status in [40270002, 40270003]: # 音频问题 + logger.bind(tag=TAG).warning(f"音频处理问题，状态码: {status}") + continue + 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).info("服务器已准备，开始发送缓存音频...") + + # 发送缓存音频 + 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 + + 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() + await self.handle_voice_stop(conn, None) + break + elif message_name == "TranscriptionCompleted": + # 识别完成 + self.is_processing = False + break + + except asyncio.TimeoutError: + continue + except websockets.exceptions.ConnectionClosed: + 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() + + async def _cleanup(self): + """清理资源""" + self.is_processing = False + self.server_ready = False # 重置服务器准备状态 + + 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: + pass + self.forward_task = None + + if self.asr_ws: + try: + await asyncio.wait_for(self.asr_ws.close(), timeout=2.0) + except: + pass + self.asr_ws = None + + 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()