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 pull request #2264 from xinnan-tech/py_xunfei_tts

Browse Source 
update:添加讯飞语音识别、大模型、语音合成供应器
This commit is contained in:
hrz
2025-09-23 11:06:00 +08:00
committed by GitHub
parent ef81a17311 86b4e61c1c
commit d28750dae0
7 changed files with 1132 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
main/manager-api/src/main/resources/db/changelog/202509191545.sql
+41
View File
@@ -0,0 +1,41 @@
-- 添加讯飞流式TTS供应器
delete from `ai_model_provider` where id = 'SYSTEM_TTS_XunFeiStreamTTS';
INSERT INTO `ai_model_provider` (`id`, `model_type`, `provider_code`, `name`, `fields`, `sort`, `creator`, `create_date`, `updater`, `update_date`) VALUES
('SYSTEM_TTS_XunFeiStreamTTS', 'TTS', 'xunfei_stream', '讯飞流式语音合成', '[{"key":"app_id","label":"APP_ID","type":"string"},{"key":"api_secret","label":"API_Secret","type":"string"},{"key":"api_key","label":"API密钥","type":"string"},{"key":"output_dir","label":"输出目录","type":"string"},{"key":"voice","label":"音色","type":"string"},{"key":"format","label":"音频格式","type":"string"},{"key":"sample_rate","label":"采样率","type":"number"},{"key": "volume", "type": "number", "label": "音量"},{"key": "speed", "type": "number", "label": "语速"},{"key": "pitch", "type": "number", "label": "音调"},{"key": "oral_level", "type": "number", "label": "口语化等级"},{"key": "spark_assist", "type": "number", "label": "是否口语化"},{"key": "stop_split", "type": "number", "label": "服务端拆句"},{"key": "remain", "type": "number", "label": "保留书面语"}]', 20, 1, NOW(), 1, NOW());
-- 添加讯飞流式TTS模型配置
delete from `ai_model_config` where id = 'TTS_XunFeiStreamTTS';
INSERT INTO `ai_model_config` VALUES ('TTS_XunFeiStreamTTS', 'TTS', 'XunFeiStreamTTS', '讯飞流式语音合成', 0, 1, '{\"type\": \"xunfei_stream\", \"app_id\": \"\", \"api_secret\": \"\", \"api_key\": \"\", \"output_dir\": \"tmp/\", \"voice\": \"x5_lingxiaoxuan_flow\", \"format\": \"raw\", \"sample_rate\": 24000, \"volume\": 50, \"speed\": 50, \"pitch\": 50, \"oral_level\": \"mid\", \"spark_assist\": 1, \"stop_split\": 0, \"remain\": 0}', NULL, NULL, 23, NULL, NULL, NULL, NULL);
-- 更新讯飞流式TTS配置说明
UPDATE `ai_model_config` SET
`doc_link` = 'https://console.xfyun.cn/app/myapp',
`remark` = '讯飞流式TTS说明：
1. 登录讯飞语音技术平台 https://console.xfyun.cn/app/myapp 创建相关应用
2. 选择需要的服务获取api相关配置 https://console.xfyun.cn/services/uts
3. 为需要使用的应用(APPID)购买相关服务 例如：超拟人合成 https://console.xfyun.cn/services/uts
5. 支持实时双流式通信，具有较低的延迟
6. 支持口语化设置和音频参数调整 注意：V5音色不支持相关口语化配置
7. 支持实时调节音量、语速、音调等参数
' WHERE `id` = 'TTS_XunFeiStreamTTS';
-- 添加讯飞流式TTS音色
delete from `ai_tts_voice` where tts_model_id = 'TTS_XunFeiStreamTTS';
-- 基础角色
INSERT INTO `ai_tts_voice` VALUES ('TTS_XunFeiStreamTTS_0001', 'TTS_XunFeiStreamTTS', '聆小璇', 'x5_lingxiaoxuan_flow', '中文', NULL, NULL, NULL, NULL, 1, NULL, NULL, NULL, NULL);
INSERT INTO `ai_tts_voice` VALUES ('TTS_XunFeiStreamTTS_0002', 'TTS_XunFeiStreamTTS', '聆飞逸', 'x5_lingfeiyi_flow', '中文', NULL, NULL, NULL, NULL, 2, NULL, NULL, NULL, NULL);
INSERT INTO `ai_tts_voice` VALUES ('TTS_XunFeiStreamTTS_0003', 'TTS_XunFeiStreamTTS', '聆小玥', 'x5_lingxiaoyue_flow', '中文', NULL, NULL, NULL, NULL, 3, NULL, NULL, NULL, NULL);
INSERT INTO `ai_tts_voice` VALUES ('TTS_XunFeiStreamTTS_0004', 'TTS_XunFeiStreamTTS', '聆玉昭', 'x5_lingyuzhao_flow', '中文', NULL, NULL, NULL, NULL, 4, NULL, NULL, NULL, NULL);
INSERT INTO `ai_tts_voice` VALUES ('TTS_XunFeiStreamTTS_0005', 'TTS_XunFeiStreamTTS', '聆玉言', 'x5_lingyuyan_flow', '中文', NULL, NULL, NULL, NULL, 5, NULL, NULL, NULL, NULL);
-- 需要添加对应的角色音色
INSERT INTO `ai_tts_voice` VALUES ('TTS_XunFeiStreamTTS_0006', 'TTS_XunFeiStreamTTS', '聆飞哲', 'x4_lingfeizhe_oral', '中文', NULL, NULL, NULL, NULL, 6, NULL, NULL, NULL, NULL);
INSERT INTO `ai_tts_voice` VALUES ('TTS_XunFeiStreamTTS_0007', 'TTS_XunFeiStreamTTS', '聆小璃', 'x4_lingxiaoli_oral', '中文', NULL, NULL, NULL, NULL, 7, NULL, NULL, NULL, NULL);
INSERT INTO `ai_tts_voice` VALUES ('TTS_XunFeiStreamTTS_0008', 'TTS_XunFeiStreamTTS', '聆小糖', 'x5_lingxiaotang_flow', '中文', NULL, NULL, NULL, NULL, 8, NULL, NULL, NULL, NULL);
INSERT INTO `ai_tts_voice` VALUES ('TTS_XunFeiStreamTTS_0009', 'TTS_XunFeiStreamTTS', '聆小琪', 'x4_lingxiaoqi_oral', '中文', NULL, NULL, NULL, NULL, 9, NULL, NULL, NULL, NULL);
INSERT INTO `ai_tts_voice` VALUES ('TTS_XunFeiStreamTTS_0010', 'TTS_XunFeiStreamTTS', '聆佑佑-童年女声', 'x4_lingyouyou_oral', '中文', NULL, NULL, NULL, NULL, 10, NULL, NULL, NULL, NULL);
INSERT INTO `ai_tts_voice` VALUES ('TTS_XunFeiStreamTTS_0011', 'TTS_XunFeiStreamTTS', '子津', 'x4_zijin_oral', '天津话', NULL, NULL, NULL, NULL, 11, NULL, NULL, NULL, NULL);
INSERT INTO `ai_tts_voice` VALUES ('TTS_XunFeiStreamTTS_0012', 'TTS_XunFeiStreamTTS', '子阳', 'x4_ziyang_oral', '东北话', NULL, NULL, NULL, NULL, 12, NULL, NULL, NULL, NULL);
INSERT INTO `ai_tts_voice` VALUES ('TTS_XunFeiStreamTTS_0013', 'TTS_XunFeiStreamTTS', 'Grant', 'x5_EnUs_Grant_flow', '英文', NULL, NULL, NULL, NULL, 13, NULL, NULL, NULL, NULL);
INSERT INTO `ai_tts_voice` VALUES ('TTS_XunFeiStreamTTS_0014', 'TTS_XunFeiStreamTTS', 'Lila', 'x5_EnUs_Lila_flow', '英文', NULL, NULL, NULL, NULL, 14, NULL, NULL, NULL, NULL);
main/manager-api/src/main/resources/db/changelog/202509220926.sql
+26
View File
@@ -0,0 +1,26 @@
-- 添加讯飞流式语音识别服务配置
delete from `ai_model_provider` where id = 'SYSTEM_ASR_XunfeiStream';
INSERT INTO `ai_model_provider` (`id`, `model_type`, `provider_code`, `name`, `fields`, `sort`, `creator`, `create_date`, `updater`, `update_date`) VALUES
('SYSTEM_ASR_XunfeiStream', 'ASR', 'xunfei_stream', '讯飞流式语音识别', '[{"key":"app_id","label":"应用ID","type":"string"},{"key":"api_key","label":"API_KEY","type":"password"},{"key":"api_secret","label":"API_SECRET","type":"password"},{"key":"domain","label":"识别领域","type":"string"},{"key":"language","label":"识别语言","type":"string"},{"key":"accent","label":"方言","type":"string"},{"key":"dwa","label":"动态修正","type":"string"},{"key":"output_dir","label":"输出目录","type":"string"}]', 18, 1, NOW(), 1, NOW());
delete from `ai_model_config` where id = 'ASR_XunfeiStream';
INSERT INTO `ai_model_config` VALUES ('ASR_XunfeiStream', 'ASR', '讯飞流式语音识别', '讯飞流式语音识别服务', 0, 1, '{"type": "xunfei_stream", "app_id": "", "api_key": "", "api_secret": "", "domain": "slm", "language": "zh_cn", "accent": "mandarin", "dwa": "wpgs", "output_dir": "tmp/"}', 'https://www.xfyun.cn/doc/spark/spark_zh_iat.html', '支持实时流式语音识别，适用于中文普通话及多种方言识别', 21, NULL, NULL, NULL, NULL);
-- 更新讯飞流式语音识别模型配置的说明文档
UPDATE `ai_model_config` SET
`doc_link` = 'https://www.xfyun.cn/doc/spark/spark_zh_iat.html',
`remark` = '讯飞流式语音识别配置说明：
1. 登录讯飞开放平台 https://www.xfyun.cn/
2. 创建语音识别应用获取APPID、APISecret、APIKey
3. 参数说明：
- app_id: 应用ID，在讯飞开放平台创建应用后获得
- api_key: API密钥，用于接口鉴权
- api_secret: API密钥，用于生成签名
- domain: 识别领域，默认slm（智能化语音转写）
- language: 识别语言，默认zh_cn（中文）
- accent: 方言类型，默认mandarin（普通话），支持cantonese（粤语）等
- dwa: 动态修正，默认wpgs（开启动态修正）
- output_dir: 音频文件输出目录，默认tmp/
4. 支持实时流式识别，适用于实时语音交互场景
5. 支持多种方言和语言识别
' WHERE `id` = 'ASR_XunfeiStream';
main/manager-api/src/main/resources/db/changelog/202509220958.sql
+19
View File
@@ -0,0 +1,19 @@
delete from `ai_model_config` where id = 'LLM_XunfeiSparkLLM';
INSERT INTO `ai_model_config` VALUES ('LLM_XunfeiSparkLLM', 'LLM', '讯飞星火认知大模型', '讯飞星火认知大模型', 0, 1, '{"type": "openai", "model_name": "generalv3.5", "base_url": "https://spark-api-open.xf-yun.com/v1", "api_password": "你的api_password", "temperature": 0.5, "max_tokens": 2048, "top_p": 1.0, "frequency_penalty": 0.0}', 'https://www.xfyun.cn/doc/spark/HTTP%E8%B0%83%E7%94%A8%E6%96%87%E6%A1%A3.html', '讯飞星火认知大模型，支持多轮对话、文本生成等功能', 14, NULL, NULL, NULL, NULL);
-- 更新讯飞星火认知大模型配置的说明文档
UPDATE `ai_model_config` SET
`doc_link` = 'https://www.xfyun.cn/doc/spark/HTTP%E8%B0%83%E7%94%A8%E6%96%87%E6%A1%A3.html',
`remark` = '讯飞星火认知大模型配置说明：
1. 登录讯飞开放平台 https://www.xfyun.cn/，每一个模型对应每一个api_password,更改模型时需要查看对应模型的api_password
2. 创建星火认知大模型应用获取API Password
3. 参数说明：
- api_password: API Password，在讯飞开放平台创建应用后获得
- model_name: 模型名称，支持generalv3.5、generalv3等版本
- base_url: API地址，默认https://spark-api-open.xf-yun.com/v1
- temperature: 温度参数，控制生成随机性，范围0-1，默认0.5
- max_tokens: 最大输出token数，默认2048
- top_p: 核心采样参数，控制词汇多样性，默认1.0
- frequency_penalty: 频率惩罚，降低重复内容，默认0.0
4. 每一个模型对应每一个api_password,更改模型时需要查看对应模型的api_password。
' WHERE `id` = 'LLM_XunfeiSparkLLM';
main/manager-api/src/main/resources/db/changelog/db.changelog-master.yaml
+24
View File
@@ -359,3 +359,27 @@ databaseChangeLog:
- sqlFile:
encoding: utf8
path: classpath:db/changelog/202509161701.sql
- changeSet:
id: 202509191545
author: RanChen
changes:
- sqlFile:
encoding: utf8
path: classpath:db/changelog/202509191545.sql
- changeSet:
id: 202509220926
author: fyb
changes:
- sqlFile:
encoding: utf8
path: classpath:db/changelog/202509220926.sql
- changeSet:
id: 202509220958
author: fyb
changes:
- sqlFile:
encoding: utf8
path: classpath:db/changelog/202509220958.sql
main/xiaozhi-server/config.yaml
+53 -2
View File
@@ -427,7 +427,26 @@ ASR:
enable_itn: true # 逆文本归一化
#language: "zh" # 语种，支持zh、en、ja、ko等
context: "" # 上下文信息，用于提高识别准确率，不超过10000 Token
XunfeiStreamASR:
# 讯飞流式语音识别服务
# 需要先在讯飞开放平台创建应用，获取以下认证信息
# 讯飞开放平台地址：https://www.xfyun.cn/
# 创建应用后，在"我的应用"中获取：
# - APPID
# - APISecret
# - APIKey
type: xunfei_stream
# 必填参数 - 讯飞开放平台应用信息
app_id: 你的APPID
api_key: 你的APIKey
api_secret: 你的APISecret
# 识别参数配置
domain: slm # 识别领域，iat:日常用语，medical:医疗，finance:金融等
language: zh_cn # 语言，zh_cn:中文，en_us:英文
accent: mandarin # 方言，mandarin:普通话
dwa: wpgs # 动态修正，wpgs:实时返回中间结果
# 调整音频处理参数以提高长语音识别质量
output_dir: tmp/
VAD:
SileroVAD:
@@ -585,6 +604,16 @@ VLLM:
url: https://dashscope.aliyuncs.com/compatible-mode/v1
# 可在这里找到你的api key https://bailian.console.aliyun.com/?apiKey=1#/api-key
api_key: 你的api_key
XunfeiSparkLLM:
# 定义LLM API类型
type: openai
# 先新建应用，在下面的地址
# 开通应用地址：https://console.xfyun.cn/app/myapp
# 有免费额度，但也要开通服务，才能获取api_key
# 每一个模型都需要单独开通，每一个模型的api_password都不同，例如Lite模型在https://console.xfyun.cn/services/cbm 开通
base_url: https://ark.cn-beijing.volces.com/api/v3
model_name: lite
api_key: 你的api_password
TTS:
# 当前支持的type为edge、doubao，可自行适配
EdgeTTS:
@@ -936,4 +965,26 @@ TTS:
# sample_rate: 24000 # 采样率：16000, 24000, 48000
# volume: 50 # 音量：0-100
# rate: 1 # 语速：0.5~2
# pitch: 1 # 语调：0.5~2
# pitch: 1 # 语调：0.5~2
XunFeiTTS:
# 讯飞TTS服务 官方网站：https://www.xfyun.cn/
# 登录讯飞语音技术平台 https://console.xfyun.cn/app/myapp 创建相关应用
# 选择需要的服务获取api相关配置 https://console.xfyun.cn/services/uts
# 为需要使用的应用(APPID)购买相关服务 例如：超拟人合成 https://console.xfyun.cn/services/uts
type: xunfei_stream
api_url: wss://cbm01.cn-huabei-1.xf-yun.com/v1/private/mcd9m97e6
app_id: 你的app_id
api_secret: 你的api_secret
api_key: 你的api_key
voice: x5_lingxiaoxuan_flow
output_dir: tmp/
# 以下可不用设置，使用默认设置，注意V5音色不支持口语化配置
# oral_level: mid # 口语化等级：high, mid, low
# spark_assist: 1 # 是否通过大模型进行口语化 开启:1, 关闭:0
# stop_split: 0 # 关闭服务端拆句 不关闭：0，关闭：1
# remain: 0 # 是否保留原书面语的样子 保留:1, 不保留:0
# format: raw # 音频格式：raw(PCM), lame(MP3), speex, opus, opus-wb, opus-swb, speex-wb
# sample_rate: 24000 # 采样率：16000, 8000, 24000
# volume: 50 # 音量：0-100
# speed: 50 # 语速：0-100
# pitch: 50 # 语调：0-100
main/xiaozhi-server/core/providers/asr/xunfei_stream.py
+442
View File
@@ -0,0 +1,442 @@
import json
import hmac
import base64
import hashlib
import asyncio
import websockets
import opuslib_next
from time import mktime
from datetime import datetime
from urllib.parse import urlencode
from typing import List
from config.logger import setup_logging
from wsgiref.handlers import format_date_time
from core.providers.asr.base import ASRProviderBase
from core.providers.asr.dto.dto import InterfaceType
TAG = __name__
logger = setup_logging()
# 帧状态常量
STATUS_FIRST_FRAME = 0 # 第一帧的标识
STATUS_CONTINUE_FRAME = 1 # 中间帧标识
STATUS_LAST_FRAME = 2 # 最后一帧的标识
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.last_frame_sent = False # 标记是否已发送最终帧
self.best_text = "" # 保存最佳识别结果
# 讯飞配置
self.app_id = config.get("app_id")
self.api_key = config.get("api_key")
self.api_secret = config.get("api_secret")
if not all([self.app_id, self.api_key, self.api_secret]):
raise ValueError("必须提供app_id、api_key和api_secret")
# 识别参数
self.iat_params = {
"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"
}
}
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'
# 生成RFC1123格式的时间戳
now = datetime.now()
date = format_date_time(mktime(now.timetuple()))
# 拼接字符串
signature_origin = "host: " + "iat.cn-huabei-1.xf-yun.com" + "\n"
signature_origin += "date: " + date + "\n"
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')
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"
}
# 拼接鉴权参数，生成url
url = url + '?' + urlencode(v)
return url
async def open_audio_channels(self, conn):
await super().open_audio_channels(conn)
async def receive_audio(self, conn, audio, audio_have_voice):
# 先调用父类方法处理基础逻辑
await super().receive_audio(conn, audio, audio_have_voice)
# 存储音频数据用于声纹识别
if not hasattr(conn, 'asr_audio_for_voiceprint'):
conn.asr_audio_for_voiceprint = []
conn.asr_audio_for_voiceprint.append(audio)
# 如果本次有声音，且之前没有建立连接
if audio_have_voice and self.asr_ws is None and not self.is_processing:
try:
await self._start_recognition(conn)
except Exception as e:
logger.bind(tag=TAG).error(f"建立ASR连接失败: {str(e)}")
await self._cleanup(conn)
return
# 发送当前音频数据
if self.asr_ws and self.is_processing and self.server_ready:
try:
pcm_frame = self.decoder.decode(audio, 960)
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)
async def _start_recognition(self, conn):
"""开始识别会话"""
try:
self.is_processing = True
# 建立WebSocket连接
ws_url = self.create_url()
logger.bind(tag=TAG).info(f"正在连接ASR服务: {ws_url[:50]}...")
self.asr_ws = await websockets.connect(
ws_url,
max_size=1000000000,
ping_interval=None,
ping_timeout=None,
close_timeout=10,
)
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))
# 发送首帧音频
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''
await self._send_audio_frame(pcm_frame, STATUS_FIRST_FRAME)
self.server_ready = True
logger.bind(tag=TAG).info("已发送首帧，开始识别")
# 发送缓存的音频数据
for cached_audio in conn.asr_audio[-10:]:
try:
pcm_frame = self.decoder.decode(cached_audio, 960)
await self._send_audio_frame(pcm_frame, STATUS_CONTINUE_FRAME)
except Exception as e:
logger.bind(tag=TAG).info(f"发送缓存音频数据时发生错误: {e}")
break
except Exception as e:
logger.bind(tag=TAG).error(f"建立ASR连接失败: {str(e)}")
if hasattr(e, "__cause__") and e.__cause__:
logger.bind(tag=TAG).error(f"错误原因: {str(e.__cause__)}")
if self.asr_ws:
await self.asr_ws.close()
self.asr_ws = None
self.is_processing = False
raise
async def _send_audio_frame(self, audio_data: bytes, status: int):
"""发送音频帧"""
if not self.asr_ws:
return
audio_b64 = base64.b64encode(audio_data).decode('utf-8')
frame_data = {
"header": {
"status": status,
"app_id": self.app_id
},
"parameter": {
"iat": self.iat_params
},
"payload": {
"audio": {
"audio": audio_b64,
"sample_rate": 16000,
"encoding": "raw"
}
}
}
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', [])
try:
# 如果已发送最终帧，增加超时时间等待完整结果
timeout = 3.0 if self.last_frame_sent else 30.0
response = await asyncio.wait_for(self.asr_ws.recv(), timeout=timeout)
result = json.loads(response)
logger.bind(tag=TAG).debug(f"收到ASR结果: {result}")
header = result.get("header", {})
payload = result.get("payload", {})
code = header.get("code", 0)
status = header.get("status", 0)
if code != 0:
logger.bind(tag=TAG).error(f"识别错误，错误码: {code}, 消息: {header.get('message', '')}")
if code in [10114, 10160]: # 连接问题
break
continue
# 处理识别结果
if payload and "result" in payload:
text_data = payload["result"]["text"]
if text_data:
# 解码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
# 更新识别文本 - 实时更新策略
if result_text and result_text.strip() not in ['', '。', '.', ',', '，']:
# 实时更新：正常情况下都更新，提高响应速度
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}")
# 如果已发送最终帧，只过滤明显的无效结果
if self.last_frame_sent:
# 最终帧后拒绝问号等明显错误的结果
if result_text.strip() in ['？', '?', '。', '.']:
should_update = False
logger.bind(tag=TAG).warning(f"最终帧后拒绝无效文本: {result_text}")
if should_update:
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 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}")
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
except websockets.ConnectionClosed:
logger.bind(tag=TAG).info("ASR服务连接已关闭")
self.is_processing = False
break
except Exception as e:
logger.bind(tag=TAG).error(f"处理ASR结果时发生错误: {str(e)}")
if hasattr(e, "__cause__") and e.__cause__:
logger.bind(tag=TAG).error(f"错误原因: {str(e.__cause__)}")
self.is_processing = False
break
except Exception as e:
logger.bind(tag=TAG).error(f"ASR结果转发任务发生错误: {str(e)}")
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
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]):
"""处理语音停止，发送最后一帧并处理识别结果"""
try:
# 先发送最后一帧表示音频结束
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("已发送最后一帧")
# 发送最终帧后，给_forward_results适当时间处理最终结果
# 减少等待时间，提高响应速度
await asyncio.sleep(0.1) # 从3秒减少到1秒
logger.bind(tag=TAG).info(f"准备处理最终识别结果: {self.text}")
except Exception as 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}")
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())
self.asr_ws = None
self.is_processing = False
async def _cleanup(self, conn):
"""清理资源"""
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 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 = ""
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
# 关闭连接
if self.asr_ws:
try:
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
# 清理连接的音频缓存
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
logger.bind(tag=TAG).info("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):
"""资源清理方法"""
if self.asr_ws:
await self.asr_ws.close()
self.asr_ws = None
if self.forward_task:
self.forward_task.cancel()
try:
await self.forward_task
except asyncio.CancelledError:
pass
self.forward_task = None
self.is_processing = False
# 清理所有连接的音频缓冲区
if hasattr(self, '_connections'):
for conn in self._connections.values():
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
main/xiaozhi-server/core/providers/tts/xunfei_stream.py
+527
View File
@@ -0,0 +1,527 @@
import os
import time
import uuid
import json
import hmac
import queue
import base64
import hashlib
import asyncio
import traceback
import websockets
from asyncio import Task
from config.logger import setup_logging
from core.utils import opus_encoder_utils
from core.utils.tts import MarkdownCleaner
from urllib.parse import urlencode, urlparse
from core.providers.tts.base import TTSProviderBase
from core.providers.tts.dto.dto import SentenceType, ContentType, InterfaceType
TAG = __name__
logger = setup_logging()
class XunfeiWSAuth:
@staticmethod
def create_auth_url(api_key, api_secret, api_url):
"""生成讯飞WebSocket认证URL"""
parsed_url = urlparse(api_url)
host = parsed_url.netloc
path = parsed_url.path
# 获取UTC时间，讯飞要求使用RFC1123格式
now = time.gmtime()
date = time.strftime('%a, %d %b %Y %H:%M:%S GMT', now)
# 构造签名字符串
signature_origin = f"host: {host}\ndate: {date}\nGET {path} HTTP/1.1"
# 计算签名
signature_sha = hmac.new(
api_secret.encode('utf-8'),
signature_origin.encode('utf-8'),
digestmod=hashlib.sha256
).digest()
signature_sha_base64 = base64.b64encode(signature_sha).decode(encoding='utf-8')
# 构造authorization
authorization_origin = f'api_key="{api_key}", algorithm="hmac-sha256", headers="host date request-line", signature="{signature_sha_base64}"'
authorization = base64.b64encode(authorization_origin.encode('utf-8')).decode(encoding='utf-8')
# 构造最终的WebSocket URL
v = {
"authorization": authorization,
"date": date,
"host": host
}
url = api_url + '?' + urlencode(v)
return url
class TTSProvider(TTSProviderBase):
def __init__(self, config, delete_audio_file):
super().__init__(config, delete_audio_file)
# 设置为流式接口类型
self.interface_type = InterfaceType.DUAL_STREAM
# 基础配置
self.app_id = config.get("app_id")
self.api_key = config.get("api_key")
self.api_secret = config.get("api_secret")
# 接口地址
self.api_url = config.get("api_url", "wss://cbm01.cn-huabei-1.xf-yun.com/v1/private/mcd9m97e6")
# 音色配置
self.voice = config.get("voice", "x5_lingxiaoxuan_flow")
if config.get("private_voice"):
self.voice = config.get("private_voice")
# 音频参数配置
speed = config.get("speed", "50")
self.speed = int(speed) if speed else 50
volume = config.get("volume", "50")
self.volume = int(volume) if volume else 50
pitch = config.get("pitch", "50")
self.pitch = int(pitch) if pitch else 50
# 音频编码配置
self.format = config.get("format", "raw")
sample_rate = config.get("sample_rate", "24000")
self.sample_rate = int(sample_rate) if sample_rate else 24000
# 口语化配置
self.oral_level = config.get("oral_level", "mid")
spark_assist = config.get("spark_assist", "1")
self.spark_assist = int(spark_assist) if spark_assist else 1
stop_split = config.get("stop_split", "0")
self.stop_split = int(stop_split) if stop_split else 0
remain = config.get("remain", "0")
self.remain = int(remain) if remain else 0
# WebSocket配置
self.ws = None
self._monitor_task = None
# 序列号管理
self.text_seq = 0
# 创建Opus编码器
self.opus_encoder = opus_encoder_utils.OpusEncoderUtils(
sample_rate=self.sample_rate, channels=1, frame_size_ms=60
)
# 验证必需参数
if not all([self.app_id, self.api_key, self.api_secret]):
raise ValueError("讯飞TTS需要配置app_id、api_key和api_secret")
async def _ensure_connection(self):
"""确保WebSocket连接可用"""
try:
logger.bind(tag=TAG).info("开始建立新连接...")
# 生成认证URL
auth_url = XunfeiWSAuth.create_auth_url(
self.api_key, self.api_secret, self.api_url
)
self.ws = await websockets.connect(
auth_url,
ping_interval=30,
ping_timeout=10,
close_timeout=10,
)
logger.bind(tag=TAG).info("WebSocket连接建立成功")
return self.ws
except Exception as e:
logger.bind(tag=TAG).error(f"建立连接失败: {str(e)}")
self.ws = None
raise
def tts_text_priority_thread(self):
"""流式文本处理线程"""
while not self.conn.stop_event.is_set():
try:
message = self.tts_text_queue.get(timeout=1)
logger.bind(tag=TAG).debug(
f"收到TTS任务｜{message.sentence_type.name} ｜ {message.content_type.name} | 会话ID: {self.conn.sentence_id}"
)
if message.sentence_type == SentenceType.FIRST:
# 重置序列号
self.text_seq = 0
self.conn.client_abort = False
# 增加序列号
self.text_seq += 1
if self.conn.client_abort:
logger.bind(tag=TAG).info("收到打断信息，终止TTS文本处理线程")
continue
if message.sentence_type == SentenceType.FIRST:
# 初始化参数
try:
if not getattr(self.conn, "sentence_id", None):
self.conn.sentence_id = uuid.uuid4().hex
logger.bind(tag=TAG).info(f"自动生成新的 会话ID: {self.conn.sentence_id}")
logger.bind(tag=TAG).info("开始启动TTS会话...")
future = asyncio.run_coroutine_threadsafe(
self.start_session(self.conn.sentence_id),
loop=self.conn.loop,
)
future.result()
self.before_stop_play_files.clear()
logger.bind(tag=TAG).info("TTS会话启动成功")
except Exception as e:
logger.bind(tag=TAG).error(f"启动TTS会话失败: {str(e)}")
continue
# 处理文本内容
if ContentType.TEXT == message.content_type:
if message.content_detail:
try:
logger.bind(tag=TAG).debug(
f"开始发送TTS文本: {message.content_detail}"
)
future = asyncio.run_coroutine_threadsafe(
self.text_to_speak(message.content_detail, None),
loop=self.conn.loop,
)
future.result()
logger.bind(tag=TAG).debug("TTS文本发送成功")
except Exception as e:
logger.bind(tag=TAG).error(f"发送TTS文本失败: {str(e)}")
# 不使用continue，确保后续处理不被中断
# 处理文件内容
if ContentType.FILE == message.content_type:
logger.bind(tag=TAG).info(
f"添加音频文件到待播放列表: {message.content_file}"
)
if message.content_file and os.path.exists(message.content_file):
# 先处理文件音频数据
self._process_audio_file_stream(message.content_file, callback=lambda audio_data: self.handle_audio_file(audio_data, message.content_detail))
# 处理会话结束
if message.sentence_type == SentenceType.LAST:
try:
logger.bind(tag=TAG).info("开始结束TTS会话...")
asyncio.run_coroutine_threadsafe(
self.finish_session(self.conn.sentence_id),
loop=self.conn.loop,
)
except Exception as e:
logger.bind(tag=TAG).error(f"结束TTS会话失败: {str(e)}")
continue
except queue.Empty:
continue
except Exception as e:
logger.bind(tag=TAG).error(
f"处理TTS文本失败: {str(e)}, 类型: {type(e).__name__}, 堆栈: {traceback.format_exc()}"
)
async def text_to_speak(self, text, _):
"""发送文本到TTS服务进行合成"""
try:
if self.ws is None:
logger.bind(tag=TAG).warning(f"WebSocket连接不存在，终止发送文本")
return
filtered_text = MarkdownCleaner.clean_markdown(text)
# 发送文本合成请求
run_request = self._build_base_request(status=1,text=filtered_text)
await self.ws.send(json.dumps(run_request))
return
except Exception as e:
logger.bind(tag=TAG).error(f"发送TTS文本失败: {str(e)}")
if self.ws:
try:
await self.ws.close()
except:
pass
self.ws = None
raise
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(
"检测到未完成的上个会话，关闭监听任务和连接..."
)
await self.close()
# 建立新连接
await self._ensure_connection()
# 启动监听任务
self._monitor_task = asyncio.create_task(self._start_monitor_tts_response())
# 发送会话启动请求
start_request = self._build_base_request(status=0)
await self.ws.send(json.dumps(start_request))
logger.bind(tag=TAG).info("会话启动请求已发送")
except Exception as e:
logger.bind(tag=TAG).error(f"启动会话失败: {str(e)}")
# 确保清理资源
await self.close()
raise
async def finish_session(self, session_id):
logger.bind(tag=TAG).info(f"关闭会话～～{session_id}")
try:
if self.ws:
# 发送会话结束请求
stop_request = self._build_base_request(status=2)
await self.ws.send(json.dumps(stop_request))
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)}")
await self.close()
raise
async def close(self):
"""资源清理"""
if self._monitor_task:
try:
self._monitor_task.cancel()
await self._monitor_task
except asyncio.CancelledError:
pass
except Exception as e:
logger.bind(tag=TAG).warning(f"关闭时取消监听任务错误: {e}")
self._monitor_task = None
if self.ws:
try:
await self.ws.close()
except:
pass
self.ws = None
async def _start_monitor_tts_response(self):
"""监听TTS响应"""
try:
while not self.conn.stop_event.is_set():
try:
msg = await self.ws.recv()
# 检查客户端是否中止
if self.conn.client_abort:
logger.bind(tag=TAG).info("收到打断信息，终止监听TTS响应")
break
try:
data = json.loads(msg)
header = data.get("header", {})
code = header.get("code")
if code == 0:
payload = data.get("payload", {})
audio_payload = payload.get("audio", {})
if audio_payload:
status = audio_payload.get("status", 0)
audio_data = audio_payload.get("audio", "")
if status == 0:
logger.bind(tag=TAG).debug("TTS合成已启动")
self.tts_audio_queue.put(
(SentenceType.FIRST, [], None)
)
elif status == 2:
logger.bind(tag=TAG).debug("收到结束状态的音频数据，TTS合成完成")
self._process_before_stop_play_files()
break
else:
if self.conn.tts_MessageText:
logger.bind(tag=TAG).info(
f"句子语音生成成功： {self.conn.tts_MessageText}"
)
self.tts_audio_queue.put(
(SentenceType.FIRST, [], self.conn.tts_MessageText)
)
self.conn.tts_MessageText = None
try:
audio_bytes = base64.b64decode(audio_data)
self.opus_encoder.encode_pcm_to_opus_stream(
audio_bytes, False, self.handle_opus
)
except Exception as e:
logger.bind(tag=TAG).error(f"处理音频数据失败: {e}")
else:
message = header.get("message", "未知错误")
logger.bind(tag=TAG).error(f"TTS合成错误: {code} - {message}")
break
except json.JSONDecodeError:
logger.bind(tag=TAG).warning("收到无效的JSON消息")
except websockets.ConnectionClosed:
logger.bind(tag=TAG).warning("WebSocket连接已关闭")
break
except Exception as e:
logger.bind(tag=TAG).error(
f"处理TTS响应时出错: {e}\n{traceback.format_exc()}"
)
break
# 链接不可复用
if self.ws:
try:
await self.ws.close()
except:
pass
self.ws = None
# 监听任务退出时清理引用
finally:
self._monitor_task = None
def to_tts(self, text: str) -> list:
"""非流式TTS处理，用于测试及保存音频文件的场景"""
try:
# 创建新的事件循环
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
# 存储音频数据
audio_data = []
async def _generate_audio():
# 生成认证URL
auth_url = XunfeiWSAuth.create_auth_url(
self.api_key, self.api_secret, self.api_url
)
# 建立WebSocket连接
ws = await websockets.connect(
auth_url,
ping_interval=30,
ping_timeout=10,
close_timeout=10,
)
try:
filtered_text = MarkdownCleaner.clean_markdown(text)
text_request = self._build_base_request(status=2,text=filtered_text)
await ws.send(json.dumps(text_request))
task_finished = False
while not task_finished:
msg = await ws.recv()
data = json.loads(msg)
header = data.get("header", {})
code = header.get("code")
if code == 0:
payload = data.get("payload", {})
audio_payload = payload.get("audio", {})
if audio_payload:
status = audio_payload.get("status", 0)
audio_base64 = audio_payload.get("audio", "")
if status == 1:
try:
audio_bytes = base64.b64decode(audio_base64)
self.opus_encoder.encode_pcm_to_opus_stream(
audio_bytes,
end_of_stream=False,
callback=lambda opus: audio_data.append(opus)
)
except Exception as e:
logger.bind(tag=TAG).error(f"处理音频数据失败: {e}")
elif status == 2:
task_finished = True
logger.bind(tag=TAG).debug("TTS任务完成")
else:
message = header.get("message", "未知错误")
raise Exception(f"合成失败: {code} - {message}")
finally:
# 清理资源
try:
await ws.close()
except:
pass
loop.run_until_complete(_generate_audio())
loop.close()
return audio_data
except Exception as e:
logger.bind(tag=TAG).error(f"生成音频数据失败: {str(e)}")
return []
def _build_base_request(self, status,text=" "):
"""构建基础请求结构"""
return {
"header": {
"app_id": self.app_id,
"status": status,
},
"parameter": {
"oral": {
"oral_level": self.oral_level,
"spark_assist": self.spark_assist,
"stop_split": self.stop_split,
"remain": self.remain
},
"tts": {
"vcn": self.voice,
"speed": self.speed,
"volume": self.volume,
"pitch": self.pitch,
"bgs": 0,
"reg": 0,
"rdn": 0,
"rhy": 0,
"audio": {
"encoding": self.format,
"sample_rate": self.sample_rate,
"channels": 1,
"bit_depth": 16,
"frame_size": 0
}
}
},
"payload": {
"text": {
"encoding": "utf8",
"compress": "raw",
"format": "plain",
"status": status,
"seq": self.text_seq,
"text": base64.b64encode(text.encode('utf-8')).decode('utf-8')
}
}
}