Compare commits

..
15 Commits
Author SHA1 Message Date
87cb0b484d Music play (#142)
* Update util.py解决返回元组报错 (#103)

解决返回元组报错

* 增加播放本地音乐功能 (#105)

* Please enter the commit message for your changes. Lines starting
 with '#' will be ignored, and an empty message aborts the commit.

 On branch Music-playback
 Your branch is up to date with 'origin/Music-playback'.

 Changes to be committed:
	modified:   core/handle/audioHandle.py
	new file:   "music/\346\234\210\344\272\256\344\273\243\350\241\250\346\210\221\347\232\204\345\277\203_\351\202\223\344\270\275\345\220\233.mp3"
	new file:   "music/\350\270\217\345\261\261\346\262\263.mp3"

* Please enter the commit message for your changes. Lines starting
with '#' will be ignored, and an empty message aborts the commit.

On branch Music-playback
Your branch is up to date with 'origin/Music-playback'.

Changes to be committed:
	modified:   config.yaml
	modified:   core/connection.py
	modified:   core/handle/abortHandle.py
	modified:   core/handle/audioHandle.py
	new file:   core/handle/musicHandler.py
	modified:   core/handle/textHandle.py
	modified:   core/websocket_server.py
	new file:   "music/\344\270\200\345\277\265\345\215\203\345\271\264_\345\233\275\351\243\216\347\211\210.mp3"
	new file:   "music/\344\270\255\347\247\213\346\234\210.mp3"
	deleted:    "music/\346\234\210\344\272\256\344\273\243\350\241\250\346\210\221\347\232\204\345\277\203_\351\202\223\344\270\275\345\220\233.mp3"
	deleted:    "music/\350\270\217\345\261\261\346\262\263.mp3"

* 增加播放本地音乐功能

* 增加播放本地音乐功能

* 让音乐配置变的优雅

On branch Music-playback
Your branch is up to date with 'origin/Music-playback'.

Changes to be committed:
	modified:   config.yaml
	modified:   core/handle/musicHandler.py
	renamed:    "music/\344\270\200\345\277\265\345\215\203\345\271\264_\345\233\275\351\243\216\347\211\210.mp3" -> "music/mp3/\344\270\200\345\277\265\345\215\203\345\271\264_\345\233\275\351\243\216\347\211\210.mp3"
	renamed:    "music/\344\270\255\347\247\213\346\234\210.mp3" -> "music/mp3/\344\270\255\347\247\213\346\234\210.mp3"
	renamed:    "music/\345\273\211\346\263\242\350\200\201\347\237\243\357\274\214\345\260\232\350\203\275\351\245\255\345\220\246.mp3" -> "music/mp3/\345\273\211\346\263\242\350\200\201\347\237\243\357\274\214\345\260\232\350\203\275\351\245\255\345\220\246.mp3"
	new file:   music/music_config.yaml

---------

Co-authored-by: 欣南科技 <huangrongzhuang@xin-nan.com>

* update:优化代码

* update:流控调试

* update:优化音乐播放

* update:优化音乐配置初始化

---------

Co-authored-by: linqingping <linqingping@users.noreply.github.com>
Co-authored-by: Chris <119588753+Chris-websketch@users.noreply.github.com>
Co-authored-by: hrz <1710360675@qq.com>
2025-02-26 01:33:05 +08:00
73803f49fc update:wechat group (#141)
Co-authored-by: hrz <1710360675@qq.com>
2025-02-25 23:51:57 +08:00
5fd87752d5 fixed:更新基础docker环境地址 (#139)
Co-authored-by: hrz <1710360675@qq.com>
2025-02-25 21:11:06 +08:00
94a90c4124 Iot (#129)
* 2025-2-17-iot配置消息处理, 以及控制iot命令发送 (#40)

* update:优化

* update:增加视频教程

* update:iot兼容性又问题,撤回

* fixed:iot boolean 参数接收bug

---------

Co-authored-by: Jiao Haoyang <108573524+XuSenfeng@users.noreply.github.com>
Co-authored-by: hrz <1710360675@qq.com>
2025-02-24 22:04:48 +08:00
c3b0347155 Iot (#128)
* 2025-2-17-iot配置消息处理, 以及控制iot命令发送 (#40)

* update:优化

* update:增加视频教程

* update:iot兼容性又问题,撤回

---------

Co-authored-by: Jiao Haoyang <108573524+XuSenfeng@users.noreply.github.com>
Co-authored-by: hrz <1710360675@qq.com>
2025-02-24 21:20:42 +08:00
600f712750 Iot (#127)
* 2025-2-17-iot配置消息处理, 以及控制iot命令发送 (#40)

* update:优化

* update:增加视频教程

---------

Co-authored-by: Jiao Haoyang <108573524+XuSenfeng@users.noreply.github.com>
Co-authored-by: hrz <1710360675@qq.com>
2025-02-24 18:18:08 +08:00
9e76869c36 Iot (#126)
* 2025-2-17-iot配置消息处理, 以及控制iot命令发送 (#40)

* update:优化

---------

Co-authored-by: Jiao Haoyang <108573524+XuSenfeng@users.noreply.github.com>
Co-authored-by: hrz <1710360675@qq.com>
2025-02-24 18:06:13 +08:00
cb540736ab 跳过使用Open ai 接口时,DeepSeek-R1 模型的深度思考内容 (#54) (#95)
* 跳过使用Open ai 接口时,DeepSeek-R1 模型的深度思考内容 (#54)

* 跳过 DeepSeek-R1 模型的深度思考内容

* 跳过 DeepSeek-R1 模型的深度思考内容

* 新增LM Studio本地大模型API接口

* 优化代码,遇到Bad Case安全处理

* update:优化

---------

Co-authored-by: Sinyo <38577585+SinyoWong@users.noreply.github.com>
Co-authored-by: hrz <1710360675@qq.com>
2025-02-24 16:16:06 +08:00
66cd5bb4a5 fixed:修复低版本docker-compose模版参数 (#124)
Co-authored-by: hrz <1710360675@qq.com>
2025-02-24 14:56:39 +08:00
1226e8be47 Fix opuslib (#122)
* fixed:ModuleNotFoundError: No module named 'opuslib'

* update:格式化文档

---------

Co-authored-by: hrz <1710360675@qq.com>
2025-02-23 20:14:46 +08:00
22ff7a92b8 fixed:ModuleNotFoundError: No module named 'opuslib' (#121)
Co-authored-by: hrz <1710360675@qq.com>
2025-02-23 19:59:26 +08:00
2d01812f8d Performance test (#118)
* update: optimize performance test files

* update:异步测试方法

* update:改回默认FunASR

---------

Co-authored-by: Alexisxty <alexisty233@gmail.com>
Co-authored-by: hrz <1710360675@qq.com>
2025-02-23 16:22:21 +08:00
3bc0b821a1 update: 修改了服务组件的性能测试代码 (#111)
* update: optimize performance test files

* update:异步测试方法

---------

Co-authored-by: hrz <1710360675@qq.com>
2025-02-23 16:07:33 +08:00
148578399f 使用配置文件设置日志参数 (#116)
* 使用配置文件设置日志参数

* update:优化

---------

Co-authored-by: 欣南科技 <huangrongzhuang@xin-nan.com>
Co-authored-by: hrz <1710360675@qq.com>
2025-02-23 15:18:59 +08:00
6c3fbd7a8f Doubao asr (#117)
* 添加豆包在线 ASR (#82)
---------
Co-authored-by: 胡垚 <qsct9501@163.com>
2025-02-23 14:38:21 +08:00
39 changed files with 1622 additions and 1003 deletions
+1
View File
@@ -1,5 +1,6 @@
# Byte-compiled / optimized / DLL files
__pycache__/
.idea/
*.py[cod]
*$py.class
+3 -3
View File
@@ -25,7 +25,7 @@ COPY requirements.txt .
# 优化apt安装
RUN pip install --no-cache-dir -r requirements.txt \
-i https://pypi.tuna.tsinghua.edu.cn/simple
-i https://mirrors.aliyun.com/pypi/simple/
# 第三阶段:生产镜像
FROM python:3.10-slim
@@ -33,8 +33,8 @@ FROM python:3.10-slim
WORKDIR /opt/xiaozhi-esp32-server
# 优化apt安装
RUN echo "deb https://mirrors.tuna.tsinghua.edu.cn/debian/ bookworm main contrib non-free non-free-firmware" > /etc/apt/sources.list && \
echo "deb https://mirrors.tuna.tsinghua.edu.cn/debian/ bookworm-updates main contrib non-free non-free-firmware" >> /etc/apt/sources.list && \
RUN echo "deb https://mirrors.aliyun.com/debian/ bookworm main contrib non-free non-free-firmware" > /etc/apt/sources.list && \
echo "deb https://mirrors.aliyun.com/debian/ bookworm-updates main contrib non-free non-free-firmware" >> /etc/apt/sources.list && \
apt-get update && \
apt-get install -y --no-install-recommends libopus0 ffmpeg && \
apt-get clean && \
+68 -17
View File
@@ -14,15 +14,49 @@
本项目需要配合 ESP32 硬件设备使用。如果您已经购买了 ESP32 相关硬件,且成功对接过虾哥部署的后端服务,并希望独立搭建自己的
`xiaozhi-esp32` 后端服务,那么本项目非常适合您。
跳转到[使用方式](#使用方式-)
想看使用效果?请猛戳视频 🎥
想看使用效果?请猛戳这个视频 🎥
<a href="https://www.bilibili.com/video/BV1FMFyejExX">
<picture>
<img alt="小智esp32连接自己的后台模型" src="docs/images/demo.png" />
</picture>
</a>
<table>
<tr>
<td>
<a href="https://www.bilibili.com/video/BV1FMFyejExX" target="_blank">
<picture>
<img alt="小智esp32连接自己的后台模型" src="docs/images/demo1.png" />
</picture>
</a>
</td>
<td>
<a href="https://www.bilibili.com/video/BV1CDKWemEU6" target="_blank">
<picture>
<img alt="自定义音色" src="docs/images/demo2.png" />
</picture>
</a>
</td>
<td>
<a href="https://www.bilibili.com/video/BV12yA2egEaC" target="_blank">
<picture>
<img alt="使用粤语交流" src="docs/images/demo3.png" />
</picture>
</a>
</td>
<td>
<a href="https://www.bilibili.com/video/av114036381327149" target="_blank">
<picture>
<img alt="控制家电开关" src="docs/images/demo5.png" />
</picture>
</a>
</td>
<td>
<a href="https://www.bilibili.com/video/BV1kgA2eYEQ9" target="_blank">
<picture>
<img alt="成本最低配置" src="docs/images/demo4.png" />
</picture>
</a>
</td>
<td>
</td>
</tr>
</table>
---
@@ -30,7 +64,7 @@
- **硬件**:一套兼容 `xiaozhi-esp32`
的硬件设备(具体型号请参考 [此处](https://rcnv1t9vps13.feishu.cn/wiki/DdgIw4BUgivWDPkhMj1cGIYCnRf))。
- **服务器**:至少 4 核 CPU、8G 内存的电脑或服务器
- **服务器**:至少 4 核 CPU、8G 内存的电脑。
- **固件编译**:请将后端服务的接口地址更新至 `xiaozhi-esp32` 项目中,再重新编译固件并烧录到设备上。
---
@@ -118,9 +152,10 @@ server:
### ASR
| 类型 | 平台名称 | 使用方式 | 收费模式 | 备注 |
|:---:|:------:|:----:|:----:|:--:|
| ASR | FunASR | 本地使用 | 免费 | |
| 类型 | 平台名称 | 使用方式 | 收费模式 | 备注 |
|:---:|:---------:|:----:|:----:|:--:|
| ASR | FunASR | 本地使用 | 免费 | |
| ASR | DoubaoASR | 接口调用 | 收费 | |
---
@@ -128,9 +163,24 @@ server:
### 一、[部署文档](./docs/Deployment.md)
1.**[本地源码运行](./docs/Deployment.md#%E6%96%B9%E5%BC%8F%E4%B8%89%E6%9C%AC%E5%9C%B0%E6%BA%90%E7%A0%81%E8%BF%90%E8%A1%8C)**
适合熟悉 Conda 环境或希望从零搭建运行环境的用户。
本项目支持以下三种部署方式,您可根据实际需求选择。
本项目的文档主要是`文字版本`的教程,如果你想要`视频版本`的教程,您可以学习一下[这个大佬的手把手教程](https://www.bilibili.com/video/BV1gePuejEvT)。
如果你能把`文字版本的教程``视频版本的教程`结合起来一起看,可以让你更快上手。
1. [Docker 快速部署](./docs/Deployment.md)
适合快速体验的普通用户,不需过多环境配置。缺点是,拉取镜像有点慢。
2. [借助 Docker 环境运行部署](./docs/Deployment.md#%E6%96%B9%E5%BC%8F%E4%BA%8C%E5%80%9F%E5%8A%A9docker%E7%8E%AF%E5%A2%83%E8%BF%90%E8%A1%8C%E9%83%A8%E7%BD%B2)
适用于已安装 Docker 且希望对代码进行自定义修改的软件工程师。
3. [本地源码运行](./docs/Deployment.md#%E6%96%B9%E5%BC%8F%E4%B8%89%E6%9C%AC%E5%9C%B0%E6%BA%90%E7%A0%81%E8%BF%90%E8%A1%8C)
适合熟悉`Conda` 环境或希望从零搭建运行环境的用户。
对于对响应速度要求较高的场景,推荐使用本地源码运行方式以降低额外开销。
### 二、[固件编译](./docs/firmware-build.md)
@@ -166,13 +216,15 @@ VAD:
### 4、为什么我说的话,小智识别出来很多韩文、日文、英文?🇰🇷
建议:检查一下`models/SenseVoiceSmall`是否已经有`model.pt`文件,如果没有就要下载,查看这里[下载语音识别模型文件](docs/Deployment.md#模型文件)
建议:检查一下`models/SenseVoiceSmall`是否已经有`model.pt`
文件,如果没有就要下载,查看这里[下载语音识别模型文件](docs/Deployment.md#模型文件)
### 5、为什么会出现“TTS 任务出错 文件不存在”?📁
建议:检查一下是否正确使用`conda` 安装了`libopus``ffmpeg`库。
如果没有安装,就安装
```
conda install conda-forge::libopus
conda install conda-forge::ffmpeg
@@ -245,7 +297,6 @@ TTS 性能排行:
## 鸣谢 🙏
- 本项目受 [百聆语音对话机器人](https://github.com/wwbin2017/bailing) 启发,并在其基础上实现。
- 感谢 [腾讯云](https://cloud.tencent.com/) 提供免费 Docker 镜像空间。
- 感谢 [十方融海](https://www.tenclass.com/) 对小智通讯协议提供的详尽文档支持。
<a href="https://star-history.com/#xinnan-tech/xiaozhi-esp32-server&Date">
+45 -2
View File
@@ -22,13 +22,28 @@ server:
# 可选:设备白名单,如果设置了白名单,那么白名单的机器无论是什么token都可以连接。
#allowed_devices:
# - "24:0A:C4:1D:3B:F0" # MAC地址列表
log:
# 设置控制台输出的日志格式,时间、日志级别、标签、消息
log_format: "<green>{time:YY-MM-DD HH:mm:ss}</green>[<light-blue>{extra[tag]}</light-blue>] - <level>{level}</level> - <light-green>{message}</light-green>"
# 设置日志文件输出的格式,时间、日志级别、标签、消息
log_format_simple: "{time:YYYY-MM-DD HH:mm:ss} - {name} - {level} - {extra[tag]} - {message}"
# 设置日志等级:INFO、DEBUG
log_level: INFO
# 设置日志路径
log_dir: tmp
# 设置日志文件
log_file: "server.log"
# 设置数据文件路径
data_dir: data
manager:
# 是否启用管理后台
# 目前这个模块还在开发中,建议:不要修改enabled选项
enabled: false
ip: 0.0.0.0
port: 8002
iot:
Speaker:
volume: 100
xiaozhi:
type: hello
version: 1
@@ -68,9 +83,15 @@ selected_module:
ASR:
FunASR:
type: fun_local
model_dir: models/SenseVoiceSmall
output_dir: tmp/
DoubaoASR:
type: doubao
appid: 你的火山引擎语音合成服务appid
access_token: 你的火山引擎语音合成服务access_token
cluster: volcengine_input_common
output_dir: tmp/
VAD:
SileroVAD:
threshold: 0.5
@@ -128,6 +149,12 @@ LLM:
user_id: 你的user_id
base_url: "https://api.coze.cn/open_api/v2/chat" # 服务地址
personal_access_token: 你的coze个人令牌
LMStudioLLM:
# 定义LLM API类型
type: openai
model_name: deepseek-r1-distill-llama-8b@q4_k_m # 使用的模型名称,需要预先在社区下载
url: http://localhost:1234/v1 # LM Studio服务地址
api_key: lm-studio # LM Studio服务的固定API Key
HomeAssistant:
# 定义LLM API类型
type: homeassistant
@@ -285,3 +312,19 @@ module_test:
- "你好,请介绍一下你自己"
- "What's the weather like today?"
- "请用100字概括量子计算的基本原理和应用前景"
# 本地音乐播放配置
music:
music_commands:
- "来一首歌"
- "唱一首歌"
- "播放音乐"
- "来点音乐"
- "背景音乐"
- "放首歌"
- "播放歌曲"
- "来点背景音乐"
- "我想听歌"
- "我要听歌"
- "放点音乐"
music_dir: "./music" # 音乐文件存放路径
+14 -12
View File
@@ -1,27 +1,29 @@
import os
import sys
from loguru import logger
from config.settings import load_config
def setup_logging():
"""从配置文件中读取日志配置,并设置日志输出格式和级别"""
config = load_config()
log_config = config["log"]
log_format = log_config.get("log_format", "<green>{time:YY-MM-DD HH:mm:ss}</green>[<light-blue>{extra[tag]}</light-blue>] - <level>{level}</level> - <light-green>{message}</light-green>")
log_format_simple = log_config.get("log_format_file", "{time:YYYY-MM-DD HH:mm:ss} - {name} - {level} - {extra[tag]} - {message}")
log_level = log_config.get("log_level", "INFO")
log_dir = log_config.get("log_dir", "tmp")
log_file = log_config.get("log_file", "server.log")
data_dir = log_config.get("data_dir", "data")
def setup_logging(log_dir='tmp', data_dir='data'):
"""配置全局彩色日志(不同区块不同标签)"""
os.makedirs(log_dir, exist_ok=True)
os.makedirs(data_dir, exist_ok=True)
# 设置日志格式,时间、日志级别、标签、消息
log_format = (
# "<green>{time:YYYY-MM-DD HH:mm:ss}</green>"
"[<light-blue>{extra[tag]}</light-blue>]"
" - <level>{level}</level> - "
"<light-green>{message}</light-green>"
)
# 配置日志输出
logger.remove()
# 输出到控制台
logger.add(sys.stdout, format=log_format, level="INFO")
logger.add(sys.stdout, format=log_format, level=log_level)
# 输出到文件
logger.add(os.path.join(log_dir, "server.log"), format="{time:YYYY-MM-DD HH:mm:ss} - {name} - {level} - {extra[tag]} - {message}", level="INFO")
logger.add(os.path.join(log_dir, log_file), format=log_format_simple, level=log_level)
return logger
+26 -14
View File
@@ -4,6 +4,7 @@ import uuid
import time
import queue
import asyncio
import traceback
from config.logger import setup_logging
import threading
import websockets
@@ -14,15 +15,18 @@ from core.utils.dialogue import Message, Dialogue
from core.handle.textHandle import handleTextMessage
from core.utils.util import get_string_no_punctuation_or_emoji
from concurrent.futures import ThreadPoolExecutor, TimeoutError
from core.handle.audioHandle import handleAudioMessage, sendAudioMessage
from core.handle.sendAudioHandle import sendAudioMessage
from core.handle.receiveAudioHandle import handleAudioMessage
from config.private_config import PrivateConfig
from core.auth import AuthMiddleware, AuthenticationError
from core.utils.auth_code_gen import AuthCodeGenerator # 添加导入
from core.utils.auth_code_gen import AuthCodeGenerator
TAG = __name__
class ConnectionHandler:
def __init__(self, config: Dict[str, Any], _vad, _asr, _llm, _tts):
def __init__(self, config: Dict[str, Any], _vad, _asr, _llm, _tts, _music):
self.config = config
self.logger = setup_logging()
self.auth = AuthMiddleware(config)
@@ -72,28 +76,33 @@ class ConnectionHandler:
self.tts_start_speak_time = None
self.tts_duration = 0
# iot相关变量
self.iot_descriptors = {}
self.cmd_exit = self.config["CMD_exit"]
self.max_cmd_length = 0
for cmd in self.cmd_exit:
if len(cmd) > self.max_cmd_length:
self.max_cmd_length = len(cmd)
self.private_config = None
self.auth_code_gen = AuthCodeGenerator.get_instance()
self.is_device_verified = False # 添加设备验证状态标志
self.music_handler = _music
async def handle_connection(self, ws):
try:
# 获取并验证headers
self.headers = dict(ws.request.headers)
self.logger.bind(tag=TAG).info(f"New connection request - Headers: {self.headers}")
# 获取客户端ip地址
client_ip = ws.remote_address[0]
self.logger.bind(tag=TAG).info(f"{client_ip} conn - Headers: {self.headers}")
# 进行认证
await self.auth.authenticate(self.headers)
device_id = self.headers.get("device-id", None)
# Load private configuration if device_id is provided
bUsePrivateConfig = self.config.get("use_private_config", False)
self.logger.bind(tag=TAG).info(f"bUsePrivateConfig: {bUsePrivateConfig}, device_id: {device_id}")
@@ -104,10 +113,10 @@ class ConnectionHandler:
# 判断是否已经绑定
owner = self.private_config.get_owner()
self.is_device_verified = owner is not None
if self.is_device_verified:
await self.private_config.update_last_chat_time()
await self.private_config.update_last_chat_time()
llm, tts = self.private_config.create_private_instances()
if all([llm, tts]):
self.llm = llm
@@ -146,7 +155,8 @@ class ConnectionHandler:
await ws.close()
return
except Exception as e:
self.logger.bind(tag=TAG).error(f"Connection error: {str(e)}")
stack_trace = traceback.format_exc()
self.logger.bind(tag=TAG).error(f"Connection error: {str(e)}-{stack_trace}")
await ws.close()
return
@@ -166,7 +176,7 @@ class ConnectionHandler:
date_time = time.strftime("%Y-%m-%d %H:%M", time.localtime())
self.prompt = self.prompt.replace("{date_time}", date_time)
self.dialogue.put(Message(role="system", content=self.prompt))
async def _check_and_broadcast_auth_code(self):
"""检查设备绑定状态并广播认证码"""
if not self.private_config.get_owner():
@@ -186,7 +196,7 @@ class ConnectionHandler:
# 如果不使用私有配置,就不需要验证
return False
return not self.is_device_verified
def chat(self, query):
# 如果设备未验证,就发送验证码
if self.isNeedAuth():
@@ -199,7 +209,7 @@ class ConnectionHandler:
finally:
loop.close()
return True
self.dialogue.put(Message(role="user", content=query))
response_message = []
start = 0
@@ -316,6 +326,8 @@ class ConnectionHandler:
async def close(self):
"""资源清理方法"""
# 清理其他资源
self.stop_event.set()
self.executor.shutdown(wait=False)
if self.websocket:
+152
View File
@@ -0,0 +1,152 @@
import json
from config.logger import setup_logging
TAG = __name__
logger = setup_logging()
class IotDescriptor:
"""
A class to represent an IoT descriptor.
Attributes:
----------
name : str
The name of the IoT descriptor.
description : str
A brief description of the IoT descriptor.
properties : dict
A dictionary containing properties of the IoT descriptor.
methods : dict
A dictionary containing methods of the IoT descriptor.
-------
"""
def __init__(self, name, description, properties, methods):
self.name = name
self.description = description
self.properties = []
self.methods = []
# 根据描述创建属性
for key, value in properties.items():
# "volume":{"description":"当前音量 值","type":"number"}
"""
等价于
{
'name': 名字,
'description': 描述,
'value': 0
}
"""
# setattr(self, key, {}) # 创建一个空字典, 名字是属性名
property_item = globals()[key] = {} # 创建一个空字典, 名字是属性名
property_item['name'] = key
property_item["description"] = value["description"]
if value["type"] == "number":
property_item["value"] = 0
elif value["type"] == "boolean":
property_item["value"] = False
else:
property_item["value"] = ""
self.properties.append(property_item)
# 根据描述创建方法
for key, value in methods.items():
# "SetVolume": {"description":"设置音量","parameters":{"volume":{"description":"0到100之间的整数","type":"number"}}}
"""
等价于
SetVolume = {
`description`: 描述,
`volume`: {
`description`: 描述,
`value`: 0
}
}
"""
# setattr(self, key, {}) # 创建一个空字典, 名字是方法名
method = globals()[key] = {} # 创建一个空字典, 名字是方法名
method["description"] = value["description"]
method['name'] = key
for k, v in value["parameters"].items():
# 不同的参数解析
method[k] = {}
method[k]["description"] = v["description"]
if v["type"] == "number":
method[k]["value"] = 0
elif v["type"] == "boolean":
method[k]["value"] = False
else:
method[k]["value"] = ""
self.methods.append(method)
async def handleIotDescriptors(conn, descriptors):
"""
处理物联网描述
示例: [{
"name":"Speaker",
"description":"当前 AI 机器人的扬声器",
"properties":{
"volume":{"description":"当前音量 值","type":"number"} 可以有boolean, number, string三种类型
},
"methods":{
"SetVolume":{
"description":"设置音量","parameters":{"volume":{"description":"0到100之间的整数","type":"number"}}
}
}
}]
descriptors: 描述列表
"""
for descriptor in descriptors:
iot_descriptor = IotDescriptor(descriptor["name"], descriptor["description"], descriptor["properties"],
descriptor["methods"])
conn.iot_descriptors[descriptor["name"]] = iot_descriptor
# 暂时从配置文件中设置音量,后期通过意图识别控制音量
default_iot_volume = 100
if "iot" in conn.config:
default_iot_volume = conn.config["iot"]["Speaker"]["volume"]
logger.bind(tag=TAG).info(f"服务端设置音量为{default_iot_volume}")
await send_iot_conn(conn, "Speaker", "SetVolume", {"volume": default_iot_volume})
async def send_iot_conn(conn, name, method_name, parameters):
"""
发送物联网指令
name: 设备名称 "Speaker"
method: 方法 "SetVolume"
parameters: 参数, 是一个字典 {"volume": 100}
发送示例:
{
"type": "iot",
"commands": [
{
"name" : "Speaker",
"method": "SetVolume",
"parameters": {
"volume": 100
}
}
]
}
"""
for key, value in conn.iot_descriptors.items():
if key == name:
# 找到了设备
for method in value.methods:
# 找到了方法
if method["name"] == method_name:
await conn.websocket.send(json.dumps({
"type": "iot",
"commands": [
{
"name": name,
"method": method_name,
"parameters": parameters
}
]
}))
return
logger.bind(tag=TAG).error(f"未找到方法{method_name}")
+109
View File
@@ -0,0 +1,109 @@
from config.logger import setup_logging
import os
import random
import difflib
import re
import traceback
from core.handle.sendAudioHandle import sendAudioMessage, send_stt_message
TAG = __name__
logger = setup_logging()
def _extract_song_name(text):
"""从用户输入中提取歌名"""
for keyword in ["", "播放", "", ""]:
if keyword in text:
parts = text.split(keyword)
if len(parts) > 1:
return parts[1].strip()
return None
def _find_best_match(potential_song, music_files):
"""查找最匹配的歌曲"""
best_match = None
highest_ratio = 0
for music_file in music_files:
song_name = os.path.splitext(music_file)[0]
ratio = difflib.SequenceMatcher(None, potential_song, song_name).ratio()
if ratio > highest_ratio and ratio > 0.4:
highest_ratio = ratio
best_match = music_file
return best_match
class MusicHandler:
def __init__(self, config):
self.config = config
self.music_related_keywords = []
if "music" in self.config:
self.music_config = self.config["music"]
self.music_dir = os.path.abspath(
self.music_config.get("music_dir", "./music") # 默认路径修改
)
self.music_related_keywords = self.music_config.get("music_commands", [])
else:
self.music_dir = os.path.abspath("./music")
self.music_related_keywords = ["来一首歌", "唱一首歌", "播放音乐", "来点音乐", "背景音乐", "放首歌",
"播放歌曲", "来点背景音乐", "我想听歌", "我要听歌", "放点音乐"]
async def handle_music_command(self, conn, text):
"""处理音乐播放指令"""
clean_text = re.sub(r'[^\w\s]', '', text).strip()
logger.bind(tag=TAG).debug(f"检查是否是音乐命令: {clean_text}")
# 尝试匹配具体歌名
if os.path.exists(self.music_dir):
music_files = [f for f in os.listdir(self.music_dir) if f.endswith('.mp3')]
logger.bind(tag=TAG).debug(f"找到的音乐文件: {music_files}")
potential_song = _extract_song_name(clean_text)
if potential_song:
best_match = _find_best_match(potential_song, music_files)
if best_match:
logger.bind(tag=TAG).info(f"找到最匹配的歌曲: {best_match}")
await self.play_local_music(conn, specific_file=best_match)
return True
# 检查是否是通用播放音乐命令
if any(cmd in clean_text for cmd in self.music_related_keywords):
await self.play_local_music(conn)
return True
return False
async def play_local_music(self, conn, specific_file=None):
"""播放本地音乐文件"""
try:
if not os.path.exists(self.music_dir):
logger.bind(tag=TAG).error(f"音乐目录不存在: {self.music_dir}")
return
# 确保路径正确性
if specific_file:
music_path = os.path.join(self.music_dir, specific_file)
if not os.path.exists(music_path):
logger.bind(tag=TAG).error(f"指定的音乐文件不存在: {music_path}")
return
selected_music = specific_file
else:
music_files = [f for f in os.listdir(self.music_dir) if f.endswith('.mp3')]
if not music_files:
logger.bind(tag=TAG).error("未找到MP3音乐文件")
return
selected_music = random.choice(music_files)
music_path = os.path.join(self.music_dir, selected_music)
text = f"正在播放{selected_music}"
await send_stt_message(conn, text)
conn.tts_first_text = selected_music
conn.tts_last_text = selected_music
conn.llm_finish_task = True
opus_packets, duration = conn.tts.wav_to_opus_data(music_path)
await sendAudioMessage(conn, opus_packets, duration, selected_music)
except Exception as e:
logger.bind(tag=TAG).error(f"播放音乐失败: {str(e)}")
logger.bind(tag=TAG).error(f"详细错误: {traceback.format_exc()}")
+81
View File
@@ -0,0 +1,81 @@
from config.logger import setup_logging
import asyncio
import time
from core.utils.util import remove_punctuation_and_length
from core.handle.sendAudioHandle import schedule_with_interrupt, send_stt_message
TAG = __name__
logger = setup_logging()
async def handleAudioMessage(conn, audio):
if not conn.asr_server_receive:
logger.bind(tag=TAG).debug(f"前期数据处理中,暂停接收")
return
if conn.client_listen_mode == "auto":
have_voice = conn.vad.is_vad(conn, audio)
else:
have_voice = conn.client_have_voice
# 如果本次没有声音,本段也没声音,就把声音丢弃了
if have_voice == False and conn.client_have_voice == False:
await no_voice_close_connect(conn)
conn.asr_audio.clear()
return
conn.client_no_voice_last_time = 0.0
conn.asr_audio.append(audio)
# 如果本段有声音,且已经停止了
if conn.client_voice_stop:
conn.client_abort = False
conn.asr_server_receive = False
# 音频太短了,无法识别
if len(conn.asr_audio) < 3:
conn.asr_server_receive = True
else:
text, file_path = await conn.asr.speech_to_text(conn.asr_audio, conn.session_id)
logger.bind(tag=TAG).info(f"识别文本: {text}")
text_len, text_without_punctuation = remove_punctuation_and_length(text)
if await conn.music_handler.handle_music_command(conn, text_without_punctuation):
conn.asr_server_receive = True
conn.asr_audio.clear()
return
if text_len <= conn.max_cmd_length and await handleCMDMessage(conn, text_without_punctuation):
return
if text_len > 0:
await startToChat(conn, text)
else:
conn.asr_server_receive = True
conn.asr_audio.clear()
conn.reset_vad_states()
async def handleCMDMessage(conn, text):
cmd_exit = conn.cmd_exit
for cmd in cmd_exit:
if text == cmd:
logger.bind(tag=TAG).info("识别到明确的退出命令".format(text))
await conn.close()
return True
return False
async def startToChat(conn, text):
# 异步发送 stt 信息
stt_task = asyncio.create_task(
schedule_with_interrupt(0, send_stt_message(conn, text))
)
conn.scheduled_tasks.append(stt_task)
conn.executor.submit(conn.chat, text)
async def no_voice_close_connect(conn):
if conn.client_no_voice_last_time == 0.0:
conn.client_no_voice_last_time = time.time() * 1000
else:
no_voice_time = time.time() * 1000 - conn.client_no_voice_last_time
close_connection_no_voice_time = conn.config.get("close_connection_no_voice_time", 120)
if no_voice_time > 1000 * close_connection_no_voice_time:
conn.client_abort = False
conn.asr_server_receive = False
prompt = "时间过得真快,我都好久没说话了。请你用十个字左右话跟我告别,以“再见”或“拜拜”为结尾"
await startToChat(conn, prompt)
@@ -8,53 +8,6 @@ TAG = __name__
logger = setup_logging()
async def handleAudioMessage(conn, audio):
if not conn.asr_server_receive:
logger.bind(tag=TAG).debug(f"前期数据处理中,暂停接收")
return
if conn.client_listen_mode == "auto":
have_voice = conn.vad.is_vad(conn, audio)
else:
have_voice = conn.client_have_voice
# 如果本次没有声音,本段也没声音,就把声音丢弃了
if have_voice == False and conn.client_have_voice == False:
await no_voice_close_connect(conn)
conn.asr_audio.clear()
return
conn.client_no_voice_last_time = 0.0
conn.asr_audio.append(audio)
# 如果本段有声音,且已经停止了
if conn.client_voice_stop:
conn.client_abort = False
conn.asr_server_receive = False
text, file_path = conn.asr.speech_to_text(conn.asr_audio, conn.session_id)
logger.bind(tag=TAG).info(f"识别文本: {text}")
text_len, text_without_punctuation = remove_punctuation_and_length(text)
if text_len <= conn.max_cmd_length and await handleCMDMessage(conn, text_without_punctuation):
return
if text_len > 0:
await startToChat(conn, text)
else:
conn.asr_server_receive = True
conn.asr_audio.clear()
conn.reset_vad_states()
async def handleCMDMessage(conn, text):
cmd_exit = conn.cmd_exit
for cmd in cmd_exit:
if text == cmd:
logger.bind(tag=TAG).info("识别到明确的退出命令".format(text))
await finishToChat(conn)
return True
return False
async def finishToChat(conn):
await conn.close()
async def isLLMWantToFinish(conn):
first_text = conn.tts_first_text
last_text = conn.tts_last_text
@@ -67,15 +20,6 @@ async def isLLMWantToFinish(conn):
return False
async def startToChat(conn, text):
# 异步发送 stt 信息
stt_task = asyncio.create_task(
schedule_with_interrupt(0, send_stt_message(conn, text))
)
conn.scheduled_tasks.append(stt_task)
conn.executor.submit(conn.chat, text)
async def sendAudioMessage(conn, audios, duration, text):
base_delay = conn.tts_duration
@@ -104,7 +48,7 @@ async def sendAudioMessage(conn, audios, duration, text):
conn.scheduled_tasks.append(stop_task)
if await isLLMWantToFinish(conn):
finish_task = asyncio.create_task(
schedule_with_interrupt(stop_duration, finishToChat(conn))
schedule_with_interrupt(stop_duration, await conn.close())
)
conn.scheduled_tasks.append(finish_task)
@@ -149,16 +93,3 @@ async def schedule_with_interrupt(delay, coro):
await coro
except asyncio.CancelledError:
pass
async def no_voice_close_connect(conn):
if conn.client_no_voice_last_time == 0.0:
conn.client_no_voice_last_time = time.time() * 1000
else:
no_voice_time = time.time() * 1000 - conn.client_no_voice_last_time
close_connection_no_voice_time = conn.config.get("close_connection_no_voice_time", 120)
if no_voice_time > 1000 * close_connection_no_voice_time:
conn.client_abort = False
conn.asr_server_receive = False
prompt = "时间过得真快,我都好久没说话了。请你用十个字左右话跟我告别,以“再见”或“拜拜拜”为结尾"
await startToChat(conn, prompt)
+5 -1
View File
@@ -2,7 +2,8 @@ from config.logger import setup_logging
import json
from core.handle.abortHandle import handleAbortMessage
from core.handle.helloHandle import handleHelloMessage
from core.handle.audioHandle import startToChat
from core.handle.receiveAudioHandle import startToChat
from core.handle.iotHandle import handleIotDescriptors
TAG = __name__
logger = setup_logging()
@@ -36,5 +37,8 @@ async def handleTextMessage(conn, message):
conn.asr_audio.clear()
if "text" in msg_json:
await startToChat(conn, msg_json["text"])
elif msg_json["type"] == "iot":
if "descriptors" in msg_json:
await handleIotDescriptors(conn, msg_json["descriptors"])
except json.JSONDecodeError:
await conn.websocket.send(message)
+19
View File
@@ -0,0 +1,19 @@
from abc import ABC, abstractmethod
from typing import Optional, Tuple, List
from config.logger import setup_logging
TAG = __name__
logger = setup_logging()
class ASRProviderBase(ABC):
@abstractmethod
def save_audio_to_file(self, opus_data: List[bytes], session_id: str) -> str:
"""解码Opus数据并保存为WAV文件"""
pass
@abstractmethod
async def speech_to_text(self, opus_data: List[bytes], session_id: str) -> Tuple[Optional[str], Optional[str]]:
"""将语音数据转换为文本"""
pass
+286
View File
@@ -0,0 +1,286 @@
import time
import io
import wave
import os
from typing import Optional, Tuple, List
import uuid
import websockets
import json
import gzip
import opuslib_next
from core.providers.asr.base import ASRProviderBase
from config.logger import setup_logging
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_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: dict, delete_audio_file: bool):
self.appid = config.get("appid")
self.cluster = config.get("cluster")
self.access_token = config.get("access_token")
self.output_dir = config.get("output_dir")
self.host = "openspeech.bytedance.com"
self.ws_url = f"wss://{self.host}/api/v2/asr"
self.success_code = 1000
self.seg_duration = 15000
# 确保输出目录存在
os.makedirs(self.output_dir, exist_ok=True)
def save_audio_to_file(self, opus_data: List[bytes], session_id: str) -> str:
"""将Opus音频数据解码并保存为WAV文件"""
file_name = f"asr_{session_id}_{uuid.uuid4()}.wav"
file_path = os.path.join(self.output_dir, file_name)
decoder = opuslib_next.Decoder(16000, 1) # 16kHz, 单声道
pcm_data = []
for opus_packet in opus_data:
try:
pcm_frame = decoder.decode(opus_packet, 960) # 960 samples = 60ms
pcm_data.append(pcm_frame)
except opuslib_next.OpusError as e:
logger.bind(tag=TAG).error(f"Opus解码错误: {e}", exc_info=True)
with wave.open(file_path, "wb") as wf:
wf.setnchannels(1)
wf.setsampwidth(2) # 2 bytes = 16-bit
wf.setframerate(16000)
wf.writeframes(b"".join(pcm_data))
return file_path
@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
def _construct_request(self, reqid) -> dict:
"""Construct the request payload."""
return {
"app": {
"appid": f"{self.appid}",
"cluster": self.cluster,
"token": self.access_token,
},
"user": {
"uid": str(uuid.uuid4()),
},
"request": {
"reqid": reqid,
"show_utterances": False,
"sequence": 1
},
"audio": {
"format": "wav",
"rate": 16000,
"language": "zh-CN",
"bits": 16,
"channel": 1,
"codec": "raw",
},
}
async def _send_request(self, audio_data: List[bytes], segment_size: int) -> Optional[str]:
"""Send request to Volcano ASR service."""
try:
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()))
print(request_params)
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
# 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
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
)
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)
# Receive response
response = await websocket.recv()
result = parse_response(response)
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 request failed: {e}", exc_info=True)
return None
@staticmethod
def decode_opus(opus_data: List[bytes], session_id: str) -> List[bytes]:
decoder = opuslib_next.Decoder(16000, 1) # 16kHz, 单声道
pcm_data = []
for opus_packet in opus_data:
try:
pcm_frame = decoder.decode(opus_packet, 960) # 960 samples = 60ms
pcm_data.append(pcm_frame)
except opuslib_next.OpusError as e:
logger.bind(tag=TAG).error(f"Opus解码错误: {e}", exc_info=True)
return pcm_data
@staticmethod
def read_wav_info(data: io.BytesIO = None) -> (int, int, int, int, int):
with io.BytesIO(data) as _f:
wave_fp = wave.open(_f, 'rb')
nchannels, sampwidth, framerate, nframes = wave_fp.getparams()[:4]
wave_bytes = wave_fp.readframes(nframes)
return nchannels, sampwidth, framerate, nframes, len(wave_bytes)
@staticmethod
def slice_data(data: bytes, chunk_size: int) -> (list, bool):
"""
slice data
:param data: wav data
:param chunk_size: the segment size in one request
:return: segment data, last flag
"""
data_len = len(data)
offset = 0
while offset + chunk_size < data_len:
yield data[offset: offset + chunk_size], False
offset += chunk_size
else:
yield data[offset: data_len], True
async def speech_to_text(self, opus_data: List[bytes], session_id: str) -> Tuple[Optional[str], Optional[str]]:
"""将语音数据转换为文本"""
try:
# 合并所有opus数据包
pcm_data = self.decode_opus(opus_data, session_id)
combined_pcm_data = b''.join(pcm_data)
wav_buffer = io.BytesIO()
with wave.open(wav_buffer, "wb") as wav_file:
wav_file.setnchannels(1) # 设置声道数
wav_file.setsampwidth(2) # 设置采样宽度
wav_file.setframerate(16000) # 设置采样率
wav_file.writeframes(combined_pcm_data) # 写入 PCM 数据
# 获取封装后的 WAV 数据
wav_data = wav_buffer.getvalue()
nchannels, sampwidth, framerate, nframes, wav_len = self.read_wav_info(wav_data)
size_per_sec = nchannels * sampwidth * framerate
segment_size = int(size_per_sec * self.seg_duration / 1000)
# 语音识别
start_time = time.time()
text = await self._send_request(wav_data, segment_size)
if text:
logger.bind(tag=TAG).debug(f"语音识别耗时: {time.time() - start_time:.3f}s | 结果: {text}")
return text, None
return "", None
except Exception as e:
logger.bind(tag=TAG).error(f"语音识别失败: {e}", exc_info=True)
return "", None
+110
View File
@@ -0,0 +1,110 @@
import time
import wave
import os
import sys
import io
from config.logger import setup_logging
from typing import Optional, Tuple, List
import uuid
import opuslib_next
from core.providers.asr.base import ASRProviderBase
from funasr import AutoModel
from funasr.utils.postprocess_utils import rich_transcription_postprocess
TAG = __name__
logger = setup_logging()
# 捕获标准输出
class CaptureOutput:
def __enter__(self):
self._output = io.StringIO()
self._original_stdout = sys.stdout
sys.stdout = self._output
def __exit__(self, exc_type, exc_value, traceback):
sys.stdout = self._original_stdout
self.output = self._output.getvalue()
self._output.close()
# 将捕获到的内容通过 logger 输出
if self.output:
logger.bind(tag=TAG).info(self.output.strip())
class ASRProvider(ASRProviderBase):
def __init__(self, config: dict, delete_audio_file: bool):
self.model_dir = config.get("model_dir")
self.output_dir = config.get("output_dir") # 修正配置键名
self.delete_audio_file = delete_audio_file
# 确保输出目录存在
os.makedirs(self.output_dir, exist_ok=True)
with CaptureOutput():
self.model = AutoModel(
model=self.model_dir,
vad_kwargs={"max_single_segment_time": 30000},
disable_update=True,
hub="hf"
# device="cuda:0", # 启用GPU加速
)
def save_audio_to_file(self, opus_data: List[bytes], session_id: str) -> str:
"""将Opus音频数据解码并保存为WAV文件"""
file_name = f"asr_{session_id}_{uuid.uuid4()}.wav"
file_path = os.path.join(self.output_dir, file_name)
decoder = opuslib_next.Decoder(16000, 1) # 16kHz, 单声道
pcm_data = []
for opus_packet in opus_data:
try:
pcm_frame = decoder.decode(opus_packet, 960) # 960 samples = 60ms
pcm_data.append(pcm_frame)
except opuslib_next.OpusError as e:
logger.bind(tag=TAG).error(f"Opus解码错误: {e}", exc_info=True)
with wave.open(file_path, "wb") as wf:
wf.setnchannels(1)
wf.setsampwidth(2) # 2 bytes = 16-bit
wf.setframerate(16000)
wf.writeframes(b"".join(pcm_data))
return file_path
async def speech_to_text(self, opus_data: List[bytes], session_id: str) -> Tuple[Optional[str], Optional[str]]:
"""语音转文本主处理逻辑"""
file_path = None
try:
# 保存音频文件
start_time = time.time()
file_path = self.save_audio_to_file(opus_data, session_id)
logger.bind(tag=TAG).debug(f"音频文件保存耗时: {time.time() - start_time:.3f}s | 路径: {file_path}")
# 语音识别
start_time = time.time()
result = self.model.generate(
input=file_path,
cache={},
language="auto",
use_itn=True,
batch_size_s=60,
)
text = rich_transcription_postprocess(result[0]["text"])
logger.bind(tag=TAG).debug(f"语音识别耗时: {time.time() - start_time:.3f}s | 结果: {text}")
return text, file_path
except Exception as e:
logger.bind(tag=TAG).error(f"语音识别失败: {e}", exc_info=True)
return "", None
finally:
# 文件清理逻辑
if self.delete_audio_file and file_path and os.path.exists(file_path):
try:
os.remove(file_path)
logger.bind(tag=TAG).debug(f"已删除临时音频文件: {file_path}")
except Exception as e:
logger.bind(tag=TAG).error(f"文件删除失败: {file_path} | 错误: {e}")
+18 -5
View File
@@ -25,12 +25,25 @@ class LLMProvider(LLMProviderBase):
messages=dialogue,
stream=True
)
is_active = True
for chunk in responses:
# 检查是否存在有效的choice且content不为空
if chunk.choices and len(chunk.choices) > 0:
delta = chunk.choices[0].delta
content = getattr(delta, 'content', '')
if content: # 仅在content非空时生成
try:
# 检查是否存在有效的choice且content不为空
delta = chunk.choices[0].delta if getattr(chunk, 'choices', None) else None
content = delta.content if hasattr(delta, 'content') else ''
except IndexError:
content = ''
if content:
# 处理标签跨多个chunk的情况
if '<think>' in content:
is_active = False
content = content.split('<think>')[0]
if '</think>' in content:
is_active = True
content = content.split('</think>')[-1]
if is_active:
yield content
except Exception as e:
logger.bind(tag=TAG).error(f"Error in response generation: {e}")
+11 -6
View File
@@ -25,7 +25,7 @@ class TTSProvider(TTSProviderBase):
async def text_to_speak(self, text, output_file):
request_json = {
"app": {
"appid": self.appid,
"appid": f"{self.appid}",
"token": "access_token",
"cluster": self.cluster
},
@@ -49,8 +49,13 @@ class TTSProvider(TTSProviderBase):
}
}
resp = requests.post(self.api_url, json.dumps(request_json), headers=self.header)
if "data" in resp.json():
data = resp.json()["data"]
file_to_save = open(output_file, "wb")
file_to_save.write(base64.b64decode(data))
try:
resp = requests.post(self.api_url, json.dumps(request_json), headers=self.header)
if "data" in resp.json():
data = resp.json()["data"]
file_to_save = open(output_file, "wb")
file_to_save.write(base64.b64decode(data))
else:
raise Exception(f"{__name__} status_code: {resp.status_code} response: {resp.content}")
except Exception as e:
raise Exception(f"{__name__} error: {e}")
+1 -1
View File
@@ -89,7 +89,7 @@ class TTSProvider(TTSProviderBase):
self.channels = config.get("channels",1)
self.rate = config.get("rate",44100)
self.api_key = config.get("api_key","YOUR_API_KEY")
if not self.api_key or "" in self.api_key:
if "" in self.api_key:
logger.bind(tag=TAG).error("你还没配置FishSpeech TTS的密钥,请在配置文件中配置密钥,否则无法正常工作")
return
self.normalize = config.get("normalize",True)
+15 -123
View File
@@ -1,132 +1,24 @@
import time
import wave
import importlib
import logging
import os
import sys
import io
from abc import ABC, abstractmethod
from config.logger import setup_logging
from typing import Optional, Tuple, List
import time
import wave
import uuid
import opuslib_next
from funasr import AutoModel
from funasr.utils.postprocess_utils import rich_transcription_postprocess
from abc import ABC, abstractmethod
from typing import Optional, Tuple, List
from core.providers.asr.base import ASRProviderBase
from config.logger import setup_logging
TAG = __name__
logger = setup_logging()
# 捕获标准输出
class CaptureOutput:
def __enter__(self):
self._output = io.StringIO()
self._original_stdout = sys.stdout
sys.stdout = self._output
def __exit__(self, exc_type, exc_value, traceback):
sys.stdout = self._original_stdout
self.output = self._output.getvalue()
self._output.close()
# 将捕获到的内容通过 logger 输出
if self.output:
logger.bind(tag=TAG).info(self.output.strip())
class ASR(ABC):
@abstractmethod
def save_audio_to_file(self, opus_data: List[bytes], session_id: str) -> str:
"""解码Opus数据并保存为WAV文件"""
pass
@abstractmethod
def speech_to_text(self, opus_data: List[bytes], session_id: str) -> Tuple[Optional[str], Optional[str]]:
"""将语音数据转换为文本"""
pass
class FunASR(ASR):
def __init__(self, config: dict, delete_audio_file: bool):
self.model_dir = config.get("model_dir")
self.output_dir = config.get("output_dir") # 修正配置键名
self.delete_audio_file = delete_audio_file
# 确保输出目录存在
os.makedirs(self.output_dir, exist_ok=True)
with CaptureOutput():
self.model = AutoModel(
model=self.model_dir,
vad_kwargs={"max_single_segment_time": 30000},
disable_update=True,
hub="hf"
# device="cuda:0", # 启用GPU加速
)
def save_audio_to_file(self, opus_data: List[bytes], session_id: str) -> str:
"""将Opus音频数据解码并保存为WAV文件"""
file_name = f"asr_{session_id}_{uuid.uuid4()}.wav"
file_path = os.path.join(self.output_dir, file_name)
decoder = opuslib_next.Decoder(16000, 1) # 16kHz, 单声道
pcm_data = []
for opus_packet in opus_data:
try:
pcm_frame = decoder.decode(opus_packet, 960) # 960 samples = 60ms
pcm_data.append(pcm_frame)
except opuslib_next.OpusError as e:
logger.bind(tag=TAG).error(f"Opus解码错误: {e}", exc_info=True)
with wave.open(file_path, "wb") as wf:
wf.setnchannels(1)
wf.setsampwidth(2) # 2 bytes = 16-bit
wf.setframerate(16000)
wf.writeframes(b"".join(pcm_data))
return file_path
def speech_to_text(self, opus_data: List[bytes], session_id: str) -> Tuple[Optional[str], Optional[str]]:
"""语音转文本主处理逻辑"""
file_path = None
try:
# 保存音频文件
start_time = time.time()
file_path = self.save_audio_to_file(opus_data, session_id)
logger.bind(tag=TAG).debug(f"音频文件保存耗时: {time.time() - start_time:.3f}s | 路径: {file_path}")
# 语音识别
start_time = time.time()
result = self.model.generate(
input=file_path,
cache={},
language="auto",
use_itn=True,
batch_size_s=60,
)
text = rich_transcription_postprocess(result[0]["text"])
logger.bind(tag=TAG).debug(f"语音识别耗时: {time.time() - start_time:.3f}s | 结果: {text}")
return text, file_path
except Exception as e:
logger.bind(tag=TAG).error(f"语音识别失败: {e}", exc_info=True)
return None, None
finally:
# 文件清理逻辑
if self.delete_audio_file and file_path and os.path.exists(file_path):
try:
os.remove(file_path)
logger.bind(tag=TAG).debug(f"已删除临时音频文件: {file_path}")
except Exception as e:
logger.bind(tag=TAG).error(f"文件删除失败: {file_path} | 错误: {e}")
def create_instance(class_name: str, *args, **kwargs) -> ASR:
def create_instance(class_name: str, *args, **kwargs) -> ASRProviderBase:
"""工厂方法创建ASR实例"""
cls_map = {
"FunASR": FunASR,
# 可扩展其他ASR实现
}
if os.path.exists(os.path.join('core', 'providers', 'asr', f'{class_name}.py')):
lib_name = f'core.providers.asr.{class_name}'
if lib_name not in sys.modules:
sys.modules[lib_name] = importlib.import_module(f'{lib_name}')
return sys.modules[lib_name].ASRProvider(*args, **kwargs)
if cls := cls_map.get(class_name):
return cls(*args, **kwargs)
raise ValueError(f"不支持的ASR类型: {class_name}")
raise ValueError(f"不支持的ASR类型: {class_name},请检查该配置的type是否设置正确")
+112 -28
View File
@@ -1,5 +1,13 @@
import os
import sys
import asyncio
from typing import List, Dict, Any
# 添加项目根目录到Python路径
current_dir = os.path.dirname(os.path.abspath(__file__))
project_root = os.path.abspath(os.path.join(current_dir, "..", ".."))
sys.path.insert(0, project_root)
from config.logger import setup_logging
import importlib
from datetime import datetime
@@ -21,39 +29,115 @@ def create_instance(class_name, *args, **kwargs):
raise ValueError(f"不支持的LLM类型: {class_name},请检查该配置的type是否设置正确")
if __name__ == "__main__":
async def test_single_model(llm_name: str, llm_config: Dict[str, Any], test_prompt: str, config: Dict[str, Any]) -> Dict[str, Any]:
"""异步测试单个模型"""
try:
# 获取实际的LLM类型
llm_type = llm_config["type"] if "type" in llm_config else llm_name
llm = create_instance(llm_type, llm_config)
# 开始测试
dialogue = []
dialogue.append({"role": "system", "content": config.get("prompt")})
dialogue.append({"role": "user", "content": test_prompt})
start_time = datetime.now()
llm_responses = llm.response("test", dialogue)
response_message = []
first_response_time = None
total_response_time = None
start = 0
full_response = ""
for content in llm_responses:
response_message.append(content)
full_response += content
if is_segment(response_message):
segment_text = "".join(response_message[start:])
segment_text = get_string_no_punctuation_or_emoji(segment_text)
if len(segment_text) > 0:
if first_response_time is None:
first_response_time = (datetime.now() - start_time).total_seconds()
start = len(response_message)
total_response_time = (datetime.now() - start_time).total_seconds()
return {
"name": llm_name,
"type": llm_type,
"first_response_time": first_response_time,
"total_response_time": total_response_time,
"response_length": len(full_response),
"status": "成功",
"response": full_response
}
except Exception as e:
print(f"测试 {llm_name} 时发生错误: {str(e)}")
return {
"name": llm_name,
"type": llm_config.get("type", llm_name),
"first_response_time": None,
"total_response_time": None,
"response_length": 0,
"status": f"失败 - {str(e)}",
"response": ""
}
async def main():
"""
响应速度测试
LLM模型响应速度测试和排行(异步版本)
"""
config = read_config(get_project_dir() + "config.yaml")
llm = create_instance(
config["selected_module"]["LLM"]
if not "type" in config["LLM"][config["selected_module"]["LLM"]]
else
config["LLM"][config["selected_module"]["LLM"]]["type"],
config["LLM"][config["selected_module"]["LLM"]]
)
test_prompt = "你好小智"
print("开始并发测试所有模型...")
# 创建所有模型的测试任务
tasks = []
for llm_name, llm_config in config["LLM"].items():
task = asyncio.create_task(test_single_model(llm_name, llm_config, test_prompt, config))
tasks.append(task)
# 等待所有测试完成
test_results = await asyncio.gather(*tasks)
# 打印测试结果排行榜
print("\n========= LLM模型性能测试排行榜 =========")
print("测试提示词:", test_prompt)
# 过滤出成功的结果,并确保数值有效
successful_results = [r for r in test_results if r["status"] == "成功" and r["first_response_time"] is not None]
if successful_results:
print("\n1. 首次响应时间排行:")
sorted_by_first = sorted(successful_results, key=lambda x: x["first_response_time"])
for i, result in enumerate(sorted_by_first, 1):
print(f"{i}. {result['name']}({result['type']}) - {result['first_response_time']:.2f}")
print(f" 响应内容: {result['response'][:50]}...") # 只显示前50个字符
start_time = datetime.now()
print("\n2. 总响应时间排行:")
sorted_by_total = sorted(successful_results, key=lambda x: x["total_response_time"] or float('inf'))
for i, result in enumerate(sorted_by_total, 1):
if result["total_response_time"] is not None:
print(f"{i}. {result['name']}({result['type']}) - {result['total_response_time']:.2f}")
dialogue = []
dialogue.append({"role": "system", "content": config.get("prompt")})
dialogue.append({"role": "user", "content": "你好小智"})
llm_responses = llm.response("test", dialogue)
response_message = []
first_text = None
start = 0
print("\n3. 响应长度比较:")
sorted_by_length = sorted(successful_results, key=lambda x: x["response_length"], reverse=True)
for i, result in enumerate(sorted_by_length, 1):
print(f"{i}. {result['name']}({result['type']}) - {result['response_length']}字符")
else:
print("\n没有成功完成测试的模型。")
for content in llm_responses:
response_message.append(content)
if len(test_results) != len(successful_results):
print("\n测试失败的模型:")
failed_results = [r for r in test_results if r["status"] != "成功" or r["first_response_time"] is None]
for result in failed_results:
print(f"- {result['name']}({result['type']}): {result['status']}")
if is_segment(response_message):
segment_text = "".join(response_message[start:])
segment_text = get_string_no_punctuation_or_emoji(segment_text)
if len(segment_text) > 0:
if first_text is None:
first_text = segment_text
print("大模型首次返回耗时:" + str(datetime.now() - start_time))
start = len(response_message)
print("大模型返回总耗时:" + str(datetime.now() - start_time))
if __name__ == "__main__":
# 运行异步主函数
asyncio.run(main())
+2 -2
View File
@@ -90,7 +90,7 @@ def remove_punctuation_and_length(text):
char not in full_width_punctuations and char not in half_width_punctuations and char not in space and char not in full_width_space])
if result == "Yeah":
return 0
return 0, ""
return len(result), result
@@ -120,4 +120,4 @@ def check_password(password):
return False
# 如果满足所有条件,则返回True
return True
return True
+17 -5
View File
@@ -2,16 +2,19 @@ import asyncio
import websockets
from config.logger import setup_logging
from core.connection import ConnectionHandler
from core.handle.musicHandler import MusicHandler
from core.utils.util import get_local_ip
from core.utils import asr, vad, llm, tts
TAG = __name__
class WebSocketServer:
def __init__(self, config: dict):
self.config = config
self.logger = setup_logging()
self._vad, self._asr, self._llm, self._tts = self._create_processing_instances()
self._vad, self._asr, self._llm, self._tts, self._music = self._create_processing_instances()
self.active_connections = set() # 添加全局连接记录
def _create_processing_instances(self):
"""创建处理模块实例"""
@@ -21,7 +24,10 @@ class WebSocketServer:
self.config["VAD"][self.config["selected_module"]["VAD"]]
),
asr.create_instance(
self.config["selected_module"]["ASR"],
self.config["selected_module"]["ASR"]
if not 'type' in self.config["ASR"][self.config["selected_module"]["ASR"]]
else
self.config["ASR"][self.config["selected_module"]["ASR"]]["type"],
self.config["ASR"][self.config["selected_module"]["ASR"]],
self.config["delete_audio"]
),
@@ -39,7 +45,8 @@ class WebSocketServer:
self.config["TTS"][self.config["selected_module"]["TTS"]]["type"],
self.config["TTS"][self.config["selected_module"]["TTS"]],
self.config["delete_audio"]
)
),
MusicHandler(self.config)
)
async def start(self):
@@ -58,5 +65,10 @@ class WebSocketServer:
async def _handle_connection(self, websocket):
"""处理新连接,每次创建独立的ConnectionHandler"""
handler = ConnectionHandler(self.config, self._vad, self._asr, self._llm, self._tts)
await handler.handle_connection(websocket)
# 创建ConnectionHandler时传入当前server实例
handler = ConnectionHandler(self.config, self._vad, self._asr, self._llm, self._tts, self._music)
self.active_connections.add(handler)
try:
await handler.handle_connection(websocket)
finally:
self.active_connections.discard(handler)
-17
View File
@@ -1,17 +0,0 @@
services:
xiaozhi-esp32-server:
image: ccr.ccs.tencentyun.com/xinnan/xiaozhi-esp32-server:latest
container_name: xiaozhi-esp32-server
restart: always
#security_opt:
# - seccomp:unconfined
ports:
- "8000:8000" # ws服务端
- "8002:8002" # 管理后台
volumes:
# 配置文件 记得创建文件夹
- ./data:/opt/xiaozhi-esp32-server/data
# 模型文件 //可以挂载本地下载好的模型
#- ./models:/opt/xiaozhi-esp32-server/models
# 临时文件 //可以看log和音频文件 记得创建文件夹
#- ./tmp:/opt/xiaozhi-esp32-server/tmp
+8 -3
View File
@@ -1,13 +1,18 @@
version: '3'
services:
xiaozhi-esp32-server:
image: ghcr.io/kalicyh/xiaozhi-esp32-server:latest
image: ghcr.nju.edu.cn/xinnan-tech/xiaozhi-esp32-server:latest
container_name: xiaozhi-esp32-server
restart: always
#security_opt:
# - seccomp:unconfined
security_opt:
- seccomp:unconfined
ports:
# ws服务端
- "8000:8000"
# 管理后台
- "8002:8002"
volumes:
# 配置文件目录
- ./data:/app/data
# 模型文件挂接,很重要
- ./models/SenseVoiceSmall/model.pt:/app/models/SenseVoiceSmall/model.pt
-217
View File
@@ -1,217 +0,0 @@
# 部署指南
本项目支持 **x86****arm64** 架构,可运行于国产操作系统。
## 方式一:快速部署(推荐)
### 1. 安装 Docker
如果未安装 Docker,可参考 [Docker 安装教程](https://www.runoob.com/docker/ubuntu-docker-install.html)。
### 2. 创建项目目录
选择一个空目录作为 **项目目录**
### 3. 下载配置文件
访问 [config.yaml](https://github.com/xinnan-tech/xiaozhi-esp32-server/blob/main/config.yaml),点击 `RAW` 旁的下载按钮,保存到 **项目目录**
### 4. 运行 Docker
打开终端,`cd` 进入 **项目目录**,执行:
```sh
docker run -d --name xiaozhi-esp32-server --restart always --security-opt seccomp:unconfined \
-p 8000:8000 \
-v $(pwd)/config.yaml:/opt/xiaozhi-esp32-server/config.yaml \
ccr.ccs.tencentyun.com/xinnan/xiaozhi-esp32-server:latest
```
### 5. 确认运行状态
```sh
docker ps # 查看容器状态
docker logs -f xiaozhi-esp32-server # 查看日志
```
## 方式二:使用 Docker Compose 部署
### 1. 创建 `docker-compose.yml`
`项目目录` 下创建 `docker-compose.yml` 文件,内容如下:
```yaml
version: '3'
services:
xiaozhi-esp32-server:
image: ccr.ccs.tencentyun.com/xinnan/xiaozhi-esp32-server:latest
container_name: xiaozhi-esp32-server
restart: always
security_opt:
- seccomp:unconfined
ports:
- "8000:8000"
volumes:
- ./config.yaml:/opt/xiaozhi-esp32-server/config.yaml
```
### 2. 启动服务
```sh
docker-compose up -d
```
### 3. 确认运行状态
```sh
docker-compose ps
docker-compose logs -f
```
### 4. 停止和删除容器
```sh
docker-compose down
```
### 5. 版本升级
#### 备份配置文件
```sh
cp config.yaml config_backup.yaml
```
#### 更新 Docker 镜像
```sh
docker-compose pull
docker-compose down
docker-compose up -d
```
## 方式三:源码部署(免环境部署/可修改代码)
### 1. 下载源码
```sh
git clone https://github.com/xinnan-tech/xiaozhi-esp32-server.git
cd xiaozhi-esp32-server
```
或手动下载 [ZIP 包](https://github.com/xinnan-tech/xiaozhi-esp32-server/archive/refs/heads/main.zip),解压后重命名为 `xiaozhi-esp32-server`
## 2.运行docker
修改完配置后,打开命令行工具,`cd`进入到你的项目目录下,执行以下命令
```sh
docker run -it --name xiaozhi-env --restart always --security-opt seccomp:unconfined \
-p 8000:8000 \
-p 8002:8002 \
-v ./:/app \
kalicyh/poetry:v3.10_latest
```
然后就和正常开发一样了
## 3.安装依赖
在刚刚的打开的终端运行
```sh
poetry install --no-root
```
```sh
apt-get update
apt-get install -y --no-install-recommends libopus0 ffmpeg
```
速度慢可以尝试使用清华镜像
```sh
echo "deb https://mirrors.tuna.tsinghua.edu.cn/debian/ bookworm main contrib non-free non-free-firmware" > /etc/apt/sources.list
echo "deb https://mirrors.tuna.tsinghua.edu.cn/debian/ bookworm-updates main contrib non-free non-free-firmware" >> /etc/apt/sources.list
echo "deb https://mirrors.tuna.tsinghua.edu.cn/debian/ bookworm-backports main contrib non-free non-free-firmware" >> /etc/apt/sources.list
echo "deb https://mirrors.tuna.tsinghua.edu.cn/debian-security bookworm-security main contrib non-free non-free-firmware" >> /etc/apt/sources.list
apt-get update
apt-get install -y --no-install-recommends libopus0 ffmpeg
```
## 4.运行项目
```sh
poetry run python app.py
```
## 方式三:本地运行(适用于开发)
### 1. 安装环境
```sh
conda create -n xiaozhi-esp32-server python=3.10 -y
conda activate xiaozhi-esp32-server
```
#### Mac/Windows:
```sh
conda install conda-forge::libopus conda-forge::ffmpeg
```
#### Ubuntu:
```sh
apt-get install libopus0 ffmpeg
```
### 2. 下载源码 & 安装依赖
```sh
git clone https://github.com/xinnan-tech/xiaozhi-esp32-server.git
cd xiaozhi-esp32-server
pip install -r requirements.txt
```
### 3. 运行项目
```sh
python app.py
```
## 版本升级
```sh
docker stop xiaozhi-esp32-server
docker rm xiaozhi-esp32-server
docker rmi ccr.ccs.tencentyun.com/xinnan/xiaozhi-esp32-server:latest
```
然后重新按照 **方式一****方式二** 运行 Docker。
## 重要配置
修改 `config.yaml` 以适配不同 LLM 和 TTS 组件。
示例:
```yaml
selected_module:
ASR: FunASR
VAD: SileroVAD
LLM: ChatGLMLLM
TTS: EdgeTTS
```
如需使用 `Dify``DeepSeekLLM`,修改 `LLM` 部分并填写密钥。
## 模型文件下载
语音转文字默认使用 `SenseVoiceSmall`,需下载 `model.pt` 并存放至 `model/SenseVoiceSmall` 目录。
- [阿里魔塔下载](https://modelscope.cn/models/iic/SenseVoiceSmall/resolve/master/model.pt)
- [百度网盘下载](https://pan.baidu.com/share/init?surl=QlgM58FHhYv1tFnUT_A8Sg&pwd=qvna) 提取码:`qvna`
---
如遇问题,可参考 `docker logs -f xiaozhi-esp32-server` 进行排查。
**至此,部署完成!** 🚀
+184 -45
View File
@@ -1,28 +1,170 @@
# 方式一:docker快速部署
# 本地源码运行
docker镜像已支持x86架构、arm64架构的CPU,支持在国产操作系统上运行。
## 1. 安装docker
如果您的电脑还没安装docker,可以按照这里的教程安装:[docker安装](https://www.runoob.com/docker/ubuntu-docker-install.html)
## 2. 创建目录
安装完后,你需要为这个项目找一个安放配置文件的目录,我们暂且称它为`项目目录`,这个目录最好是一个新建的空的目录。
创建好目录后,你需要在`项目目录`下面创建`data`文件夹和`models`文件夹,`models`下面还要再创建`SenseVoiceSmall`文件夹。
最终目录结构如下所示:
```
你的项目根目录
├─ data
├─ models
├─ SenseVoiceSmall
```
## 4. 下载语音识别模型文件
你需要下载语音识别的模型文件,因为本项目的默认语音识别用的是本地离线语音识别方案。可通过这个方式下载
[跳转到下载语音识别模型文件](#模型文件)
下载完后,回到本教程。
## 3. 下载docker-compose.yaml
用浏览器打开[这个链接](https://github.com/xinnan-tech/xiaozhi-esp32-server/blob/main/docker-compose.yml)。
在页面的右侧找到名称为`RAW`按钮,在`RAW`按钮的旁边,找到下载的图标,点击下载按钮,下载`docker-compose.yml`文件。 把文件下载到你的
`项目目录`中。
下载完后,回到本教程继续往下。
## 3. 下载配置文件
用浏览器打开[这个链接](https://github.com/xinnan-tech/xiaozhi-esp32-server/blob/main/config.yaml)。
在页面的右侧找到名称为`RAW`按钮,在`RAW`按钮的旁边,找到下载的图标,点击下载按钮,下载`config.yaml`文件。 把文件下载到你的
`项目目录`下面的`data`文件夹中,然后把`config.yaml`文件重命名为`.config.yaml`
下载完配置文件后,我们确认一下整个`项目目录`里面的文件如下所示:
```
你的项目根目录
├─ docker-compose.yml
├─ data
├─ .config.yaml
├─ models
├─ SenseVoiceSmall
├─ model.pt
```
如果你的文件目录结构也是上面的,就继续往下。如果不是,你就再仔细看看是不是漏操作了什么。
## 4. 配置项目文件
接下里,程序还不能直接运行,你需要配置一下,你到底使用的是什么模型。你可以看这个教程:
[跳转到配置项目文件](#配置项目)
配置完项目文件后,回到本教程继续往下。
## 5. 执行docker命令
打开命令行工具,使用`终端``命令行`工具 进入到你的`项目目录`,执行以下命令
```
docker-compose up -d
```
执行完后,再执行以下命令,查看日志信息。
```
docker logs -f xiaozhi-esp32-server
```
这时,你就要留意日志信息,可以根据这个教程,判断是否成功了。[跳转到运行状态确认](#运行状态确认)
## 6.版本升级操作
如果后期想升级版本,可以这么操作
1、备份好`data`文件夹中的`.config.yaml`文件,一些关键的配置到时复制到新的`.config.yaml`文件里。
请注意是对关键密钥逐个复制,不要直接覆盖。因为新的`.config.yaml`文件可能有一些新的配置项,旧的`.config.yaml`文件不一定有。
2、执行以下命令
```
docker stop xiaozhi-esp32-server
docker rm xiaozhi-esp32-server
docker rmi ghcr.nju.edu.cn/xinnan-tech/xiaozhi-esp32-server:latest
```
3、重新按docker方式部署
# 方式二:借助Docker环境运行部署
开发人员如果不想安装`conda`环境,可以使用这种方法管理好依赖。
## 1.克隆项目
## 2.[跳转到下载语音识别模型文件](#模型文件)
## 3.[跳转到配置项目文件](#配置项目)
## 4.运行docker
修改完配置后,打开命令行工具,`cd`进入到你的项目目录下,执行以下命令
```sh
docker run -it --name xiaozhi-env --restart always --security-opt seccomp:unconfined \
-p 8000:8000 \
-p 8002:8002 \
-v ./:/app \
kalicyh/poetry:v3.10_xiaozhi
```
然后就和正常开发一样了
## 5.安装依赖
在刚刚的打开的终端运行
```sh
poetry install --no-root
```
```sh
apt-get update
apt-get install -y --no-install-recommends libopus0 ffmpeg
```
速度慢可以尝试使用清华镜像
```sh
echo "deb https://mirrors.tuna.tsinghua.edu.cn/debian/ bookworm main contrib non-free non-free-firmware" > /etc/apt/sources.list
echo "deb https://mirrors.tuna.tsinghua.edu.cn/debian/ bookworm-updates main contrib non-free non-free-firmware" >> /etc/apt/sources.list
echo "deb https://mirrors.tuna.tsinghua.edu.cn/debian/ bookworm-backports main contrib non-free non-free-firmware" >> /etc/apt/sources.list
echo "deb https://mirrors.tuna.tsinghua.edu.cn/debian-security bookworm-security main contrib non-free non-free-firmware" >> /etc/apt/sources.list
apt-get update
apt-get install -y --no-install-recommends libopus0 ffmpeg
```
## 6.运行项目
```sh
poetry run python app.py
```
# 方式三:本地源码运行
## 1.安装基础环境
本项目使用`conda`管理依赖环境,安装好后,开始执行以下命令
本项目使用`conda`管理依赖环境。如果不方便安装`conda`,需要根据实际的操作系统安装好`libopus``ffmpeg`
如果确定使用`conda`,则安装好后,开始执行以下命令。
```
conda remove -n xiaozhi-esp32-server --all -y
conda create -n xiaozhi-esp32-server python=3.10 -y
conda activate xiaozhi-esp32-server
```
执行以上命令后, 如果你的电脑是Windows或Mac,执行下面的语句:
```
conda activate xiaozhi-esp32-server
conda install conda-forge::libopus
conda install conda-forge::ffmpeg
```
如果你的电脑是ubuntu,执行下面的语句:
```
apt-get install libopus0 ffmpeg
conda install conda-forge::libopus -y
conda install conda-forge::ffmpeg -y
```
## 2.安装本项目依赖
@@ -43,9 +185,17 @@ pip config set global.index-url https://mirrors.aliyun.com/pypi/simple/
pip install -r requirements.txt
```
## 3.[跳转到下载语音识别模型文件](#模型文件)
## 3.下载语音识别模型文件
## 4.[跳转到配置项目文件](#配置项目)
你需要下载语音识别的模型文件,因为本项目的默认语音识别用的是本地离线语音识别方案。可通过这个方式下载
[跳转到下载语音识别模型文件](#模型文件)
下载完后,回到本教程。
## 4.配置项目文件
接下里,程序还不能直接运行,你需要配置一下,你到底使用的是什么模型。你可以看这个教程:
[跳转到配置项目文件](#配置项目)
## 5.运行项目
@@ -55,25 +205,20 @@ conda activate xiaozhi-esp32-server
python app.py
```
## 6.[跳转到运行状态确认](#运行状态确认)
这时,你就要留意日志信息,可以根据这个教程,判断是否成功了。[跳转到运行状态确认](#运行状态确认)
# 汇总
## 配置项目
修改`config.yaml`文件,配置本项目所需的各种参数。默认的LLM使用的是`ChatGLMLLM`
如果你的`项目目录`目录没有`data`,你需要创建`data`目录。
如果你的`data`下面没有`.config.yaml`文件,你可以把源码目录下的`config.yaml`文件复制一份,重命名为`.config.yaml`
修改`项目目录``data`目录下的`.config.yaml`文件,配置本项目所需的各种参数。默认的LLM使用的是`ChatGLMLLM`
,你需要配置密钥,因为他们的模型,虽然有免费的,但是仍要去[官网](https://bigmodel.cn/usercenter/proj-mgmt/apikeys)注册密钥,才能启动。
默认的TTS使用的是`EdgeTTS`,这个无需配置,如果你需要更换成`豆包TTS`,则需要配置密钥。
```
# 如果您是一名开发者,建议阅读以下内容。如果不是开发者,可以忽略这部分内容。
# 在开发中,在项目根目录创建data目录,将【config.yaml】复制一份,改成【.config.yaml】,放进data目录中
# 系统会优先读取【data/.config.yaml】文件的配置。
# 这样做,可以避免在提交代码的时候,错误地提交密钥信息,保护您的密钥安全。
```
配置说明:这里是各个功能使用的默认组件,例如LLM默认使用`ChatGLMLLM`模型。如果需要切换模型,就是改对应的名称。
本项目的默认配置仅是成本最低配置(`glm-4-flash``EdgeTTS`都是免费的),如果需要更优的更快的搭配,需要自己结合部署环境切换各组件的使用。
```
@@ -104,7 +249,7 @@ LLM:
## 模型文件
下载源码后,需要下载模型文件。 默认使用`SenseVoiceSmall`模型,进行语音转文字。因为模型较大,需要独立下载,下载后把`model.pt`
本项目语音识别模型,默认使用`SenseVoiceSmall`模型,进行语音转文字。因为模型较大,需要独立下载,下载后把`model.pt`
文件放在`model/SenseVoiceSmall`
目录下。下面两个下载路线任选一个。
@@ -112,27 +257,21 @@ LLM:
- 线路二:百度网盘下载[SenseVoiceSmall](https://pan.baidu.com/share/init?surl=QlgM58FHhYv1tFnUT_A8Sg&pwd=qvna) 提取码:
`qvna`
huggingface地址:
```
https://huggingface.co/jinaai/jina-embeddings-v2-base-zh?clone=true
https://huggingface.co/FunAudioLLM/SenseVoiceSmall/tree/main?clone=true
```
镜像地址`https://hf-mirror.com/`
## 运行状态确认
如果你能看到,类似以下日志,则是本项目服务启动成功的标志。
```
2025-xx-xx xx:51:59,492 - core.server - INFO - Server is running at ws://xx.xx.xx.xxx:8000
2025-xx-xx xx:51:59,516 - websockets.server - INFO - server listening on 0.0.0.0:8000
25-02-23 12:01:09[core.websocket_server] - INFO - Server is running at ws://xxx.xx.xx.xx:8000
25-02-23 12:01:09[core.websocket_server] - INFO - =======上面的地址是websocket协议地址,请勿用浏览器访问=======
```
接下来,你就可以开始 `编译esp32固件`了,请往下翻,翻到编译`esp32固件`相关章节
正常来说,假设你的ip是`192.168.1.25`,那么你的接口地址就是:`ws://192.168.1.25:8000`。这个信息很有用的,后面`编译esp32固件`
需要用到。
正常来说,如果您是通过源码运行本项目,日志会有你的接口地址信息
但是如果你用docker部署,那么你的日志里给出的接口地址信息就不是真实的接口地址。
请注意,你的接口地址是`websocket`协议的地址,你可以使用`apifox`等工具调试。但是不能直接用浏览器打开访问,如果用浏览器打开,日志会显示错误,会让你怀疑是否部署成功了
最正确的方法,是根据电脑的局域网IP来确定你的接口地址
如果你的电脑的局域网IP比如是`192.168.1.25`,那么你的接口地址就是:`ws://192.168.1.25:8000`
这个信息很有用的,后面`编译esp32固件`需要用到。
接下来,你就可以开始 [编译esp32固件](firmware-build.md)了。
-250
View File
@@ -1,250 +0,0 @@
# 方式一:docker快速部署
docker镜像已支持x86架构、arm64架构的CPU,支持在国产操作系统上运行。
## 1. 安装docker
如果您的电脑还没安装docker,可以按照这里的教程安装:[docker安装](https://www.runoob.com/docker/ubuntu-docker-install.html)
## 2. 创建目录
安装完后,你需要为这个项目找一个安放配置文件的目录,我们暂且称它为`项目目录`,这个目录最好是一个新建的空的目录。
## 3. 下载配置文件
用浏览器打开[这个链接](https://github.com/xinnan-tech/xiaozhi-esp32-server/blob/main/config.yaml)。
在页面的右侧找到名称为`RAW`按钮,在`RAW`按钮的旁边,找到下载的图标,点击下载按钮,下载`config.yaml`文件。 把文件下载到你的
`项目目录`
## 4.[跳转到配置项目文件](#配置项目)
## 5. 执行docker命令
打开命令行工具,`cd` 进入到你的`项目目录`,执行以下命令
```
#如果你是linux,执行
ls
#如果你是windows,执行
dir
```
如果你能看到`config.yaml`文件,确确实实进入到了`项目目录`,接着执行以下命令:
```
docker run -d --name xiaozhi-esp32-server --restart always --security-opt seccomp:unconfined -p 8000:8000 -v $(pwd)/config.yaml:/opt/xiaozhi-esp32-server/config.yaml ccr.ccs.tencentyun.com/xinnan/xiaozhi-esp32-server:latest
```
## 6.[跳转到运行状态确认](#运行状态确认)
## [跳转到版本升级操作](#版本升级操作)
# 方式二:借助docker环境运行部署(仅限开发人员/小白勿用)
## 1.克隆项目
## 2.[跳转到下载语音识别模型文件](#模型文件)
## 3.[跳转到配置项目文件](#配置项目)
## 4.运行docker
修改完配置后,打开命令行工具,`cd`进入到你的项目目录下,执行以下命令
```sh
docker run -it --name xiaozhi-env --restart always --security-opt seccomp:unconfined \
-p 8000:8000 \
-p 8002:8002 \
-v ./:/app \
ccr.ccs.tencentyun.com/kalicyh/poetry:v3.10_latest
```
然后就和正常开发一样了
## 5.安装依赖
在刚刚的打开的终端运行
```sh
poetry install --no-root
```
```sh
apt-get update
apt-get install -y --no-install-recommends libopus0 ffmpeg
```
速度慢可以尝试使用清华镜像
```sh
echo "deb https://mirrors.tuna.tsinghua.edu.cn/debian/ bookworm main contrib non-free non-free-firmware" > /etc/apt/sources.list
echo "deb https://mirrors.tuna.tsinghua.edu.cn/debian/ bookworm-updates main contrib non-free non-free-firmware" >> /etc/apt/sources.list
echo "deb https://mirrors.tuna.tsinghua.edu.cn/debian/ bookworm-backports main contrib non-free non-free-firmware" >> /etc/apt/sources.list
echo "deb https://mirrors.tuna.tsinghua.edu.cn/debian-security bookworm-security main contrib non-free non-free-firmware" >> /etc/apt/sources.list
apt-get update
apt-get install -y --no-install-recommends libopus0 ffmpeg
```
## 6.运行项目
```sh
poetry run python app.py
```
# 方式三:本地源码运行
## 1.安装基础环境
本项目使用`conda`管理依赖环境,安装好后,开始执行以下命令。
```
conda remove -n xiaozhi-esp32-server --all -y
conda create -n xiaozhi-esp32-server python=3.10 -y
conda activate xiaozhi-esp32-server
```
执行以上命令后, 如果你的电脑是Windows或Mac,执行下面的语句:
```
conda activate xiaozhi-esp32-server
conda install conda-forge::libopus
conda install conda-forge::ffmpeg
```
如果你的电脑是ubuntu,执行下面的语句:
```
apt-get install libopus0 ffmpeg
```
## 2.安装本项目依赖
你先要下载本项目源码,源码可以通过`git clone`命令下载,如果你不熟悉`git clone`命令。
你可以用浏览器打开这个地址`https://github.com/xinnan-tech/xiaozhi-esp32-server.git`
打开完,找到页面中一个绿色的按钮,写着`Code`的按钮,点开它,然后你就看到`Download ZIP`的按钮。
点击它,下载本项目源码压缩包。下载到你电脑后,解压它,此时它的名字可能叫`xiaozhi-esp32-server-main`
你需要把它重命名成`xiaozhi-esp32-server`,好了请记住这个目录,我们暂且称它为`项目目录`
```
# 使用dos或者终端,进入到你的项目目录,执行以下命令
conda activate xiaozhi-esp32-server
pip config set global.index-url https://mirrors.aliyun.com/pypi/simple/
pip install -r requirements.txt
```
## 3.[跳转到下载语音识别模型文件](#模型文件)
## 4.[跳转到配置项目文件](#配置项目)
## 5.运行项目
```
# 确保在本项目的根目录下执行
conda activate xiaozhi-esp32-server
python app.py
```
## 6.[跳转到运行状态确认](#运行状态确认)
# 汇总
## 配置项目
修改`config.yaml`文件,配置本项目所需的各种参数。默认的LLM使用的是`ChatGLMLLM`
,你需要配置密钥,因为他们的模型,虽然有免费的,但是仍要去[官网](https://bigmodel.cn/usercenter/proj-mgmt/apikeys)注册密钥,才能启动。
默认的TTS使用的是`EdgeTTS`,这个无需配置,如果你需要更换成`豆包TTS`,则需要配置密钥。
```
# 如果您是一名开发者,建议阅读以下内容。如果不是开发者,可以忽略这部分内容。
# 在开发中,在项目根目录创建data目录,将【config.yaml】复制一份,改成【.config.yaml】,放进data目录中
# 系统会优先读取【data/.config.yaml】文件的配置。
# 这样做,可以避免在提交代码的时候,错误地提交密钥信息,保护您的密钥安全。
```
配置说明:这里是各个功能使用的默认组件,例如LLM默认使用`ChatGLMLLM`模型。如果需要切换模型,就是改对应的名称。
本项目的默认配置仅是成本最低配置(`glm-4-flash``EdgeTTS`都是免费的),如果需要更优的更快的搭配,需要自己结合部署环境切换各组件的使用。
```
selected_module:
ASR: FunASR
VAD: SileroVAD
LLM: ChatGLMLLM
TTS: EdgeTTS
```
比如修改`LLM`使用的组件,就看本项目支持哪些`LLM` API接口,当前支持的是`openai``dify`。欢迎验证和支持更多LLM平台的接口。
使用时,在`selected_module`修改成对应的如下LLM配置的名称:
```
LLM:
DeepSeekLLM:
type: openai
...
ChatGLMLLM:
type: openai
...
DifyLLM:
type: dify
...
```
有些服务,比如如果你使用`Dify``豆包的TTS`,是需要密钥的,记得在配置文件加上哦!
## 模型文件
下载源码后,需要下载模型文件。 默认使用`SenseVoiceSmall`模型,进行语音转文字。因为模型较大,需要独立下载,下载后把`model.pt`
文件放在`model/SenseVoiceSmall`
目录下。下面两个下载路线任选一个。
- 线路一:阿里魔塔下载[SenseVoiceSmall](https://modelscope.cn/models/iic/SenseVoiceSmall/resolve/master/model.pt)
- 线路二:百度网盘下载[SenseVoiceSmall](https://pan.baidu.com/share/init?surl=QlgM58FHhYv1tFnUT_A8Sg&pwd=qvna) 提取码:
`qvna`
## 运行状态确认
如果首次执行,可能需要几分钟时间,你要耐心等待他完成拉取。正常拉取完成后,你可以在命令行执行以下命令查看服务是否启动成功
```
docker ps
```
如果你能看到`xiaozhi-server`,说明服务启动成功。那你还可以进一步执行以下命令,查看服务的日志
```
docker logs -f xiaozhi-esp32-server
```
如果你能看到,类似以下日志,则是本项目服务启动成功的标志。
```
2025-xx-xx xx:51:59,492 - core.server - INFO - Server is running at ws://xx.xx.xx.xxx:8000
2025-xx-xx xx:51:59,516 - websockets.server - INFO - server listening on 0.0.0.0:8000
```
接下来,你就可以开始 `编译esp32固件`了,请往下翻,翻到编译`esp32固件`相关章节。那么由于你是用docker部署,你要自己查看自己本机电脑的ip是多少。
正常来说,假设你的ip是`192.168.1.25`,那么你的接口地址就是:`ws://192.168.1.25:8000`。这个信息很有用的,后面`编译esp32固件`
需要用到。
请注意,你的接口地址是`websocket`协议的地址,你可以使用`apifox`等工具调试。但是不能直接用浏览器打开访问,如果用浏览器打开,日志会显示错误,会让你怀疑是否部署成功了。
## 版本升级操作
如果想升级版本,可以这么操作
1、备份好`config.yaml`文件,一些关键的配置到时复制到新的`config.yaml`文件里。
2、执行以下命令
```
docker stop xiaozhi-esp32-server
docker rm xiaozhi-esp32-server
docker rmi ccr.ccs.tencentyun.com/xinnan/xiaozhi-esp32-server:latest
```
3、重新开始安装
+5 -2
View File
@@ -1,4 +1,7 @@
# 编译docker镜像
# 本地编译docker镜像方法
现在本项目已经使用github自动编译docker功能,本文档是提供给有本地编译docker镜像需求的朋友准备的。
1、安装docker
```
sudo apt-get install docker-ce docker-ce-cli containerd.io docker-buildx-plugin docker-compose-plugin
@@ -6,7 +9,7 @@ sudo apt-get install docker-ce docker-ce-cli containerd.io docker-buildx-plugin
2、编译docker镜像
```
# 普通编译
docker build -t xiaozhi-esp32-server:local -f ./Dockerfile .
docker build -t xiaozhi-esp32-server:local -f ./Dockerfile-pip .
```
3、测试本地镜像
```
Binary file not shown.

Before

Width:  |  Height:  |  Size: 249 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 258 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 269 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 387 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 404 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 288 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 446 KiB

After

Width:  |  Height:  |  Size: 427 KiB

Binary file not shown.
Binary file not shown.
Binary file not shown.
+328 -179
View File
@@ -1,27 +1,27 @@
import time
import aiohttp
import asyncio
from tabulate import tabulate
from typing import Dict
from typing import Dict, List
from core.utils.llm import create_instance as create_llm_instance
from core.utils.tts import create_instance as create_tts_instance
from core.utils.util import read_config
import statistics
from config.settings import get_config_file
from concurrent.futures import ThreadPoolExecutor
import inspect
import os
import requests
import logging
# 设置全局日志级别为WARNING,抑制INFO级别日志
logging.basicConfig(level=logging.WARNING)
class PerformanceTester:
class AsyncPerformanceTester:
def __init__(self):
self.config = read_config(get_config_file())
# 从配置读取测试句子,如果不存在则使用默认
self.test_sentences = self.config.get("module_test", {}).get(
"test_sentences",
["你好,请介绍一下你自己", "What's the weather like today?",
"test_sentences",
["你好,请介绍一下你自己", "What's the weather like today?",
"请用100字概括量子计算的基本原理和应用前景"]
)
self.results = {
@@ -30,258 +30,407 @@ class PerformanceTester:
"combinations": []
}
def _test_llm(self, llm_name: str, config: Dict) -> Dict:
"""测试单个LLM性能"""
async def _check_ollama_service(self, base_url: str, model_name: str) -> bool:
"""异步检查Ollama服务状态"""
async with aiohttp.ClientSession() as session:
try:
# 检查服务是否可用
async with session.get(f"{base_url}/api/version") as response:
if response.status != 200:
print(f"🚫 Ollama服务未启动或无法访问: {base_url}")
return False
# 检查模型是否存在
async with session.get(f"{base_url}/api/tags") as response:
if response.status == 200:
data = await response.json()
models = data.get("models", [])
if not any(model["name"] == model_name for model in models):
print(f"🚫 Ollama模型 {model_name} 未找到,请先使用 ollama pull {model_name} 下载")
return False
else:
print(f"🚫 无法获取Ollama模型列表")
return False
return True
except Exception as e:
print(f"🚫 无法连接到Ollama服务: {str(e)}")
return False
async def _test_tts(self, tts_name: str, config: Dict) -> Dict:
"""异步测试单个TTS性能"""
try:
# 跳过未配置密钥的模块
if "api_key" in config and any(x in config["api_key"] for x in ["你的", "placeholder", "sk-xxx"]):
print(f"🚫 跳过未配置的LLM: {llm_name}")
return {"errors": 1}
logging.getLogger("core.providers.tts.base").setLevel(logging.WARNING)
token_fields = ["access_token", "api_key", "token"]
if any(field in config and any(x in config[field] for x in ["你的", "placeholder"]) for field in
token_fields):
print(f"⏭️ TTS {tts_name} 未配置access_token/api_key,已跳过")
return {"name": tts_name, "type": "tts", "errors": 1}
module_type = config.get('type', tts_name)
tts = create_tts_instance(
module_type,
config,
delete_audio_file=True
)
print(f"🎵 测试 TTS: {tts_name}")
tmp_file = tts.generate_filename()
await tts.text_to_speak("连接测试", tmp_file)
if not tmp_file or not os.path.exists(tmp_file):
print(f"{tts_name} 连接失败")
return {"name": tts_name, "type": "tts", "errors": 1}
total_time = 0
test_count = len(self.test_sentences[:2])
for i, sentence in enumerate(self.test_sentences[:2], 1):
start = time.time()
tmp_file = tts.generate_filename()
await tts.text_to_speak(sentence, tmp_file)
duration = time.time() - start
total_time += duration
if tmp_file and os.path.exists(tmp_file):
print(f"{tts_name} [{i}/{test_count}]")
else:
print(f"{tts_name} [{i}/{test_count}]")
return {"name": tts_name, "type": "tts", "errors": 1}
return {
"name": tts_name,
"type": "tts",
"avg_time": total_time / test_count,
"errors": 0
}
except Exception as e:
print(f"⚠️ {tts_name} 测试失败: {str(e)}")
return {"name": tts_name, "type": "tts", "errors": 1}
async def _test_llm(self, llm_name: str, config: Dict) -> Dict:
"""异步测试单个LLM性能"""
try:
# 对于Ollama,跳过api_key检查并进行特殊处理
if llm_name == "Ollama":
base_url = config.get('base_url', 'http://localhost:11434')
model_name = config.get('model_name')
if not model_name:
print(f"🚫 Ollama未配置model_name")
return {"name": llm_name, "type": "llm", "errors": 1}
if not await self._check_ollama_service(base_url, model_name):
return {"name": llm_name, "type": "llm", "errors": 1}
else:
if "api_key" in config and any(x in config["api_key"] for x in ["你的", "placeholder", "sk-xxx"]):
print(f"🚫 跳过未配置的LLM: {llm_name}")
return {"name": llm_name, "type": "llm", "errors": 1}
# 获取实际类型(兼容旧配置)
module_type = config.get('type', llm_name)
llm = create_llm_instance(module_type, config)
# 统一使用UTF-8编码
test_sentences = [s.encode('utf-8').decode('utf-8') for s in self.test_sentences]
total_time = 0
first_token_times = []
valid_times = []
# 创建所有句子的测试任务
sentence_tasks = []
for sentence in test_sentences:
sentence_start = time.time() # 记录整句开始时间
first_token_received = False
# 遍历响应流
for chunk in llm.response("perf_test", [{"role": "user", "content": sentence}]):
if not first_token_received and chunk.strip() != '':
first_token_times.append(time.time() - sentence_start)
first_token_received = True
# 计算整句耗时
sentence_duration = time.time() - sentence_start
total_time += sentence_duration
valid_times.append(sentence_duration)
# 新增有效性检查
if len(first_token_times) == 0 or len(valid_times) == 0:
sentence_tasks.append(self._test_single_sentence(llm_name, llm, sentence))
# 并发执行所有句子测试
sentence_results = await asyncio.gather(*sentence_tasks)
# 处理结果
valid_results = [r for r in sentence_results if r is not None]
if not valid_results:
print(f"⚠️ {llm_name} 无有效数据,可能配置错误")
return {"errors": 1}
# 过滤异常数据(超过3倍标准差)
mean = statistics.mean(valid_times)
stdev = statistics.stdev(valid_times) if len(valid_times) > 1 else 0
filtered_times = [t for t in valid_times if t <= mean + 3*stdev]
# 当有效数据不足时标记错误
return {"name": llm_name, "type": "llm", "errors": 1}
first_token_times = [r["first_token_time"] for r in valid_results]
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_sentences) * 0.5:
print(f"⚠️ {llm_name} 有效数据不足,可能网络不稳定")
return {"errors": 1}
return {"name": llm_name, "type": "llm", "errors": 1}
return {
"avg_response": total_time / len(test_sentences),
"avg_first_token": sum(first_token_times)/len(first_token_times),
"name": llm_name,
"type": "llm",
"avg_response": sum(response_times) / len(response_times),
"avg_first_token": sum(first_token_times) / len(first_token_times),
"std_first_token": statistics.stdev(first_token_times) if len(first_token_times) > 1 else 0,
"std_response": statistics.stdev(valid_times) if len(valid_times) > 1 else 0,
"std_response": statistics.stdev(response_times) if len(response_times) > 1 else 0,
"errors": 0
}
except Exception as e:
print(f"LLM {llm_name} 测试失败: {str(e)}")
return {"errors": 1}
return {"name": llm_name, "type": "llm", "errors": 1}
def _test_tts(self, tts_name: str, config: Dict) -> Dict:
"""测试单个TTS性能"""
async def _test_single_sentence(self, llm_name: str, llm, sentence: str) -> Dict:
"""测试单个句子的性能"""
try:
# 关闭详细日志
logging.getLogger("core.providers.tts.base").setLevel(logging.WARNING)
# 跳过未配置密钥的模块
token_fields = ["access_token", "api_key", "token"]
if any(field in config and any(x in config[field] for x in ["你的", "placeholder"]) for field in token_fields):
print(f"⏭️ TTS {tts_name} 未配置access_token/api_key,已跳过")
return {"errors": 1}
# 获取实际类型(兼容旧配置)
module_type = config.get('type', tts_name)
tts = create_tts_instance(
module_type,
config,
delete_audio_file=True # 确保参数名称正确
)
# 简化后的输出
print(f"\n🎵 正在测试 TTS: {tts_name}")
print(f"🔊 测试 {tts_name}", end="", flush=True)
# 连接测试
test_conn = tts.to_tts("连接测试")
if not os.path.exists(test_conn):
print("❌ 连接失败")
return {"errors": 1}
else:
print("")
total_time = 0
test_count = len(self.test_sentences[:2])
for i, sentence in enumerate(self.test_sentences[:2], 1):
start = time.time()
file_path = tts.to_tts(sentence)
duration = time.time() - start
total_time += duration
# 显示简单的进度标识
if os.path.exists(file_path):
print(f"✓[{i}/{test_count}]", end="", flush=True)
else:
print(f"✗[{i}/{test_count}]", end="", flush=True)
print() # 换行
return {"avg_time": total_time / test_count, "errors": 0}
except requests.exceptions.ConnectionError:
print(f"\n{tts_name} 无法连接服务端")
return {"errors": 1}
except Exception as e:
print(f"\n⚠️ {tts_name} 测试失败: {str(e)}")
return {"errors": 1}
print(f"📝 {llm_name} 开始测试: {sentence[:20]}...")
sentence_start = time.time()
first_token_received = False
first_token_time = None
def run(self):
"""执行全量测试并自动跳过未配置的模块"""
print("🔍 开始自动检测已配置的模块...")
# 测试所有LLM
for llm_name, config in self.config.get("LLM", {}).items():
# 特殊处理CozeLLM的配置检查
if llm_name == "CozeLLM":
if any(x in config.get("bot_id", "") for x in ["你的"]) \
or any(x in config.get("user_id", "") for x in ["你的"]):
print(f"⏭️ LLM {llm_name} 未配置bot_id/user_id,已跳过")
continue
# 通用的api_key配置检查
if "api_key" in config and any(x in config["api_key"] for x in ["你的", "placeholder"]):
print(f"⏭️ LLM {llm_name} 未配置api_key,已跳过")
continue
print(f"🚀 正在测试 LLM: {llm_name}")
self.results["llm"][llm_name] = self._test_llm(llm_name, config)
# 测试所有TTS
for tts_name, config in self.config.get("TTS", {}).items():
# 根据不同服务的token字段检测
token_fields = ["access_token", "api_key", "token"]
if any(field in config and any(x in config[field] for x in ["你的", "placeholder"]) for field in token_fields):
print(f"⏭️ TTS {tts_name} 未配置access_token/api_key,已跳过")
continue
print(f"🎵 正在测试 TTS: {tts_name}")
self.results["tts"][tts_name] = self._test_tts(tts_name, config)
# 生成组合建议
self._generate_combinations()
self._print_results()
async def process_response():
nonlocal first_token_received, first_token_time
for chunk in llm.response("perf_test", [{"role": "user", "content": sentence}]):
if not first_token_received and chunk.strip() != '':
first_token_time = time.time() - sentence_start
first_token_received = True
print(f"{llm_name} 首个Token: {first_token_time:.3f}s")
yield chunk
response_chunks = []
async for chunk in process_response():
response_chunks.append(chunk)
response_time = time.time() - sentence_start
print(f" {llm_name} 完成响应: {response_time:.3f}s")
if first_token_time is None:
first_token_time = response_time # 如果没有检测到first token,使用总响应时间
return {
"name": llm_name,
"type": "llm",
"first_token_time": first_token_time,
"response_time": response_time
}
except Exception as e:
print(f"⚠️ {llm_name} 句子测试失败: {str(e)}")
return None
def _generate_combinations(self):
"""生成最佳组合建议"""
# 调整过滤条件,例如设为 >= 0.05
valid_llms = [
k for k, v in self.results["llm"].items()
k for k, v in self.results["llm"].items()
if v["errors"] == 0 and v["avg_first_token"] >= 0.05
]
valid_tts = [k for k, v in self.results["tts"].items() if v["errors"] == 0]
# 找出基准值
min_first_token = min([self.results["llm"][llm]["avg_first_token"] for llm in valid_llms]) if valid_llms else 1
min_tts_time = min([self.results["tts"][tts]["avg_time"] for tts in valid_tts]) if valid_tts else 1
for llm in valid_llms:
for tts in valid_tts:
llm_weight = 0.8 if self.results["llm"][llm]["avg_first_token"] < 1.0 else 0.6
tts_weight = 1 - llm_weight
score = (
self.results["llm"][llm]["avg_first_token"] * llm_weight +
self.results["tts"][tts]["avg_time"] * tts_weight
)
# 计算相对性能分数(越小越好)
llm_score = self.results["llm"][llm]["avg_first_token"] / min_first_token
tts_score = self.results["tts"][tts]["avg_time"] / min_tts_time
# 计算稳定性分数(标准差/平均值,越小越稳定)
llm_stability = self.results["llm"][llm]["std_first_token"] / self.results["llm"][llm][
"avg_first_token"]
# 综合得分(考虑性能和稳定性)
# 性能权重0.7,稳定性权重0.3
llm_final_score = llm_score * 0.7 + llm_stability * 0.3
# 总分 = LLM得分(70%) + TTS得分(30%)
total_score = llm_final_score * 0.7 + tts_score * 0.3
self.results["combinations"].append({
"llm": llm,
"tts": tts,
"score": score,
"score": total_score,
"details": {
"llm_first_token": self.results["llm"][llm]["avg_first_token"],
"llm_stability": llm_stability,
"tts_time": self.results["tts"][tts]["avg_time"]
}
})
# 按综合得分排序
# 分数越小越好
self.results["combinations"].sort(key=lambda x: x["score"])
def _print_results(self):
"""控制台输出结果"""
# LLM结果表格
"""打印测试结果"""
llm_table = []
for name, data in self.results["llm"].items():
if data["errors"] == 0:
stability = data["std_first_token"] / data["avg_first_token"]
llm_table.append([
name,
f"{data['avg_first_token']:.3f}s",
f"{data['avg_response']:.3f}s"
name, # 不需要固定宽度,让tabulate自己处理对齐
f"{data['avg_first_token']:.3f}",
f"{data['avg_response']:.3f}",
f"{stability:.3f}"
])
if llm_table:
print("\nLLM 性能排行:")
print(tabulate(
llm_table,
headers=["名称", "平均首Token时间", "平均总响应时间"],
tablefmt="github"
headers=["名称", "首字耗时", "总耗时", "稳定性"],
tablefmt="github",
colalign=("left", "right", "right", "right"),
disable_numparse=True
))
else:
print("\n⚠️ 没有可用的LLM模块进行测试。")
# TTS结果表格
tts_table = []
for name, data in self.results["tts"].items():
if data["errors"] == 0:
tts_table.append([
name,
f"{data['avg_time']:.3f}s"
name, # 不需要固定宽度
f"{data['avg_time']:.3f}"
])
if tts_table:
print("\nTTS 性能排行:")
print(tabulate(
tts_table,
headers=["名称", "平均合成时间"],
tablefmt="github"
headers=["名称", "合成耗时"],
tablefmt="github",
colalign=("left", "right"),
disable_numparse=True
))
else:
print("\n⚠️ 没有可用的TTS模块进行测试。")
# 最佳组合建议
if self.results["combinations"]:
print("\n推荐配置组合 (综合响应速度):")
print("\n推荐配置组合 (得分越小越好):")
combo_table = []
for combo in self.results["combinations"][:5]: # 显示前5名
for combo in self.results["combinations"][:5]:
combo_table.append([
f"{combo['llm']} + {combo['tts']}",
f"{combo['llm']} + {combo['tts']}", # 不需要固定宽度
f"{combo['score']:.3f}",
f"{combo['details']['llm_first_token']:.3f}s",
f"{combo['details']['tts_time']:.3f}s"
f"{combo['details']['llm_first_token']:.3f}",
f"{combo['details']['llm_stability']:.3f}",
f"{combo['details']['tts_time']:.3f}"
])
print(tabulate(
combo_table,
headers=["组合方案", "综合得分", "LLM首Token", "TTS合成"],
tablefmt="github"
headers=["组合方案", "综合得分", "LLM首字耗时", "稳定性", "TTS合成耗时"],
tablefmt="github",
colalign=("left", "right", "right", "right", "right"),
disable_numparse=True
))
else:
print("\n⚠️ 没有可用的模块组合建议。")
def _execute_with_timeout(self, func, args=(), kwargs={}, timeout=None):
with ThreadPoolExecutor(max_workers=1) as executor:
future = executor.submit(func, *args, **kwargs)
try:
result = future.result(timeout)
return list(result) if inspect.isgenerator(result) else result
except TimeoutError:
raise Exception("操作超时")
def _process_results(self, all_results):
"""处理测试结果"""
for result in all_results:
if result["errors"] == 0:
if result["type"] == "llm":
self.results["llm"][result["name"]] = result
else:
self.results["tts"][result["name"]] = result
async def run(self):
"""执行全量异步测试"""
print("🔍 开始筛选可用模块...")
# 创建所有测试任务
all_tasks = []
# LLM测试任务
for llm_name, config in self.config.get("LLM", {}).items():
# 检查配置有效性
if llm_name == "CozeLLM":
if any(x in config.get("bot_id", "") for x in ["你的"]) \
or any(x in config.get("user_id", "") for x in ["你的"]):
print(f"⏭️ LLM {llm_name} 未配置bot_id/user_id,已跳过")
continue
elif "api_key" in config and any(x in config["api_key"] for x in ["你的", "placeholder", "sk-xxx"]):
print(f"⏭️ LLM {llm_name} 未配置api_key,已跳过")
continue
# 对于Ollama,先检查服务状态
if llm_name == "Ollama":
base_url = config.get('base_url', 'http://localhost:11434')
model_name = config.get('model_name')
if not model_name:
print(f"🚫 Ollama未配置model_name")
continue
if not await self._check_ollama_service(base_url, model_name):
continue
print(f"📋 添加LLM测试任务: {llm_name}")
module_type = config.get('type', llm_name)
llm = create_llm_instance(module_type, config)
# 为每个句子创建独立任务
for sentence in self.test_sentences:
sentence = sentence.encode('utf-8').decode('utf-8')
all_tasks.append(self._test_single_sentence(llm_name, llm, sentence))
# TTS测试任务
for tts_name, config in self.config.get("TTS", {}).items():
token_fields = ["access_token", "api_key", "token"]
if any(field in config and any(x in config[field] for x in ["你的", "placeholder"]) for field in
token_fields):
print(f"⏭️ TTS {tts_name} 未配置access_token/api_key,已跳过")
continue
print(f"🎵 添加TTS测试任务: {tts_name}")
all_tasks.append(self._test_tts(tts_name, config))
print(
f"\n✅ 找到 {len([t for t in all_tasks if 'test_single_sentence' in str(t)]) / len(self.test_sentences):.0f} 个可用LLM模块")
print(f"✅ 找到 {len([t for t in all_tasks if '_test_tts' in str(t)])} 个可用TTS模块")
print("\n⏳ 开始并发测试所有模块...\n")
# 并发执行所有测试任务
all_results = await asyncio.gather(*all_tasks, return_exceptions=True)
# 处理LLM结果
llm_results = {}
for result in [r for r in all_results if r and isinstance(r, dict) and r.get("type") == "llm"]:
llm_name = result["name"]
if llm_name not in llm_results:
llm_results[llm_name] = {
"name": llm_name,
"type": "llm",
"first_token_times": [],
"response_times": [],
"errors": 0
}
llm_results[llm_name]["first_token_times"].append(result["first_token_time"])
llm_results[llm_name]["response_times"].append(result["response_time"])
# 计算LLM平均值和标准差
for llm_name, data in llm_results.items():
if len(data["first_token_times"]) >= len(self.test_sentences) * 0.5:
self.results["llm"][llm_name] = {
"name": llm_name,
"type": "llm",
"avg_response": sum(data["response_times"]) / len(data["response_times"]),
"avg_first_token": sum(data["first_token_times"]) / len(data["first_token_times"]),
"std_first_token": statistics.stdev(data["first_token_times"]) if len(
data["first_token_times"]) > 1 else 0,
"std_response": statistics.stdev(data["response_times"]) if len(data["response_times"]) > 1 else 0,
"errors": 0
}
# 处理TTS结果
for result in [r for r in all_results if r and isinstance(r, dict) and r.get("type") == "tts"]:
if result["errors"] == 0:
self.results["tts"][result["name"]] = result
# 生成组合建议并打印结果
print("\n📊 生成测试报告...")
self._generate_combinations()
self._print_results()
async def main():
tester = AsyncPerformanceTester()
await tester.run()
if __name__ == "__main__":
tester = PerformanceTester()
tester.run()
asyncio.run(main())
+1 -1
View File
@@ -16,4 +16,4 @@ aiohttp_cors==0.7.0
ormsgpack==1.7.0
ruamel.yaml==0.18.10
loguru==0.7.3
requests>=2.0.0
requests==2.32.3