Hass/xiaozhi-esp32-server
1
0
Fork 0
mirror of https://github.com/xinnan-tech/xiaozhi-esp32-server.git synced 2026-07-24 16:13:54 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'main' into py_Text

Browse Source 
# Conflicts:
#	main/xiaozhi-server/core/handle/helloHandle.py
#	main/xiaozhi-server/core/providers/tts/base.py
This commit is contained in:
hrz
2025-06-07 22:48:23 +08:00
parent 50e6e4817b 6023af2f21
commit 29d90df9ea
37 changed files with 1968 additions and 581 deletions
Download Patch File Download Diff File Expand all files Collapse all files
main/manager-api/src/main/java/xiaozhi/common/constant/Constant.java
+1 -1
View File
@@ -227,7 +227,7 @@ public interface Constant {
/**
* 版本号
*/
public static final String VERSION = "0.5.2";
public static final String VERSION = "0.5.4";
/**
* 无效固件URL
main/manager-api/src/main/java/xiaozhi/modules/agent/controller/AgentController.java
+3
View File
@@ -171,6 +171,9 @@ public class AgentController {
if (dto.getLlmModelId() != null) {
existingEntity.setLlmModelId(dto.getLlmModelId());
}
if (dto.getVllmModelId() != null) {
existingEntity.setVllmModelId(dto.getVllmModelId());
}
if (dto.getTtsModelId() != null) {
existingEntity.setTtsModelId(dto.getTtsModelId());
}
main/manager-api/src/main/java/xiaozhi/modules/agent/dto/AgentUpdateDTO.java
+3
View File
@@ -30,6 +30,9 @@ public class AgentUpdateDTO implements Serializable {
@Schema(description = "大语言模型标识", example = "llm_model_02", required = false)
private String llmModelId;
@Schema(description = "VLLM模型标识", example = "vllm_model_02", required = false)
private String vllmModelId;
@Schema(description = "语音合成模型标识", example = "tts_model_02", required = false)
private String ttsModelId;
main/manager-api/src/main/resources/db/changelog/202506031639.sql
+45
View File
@@ -0,0 +1,45 @@
-- VLLM模型供应器
delete from `ai_model_provider` where id = 'SYSTEM_ASR_DoubaoStreamASR';
INSERT INTO `ai_model_provider` (`id`, `model_type`, `provider_code`, `name`, `fields`, `sort`, `creator`, `create_date`, `updater`, `update_date`) VALUES
('SYSTEM_ASR_DoubaoStreamASR', 'ASR', 'doubao_stream', '火山引擎语音识别(流式)', '[{"key":"appid","label":"应用ID","type":"string"},{"key":"access_token","label":"访问令牌","type":"string"},{"key":"cluster","label":"集群","type":"string"},{"key":"boosting_table_name","label":"热词文件名称","type":"string"},{"key":"correct_table_name","label":"替换词文件名称","type":"string"},{"key":"output_dir","label":"输出目录","type":"string"}]', 3, 1, NOW(), 1, NOW());
-- VLLM模型配置
delete from `ai_model_config` where id = 'ASR_DoubaoStreamASR';
INSERT INTO `ai_model_config` VALUES ('ASR_DoubaoStreamASR', 'ASR', 'DoubaoStreamASR', '豆包语音识别(流式)', 0, 1, '{\"type\": \"doubao_stream\", \"appid\": \"\", \"access_token\": \"\", \"cluster\": \"volcengine_input_common\", \"output_dir\": \"tmp/\"}', NULL, NULL, 3, NULL, NULL, NULL, NULL);
-- 更新豆包ASR配置说明
UPDATE `ai_model_config` SET
`doc_link` = 'https://console.volcengine.com/speech/app',
`remark` = '豆包ASR配置说明：
1. 豆包ASR和豆包(流式)ASR的区别是：豆包ASR是按次收费，豆包(流式)ASR是按时收费
2. 一般来说按次收费的更便宜，但是豆包(流式)ASR使用了大模型技术，效果更好
3. 需要在火山引擎控制台创建应用并获取appid和access_token
4. 支持中文语音识别
5. 需要网络连接
6. 输出文件保存在tmp/目录
申请步骤：
1. 访问 https://console.volcengine.com/speech/app
2. 创建新应用
3. 获取appid和access_token
4. 填入配置文件中
如需设置热词，请参考：https://www.volcengine.com/docs/6561/155738
' WHERE `id` = 'ASR_DoubaoASR';
UPDATE `ai_model_config` SET
`doc_link` = 'https://console.volcengine.com/speech/app',
`remark` = '豆包ASR配置说明：
1. 豆包ASR和豆包(流式)ASR的区别是：豆包ASR是按次收费，豆包(流式)ASR是按时收费
2. 一般来说按次收费的更便宜，但是豆包(流式)ASR使用了大模型技术，效果更好
3. 需要在火山引擎控制台创建应用并获取appid和access_token
4. 支持中文语音识别
5. 需要网络连接
6. 输出文件保存在tmp/目录
申请步骤：
1. 访问 https://console.volcengine.com/speech/app
2. 创建新应用
3. 获取appid和access_token
4. 填入配置文件中
如需设置热词，请参考：https://www.volcengine.com/docs/6561/155738
' WHERE `id` = 'ASR_DoubaoStreamASR';
main/manager-api/src/main/resources/db/changelog/202506032232.sql
+14
View File
@@ -0,0 +1,14 @@
-- VLLM模型配置
delete from `ai_model_config` where id = 'VLLM_QwenVLVLLM';
INSERT INTO `ai_model_config` VALUES ('VLLM_QwenVLVLLM', 'VLLM', 'QwenVLVLLM', '千问视觉模型', 0, 1, '{\"type\": \"openai\", \"model_name\": \"qwen2.5-vl-3b-instruct\", \"base_url\": \"https://dashscope.aliyuncs.com/compatible-mode/v1\", \"api_key\": \"你的api_key\"}', NULL, NULL, 2, NULL, NULL, NULL, NULL);
-- 更新文档
UPDATE `ai_model_config` SET
`doc_link` = 'https://bailian.console.aliyun.com/?tab=api#/api/?type=model&url=https%3A%2F%2Fhelp.aliyun.com%2Fdocument_detail%2F2845564.html&renderType=iframe',
`remark` = '千问视觉模型配置说明：
1. 访问 https://bailian.console.aliyun.com/?tab=model#/api-key
2. 注册并获取API密钥
3. 填入配置文件中' WHERE `id` = 'VLLM_QwenVLVLLM';
-- 删除参数，这两个参数已挪至python配置文件
delete from `sys_params` where id in (113,114);
main/manager-api/src/main/resources/db/changelog/202506051538.sql
+20
View File
@@ -0,0 +1,20 @@
-- 增加LinkeraiTTS供应器和模型配置
delete from `ai_model_provider` where id = 'SYSTEM_TTS_LinkeraiTTS';
INSERT INTO `ai_model_provider` (`id`, `model_type`, `provider_code`, `name`, `fields`, `sort`, `creator`, `create_date`, `updater`, `update_date`) VALUES
('SYSTEM_TTS_LinkeraiTTS', 'TTS', 'linkerai', 'Linkerai语音合成', '[{"key":"api_url","label":"API地址","type":"string"},{"key":"audio_format","label":"音频格式","type":"string"},{"key":"access_token","label":"访问令牌","type":"string"},{"key":"voice","label":"默认音色","type":"string"}]', 14, 1, NOW(), 1, NOW());
delete from `ai_model_config` where id = 'TTS_LinkeraiTTS';
INSERT INTO `ai_model_config` VALUES ('TTS_LinkeraiTTS', 'TTS', 'LinkeraiTTS', 'Linkerai语音合成', 0, 1, '{\"type\": \"linkerai\", \"api_url\": \"https://tts.linkerai.cn/tts\", \"audio_format\": \"pcm\", \"access_token\": \"U4YdYXVfpwWnk2t5Gp822zWPCuORyeJL\", \"voice\": \"OUeAo1mhq6IBExi\"}', NULL, NULL, 17, NULL, NULL, NULL, NULL);
-- LinkeraiTTS模型配置说明文档
UPDATE `ai_model_config` SET
`doc_link` = 'https://tts.linkerai.cn/docs',
`remark` = 'Linkerai语音合成服务配置说明：
1. 访问 https://linkerai.cn 注册并获取访问令牌
2. 默认的access_token供测试使用，请勿用于商业用途
3. 支持声音克隆功能，可自行上传音频，填入voice参数
4. 如果voice参数为空，将使用默认声音' WHERE `id` = 'TTS_LinkeraiTTS';
delete from `ai_tts_voice` where tts_model_id = 'TTS_LinkeraiTTS';
INSERT INTO `ai_tts_voice` VALUES ('TTS_LinkeraiTTS_0001', 'TTS_LinkeraiTTS', '芷若', 'OUeAo1mhq6IBExi', '中文', NULL, NULL, 1, NULL, NULL, NULL, NULL);
main/manager-api/src/main/resources/db/changelog/db.changelog-master.yaml
+22 -1
View File
@@ -176,4 +176,25 @@ databaseChangeLog:
changes:
- sqlFile:
encoding: utf8
path: classpath:db/changelog/202506010920.sql
path: classpath:db/changelog/202506010920.sql
- changeSet:
id: 202506031639
author: hrz
changes:
- sqlFile:
encoding: utf8
path: classpath:db/changelog/202506031639.sql
- changeSet:
id: 202506032232
author: hrz
changes:
- sqlFile:
encoding: utf8
path: classpath:db/changelog/202506032232.sql
- changeSet:
id: 202506051538
author: hrz
changes:
- sqlFile:
encoding: utf8
path: classpath:db/changelog/202506051538.sql
main/manager-web/src/views/ModelConfig.vue
+2 -2
View File
@@ -31,7 +31,7 @@
<span class="menu-text">大语言模型</span>
</el-menu-item>
<el-menu-item index="vllm">
<span class="menu-text">视觉大语言模型</span>
<span class="menu-text">视觉大模型</span>
</el-menu-item>
<el-menu-item index="intent">
<span class="menu-text">意图识别</span>
@@ -176,7 +176,7 @@ export default {
vad: '语言活动检测模型(VAD)',
asr: '语音识别模型(ASR)',
llm: '大语言模型（LLM）',
vllm: '视觉大语言模型（VLLM）',
vllm: '视觉大模型（VLLM）',
intent: '意图识别模型(Intent)',
tts: '语音合成模型(TTS)',
memory: '记忆模型(Memory)'
main/manager-web/src/views/roleConfig.vue
+1 -1
View File
@@ -177,7 +177,7 @@ export default {
{ label: '语音活动检测(VAD)', key: 'vadModelId', type: 'VAD' },
{ label: '语音识别(ASR)', key: 'asrModelId', type: 'ASR' },
{ label: '大语言模型(LLM)', key: 'llmModelId', type: 'LLM' },
{ label: '视觉大语言模型(VLLM)', key: 'vllmModelId', type: 'VLLM' },
{ label: '视觉大模型(VLLM)', key: 'vllmModelId', type: 'VLLM' },
{ label: '意图识别(Intent)', key: 'intentModelId', type: 'Intent' },
{ label: '记忆(Memory)', key: 'memModelId', type: 'Memory' },
{ label: '语音合成(TTS)', key: 'ttsModelId', type: 'TTS' },
main/xiaozhi-server/config.yaml
+34 -1
View File
@@ -264,6 +264,8 @@ ASR:
DoubaoASR:
# 可以在这里申请相关Key等信息
# https://console.volcengine.com/speech/app
# DoubaoASR和DoubaoStreamASR的区别是：DoubaoASR是按次收费，DoubaoStreamASR是按时收费
# 一般来说按次收费的更便宜，但是DoubaoStreamASR使用了大模型技术，效果更好
type: doubao
appid: 你的火山引擎语音合成服务appid
access_token: 你的火山引擎语音合成服务access_token
@@ -272,6 +274,20 @@ ASR:
boosting_table_name: （选填）你的热词文件名称
correct_table_name: （选填）你的替换词文件名称
output_dir: tmp/
DoubaoStreamASR:
# 可以在这里申请相关Key等信息
# https://console.volcengine.com/speech/app
# DoubaoASR和DoubaoStreamASR的区别是：DoubaoASR是按次收费，DoubaoStreamASR是按时收费
# 开通地址https://console.volcengine.com/speech/service/10011
# 一般来说按次收费的更便宜，但是DoubaoStreamASR使用了大模型技术，效果更好
type: doubao_stream
appid: 你的火山引擎语音合成服务appid
access_token: 你的火山引擎语音合成服务access_token
cluster: volcengine_input_common
# 热词、替换词使用流程：https://www.volcengine.com/docs/6561/155738
boosting_table_name: （选填）你的热词文件名称
correct_table_name: （选填）你的替换词文件名称
output_dir: tmp/
TencentASR:
# token申请地址：https://console.cloud.tencent.com/cam/capi
# 免费领取资源：https://console.cloud.tencent.com/asr/resourcebundle
@@ -308,7 +324,7 @@ VAD:
type: silero
threshold: 0.5
model_dir: models/snakers4_silero-vad
min_silence_duration_ms: 700 # 如果说话停顿比较长，可以把这个值设置大一些
min_silence_duration_ms: 200 # 如果说话停顿比较长，可以把这个值设置大一些
LLM:
# 所有openai类型均可以修改超参，以AliLLM为例
@@ -452,6 +468,12 @@ VLLM:
model_name: glm-4v-flash # 智谱AI的视觉模型
url: https://open.bigmodel.cn/api/paas/v4/
api_key: 你的api_key
QwenVLVLLM:
type: openai
model_name: qwen2.5-vl-3b-instruct
url: https://dashscope.aliyuncs.com/compatible-mode/v1
# 可在这里找到你的api key https://bailian.console.aliyun.com/?apiKey=1#/api-key
api_key: 你的api_key
TTS:
# 当前支持的type为edge、doubao，可自行适配
EdgeTTS:
@@ -733,4 +755,15 @@ TTS:
headers: # 自定义请求头
# Authorization: Bearer xxxx
format: mp3 # 接口返回的音频格式
output_dir: tmp/
LinkeraiTTS:
type: linkerai
api_url: https://tts.linkerai.cn/tts
audio_format: "pcm"
# 默认的access_token供大家测试时免费使用的，此access_token请勿用于商业用途
# 如果效果不错，可自行申请token，申请地址：https://linkerai.cn
# 各参数意义见开发文档：https://tts.linkerai.cn/docs
# 支持声音克隆，可自行上传音频，填入voice参数，voice参数为空时，使用默认声音
access_token: "U4YdYXVfpwWnk2t5Gp822zWPCuORyeJL"
voice: "OUeAo1mhq6IBExi"
output_dir: tmp/
main/xiaozhi-server/core/connection.py
+10 -4
View File
@@ -35,7 +35,6 @@ from plugins_func.loadplugins import auto_import_modules
from plugins_func.register import Action, ActionResponse
from core.auth import AuthMiddleware, AuthenticationError
from config.config_loader import get_private_config_from_api
from core.handle.receiveAudioHandle import handleAudioMessage
from core.providers.tts.dto.dto import ContentType, TTSMessageDTO, SentenceType
from config.logger import setup_logging, build_module_string, update_module_string
from config.manage_api_client import DeviceNotFoundException, DeviceBindException
@@ -81,6 +80,7 @@ class ConnectionHandler:
self.welcome_msg = None
self.max_output_size = 0
self.chat_history_conf = 0
self.audio_format = "opus"
# 客户端状态相关
self.client_abort = False
@@ -117,7 +117,10 @@ class ConnectionHandler:
self.client_voice_stop = False
# asr相关变量
# 因为实际部署时可能会用到公共的本地ASR，不能把变量暴露给公共ASR
# 所以涉及到ASR的变量，需要在这里定义，属于connection的私有变量
self.asr_audio = []
self.asr_audio_queue = queue.Queue()
# llm相关变量
self.llm_finish_task = True
@@ -146,7 +149,6 @@ class ConnectionHandler:
int(self.config.get("close_connection_no_voice_time", 120)) + 60
) # 在原来第一道关闭的基础上加60秒，进行二道关闭
self.audio_format = "opus"
# {"mcp":true} 表示启用MCP功能
self.features = None
@@ -254,7 +256,11 @@ class ConnectionHandler:
if isinstance(message, str):
await handleTextMessage(self, message)
elif isinstance(message, bytes):
await handleAudioMessage(self, message)
if self.vad is None:
return
if self.asr is None:
return
self.asr_audio_queue.put(message)
async def handle_restart(self, message):
"""处理服务器重启请求"""
@@ -645,7 +651,7 @@ class ConnectionHandler:
# print("content_arguments", content_arguments)
tool_call_flag = True
if tools_call is not None:
if tools_call is not None and len(tools_call) > 0:
tool_call_flag = True
if tools_call[0].id is not None:
function_id = tools_call[0].id
main/xiaozhi-server/core/handle/abortHandle.py
-2
View File
@@ -1,6 +1,4 @@
import json
import queue
from config.logger import setup_logging
TAG = __name__
main/xiaozhi-server/core/handle/helloHandle.py
+1 -2
View File
@@ -35,8 +35,6 @@ async def handleHelloMessage(conn, msg_json):
format = audio_params.get("format")
conn.logger.bind(tag=TAG).info(f"客户端音频格式: {format}")
conn.audio_format = format
if conn.asr is not None:
conn.asr.set_audio_format(format)
conn.welcome_msg["audio_params"] = audio_params
features = msg_json.get("features")
if features:
@@ -68,6 +66,7 @@ async def checkWakeupWords(conn, text):
if filtered_text not in conn.config.get("wakeup_words"):
return False
conn.just_woken_up = True
await send_stt_message(conn, text)
# 获取当前音色
main/xiaozhi-server/core/handle/intentHandler.py
+1
View File
@@ -96,6 +96,7 @@ async def process_intent_result(conn, intent_result, original_text):
}
await send_stt_message(conn, original_text)
conn.client_abort = False
# 使用executor执行函数调用和结果处理
def process_function_call():
main/xiaozhi-server/core/handle/receiveAudioHandle.py
+16 -5
View File
@@ -3,6 +3,7 @@ from core.handle.intentHandler import handle_user_intent
from core.utils.output_counter import check_device_output_limit
from core.handle.abortHandle import handleAbortMessage
import time
import asyncio
from core.handle.sendAudioHandle import SentenceType
from core.utils.util import audio_to_data
@@ -10,19 +11,29 @@ TAG = __name__
async def handleAudioMessage(conn, audio):
if conn.vad is None:
return
if conn.asr is None or not hasattr(conn.asr, "conn") or conn.asr.conn is None:
return
# 当前片段是否有人说话
have_voice = conn.vad.is_vad(conn, audio)
# 如果设备刚刚被唤醒，短暂忽略VAD检测
if hasattr(conn, "just_woken_up") and conn.just_woken_up:
have_voice = False
# 设置一个短暂延迟后恢复VAD检测
conn.asr_audio.clear()
asyncio.create_task(resume_vad_detection(conn))
if have_voice:
if conn.client_is_speaking:
await handleAbortMessage(conn)
# 设备长时间空闲检测，用于say goodbye
await no_voice_close_connect(conn, have_voice)
# 接收音频
await conn.asr.receive_audio(audio, have_voice)
await conn.asr.receive_audio(conn, audio, have_voice)
async def resume_vad_detection(conn):
# 等待2秒后恢复VAD检测
await asyncio.sleep(2)
conn.just_woken_up = False
async def startToChat(conn, text):
main/xiaozhi-server/core/handle/sendAudioHandle.py
+1 -2
View File
@@ -89,8 +89,7 @@ async def sendAudio(conn, audios, pre_buffer=True):
# 播放剩余音频帧
for opus_packet in remaining_audios:
if conn.client_abort:
conn.client_abort = False
return
break
# 每分钟重置一次计时器
if time.perf_counter() - last_reset_time > 60:
main/xiaozhi-server/core/handle/textHandle.py
+4 -1
View File
@@ -61,6 +61,7 @@ async def handleTextMessage(conn, message):
await send_tts_message(conn, "stop", None)
conn.client_is_speaking = False
elif is_wakeup_words:
conn.just_woken_up = True
# 上报纯文字数据（复用ASR上报功能，但不提供音频数据）
enqueue_asr_report(conn, "嘿，你好呀", [])
await startToChat(conn, "嘿，你好呀")
@@ -78,7 +79,9 @@ async def handleTextMessage(conn, message):
elif msg_json["type"] == "mcp":
conn.logger.bind(tag=TAG).info(f"收到mcp消息：{message}")
if "payload" in msg_json:
asyncio.create_task(handle_mcp_message(conn, conn.mcp_client, msg_json["payload"]))
asyncio.create_task(
handle_mcp_message(conn, conn.mcp_client, msg_json["payload"])
)
elif msg_json["type"] == "server":
# 记录日志时过滤敏感信息
conn.logger.bind(tag=TAG).info(
main/xiaozhi-server/core/providers/asr/aliyun.py
+2 -2
View File
@@ -213,7 +213,7 @@ class ASRProvider(ASRProviderBase):
return None
async def speech_to_text(
self, opus_data: List[bytes], session_id: str
self, opus_data: List[bytes], session_id: str, audio_format="opus"
) -> Tuple[Optional[str], Optional[str]]:
"""将语音数据转换为文本"""
if self._is_token_expired():
@@ -223,7 +223,7 @@ class ASRProvider(ASRProviderBase):
file_path = None
try:
# 解码Opus为PCM
if self.audio_format == "pcm":
if audio_format == "pcm":
pcm_data = opus_data
else:
pcm_data = self.decode_opus(opus_data)
main/xiaozhi-server/core/providers/asr/baidu.py
+2 -2
View File
@@ -30,7 +30,7 @@ class ASRProvider(ASRProviderBase):
os.makedirs(self.output_dir, exist_ok=True)
async def speech_to_text(
self, opus_data: List[bytes], session_id: str
self, opus_data: List[bytes], session_id: str, audio_format="opus"
) -> Tuple[Optional[str], Optional[str]]:
"""将语音数据转换为文本"""
if not opus_data:
@@ -45,7 +45,7 @@ class ASRProvider(ASRProviderBase):
return None, file_path
# 将Opus音频数据解码为PCM
if self.audio_format == "pcm":
if audio_format == "pcm":
pcm_data = opus_data
else:
pcm_data = self.decode_opus(opus_data)
main/xiaozhi-server/core/providers/asr/base.py
+53 -31
View File
@@ -1,15 +1,19 @@
import os
import time
import wave
import copy
import uuid
import wave
import queue
import asyncio
import traceback
import threading
import opuslib_next
from abc import ABC, abstractmethod
from config.logger import setup_logging
from typing import Optional, Tuple, List
from core.utils.util import remove_punctuation_and_length
from core.handle.reportHandle import enqueue_asr_report
from core.handle.receiveAudioHandle import startToChat
from core.handle.reportHandle import enqueue_asr_report
from core.utils.util import remove_punctuation_and_length
from core.handle.receiveAudioHandle import handleAudioMessage
TAG = __name__
logger = setup_logging()
@@ -17,53 +21,75 @@ logger = setup_logging()
class ASRProviderBase(ABC):
def __init__(self):
self.audio_format = "opus"
self.conn = None
pass
# 打开音频通道
# 这里默认是非流式的处理方式
# 流式处理方式请在子类中重写
async def open_audio_channels(self, conn):
self.conn = conn
# tts 消化线程
conn.asr_priority_thread = threading.Thread(
target=self.asr_text_priority_thread, args=(conn,), daemon=True
)
conn.asr_priority_thread.start()
# 有序处理ASR音频
def asr_text_priority_thread(self, conn):
while not conn.stop_event.is_set():
try:
message = conn.asr_audio_queue.get(timeout=1)
future = asyncio.run_coroutine_threadsafe(
handleAudioMessage(conn, message),
conn.loop,
)
future.result()
except queue.Empty:
continue
except Exception as e:
logger.bind(tag=TAG).error(
f"处理ASR文本失败: {str(e)}, 类型: {type(e).__name__}, 堆栈: {traceback.format_exc()}"
)
continue
# 接收音频
# 这里默认是非流式的处理方式
# 流式处理方式请在子类中重写
async def receive_audio(self, audio, audio_have_voice):
if (
self.conn.client_listen_mode == "auto"
or self.conn.client_listen_mode == "realtime"
):
async def receive_audio(self, conn, audio, audio_have_voice):
if conn.client_listen_mode == "auto" or conn.client_listen_mode == "realtime":
have_voice = audio_have_voice
else:
have_voice = self.conn.client_have_voice
have_voice = conn.client_have_voice
# 如果本次没有声音，本段也没声音，就把声音丢弃了
self.conn.asr_audio.append(audio)
if have_voice == False and self.conn.client_have_voice == False:
self.conn.asr_audio = self.conn.asr_audio[-10:]
conn.asr_audio.append(audio)
if have_voice == False and conn.client_have_voice == False:
conn.asr_audio = conn.asr_audio[-10:]
return
# 如果本段有声音，且已经停止了
if self.conn.client_voice_stop:
asr_audio_task = copy.deepcopy(self.conn.asr_audio)
self.conn.asr_audio.clear()
if conn.client_voice_stop:
asr_audio_task = copy.deepcopy(conn.asr_audio)
conn.asr_audio.clear()
# 音频太短了，无法识别
self.conn.reset_vad_states()
conn.reset_vad_states()
if len(asr_audio_task) > 15:
await self.handle_voice_stop(asr_audio_task)
await self.handle_voice_stop(conn, asr_audio_task)
# 处理语音停止
async def handle_voice_stop(self, asr_audio_task):
async def handle_voice_stop(self, conn, asr_audio_task):
raw_text, _ = await self.speech_to_text(
asr_audio_task, self.conn.session_id
asr_audio_task, conn.session_id, conn.audio_format
) # 确保ASR模块返回原始文本
self.conn.logger.bind(tag=TAG).info(f"识别文本: {raw_text}")
conn.logger.bind(tag=TAG).info(f"识别文本: {raw_text}")
text_len, _ = remove_punctuation_and_length(raw_text)
self.stop_ws_connection()
if text_len > 0:
# 使用自定义模块进行上报
await startToChat(self.conn, raw_text)
enqueue_asr_report(self.conn, raw_text, asr_audio_task)
await startToChat(conn, raw_text)
enqueue_asr_report(conn, raw_text, asr_audio_task)
def stop_ws_connection(self):
pass
def save_audio_to_file(self, pcm_data: List[bytes], session_id: str) -> str:
"""PCM数据保存为WAV文件"""
@@ -81,15 +107,11 @@ class ASRProviderBase(ABC):
@abstractmethod
async def speech_to_text(
self, opus_data: List[bytes], session_id: str
self, opus_data: List[bytes], session_id: str, audio_format="opus"
) -> Tuple[Optional[str], Optional[str]]:
"""将语音数据转换为文本"""
pass
def set_audio_format(self, format: str) -> None:
"""设置音频格式"""
self.audio_format = format
@staticmethod
def decode_opus(opus_data: List[bytes]) -> bytes:
"""将Opus音频数据解码为PCM数据"""
main/xiaozhi-server/core/providers/asr/doubao.py
+206 -473
View File
@@ -1,534 +1,267 @@
import time
import os
import uuid
import json
import gzip
import uuid
import asyncio
import websockets
import opuslib_next
from core.providers.asr.base import ASRProviderBase
from config.logger import setup_logging
from typing import Optional, Tuple, List
from core.providers.asr.base import ASRProviderBase
from core.providers.asr.dto.dto import InterfaceType
import threading
TAG = __name__
logger = setup_logging()
CLIENT_FULL_REQUEST = 0b0001
CLIENT_AUDIO_ONLY_REQUEST = 0b0010
NO_SEQUENCE = 0b0000
NEG_SEQUENCE = 0b0010
SERVER_FULL_RESPONSE = 0b1001
SERVER_ACK = 0b1011
SERVER_ERROR_RESPONSE = 0b1111
NO_SEQUENCE = 0b0000
NEG_SEQUENCE = 0b0010
JSON_SERIALIZATION = 0b0001
GZIP_COMPRESSION = 0b0001
PROTOCOL_VERSION = 0b0001
NO_SERIALIZATION = 0b0000
JSON = 0b0001
THRIFT = 0b0011
CUSTOM_TYPE = 0b1111
NO_COMPRESSION = 0b0000
GZIP = 0b0001
CUSTOM_COMPRESSION = 0b1111
def parse_response(res):
"""
protocol_version(4 bits), header_size(4 bits),
message_type(4 bits), message_type_specific_flags(4 bits)
serialization_method(4 bits) message_compression(4 bits)
reserved （8bits) 保留字段
header_extensions 扩展头(大小等于 8 * 4 * (header_size - 1) )
payload 类似与http 请求体
"""
protocol_version = res[0] >> 4
header_size = res[0] & 0x0F
message_type = res[1] >> 4
message_type_specific_flags = res[1] & 0x0F
serialization_method = res[2] >> 4
message_compression = res[2] & 0x0F
reserved = res[3]
header_extensions = res[4 : header_size * 4]
payload = res[header_size * 4 :]
result = {}
payload_msg = None
payload_size = 0
if message_type == SERVER_FULL_RESPONSE:
payload_size = int.from_bytes(payload[:4], "big", signed=True)
payload_msg = payload[4:]
elif message_type == SERVER_ACK:
seq = int.from_bytes(payload[:4], "big", signed=True)
result["seq"] = seq
if len(payload) >= 8:
payload_size = int.from_bytes(payload[4:8], "big", signed=False)
payload_msg = payload[8:]
elif message_type == SERVER_ERROR_RESPONSE:
code = int.from_bytes(payload[:4], "big", signed=False)
result["code"] = code
payload_size = int.from_bytes(payload[4:8], "big", signed=False)
payload_msg = payload[8:]
if payload_msg is None:
return result
if message_compression == GZIP:
payload_msg = gzip.decompress(payload_msg)
if serialization_method == JSON:
payload_msg = json.loads(str(payload_msg, "utf-8"))
elif serialization_method != NO_SERIALIZATION:
payload_msg = str(payload_msg, "utf-8")
result["payload_msg"] = payload_msg
result["payload_size"] = payload_size
return result
class ASRProvider(ASRProviderBase):
def __init__(self, config, delete_audio_file):
def __init__(self, config: dict, delete_audio_file: bool):
super().__init__()
self.interface_type = InterfaceType.STREAM
self.config = config
self.text = ""
self.max_retries = 3
self.retry_delay = 2 # 重试延迟秒数
self.recv_lock = asyncio.Lock() # 添加接收锁
self.reconnect_lock = asyncio.Lock() # 添加重连锁
self.last_reconnect_time = 0 # 上次重连时间
self.reconnect_cooldown = 1 # 增加重连冷却时间到10秒
self.reconnect_count = 0 # 当前重连次数
self.max_reconnect_count = 3 # 减少最大重连次数到3次
self.asr_thread = None # ASR监听线程
self.thread_lock = threading.Lock() # 线程管理锁
self.is_reconnecting = False # 添加重连状态标志
# 添加会话管理相关属性
self._session_lock = asyncio.Lock() # 会话操作的并发锁
self._current_session_id = None # 当前会话ID
self._session_started = False # 会话是否已开始
self._session_finished = False # 会话是否已结束
self._session_close_event = asyncio.Event() # 添加会话关闭事件
self.appid = str(config.get("appid"))
self.interface_type = InterfaceType.NON_STREAM
self.appid = config.get("appid")
self.cluster = config.get("cluster")
self.access_token = config.get("access_token")
self.boosting_table_name = config.get("boosting_table_name", "")
self.correct_table_name = config.get("correct_table_name", "")
self.output_dir = config.get("output_dir", "temp/")
self.output_dir = config.get("output_dir")
self.delete_audio_file = delete_audio_file
self.ws_url = "wss://openspeech.bytedance.com/api/v2/asr"
self.uid = config.get("uid", "streaming_asr_service")
self.workflow = config.get(
"workflow", "audio_in,resample,partition,vad,fe,decode,itn,nlu_punctuate"
)
self.result_type = config.get("result_type", "single")
self.format = config.get("format", "raw")
self.codec = config.get("codec", "pcm")
self.rate = config.get("sample_rate", 16000)
self.language = config.get("language", "zh-CN")
self.bits = config.get("bits", 16)
self.channel = config.get("channel", 1)
self.auth_method = config.get("auth_method", "token")
self.secret = config.get("secret", "access_secret")
self.decoder = opuslib_next.Decoder(16000, 1)
self.asr_ws = None
self.forward_task = None
self.conn = None
self.host = "openspeech.bytedance.com"
self.ws_url = f"wss://{self.host}/api/v2/asr"
self.success_code = 1000
self.seg_duration = 15000
###################################################################################
# 豆包流式ASR重写父类的方法--开始
###################################################################################
async def open_audio_channels(self, conn):
await super().open_audio_channels(conn)
# 确保输出目录存在
os.makedirs(self.output_dir, exist_ok=True)
async with self._session_lock:
# 如果正在重连，等待重连完成
if self.is_reconnecting:
logger.bind(tag=TAG).info("等待当前重连完成...")
await self._session_close_event.wait()
self._session_close_event.clear()
@staticmethod
def _generate_header(
message_type=CLIENT_FULL_REQUEST, message_type_specific_flags=NO_SEQUENCE
) -> bytearray:
"""Generate protocol header."""
header = bytearray()
header_size = 1
header.append((0b0001 << 4) | header_size) # Protocol version
header.append((message_type << 4) | message_type_specific_flags)
header.append((0b0001 << 4) | 0b0001) # JSON serialization & GZIP compression
header.append(0x00) # reserved
return header
# 如果已有会话未结束，先关闭它
if self._session_started and not self._session_finished:
logger.bind(tag=TAG).warning(
f"发现未关闭的会话 {self._current_session_id}，正在关闭..."
)
if self.asr_ws is not None:
try:
await self.asr_ws.close()
except Exception as e:
logger.bind(tag=TAG).warning(f"关闭旧连接时发生错误: {e}")
finally:
self.asr_ws = None
self._session_finished = True
self._session_close_event.set()
# 重置会话状态
self._current_session_id = str(uuid.uuid4())
self._session_started = True
self._session_finished = False
self.is_reconnecting = True
try:
retry_count = 0
while retry_count < self.max_retries:
try:
headers = (
self.token_auth() if self.auth_method == "token" else None
)
self.asr_ws = await websockets.connect(
self.ws_url,
additional_headers=headers,
max_size=1000000000,
ping_interval=None,
ping_timeout=None,
close_timeout=10,
)
# 发送初始化请求
request_params = self.construct_request(
self._current_session_id
)
try:
payload_bytes = str.encode(json.dumps(request_params))
payload_bytes = gzip.compress(payload_bytes)
full_client_request = self.generate_header()
full_client_request.extend(
(len(payload_bytes)).to_bytes(4, "big")
)
full_client_request.extend(payload_bytes)
await self.asr_ws.send(full_client_request)
except Exception as e:
logger.bind(tag=TAG).error(f"发送初始化请求失败: {e}")
raise e
# 等待初始化响应
try:
init_res = await self.asr_ws.recv()
self.parse_response(init_res)
except Exception as e:
logger.bind(tag=TAG).error(f"ASR服务初始化失败: {e}")
raise e
# 启动接收ASR结果的异步任务
with self.thread_lock:
if (
self.asr_thread is None
or not self.asr_thread.is_alive()
):
logger.bind(tag=TAG).info("创建新的ASR监听线程...")
self.asr_thread = threading.Thread(
target=self._start_monitor_asr_response_thread,
daemon=True,
)
self.asr_thread.start()
# 等待一小段时间确保线程启动
await asyncio.sleep(0.1)
if not self.asr_thread.is_alive():
logger.bind(tag=TAG).error("ASR监听线程启动失败")
raise Exception("ASR监听线程启动失败")
logger.bind(tag=TAG).info("ASR监听线程已启动")
return
except websockets.exceptions.WebSocketException as e:
retry_count += 1
if retry_count < self.max_retries:
logger.bind(tag=TAG).warning(
f"WebSocket连接失败，正在进行第{retry_count}次重试: {e}"
)
await asyncio.sleep(self.retry_delay)
else:
logger.bind(tag=TAG).warning(
f"WebSocket连接失败，已达到最大重试次数: {e}"
)
raise
except Exception as e:
logger.bind(tag=TAG).error(f"WebSocket连接发生未知错误: {e}")
raise
finally:
self.is_reconnecting = False
self._session_close_event.set()
async def receive_audio(self, audio, _):
if not isinstance(audio, bytes):
return
try:
# 解码opus得到PCM数据
pcm_frame = self.decoder.decode(audio, 960)
payload = gzip.compress(pcm_frame)
audio_request = bytearray(self.generate_audio_default_header())
audio_request.extend(len(payload).to_bytes(4, "big"))
audio_request.extend(payload)
if self.asr_ws:
await self.asr_ws.send(audio_request)
except Exception as e:
logger.bind(tag=TAG).debug(f"发送音频数据时发生错误: {e}")
###################################################################################
# 豆包流式ASR重写父类的方法--结束
###################################################################################
def construct_request(self, reqid):
req = {
def _construct_request(self, reqid) -> dict:
"""Construct the request payload."""
return {
"app": {
"appid": self.appid,
"appid": f"{self.appid}",
"cluster": self.cluster,
"token": self.access_token,
},
"user": {"uid": self.uid},
"user": {
"uid": str(uuid.uuid4()),
},
"request": {
"reqid": reqid,
"workflow": self.workflow,
"show_utterances": True,
"result_type": self.result_type,
"show_utterances": False,
"sequence": 1,
"boosting_table_name": self.boosting_table_name,
"correct_table_name": self.correct_table_name,
},
"audio": {
"format": self.format,
"codec": self.codec,
"rate": self.rate,
"language": self.language,
"bits": self.bits,
"channel": self.channel,
"format": "raw",
"rate": 16000,
"language": "zh-CN",
"bits": 16,
"channel": 1,
"codec": "raw",
},
}
return req
def token_auth(self):
return {"Authorization": f"Bearer; {self.access_token}"}
def generate_header(
self,
version=PROTOCOL_VERSION,
message_type=CLIENT_FULL_REQUEST,
message_type_specific_flags=NO_SEQUENCE,
serial_method=JSON_SERIALIZATION,
compression_type=GZIP_COMPRESSION,
reserved_data=0x00,
extension_header: bytes = b"",
):
"""
生成协议头：
- 第1字节：高4位：协议版本，低4位：头部大小（单位 4 字节）
- 第2字节：高4位：消息类型，低4位：消息类型特定标志
- 第3字节：高4位：序列化方式，低4位：压缩方式
- 第4字节：保留字段
- 后续：扩展头（如果有）
"""
header = bytearray()
header_size = int(len(extension_header) / 4) + 1
header.append((version << 4) | header_size)
header.append((message_type << 4) | message_type_specific_flags)
header.append((serial_method << 4) | compression_type)
header.append(reserved_data)
header.extend(extension_header)
return header
def generate_full_default_header(self):
# full client request 默认头
return self.generate_header(
version=PROTOCOL_VERSION,
message_type=CLIENT_FULL_REQUEST,
message_type_specific_flags=NO_SEQUENCE,
serial_method=JSON_SERIALIZATION,
compression_type=GZIP_COMPRESSION,
)
def generate_audio_default_header(self):
# 普通音频片段请求
return self.generate_header(
version=PROTOCOL_VERSION,
message_type=CLIENT_AUDIO_ONLY_REQUEST,
message_type_specific_flags=NO_SEQUENCE,
serial_method=JSON_SERIALIZATION,
compression_type=GZIP_COMPRESSION,
)
def generate_last_audio_default_header(self):
# 最后一个音频片段标志
return self.generate_header(
version=PROTOCOL_VERSION,
message_type=CLIENT_AUDIO_ONLY_REQUEST,
message_type_specific_flags=NEG_SEQUENCE, # 用 NEG_SEQUENCE 表示结束
serial_method=JSON_SERIALIZATION,
compression_type=GZIP_COMPRESSION,
)
def _start_monitor_asr_response_thread(self):
# 初始化链接
async def _send_request(
self, audio_data: List[bytes], segment_size: int
) -> Optional[str]:
"""Send request to Volcano ASR service."""
try:
with self.thread_lock:
if self.conn is None or self.conn.loop is None:
logger.bind(tag=TAG).error(
"无法启动ASR监听线程：conn或loop未初始化"
)
return
auth_header = {"Authorization": "Bearer; {}".format(self.access_token)}
async with websockets.connect(
self.ws_url, additional_headers=auth_header
) as websocket:
# Prepare request data
request_params = self._construct_request(str(uuid.uuid4()))
payload_bytes = str.encode(json.dumps(request_params))
payload_bytes = gzip.compress(payload_bytes)
full_client_request = self._generate_header()
full_client_request.extend(
(len(payload_bytes)).to_bytes(4, "big")
) # payload size(4 bytes)
full_client_request.extend(payload_bytes) # payload
try:
logger.bind(tag=TAG).info("开始启动ASR监听...")
asyncio.run_coroutine_threadsafe(
self._forward_asr_results(), loop=self.conn.loop
)
logger.bind(tag=TAG).info("ASR监听已启动")
except Exception as e:
logger.bind(tag=TAG).error(f"启动ASR监听线程失败: {e}")
except Exception as e:
logger.bind(tag=TAG).error(f"ASR监听线程发生未预期的错误: {e}")
# Send header and metadata
# full_client_request
await websocket.send(full_client_request)
res = await websocket.recv()
result = parse_response(res)
if (
"payload_msg" in result
and result["payload_msg"]["code"] != self.success_code
):
logger.bind(tag=TAG).error(f"ASR error: {result}")
return None
async def _forward_asr_results(self):
try:
while not self.conn.stop_event.is_set():
try:
if self.asr_ws is None:
# 检查是否需要重连
async with self.reconnect_lock:
current_time = asyncio.get_event_loop().time()
if (
current_time - self.last_reconnect_time
< self.reconnect_cooldown
):
await asyncio.sleep(1)
continue
if self.reconnect_count >= self.max_reconnect_count:
logger.bind(tag=TAG).error(
"达到最大重连次数限制，停止重连"
)
await asyncio.sleep(self.reconnect_cooldown)
self.reconnect_count = 0
continue
self.last_reconnect_time = current_time
self.reconnect_count += 1
logger.bind(tag=TAG).info(
f"尝试重新连接ASR服务... (第{self.reconnect_count}次)"
)
await self.open_audio_channels(self.conn)
continue
# 使用锁来确保同一时间只有一个协程在接收数据
async with self.recv_lock:
response = await self.asr_ws.recv()
result = self.parse_response(response)
# 检查是否需要重连
if result.get("need_reconnect", False):
logger.bind(tag=TAG).info(
"检测到需要重连的错误，准备重新连接..."
for seq, (chunk, last) in enumerate(
self.slice_data(audio_data, segment_size), 1
):
if last:
audio_only_request = self._generate_header(
message_type=CLIENT_AUDIO_ONLY_REQUEST,
message_type_specific_flags=NEG_SEQUENCE,
)
if self.asr_ws is not None:
try:
await self.asr_ws.close()
except Exception as e:
logger.bind(tag=TAG).warning(
f"关闭旧连接时发生错误: {e}"
)
finally:
self.asr_ws = None
continue
else:
audio_only_request = self._generate_header(
message_type=CLIENT_AUDIO_ONLY_REQUEST
)
payload_bytes = gzip.compress(chunk)
audio_only_request.extend(
(len(payload_bytes)).to_bytes(4, "big")
) # payload size(4 bytes)
audio_only_request.extend(payload_bytes) # payload
# Send audio data
await websocket.send(audio_only_request)
if "payload_msg" in result:
if "result" in result["payload_msg"]:
# 检查是否有utterances并且definite为True
utterances = result["payload_msg"]["result"][0].get(
"utterances", []
)
for utterance in utterances:
if utterance.get("definite", False):
self.text = utterance["text"]
await self.handle_voice_stop(None)
break
# Receive response
response = await websocket.recv()
result = parse_response(response)
except websockets.ConnectionClosed:
logger.bind(tag=TAG).debug("ASR服务连接已关闭，准备重连...")
# 确保关闭旧连接
if self.asr_ws is not None:
try:
await self.asr_ws.close()
except Exception as e:
logger.bind(tag=TAG).warning(f"关闭旧连接时发生错误: {e}")
finally:
self.asr_ws = None
# 等待冷却时间
await asyncio.sleep(self.reconnect_cooldown)
continue
except Exception as e:
if not self.conn.stop_event.is_set():
logger.bind(tag=TAG).error(f"ASR监听发生错误: {e}")
await asyncio.sleep(self.retry_delay)
continue
if (
"payload_msg" in result
and result["payload_msg"]["code"] == self.success_code
):
if len(result["payload_msg"]["result"]) > 0:
return result["payload_msg"]["result"][0]["text"]
return None
else:
logger.bind(tag=TAG).error(f"ASR error: {result}")
return None
except Exception as e:
logger.bind(tag=TAG).error(f"ASR监听线程发生错误: {e}")
# 确保在发生严重错误时也能继续尝试重连
if not self.conn.stop_event.is_set():
await asyncio.sleep(self.retry_delay)
await self._forward_asr_results() # 递归重试
logger.bind(tag=TAG).error(f"ASR request failed: {e}", exc_info=True)
return None
async def speech_to_text(self, opus_data, session_id):
result = self.text
self.text = "" # 清空text
return result, None
def parse_response(self, res: bytes) -> dict:
@staticmethod
def slice_data(data: bytes, chunk_size: int) -> (list, bool):
"""
解析 ASR 服务返回的二进制响应。
根据协议格式解析头部和 payload，若采用 GZIP 压缩则先解压，再根据 JSON 反序列化。
slice data
:param data: wav data
:param chunk_size: the segment size in one request
:return: segment data, last flag
"""
protocol_version = res[0] >> 4
header_size = res[0] & 0x0F
message_type = res[1] >> 4
serialization_method = res[2] >> 4
message_compression = res[2] & 0x0F
payload = res[header_size * 4 :]
result = {}
payload_msg = None
payload_size = 0
if message_type == SERVER_FULL_RESPONSE:
payload_size = int.from_bytes(payload[:4], "big", signed=True)
payload_msg = payload[4:]
elif message_type == SERVER_ACK:
seq = int.from_bytes(payload[:4], "big", signed=True)
result["seq"] = seq
if len(payload) >= 8:
payload_size = int.from_bytes(payload[4:8], "big", signed=False)
payload_msg = payload[8:]
elif message_type == SERVER_ERROR_RESPONSE:
code = int.from_bytes(payload[:4], "big", signed=False)
result["code"] = code
payload_size = int.from_bytes(payload[4:8], "big", signed=False)
payload_msg = payload[8:]
if payload_msg is None:
return result
if message_compression == GZIP_COMPRESSION:
payload_msg = gzip.decompress(payload_msg)
if serialization_method == JSON_SERIALIZATION:
payload_msg = json.loads(payload_msg.decode("utf-8"))
data_len = len(data)
offset = 0
while offset + chunk_size < data_len:
yield data[offset : offset + chunk_size], False
offset += chunk_size
else:
payload_msg = payload_msg.decode("utf-8")
result["payload_msg"] = payload_msg
result["payload_size"] = payload_size
yield data[offset:data_len], True
# 错误码处理
if "code" in result:
error_code = result["code"]
error_message = ""
async def speech_to_text(
self, opus_data: List[bytes], session_id: str, audio_format="opus"
) -> Tuple[Optional[str], Optional[str]]:
"""将语音数据转换为文本"""
if error_code == 1000:
error_message = "成功"
elif error_code == 1001:
error_message = "请求参数无效：请求参数缺失必需字段/字段值无效/重复请求"
elif error_code == 1002:
error_message = "无访问权限：token无效/过期/无权访问指定服务"
elif error_code == 1003:
error_message = "访问超频：当前appid访问QPS超出设定阈值"
elif error_code == 1004:
error_message = "访问超额：当前appid访问次数超出限制"
elif error_code == 1005:
error_message = "服务器繁忙：服务过载，无法处理当前请求"
elif error_code == 1010:
error_message = "音频过长：音频数据时长超出阈值"
elif error_code == 1011:
error_message = "音频过大：音频数据大小超出阈值"
elif error_code == 1012:
error_message = "音频格式无效：音频header有误/无法进行音频解码"
elif error_code == 1013:
error_message = "音频静音：音频未识别出任何文本结果"
elif error_code >= 1020 and error_code <= 1022:
error_message = "识别相关错误：需要重连"
if error_code == 1020:
error_message = "识别等待超时：等待下一包就绪超时"
elif error_code == 1021:
error_message = "识别处理超时：识别处理过程超时"
elif error_code == 1022:
error_message = "识别错误：识别过程中发生错误"
file_path = None
try:
# 合并所有opus数据包
if audio_format == "pcm":
pcm_data = opus_data
else:
error_message = "未知错误：未归类错误"
pcm_data = self.decode_opus(opus_data)
combined_pcm_data = b"".join(pcm_data)
logger.bind(tag=TAG).debug(
f"ASR错误: {error_message} (错误码: {error_code})"
)
# 判断是否保存为WAV文件
if self.delete_audio_file:
pass
else:
file_path = self.save_audio_to_file(pcm_data, session_id)
# 如果是识别相关错误，标记需要重连
if error_code >= 1020 or error_code == 1001:
result["need_reconnect"] = True
# 直接使用PCM数据
# 计算分段大小 (单声道, 16bit, 16kHz采样率)
size_per_sec = 1 * 2 * 16000 # nchannels * sampwidth * framerate
segment_size = int(size_per_sec * self.seg_duration / 1000)
return result
async def close_session(self):
"""关闭当前会话"""
async with self._session_lock:
if not self._session_started:
logger.bind(tag=TAG).warning("尝试关闭未开始的会话")
return
if self._session_finished:
logger.bind(tag=TAG).warning(
f"会话 {self._current_session_id} 已经关闭"
# 语音识别
start_time = time.time()
text = await self._send_request(combined_pcm_data, segment_size)
if text:
logger.bind(tag=TAG).debug(
f"语音识别耗时: {time.time() - start_time:.3f}s | 结果: {text}"
)
return
return text, file_path
return "", file_path
try:
if self.asr_ws is not None:
await self.asr_ws.close()
except Exception as e:
logger.bind(tag=TAG).warning(f"关闭WebSocket连接时发生错误: {e}")
finally:
self.asr_ws = None
self._session_finished = True
self._session_started = False
self._current_session_id = None
# 重置重连计数
self.reconnect_count = 0
async def close(self):
"""资源清理方法"""
await self.close_session()
except Exception as e:
logger.bind(tag=TAG).error(f"语音识别失败: {e}", exc_info=True)
return "", file_path
main/xiaozhi-server/core/providers/asr/doubao_stream.py
+344
View File
@@ -0,0 +1,344 @@
import json
import gzip
import uuid
import asyncio
import websockets
import opuslib_next
from core.providers.asr.base import ASRProviderBase
from config.logger import setup_logging
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.max_retries = 3
self.retry_delay = 2
self.decoder = opuslib_next.Decoder(16000, 1)
self.asr_ws = None
self.forward_task = None
self.is_processing = False # 添加处理状态标志
# 配置参数
self.appid = str(config.get("appid"))
self.cluster = config.get("cluster")
self.access_token = config.get("access_token")
self.boosting_table_name = config.get("boosting_table_name", "")
self.correct_table_name = config.get("correct_table_name", "")
self.output_dir = config.get("output_dir", "tmp/")
self.delete_audio_file = delete_audio_file
# 火山引擎ASR配置
self.ws_url = "wss://openspeech.bytedance.com/api/v3/sauc/bigmodel"
self.uid = config.get("uid", "streaming_asr_service")
self.workflow = config.get(
"workflow", "audio_in,resample,partition,vad,fe,decode,itn,nlu_punctuate"
)
self.result_type = config.get("result_type", "single")
self.format = config.get("format", "pcm")
self.codec = config.get("codec", "pcm")
self.rate = config.get("sample_rate", 16000)
self.language = config.get("language", "zh-CN")
self.bits = config.get("bits", 16)
self.channel = config.get("channel", 1)
self.auth_method = config.get("auth_method", "token")
self.secret = config.get("secret", "access_secret")
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:]
# 如果本次有声音，且之前没有建立连接
if audio_have_voice and self.asr_ws is None and not self.is_processing:
try:
self.is_processing = True
# 建立新的WebSocket连接
headers = self.token_auth() if self.auth_method == "token" else None
logger.bind(tag=TAG).info(f"正在连接ASR服务，headers: {headers}")
self.asr_ws = await websockets.connect(
self.ws_url,
additional_headers=headers,
max_size=1000000000,
ping_interval=None,
ping_timeout=None,
close_timeout=10,
)
# 发送初始化请求
request_params = self.construct_request(str(uuid.uuid4()))
try:
payload_bytes = str.encode(json.dumps(request_params))
payload_bytes = gzip.compress(payload_bytes)
full_client_request = self.generate_header()
full_client_request.extend((len(payload_bytes)).to_bytes(4, "big"))
full_client_request.extend(payload_bytes)
logger.bind(tag=TAG).info(f"发送初始化请求: {request_params}")
await self.asr_ws.send(full_client_request)
# 等待初始化响应
init_res = await self.asr_ws.recv()
result = self.parse_response(init_res)
logger.bind(tag=TAG).info(f"收到初始化响应: {result}")
# 检查初始化响应
if "code" in result and result["code"] != 1000:
error_msg = f"ASR服务初始化失败: {result.get('payload_msg', {}).get('message', '未知错误')}"
if "payload_msg" in result:
error_msg += f"\n详细错误信息: {json.dumps(result['payload_msg'], ensure_ascii=False)}"
logger.bind(tag=TAG).error(error_msg)
raise Exception(error_msg)
except Exception as e:
logger.bind(tag=TAG).error(f"发送初始化请求失败: {str(e)}")
if hasattr(e, "__cause__") and e.__cause__:
logger.bind(tag=TAG).error(f"错误原因: {str(e.__cause__)}")
raise e
# 启动接收ASR结果的异步任务
self.forward_task = asyncio.create_task(self._forward_asr_results(conn))
# 发送缓存的音频数据
if conn.asr_audio and len(conn.asr_audio) > 0:
for cached_audio in conn.asr_audio[-10:]:
try:
pcm_frame = self.decoder.decode(cached_audio, 960)
payload = gzip.compress(pcm_frame)
audio_request = bytearray(
self.generate_audio_default_header()
)
audio_request.extend(len(payload).to_bytes(4, "big"))
audio_request.extend(payload)
await self.asr_ws.send(audio_request)
except Exception as e:
logger.bind(tag=TAG).info(
f"发送缓存音频数据时发生错误: {e}"
)
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
return
# 发送当前音频数据
if self.asr_ws and self.is_processing:
try:
pcm_frame = self.decoder.decode(audio, 960)
payload = gzip.compress(pcm_frame)
audio_request = bytearray(self.generate_audio_default_header())
audio_request.extend(len(payload).to_bytes(4, "big"))
audio_request.extend(payload)
await self.asr_ws.send(audio_request)
except Exception as e:
logger.bind(tag=TAG).info(f"发送音频数据时发生错误: {e}")
async def _forward_asr_results(self, conn):
try:
while self.asr_ws and not conn.stop_event.is_set():
try:
response = await self.asr_ws.recv()
result = self.parse_response(response)
logger.bind(tag=TAG).debug(f"收到ASR结果: {result}")
if "payload_msg" in result:
payload = result["payload_msg"]
if "result" in payload:
utterances = payload["result"].get("utterances", [])
# 检查duration和空文本的情况
if (
payload.get("audio_info", {}).get("duration", 0) > 2000
and not utterances
and not payload["result"].get("text")
):
logger.bind(tag=TAG).error(f"识别文本：空")
self.text = ""
conn.reset_vad_states()
await self.handle_voice_stop(conn, None)
break
for utterance in utterances:
if utterance.get("definite", False):
self.text = utterance["text"]
logger.bind(tag=TAG).info(
f"识别到文本: {self.text}"
)
conn.reset_vad_states()
await self.handle_voice_stop(conn, None)
break
elif "error" in payload:
error_msg = payload.get("error", "未知错误")
logger.bind(tag=TAG).error(f"ASR服务返回错误: {error_msg}")
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"处理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
def stop_ws_connection(self):
if self.asr_ws:
asyncio.create_task(self.asr_ws.close())
self.asr_ws = None
self.is_processing = False
def construct_request(self, reqid):
req = {
"app": {
"appid": self.appid,
"cluster": self.cluster,
"token": self.access_token,
},
"user": {"uid": self.uid},
"request": {
"reqid": reqid,
"workflow": self.workflow,
"show_utterances": True,
"result_type": self.result_type,
"sequence": 1,
"boosting_table_name": self.boosting_table_name,
"correct_table_name": self.correct_table_name,
"end_window_size": 200,
},
"audio": {
"format": self.format,
"codec": self.codec,
"rate": self.rate,
"language": self.language,
"bits": self.bits,
"channel": self.channel,
"sample_rate": self.rate,
},
}
logger.bind(tag=TAG).debug(
f"构造请求参数: {json.dumps(req, ensure_ascii=False)}"
)
return req
def token_auth(self):
return {
"X-Api-App-Key": self.appid,
"X-Api-Access-Key": self.access_token,
"X-Api-Resource-Id": "volc.bigasr.sauc.duration",
"X-Api-Connect-Id": str(uuid.uuid4()),
"Host": "openspeech.bytedance.com",
}
def generate_header(
self,
version=0x01,
message_type=0x01,
message_type_specific_flags=0x00,
serial_method=0x01,
compression_type=0x01,
reserved_data=0x00,
extension_header: bytes = b"",
):
header = bytearray()
header_size = int(len(extension_header) / 4) + 1
header.append((version << 4) | header_size)
header.append((message_type << 4) | message_type_specific_flags)
header.append((serial_method << 4) | compression_type)
header.append(reserved_data)
header.extend(extension_header)
return header
def generate_audio_default_header(self):
return self.generate_header(
version=0x01,
message_type=0x02,
message_type_specific_flags=0x00,
serial_method=0x01,
compression_type=0x01,
)
def generate_last_audio_default_header(self):
return self.generate_header(
version=0x01,
message_type=0x02,
message_type_specific_flags=0x02,
serial_method=0x01,
compression_type=0x01,
)
def parse_response(self, res: bytes) -> dict:
try:
# 检查响应长度
if len(res) < 4:
logger.bind(tag=TAG).error(f"响应数据长度不足: {len(res)}")
return {"error": "响应数据长度不足"}
# 获取消息头
header = res[:4]
message_type = header[1] >> 4
# 如果是错误响应
if message_type == 0x0F: # SERVER_ERROR_RESPONSE
code = int.from_bytes(header[4:8], "big", signed=False)
error_msg = res[8:].decode("utf-8")
return {"code": code, "error": error_msg}
# 获取JSON数据（跳过12字节头部）
try:
json_data = res[12:].decode("utf-8")
result = json.loads(json_data)
logger.bind(tag=TAG).debug(f"成功解析JSON响应: {result}")
return {"payload_msg": result}
except (UnicodeDecodeError, json.JSONDecodeError) as e:
logger.bind(tag=TAG).error(f"JSON解析失败: {str(e)}")
logger.bind(tag=TAG).error(f"原始数据: {res}")
raise
except Exception as e:
logger.bind(tag=TAG).error(f"解析响应失败: {str(e)}")
logger.bind(tag=TAG).error(f"原始响应数据: {res.hex()}")
raise
async def speech_to_text(self, opus_data, session_id, audio_format):
result = self.text
self.text = "" # 清空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
main/xiaozhi-server/core/providers/asr/fun_local.py
+2 -2
View File
@@ -62,7 +62,7 @@ class ASRProvider(ASRProviderBase):
)
async def speech_to_text(
self, opus_data: List[bytes], session_id: str
self, opus_data: List[bytes], session_id: str, audio_format="opus"
) -> Tuple[Optional[str], Optional[str]]:
"""语音转文本主处理逻辑"""
file_path = None
@@ -71,7 +71,7 @@ class ASRProvider(ASRProviderBase):
while retry_count < MAX_RETRIES:
try:
# 合并所有opus数据包
if self.audio_format == "pcm":
if audio_format == "pcm":
pcm_data = opus_data
else:
pcm_data = self.decode_opus(opus_data)
main/xiaozhi-server/core/providers/asr/fun_server.py
+2 -2
View File
@@ -100,7 +100,7 @@ class ASRProvider(ASRProviderBase):
logger.bind(tag=TAG).debug(f"Sent end message: {end_message}")
async def speech_to_text(
self, opus_data: List[bytes], session_id: str
self, opus_data: List[bytes], session_id: str, audio_format="opus"
) -> Tuple[Optional[str], Optional[str]]:
"""
Convert speech data to text using FunASR.
@@ -109,7 +109,7 @@ class ASRProvider(ASRProviderBase):
:return: Tuple containing recognized text and optional timestamp.
"""
file_path = None
if self.audio_format == "pcm":
if audio_format == "pcm":
pcm_data = opus_data
else:
pcm_data = self.decode_opus(opus_data)
main/xiaozhi-server/core/providers/asr/sherpa_onnx_local.py
+2 -2
View File
@@ -109,14 +109,14 @@ class ASRProvider(ASRProviderBase):
return samples_float32, f.getframerate()
async def speech_to_text(
self, opus_data: List[bytes], session_id: str
self, opus_data: List[bytes], session_id: str, audio_format="opus"
) -> Tuple[Optional[str], Optional[str]]:
"""语音转文本主处理逻辑"""
file_path = None
try:
# 保存音频文件
start_time = time.time()
if self.audio_format == "pcm":
if audio_format == "pcm":
pcm_data = opus_data
else:
pcm_data = self.decode_opus(opus_data)
main/xiaozhi-server/core/providers/asr/tencent.py
+2 -2
View File
@@ -32,7 +32,7 @@ class ASRProvider(ASRProviderBase):
os.makedirs(self.output_dir, exist_ok=True)
async def speech_to_text(
self, opus_data: List[bytes], session_id: str
self, opus_data: List[bytes], session_id: str, audio_format="opus"
) -> Tuple[Optional[str], Optional[str]]:
"""将语音数据转换为文本"""
if not opus_data:
@@ -47,7 +47,7 @@ class ASRProvider(ASRProviderBase):
return None, file_path
# 将Opus音频数据解码为PCM
if self.audio_format == "pcm":
if audio_format == "pcm":
pcm_data = opus_data
else:
pcm_data = self.decode_opus(opus_data)
main/xiaozhi-server/core/providers/intent/intent_llm/intent_llm.py
+4 -3
View File
@@ -122,6 +122,8 @@ class IntentProvider(IntentProviderBase):
async def detect_intent(self, conn, dialogue_history: List[Dict], text: str) -> str:
if not self.llm:
raise ValueError("LLM provider not set")
if conn.func_handler is None:
return '{"function_call": {"name": "continue_chat"}}'
# 记录整体开始时间
total_start_time = time.time()
@@ -148,9 +150,8 @@ class IntentProvider(IntentProviderBase):
self.clean_cache()
if self.promot == "":
if hasattr(conn, "func_handler"):
functions = conn.func_handler.get_functions()
self.promot = self.get_intent_system_prompt(functions)
functions = conn.func_handler.get_functions()
self.promot = self.get_intent_system_prompt(functions)
music_config = initialize_music_handler(conn)
music_file_names = music_config["music_file_names"]
main/xiaozhi-server/core/providers/tts/base.py
+13 -1
View File
@@ -37,6 +37,7 @@ class TTSProviderBase(ABC):
self.tts_text_queue = queue.Queue()
self.tts_audio_queue = queue.Queue()
self.tts_audio_first_sentence = True
self.before_stop_play_files = []
self.tts_text_buff = []
self.punctuations = (
@@ -52,6 +53,7 @@ class TTSProviderBase(ABC):
)
self.first_sentence_punctuations = (
"，",
"～",
"~",
"、",
",",
@@ -84,7 +86,9 @@ class TTSProviderBase(ABC):
try:
audio_bytes = asyncio.run(self.text_to_speak(text, None))
if audio_bytes:
audio_datas, _ = audio_bytes_to_data(audio_bytes, file_type=self.audio_file_type, is_opus=True)
audio_datas, _ = audio_bytes_to_data(
audio_bytes, file_type=self.audio_file_type, is_opus=True
)
return audio_datas
else:
max_repeat_time -= 1
@@ -355,6 +359,14 @@ class TTSProviderBase(ABC):
os.remove(tts_file)
return audio_datas
def _process_before_stop_play_files(self):
for tts_file, text in self.before_stop_play_files:
if tts_file and os.path.exists(tts_file):
audio_datas = self._process_audio_file(tts_file)
self.tts_audio_queue.put((SentenceType.MIDDLE, audio_datas, text))
self.before_stop_play_files.clear()
self.tts_audio_queue.put((SentenceType.LAST, [], None))
def _process_remaining_text(self):
"""处理剩余的文本并生成语音
main/xiaozhi-server/core/providers/tts/huoshan_double_stream.py
+4 -13
View File
@@ -153,7 +153,6 @@ class TTSProvider(TTSProviderBase):
self.header = {"Authorization": f"{self.authorization}{self.access_token}"}
self.enable_two_way = True
self.tts_text = ""
self.before_stop_play_files = []
self.opus_encoder = opus_encoder_utils.OpusEncoderUtils(
sample_rate=16000, channels=1, frame_size_ms=60
)
@@ -194,7 +193,7 @@ class TTSProvider(TTSProviderBase):
try:
logger.bind(tag=TAG).debug("等待TTS文本队列消息...")
message = self.tts_text_queue.get(timeout=1)
logger.bind(tag=TAG).info(
logger.bind(tag=TAG).debug(
f"收到TTS任务｜{message.sentence_type.name} ｜ {message.content_type.name} | 会话ID: {self.conn.sentence_id}"
)
if self.conn.client_abort:
@@ -220,7 +219,7 @@ class TTSProvider(TTSProviderBase):
elif ContentType.TEXT == message.content_type:
if message.content_detail:
try:
logger.bind(tag=TAG).info(
logger.bind(tag=TAG).debug(
f"开始发送TTS文本: {message.content_detail}"
)
future = asyncio.run_coroutine_threadsafe(
@@ -228,7 +227,7 @@ class TTSProvider(TTSProviderBase):
loop=self.conn.loop,
)
future.result()
logger.bind(tag=TAG).info("TTS文本发送成功")
logger.bind(tag=TAG).debug("TTS文本发送成功")
except Exception as e:
logger.bind(tag=TAG).error(f"发送TTS文本失败: {str(e)}")
continue
@@ -249,7 +248,6 @@ class TTSProvider(TTSProviderBase):
loop=self.conn.loop,
)
future.result()
logger.bind(tag=TAG).info("TTS会话结束成功")
except Exception as e:
logger.bind(tag=TAG).error(f"结束TTS会话失败: {str(e)}")
continue
@@ -432,14 +430,7 @@ class TTSProvider(TTSProviderBase):
is_first_sentence = False
elif res.optional.event == EVENT_SessionFinished:
logger.bind(tag=TAG).debug(f"会话结束～～")
for tts_file, text in self.before_stop_play_files:
if tts_file and os.path.exists(tts_file):
audio_datas = self._process_audio_file(tts_file)
self.tts_audio_queue.put(
(SentenceType.MIDDLE, audio_datas, text)
)
self.before_stop_play_files.clear()
self.tts_audio_queue.put((SentenceType.LAST, [], None))
self._process_before_stop_play_files()
break
except websockets.ConnectionClosed:
logger.bind(tag=TAG).warning("WebSocket连接已关闭")
main/xiaozhi-server/core/providers/tts/linkerai.py
+227
View File
@@ -0,0 +1,227 @@
import queue
import asyncio
import traceback
import aiohttp
from config.logger import setup_logging
from core.utils.tts import MarkdownCleaner
from core.providers.tts.base import TTSProviderBase
from core.utils import opus_encoder_utils, textUtils
from core.providers.tts.dto.dto import SentenceType, ContentType, InterfaceType
TAG = __name__
logger = setup_logging()
class TTSProvider(TTSProviderBase):
def __init__(self, config, delete_audio_file):
super().__init__(config, delete_audio_file)
self.interface_type = InterfaceType.SINGLE_STREAM
self.access_token = config.get("access_token")
self.voice = config.get("voice")
self.api_url = config.get("api_url")
self.audio_format = "pcm"
self.before_stop_play_files = []
self.segment_count = 0 # 添加片段计数器
# 创建Opus编码器
self.opus_encoder = opus_encoder_utils.OpusEncoderUtils(
sample_rate=16000, channels=1, frame_size_ms=60
)
# 添加文本缓冲区
self.text_buffer = ""
# PCM缓冲区
self.pcm_buffer = bytearray()
###################################################################################
# linkerai单流式TTS重写父类的方法--开始
###################################################################################
def tts_text_priority_thread(self):
"""流式文本处理线程"""
while not self.conn.stop_event.is_set():
try:
message = self.tts_text_queue.get(timeout=1)
if message.sentence_type == SentenceType.FIRST:
# 初始化参数
self.tts_stop_request = False
self.processed_chars = 0
self.tts_text_buff = []
self.segment_count = 0
self.tts_audio_first_sentence = True
self.before_stop_play_files.clear()
elif ContentType.TEXT == message.content_type:
self.tts_text_buff.append(message.content_detail)
segment_text = self._get_segment_text()
if segment_text:
self.to_tts(segment_text)
elif ContentType.FILE == message.content_type:
logger.bind(tag=TAG).info(
f"添加音频文件到待播放列表: {message.content_file}"
)
self.before_stop_play_files.append(
(message.content_file, message.content_detail)
)
if message.sentence_type == SentenceType.LAST:
# 处理剩余的文本
self._process_remaining_text(True)
except queue.Empty:
continue
except Exception as e:
logger.bind(tag=TAG).error(
f"处理TTS文本失败: {str(e)}, 类型: {type(e).__name__}, 堆栈: {traceback.format_exc()}"
)
def _process_remaining_text(self, is_last=False):
"""处理剩余的文本并生成语音
Returns:
bool: 是否成功处理了文本
"""
full_text = "".join(self.tts_text_buff)
remaining_text = full_text[self.processed_chars :]
if remaining_text:
segment_text = textUtils.get_string_no_punctuation_or_emoji(remaining_text)
if segment_text:
self.to_tts(segment_text, is_last)
self.processed_chars += len(full_text)
else:
self._process_before_stop_play_files()
def to_tts(self, text, is_last=False):
try:
max_repeat_time = 5
text = MarkdownCleaner.clean_markdown(text)
try:
asyncio.run(self.text_to_speak(text, is_last))
except Exception as e:
logger.bind(tag=TAG).warning(
f"语音生成失败{5 - max_repeat_time + 1}次: {text}，错误: {e}"
)
max_repeat_time -= 1
if max_repeat_time > 0:
logger.bind(tag=TAG).info(
f"语音生成成功: {text}，重试{5 - max_repeat_time}次"
)
else:
logger.bind(tag=TAG).error(
f"语音生成失败: {text}，请检查网络或服务是否正常"
)
except Exception as e:
logger.bind(tag=TAG).error(f"Failed to generate TTS file: {e}")
finally:
return None
###################################################################################
# linkerai单流式TTS重写父类的方法--结束
###################################################################################
async def text_to_speak(self, text, is_last):
"""流式处理TTS音频，每句只推送一次音频列表"""
await self._tts_request(text, is_last)
async def close(self):
"""资源清理"""
await super().close()
if hasattr(self, "opus_encoder"):
self.opus_encoder.close()
async def _tts_request(self, text: str, is_last: bool) -> None:
params = {
"tts_text": text,
"spk_id": self.voice,
"frame_durition": 60,
"stream": "true",
"target_sr": 16000,
"audio_format": "pcm",
"instruct_text": "请生成一段自然流畅的语音",
}
headers = {
"Authorization": f"Bearer {self.access_token}",
"Content-Type": "application/json",
}
# 一帧 PCM 所需字节数：60 ms &times; 16 kHz &times; 1 ch &times; 2 B = 1 920
frame_bytes = int(
self.opus_encoder.sample_rate
* self.opus_encoder.channels # 1
* self.opus_encoder.frame_size_ms
/ 1000
* 2
) # 16-bit = 2 bytes
try:
async with aiohttp.ClientSession() as session:
async with session.get(
self.api_url, params=params, headers=headers, timeout=10
) as resp:
if resp.status != 200:
logger.error(f"TTS请求失败: {resp.status}, {await resp.text()}")
self.tts_audio_queue.put((SentenceType.LAST, [], None))
return
self.pcm_buffer.clear()
opus_datas_cache = []
# 兼容 iter_chunked / iter_chunks / iter_any
async for chunk in resp.content.iter_any():
data = chunk[0] if isinstance(chunk, (list, tuple)) else chunk
if not data:
continue
# 拼到 buffer
self.pcm_buffer.extend(data)
# 够一帧就编码
while len(self.pcm_buffer) >= frame_bytes:
frame = bytes(self.pcm_buffer[:frame_bytes])
del self.pcm_buffer[:frame_bytes]
opus = self.opus_encoder.encode_pcm_to_opus(
frame, end_of_stream=False
)
if opus:
if self.segment_count < 10: # 前10个片段直接发送
self.tts_audio_queue.put(
(SentenceType.MIDDLE, opus, text)
)
self.segment_count += 1
else:
opus_datas_cache.extend(opus)
# flush 剩余不足一帧的数据
if self.pcm_buffer:
opus = self.opus_encoder.encode_pcm_to_opus(
bytes(self.pcm_buffer), end_of_stream=True
)
if opus:
if self.segment_count < 10: # 前10个片段直接发送
# 直接发送
self.tts_audio_queue.put(
(SentenceType.MIDDLE, opus, text)
)
self.segment_count += 1
else:
# 后续片段缓存
opus_datas_cache.extend(opus)
self.pcm_buffer.clear()
# 如果不是前10个片段，发送缓存的数据
if self.segment_count >= 10 and opus_datas_cache:
self.tts_audio_queue.put(
(SentenceType.MIDDLE, opus_datas_cache, text)
)
# 如果是最后一段，输出音频获取完毕
if is_last:
self._process_before_stop_play_files()
except Exception as e:
logger.error(f"TTS请求异常: {e}")
self.tts_audio_queue.put((SentenceType.LAST, [], None))
main/xiaozhi-server/core/providers/vllm/openai.py
+3 -1
View File
@@ -46,7 +46,9 @@ class VLLMProvider(VLLMProviderBase):
{"type": "text", "text": question},
{
"type": "image_url",
"image_url": {"url": f"{base64_image}"},
"image_url": {
"url": f"data:image/jpeg;base64,{base64_image}"
},
},
],
}
main/xiaozhi-server/performance_tester.py
+4 -2
View File
@@ -140,7 +140,9 @@ class AsyncPerformanceTester:
print(f"🎵 测试 STT: {stt_name}")
text, _ = await stt.speech_to_text([self.test_wav_list[0]], "1")
text, _ = await stt.speech_to_text(
[self.test_wav_list[0]], "1", stt.audio_format
)
if text is None:
print(f"❌ {stt_name} 连接失败")
@@ -151,7 +153,7 @@ class AsyncPerformanceTester:
for i, sentence in enumerate(self.test_wav_list, 1):
start = time.time()
text, _ = await stt.speech_to_text([sentence], "1")
text, _ = await stt.speech_to_text([sentence], "1", stt.audio_format)
duration = time.time() - start
total_time += duration
main/xiaozhi-server/performance_tester_vllm.py
+189
View File
@@ -0,0 +1,189 @@
import time
import asyncio
import logging
import statistics
import base64
from typing import Dict
from tabulate import tabulate
from config.settings import load_config
from core.utils.vllm import create_instance
# 设置全局日志级别为WARNING，抑制INFO级别日志
logging.basicConfig(level=logging.WARNING)
class AsyncVisionPerformanceTester:
def __init__(self):
self.config = load_config()
self.test_images = [
"../../docs/images/demo1.png",
"../../docs/images/demo2.png",
]
self.test_questions = [
"这张图片里有什么？",
"请详细描述这张图片的内容",
]
# 加载测试图片
self.results = {"vllm": {}}
async def _test_vllm(self, vllm_name: str, config: Dict) -> Dict:
"""异步测试单个视觉大模型性能"""
try:
# 检查API密钥配置
if "api_key" in config and any(
x in config["api_key"] for x in ["你的", "placeholder", "sk-xxx"]
):
print(f"⏭️ VLLM {vllm_name} 未配置api_key，已跳过")
return {"name": vllm_name, "type": "vllm", "errors": 1}
# 获取实际类型（兼容旧配置）
module_type = config.get("type", vllm_name)
vllm = create_instance(module_type, config)
print(f"🖼️ 测试 VLLM: {vllm_name}")
# 创建所有测试任务
test_tasks = []
for question in self.test_questions:
for image in self.test_images:
test_tasks.append(
self._test_single_vision(vllm_name, vllm, question, image)
)
# 并发执行所有测试
test_results = await asyncio.gather(*test_tasks)
# 处理结果
valid_results = [r for r in test_results if r is not None]
if not valid_results:
print(f"⚠️ {vllm_name} 无有效数据，可能配置错误")
return {"name": vllm_name, "type": "vllm", "errors": 1}
response_times = [r["response_time"] for r in valid_results]
# 过滤异常数据
mean = statistics.mean(response_times)
stdev = statistics.stdev(response_times) if len(response_times) > 1 else 0
filtered_times = [t for t in response_times if t <= mean + 3 * stdev]
if len(filtered_times) < len(test_tasks) * 0.5:
print(f"⚠️ {vllm_name} 有效数据不足，可能网络不稳定")
return {"name": vllm_name, "type": "vllm", "errors": 1}
return {
"name": vllm_name,
"type": "vllm",
"avg_response": sum(response_times) / len(response_times),
"std_response": (
statistics.stdev(response_times) if len(response_times) > 1 else 0
),
"errors": 0,
}
except Exception as e:
print(f"⚠️ VLLM {vllm_name} 测试失败: {str(e)}")
return {"name": vllm_name, "type": "vllm", "errors": 1}
async def _test_single_vision(
self, vllm_name: str, vllm, question: str, image: str
) -> Dict:
"""测试单个视觉问题的性能"""
try:
print(f"📝 {vllm_name} 开始测试: {question[:20]}...")
start_time = time.time()
# 读取图片并转换为base64
with open(image, "rb") as image_file:
image_data = image_file.read()
image_base64 = base64.b64encode(image_data).decode("utf-8")
# 直接获取响应
response = vllm.response(question, image_base64)
response_time = time.time() - start_time
print(f"✓ {vllm_name} 完成响应: {response_time:.3f}s")
return {
"name": vllm_name,
"type": "vllm",
"response_time": response_time,
}
except Exception as e:
print(f"⚠️ {vllm_name} 测试失败: {str(e)}")
return None
def _print_results(self):
"""打印测试结果"""
vllm_table = []
for name, data in self.results["vllm"].items():
if data["errors"] == 0:
stability = data["std_response"] / data["avg_response"]
vllm_table.append(
[
name,
f"{data['avg_response']:.3f}秒",
f"{stability:.3f}",
]
)
if vllm_table:
print("\n视觉大模型性能排行:\n")
print(
tabulate(
vllm_table,
headers=["模型名称", "响应耗时", "稳定性"],
tablefmt="github",
colalign=("left", "right", "right"),
disable_numparse=True,
)
)
else:
print("\n⚠️ 没有可用的视觉大模型进行测试。")
async def run(self):
"""执行全量异步测试"""
print("🔍 开始筛选可用视觉大模型...")
if not self.test_images:
print(f"\n⚠️ {self.image_root} 路径下没有图片文件，无法进行测试")
return
# 创建所有测试任务
all_tasks = []
# VLLM测试任务
if self.config.get("VLLM") is not None:
for vllm_name, config in self.config.get("VLLM", {}).items():
if "api_key" in config and any(
x in config["api_key"] for x in ["你的", "placeholder", "sk-xxx"]
):
print(f"⏭️ VLLM {vllm_name} 未配置api_key，已跳过")
continue
print(f"🖼️ 添加VLLM测试任务: {vllm_name}")
all_tasks.append(self._test_vllm(vllm_name, config))
print(f"\n✅ 找到 {len(all_tasks)} 个可用视觉大模型")
print(f"✅ 使用 {len(self.test_images)} 张测试图片")
print(f"✅ 使用 {len(self.test_questions)} 个测试问题")
print("\n⏳ 开始并发测试所有模型...\n")
# 并发执行所有测试任务
all_results = await asyncio.gather(*all_tasks, return_exceptions=True)
# 处理结果
for result in all_results:
if isinstance(result, dict) and result["errors"] == 0:
self.results["vllm"][result["name"]] = result
# 打印结果
print("\n📊 生成测试报告...")
self._print_results()
async def main():
tester = AsyncVisionPerformanceTester()
await tester.run()
if __name__ == "__main__":
asyncio.run(main())
main/xiaozhi-server/requirements.txt
+1 -1
View File
@@ -21,7 +21,7 @@ cozepy==0.12.0
mem0ai==0.1.62
bs4==0.0.2
modelscope==1.23.2
sherpa_onnx==1.11.0
sherpa_onnx==1.12.0
mcp==1.8.1
cnlunar==0.2.0
PySocks==1.7.1