py-xiaozhi 二次开发指南
本文档适用于希望对 py-xiaozhi 进行功能扩展、插件开发或 UI 定制的开发者。
目录
项目架构概览
目录结构
py-xiaozhi/
├── main.py # 应用入口
├── src/
│ ├── activation/ # 设备激活模块
│ ├── audio_codecs/ # 音频编解码器
│ ├── audio_processing/ # 音频处理(AEC、VAD等)
│ ├── bootstrap/ # 服务容器与启动逻辑
│ │ ├── container.py # ServiceContainer 核心
│ │ └── protocols.py # 插件接口协议
│ ├── constants/ # 常量定义
│ ├── core/ # 核心模块
│ │ ├── event_bus.py # 事件总线
│ │ ├── state_manager.py# 状态管理器
│ │ └── task_manager.py # 任务管理器
│ ├── logging/ # 日志系统
│ ├── mcp/ # MCP 工具系统
│ │ ├── decorators.py # 工具装饰器
│ │ ├── mcp_server.py # MCP 服务器
│ │ ├── tooling.py # 工具基础设施
│ │ └── tools/ # 工具实现目录
│ ├── plugins/ # 插件系统
│ │ ├── base.py # 插件基类
│ │ ├── manager.py # 插件管理器
│ │ ├── audio.py # 音频插件
│ │ ├── ui.py # UI 插件
│ │ ├── mcp.py # MCP 插件
│ │ └── wake_word.py # 唤醒词插件
│ ├── protocols/ # 通信协议
│ ├── ui/ # UI 层
│ │ ├── cli/ # 命令行界面
│ │ ├── gui/ # 图形界面
│ │ │ ├── manager.py # ViewManager
│ │ │ ├── qml/ # QML 文件
│ │ │ └── services/ # GUI 服务
│ │ ├── gpio/ # GPIO 按键界面(仅 Linux)
│ │ │ ├── manager.py # GPIOViewManager
│ │ │ └── input.py # GPIO 按键输入
│ │ └── shared/ # 共享组件
│ │ ├── bridge.py # EventBridge
│ │ └── models/ # 数据模型
│ └── utils/ # 工具类核心组件关系
┌─────────────────────────────────────────────────────────┐
│ ServiceContainer │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────────┐ │
│ │ TaskManager │ │StateManager │ │ EventBus │ │
│ └─────────────┘ └─────────────┘ └─────────────────┘ │
│ │ │
│ ┌───────────────────────────────────────────────────┐ │
│ │ PluginManager │ │
│ │ ┌────────┐ ┌────────┐ ┌────────┐ ┌────────────┐ │ │
│ │ │ Audio │ │ UI │ │ MCP │ │ WakeWord │ │ │
│ │ │ Plugin │ │ Plugin │ │ Plugin │ │ Plugin │ │ │
│ │ └────────┘ └────────┘ └────────┘ └────────────┘ │ │
│ └───────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────┘快速开始
环境准备
# 克隆项目
git clone https://github.com/your-repo/py-xiaozhi.git
cd py-xiaozhi
# 创建虚拟环境
python -m venv venv
source venv/bin/activate # Linux/macOS
# 或 venv\Scripts\activate # Windows
# 安装依赖
pip install -r requirements.txt运行应用
# GUI 模式
python main.py --mode gui
# CLI 模式
python main.py --mode cli
# 跳过激活流程(调试用)
python main.py --mode gui --skip-activation修改项目名称
如果你需要修改项目名称(例如二次开发),需要修改 src/constants/system.py 中的 APP_NAME:
# src/constants/system.py
class SystemConstants:
APP_NAME = "my-xiaozhi" # 修改为你的项目名称
APP_VERSION = "2.0.0"
# ...修改后的影响:
- 用户数据目录会变更为
~/Library/Application Support/my-xiaozhi/(macOS) - 配置文件、缓存、日志等都会存储在新目录下
- 不同名称的项目数据相互隔离,不会冲突
注意:修改项目名称后,原有的用户配置不会自动迁移,需要手动复制或重新配置。
自定义验证码播报
设备激活时会播报验证码,使用预录制的音频文件。音频资源位于 assets/sounds/ 目录。
支持的语言(37种):
assets/sounds/
├── zh-CN/ # 简体中文(默认)
├── zh-TW/ # 繁体中文
├── en-US/ # 英语
├── ja-JP/ # 日语
├── ko-KR/ # 韩语
├── de-DE/ # 德语
├── fr-FR/ # 法语
├── es-ES/ # 西班牙语
├── ... (更多语言)
└── vi-VN/ # 越南语每个语言包含的音频文件:
| 文件 | 说明 |
|---|---|
0.ogg ~ 9.ogg | 数字 0-9 的发音 |
activation.ogg | 激活提示语(如"请输入验证码") |
修改播报语言:
在 src/activation/service.py 中修改 locale 参数:
from src.utils.activation_announcer import announce_activation_code
# 使用英语播报
announce_activation_code(code, locale="en-US")
# 使用日语播报
announce_activation_code(code, locale="ja-JP")替换音频文件:
- 准备新的 OGG 格式音频文件(建议采样率 24kHz)
- 替换对应语言目录下的文件
- 文件名必须保持一致(
0.ogg~9.ogg,activation.ogg)
添加新语言:
- 在
assets/sounds/下创建新的语言目录(如my-lang/) - 添加 11 个必需的 OGG 音频文件
- 调用时指定新的 locale:
announce_activation_code(code, locale="my-lang")
自定义 GPIO 引脚
GPIO 模式仅支持 Linux 系统(树莓派),用于通过物理按键控制设备。
推荐硬件:四按钮独立按键模块


硬件接线说明
按钮模块 6Pin(1 2 3 4 V G)接树莓派 5:

电源接线:
| 模块引脚 | 树莓派引脚 | 说明 |
|---|---|---|
| V | 3.3V(Pin 1 或 Pin 17) | 电源正极 |
| G | GND(Pin 6/9/14/20/25/30/34/39) | 电源负极,推荐 Pin 6 |
⚠️ 注意:不要接 5V(Pin 2/4),会损坏模块!
按键接线:
| 模块引脚 | GPIO(BCM) | 物理引脚 | 功能 |
|---|---|---|---|
| 1 | GPIO 17 | Pin 11 | 开始/停止对话 |
| 2 | GPIO 27 | Pin 13 | 中断当前语音 |
| 3 | GPIO 22 | Pin 15 | 切换手动/自动模式 |
| 4 | GPIO 23 | Pin 16 | 退出程序 |
模块特性:按下输出低电平,代码使用 BCM 编号:17/27/22/23
修改引脚配置:
编辑 src/ui/gpio/input.py 中的 DEFAULT_PINS:
# src/ui/gpio/input.py
# 默认 GPIO 引脚配置(BCM 编号)
# 按顺序对应 KEY1, KEY2, KEY3, KEY4
DEFAULT_PINS: List[int] = [17, 27, 22, 23]
# 修改为你实际的接线引脚,例如:
# DEFAULT_PINS: List[int] = [5, 6, 13, 19]依赖安装:
# 安装 gpiozero 库
sudo apt install python3-gpiozero python3-rpi.gpio
# 可选:如果系统有这个包,也装上(gpiozero 会优先用它)
sudo apt install -y python3-lgpio || true
# 把当前用户加入 gpio 组(避免每次 sudo)
sudo usermod -aG gpio $USER
# 重新登录一次或重启后生效运行 GPIO 模式:
python main.py --mode gpio --protocol websocket购买按钮模块:

插件开发
插件是扩展 py-xiaozhi 功能的核心方式。通过继承 Plugin 基类,你可以:
- 响应系统事件(音频数据、JSON 消息、状态变更等)
- 调用核心命令(发送音频、启动监听等)
- 与其他插件交互
创建插件
在 src/plugins/ 目录下创建新文件:
# src/plugins/my_plugin.py
from typing import Any, TYPE_CHECKING
from src.plugins.base import Plugin
from src.logging import get_logger
if TYPE_CHECKING:
from src.bootstrap.protocols import PluginCommands, PluginContext
logger = get_logger()
class MyPlugin(Plugin):
"""我的自定义插件"""
# 插件唯一标识名(必需)
name = "my_plugin"
# 优先级:1-100,数值越小越优先初始化
priority = 50
# 依赖的其他插件(可选)
requires = ["audio"] # 声明依赖 AudioPlugin
def __init__(self):
super().__init__()
self._my_state = None
async def setup(self, ctx: "PluginContext", cmd: "PluginCommands") -> None:
"""插件初始化阶段
Args:
ctx: 插件上下文,用于获取状态(只读)
cmd: 插件命令接口,用于执行操作
"""
await super().setup(ctx, cmd)
# 通过 self.ctx 访问上下文
config = self.ctx.get_config()
# 通过 self.deps 访问依赖的插件
audio_plugin = self.deps.get("audio")
logger.info("MyPlugin 初始化完成")
async def start(self) -> None:
"""插件启动(协议连接建立后调用)"""
await super().start()
# 订阅事件
from src.core.event_bus import Events
self.ctx.event_bus.on(Events.DEVICE_STATE_CHANGED, self._on_state_changed)
logger.info("MyPlugin 已启动")
async def on_protocol_connected(self, protocol: Any) -> None:
"""协议通道建立后的通知"""
logger.info("协议已连接")
async def on_incoming_json(self, message: Any) -> None:
"""收到 JSON 消息时的通知
Args:
message: JSON 消息字典
"""
if not isinstance(message, dict):
return
msg_type = message.get("type")
if msg_type == "my_custom_type":
# 处理自定义消息
pass
async def on_incoming_audio(self, data: bytes) -> None:
"""收到音频数据时的通知
Args:
data: 原始音频字节数据
"""
# 处理音频数据
pass
async def on_device_state_changed(self, state: Any) -> None:
"""设备状态变更通知"""
from src.constants.constants import DeviceState
if state == DeviceState.LISTENING:
logger.info("设备开始监听")
elif state == DeviceState.SPEAKING:
logger.info("设备正在说话")
async def _on_state_changed(self, state):
"""EventBus 事件处理器"""
logger.debug(f"状态变更: {state}")
async def stop(self) -> None:
"""插件停止"""
await super().stop()
logger.info("MyPlugin 已停止")
async def shutdown(self) -> None:
"""插件最终清理"""
# 取消订阅事件
from src.core.event_bus import Events
self.ctx.event_bus.off(Events.DEVICE_STATE_CHANGED, self._on_state_changed)
await super().shutdown()
logger.info("MyPlugin 已清理")注册插件
在 src/bootstrap/container.py 中注册插件:
from src.plugins.my_plugin import MyPlugin
# 在 _setup_plugins 方法中添加
self.plugins.register(
AudioPlugin(),
UIPlugin(mode=mode),
McpPlugin(),
WakeWordPlugin(),
MyPlugin(), # 添加你的插件
)PluginContext 接口
PluginContext 提供只读状态访问:
| 方法 | 说明 |
|---|---|
get_device_state() | 获取当前设备状态 |
get_listening_mode() | 获取当前监听模式 |
is_listening() | 是否正在监听 |
is_speaking() | 是否正在说话 |
is_idle() | 是否空闲 |
is_audio_channel_opened() | 音频通道是否已打开 |
should_capture_audio() | 是否应该采集音频 |
get_config() | 获取配置管理器 |
event_bus | 获取事件总线 |
PluginCommands 接口
PluginCommands 提供操作执行:
| 方法 | 说明 |
|---|---|
start_listening(mode) | 开始监听 |
stop_listening() | 停止监听 |
abort_speaking(reason) | 中断说话 |
send_audio(data) | 发送音频数据 |
send_text(text) | 发送文本 |
send_wake_word_detected(text) | 发送唤醒词检测 |
send_mcp_message(payload) | 发送 MCP 消息 |
connect_protocol() | 连接协议 |
spawn(coro, name) | 异步任务派生 |
request_shutdown() | 请求应用关闭 |
MCP 工具开发
MCP (Model Context Protocol) 是与 AI 模型交互的工具协议。通过定义 MCP 工具,AI 可以调用你的功能。
创建 MCP 工具
方式一:使用装饰器(推荐)
在 src/mcp/tools/ 目录下创建工具:
# src/mcp/tools/my_tools/_tools.py
from typing import Any, Dict
from src.mcp.decorators import Prop, PropType, mcp_tool
from src.logging import get_logger
logger = get_logger()
@mcp_tool(
name="my_tools.greet",
description=(
"向用户打招呼。"
"当用户说'你好'、'Hi'、'打个招呼'时调用此工具。"
),
props=[
Prop("name", PropType.STR), # 必需的字符串参数
],
)
async def tool_greet(args: Dict[str, Any]) -> str:
"""打招呼工具实现"""
name = args.get("name", "朋友")
return f"你好,{name}!很高兴认识你!"
@mcp_tool(
name="my_tools.calculate",
description=(
"执行简单的数学计算。"
"支持两个数字的加减乘除运算。"
),
props=[
Prop("num1", PropType.INT),
Prop("num2", PropType.INT),
Prop("operation", PropType.STR), # add, subtract, multiply, divide
],
)
async def tool_calculate(args: Dict[str, Any]) -> str:
"""计算工具实现"""
num1 = args.get("num1", 0)
num2 = args.get("num2", 0)
operation = args.get("operation", "add")
operations = {
"add": num1 + num2,
"subtract": num1 - num2,
"multiply": num1 * num2,
"divide": num1 / num2 if num2 != 0 else "除数不能为零",
}
result = operations.get(operation, "未知操作")
return f"计算结果: {num1} {operation} {num2} = {result}"
@mcp_tool(
name="my_tools.set_reminder",
description=(
"设置一个提醒。"
"用于用户说'提醒我...'、'设置提醒...'等场景。"
),
props=[
Prop("message", PropType.STR),
Prop("minutes", PropType.INT, min_val=1, max_val=1440), # 带范围限制
Prop("repeat", PropType.BOOL, default=False), # 带默认值
],
)
async def tool_set_reminder(args: Dict[str, Any]) -> str:
"""设置提醒工具"""
message = args.get("message", "")
minutes = args.get("minutes", 5)
repeat = args.get("repeat", False)
# 实现提醒逻辑...
repeat_text = "(每天重复)" if repeat else ""
return f"已设置提醒: {minutes}分钟后提醒您「{message}」{repeat_text}"目录结构要求:
src/mcp/tools/
├── __init__.py
├── my_tools/
│ ├── __init__.py # 必需,可以为空
│ └── _tools.py # 工具定义文件(会被自动发现)方式二:手动注册
from src.mcp.tooling import McpTool, Property, PropertyList, PropertyType
# 创建属性列表
props = PropertyList([
Property("param1", PropertyType.STRING),
Property("param2", PropertyType.INTEGER, min_value=0, max_value=100),
])
# 创建工具
tool = McpTool(
name="my_tool",
description="我的工具描述",
properties=props,
callback=my_callback_function,
)
# 手动添加到 MCP 服务器
mcp_server = McpServer.get_instance()
mcp_server.add_tool(tool)属性类型
| PropType | 说明 | 对应参数 |
|---|---|---|
PropType.STR | 字符串 | name: str |
PropType.INT | 整数 | min_val, max_val |
PropType.BOOL | 布尔值 | default: bool |
工具描述最佳实践
工具的 description 非常重要,AI 会根据描述决定何时调用工具:
@mcp_tool(
name="music_player.play",
description=(
"播放指定的音乐。" # 简短说明
"当用户说'播放...'、'放一首...'时调用。" # 触发场景
"如果已有音乐在播放,会先停止当前音乐。" # 行为说明
"注意:不要在 TTS 说话时主动调用。" # 注意事项
),
...
)UI 页面开发
py-xiaozhi 的 GUI 使用 PySide6 + QML (QtQuick) 架构,遵循 MVVM 模式。
架构说明
┌─────────────────────────────────────────────────────────┐
│ QML (View) │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────────┐ │
│ │ MainWindow │ │Settings │ │ MyWindow │ │
│ │ .qml │ │ Window.qml │ │ .qml │ │
│ └──────┬──────┘ └──────┬──────┘ └────────┬────────┘ │
│ │ │ │ │
│ ▼ ▼ ▼ │
│ ┌───────────────────────────────────────────────────┐ │
│ │ EventBridge (Signal/Slot) │ │
│ └───────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────┐
│ Python (ViewModel) │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────────┐ │
│ │ MainModel │ │ Settings │ │ MyModel │ │
│ │ │ │ Model │ │ │ │
│ └─────────────┘ └─────────────┘ └─────────────────┘ │
└─────────────────────────────────────────────────────────┘创建新页面
1. 创建 QML 文件
// src/ui/gui/qml/windows/MyWindow.qml
import QtQuick
import QtQuick.Controls
import QtQuick.Layouts
import "../theme"
import "../components"
AppWindow {
id: root
width: 400
height: 300
minimumWidth: 300
minimumHeight: 200
title: "我的窗口"
Rectangle {
id: content
anchors.fill: parent
anchors.margins: root.isMaximized ? 0 : 1
color: Theme.background
ColumnLayout {
anchors.fill: parent
anchors.margins: Theme.spacingMd
spacing: Theme.spacingMd
// 标题
Text {
text: "欢迎使用"
font.pixelSize: Theme.fontSizeLg
font.weight: Font.Bold
color: Theme.textPrimary
}
// 动态数据绑定
Text {
text: myModel ? myModel.message : "加载中..."
font.pixelSize: Theme.fontSizeMd
color: Theme.textSecondary
wrapMode: Text.WordWrap
Layout.fillWidth: true
}
// 输入框
Rectangle {
Layout.fillWidth: true
Layout.preferredHeight: 40
color: Theme.background
radius: Theme.radiusMd
border.color: inputField.activeFocus ? Theme.primary : Theme.border
border.width: 1
TextInput {
id: inputField
anchors.fill: parent
anchors.margins: 10
font.pixelSize: Theme.fontSizeMd
color: Theme.textPrimary
verticalAlignment: TextInput.AlignVCenter
// 占位符
Text {
anchors.fill: parent
text: "请输入内容..."
font: parent.font
color: Theme.textPlaceholder
verticalAlignment: Text.AlignVCenter
visible: !parent.text && !parent.activeFocus
}
}
}
// 按钮
Button {
id: actionBtn
Layout.preferredWidth: 120
Layout.preferredHeight: 40
text: "确认"
background: Rectangle {
color: actionBtn.pressed
? Theme.primaryPressed
: (actionBtn.hovered ? Theme.primaryHover : Theme.primary)
radius: Theme.radiusMd
}
contentItem: Text {
text: actionBtn.text
font.pixelSize: Theme.fontSizeMd
color: "white"
horizontalAlignment: Text.AlignHCenter
verticalAlignment: Text.AlignVCenter
}
onClicked: {
if (eventBridge) {
eventBridge.onMyAction(inputField.text)
}
}
}
// 填充剩余空间
Item { Layout.fillHeight: true }
}
}
}2. 创建 Python Model
# src/ui/shared/models/my_model.py
from PySide6.QtCore import QObject, Property, Signal, Slot
class MyModel(QObject):
"""我的窗口数据模型"""
# 信号定义
messageChanged = Signal()
dataListChanged = Signal()
def __init__(self, parent=None):
super().__init__(parent)
self._message = "Hello World"
self._data_list = []
# ========== message 属性 ==========
def get_message(self) -> str:
return self._message
def set_message(self, value: str) -> None:
if self._message != value:
self._message = value
self.messageChanged.emit()
message = Property(str, get_message, set_message, notify=messageChanged)
# ========== dataList 属性 ==========
def get_data_list(self) -> list:
return self._data_list
def set_data_list(self, value: list) -> None:
self._data_list = value
self.dataListChanged.emit()
dataList = Property(list, get_data_list, set_data_list, notify=dataListChanged)
# ========== Slot 方法(可从 QML 调用)==========
@Slot(str, result=bool)
def validate_input(self, text: str) -> bool:
"""验证输入"""
return len(text) > 0
@Slot()
def refresh(self) -> None:
"""刷新数据"""
# 实现刷新逻辑
pass3. 扩展 EventBridge
# src/ui/shared/bridge.py 中添加
class EventBridge(QObject):
# ... 现有代码 ...
# 添加新的信号
myActionRequested = Signal(str)
@Slot(str)
def onMyAction(self, data: str) -> None:
"""处理我的操作"""
asyncio.create_task(
self._event_bus.emit(Events.MY_CUSTOM_ACTION, {"data": data})
)4. 在 ViewManager 中注册
# src/ui/gui/manager.py
from src.ui.shared.models.my_model import MyModel
class ViewManager(QObject):
def __init__(self, event_bus: EventBus):
# ... 现有代码 ...
self._my_model = MyModel()
def _inject_context(self):
ctx = self._engine.rootContext()
# ... 现有代码 ...
ctx.setContextProperty("myModel", self._my_model)
@property
def my_model(self) -> MyModel:
return self._my_model主题变量
在 QML 中使用 Theme 对象访问主题变量:
颜色
| 变量 | 说明 |
|---|---|
| 主色系 | |
Theme.primary | 主色 #165DFF |
Theme.primaryHover | 主色悬停 #4080FF |
Theme.primaryPressed | 主色按下 #0E42D2 |
Theme.primaryLight | 浅蓝背景 #E8F3FF |
Theme.primaryText | 蓝色文字 #2196F3 |
| 功能色 | |
Theme.success | 成功色 #00B42A |
Theme.successLight | 成功浅背景 #E8FFEA |
Theme.successBorder | 成功边框 #B7EB8F |
Theme.warning | 警告色 #FF7D00 |
Theme.warningLight | 警告浅背景 #FFF7E8 |
Theme.warningBorder | 警告边框 #FFE58F |
Theme.error | 错误色 #F53F3F |
Theme.errorHover | 错误悬停 #FF7875 |
Theme.errorLight | 错误浅背景 #FFF2F0 |
Theme.errorBorder | 错误边框 #FFCCC7 |
| 背景色 | |
Theme.background | 背景色 #FFFFFF |
Theme.backgroundSecondary | 次要背景色 #F7F8FA |
Theme.backgroundHover | 悬停背景色 #F2F3F5 |
| 文字色 | |
Theme.textPrimary | 主要文字颜色 #1D2129 |
Theme.textSecondary | 次要文字颜色 #4E5969 |
Theme.textPlaceholder | 占位符文字颜色 #86909C |
| 边框分割线 | |
Theme.border | 边框颜色 #E5E6EB |
Theme.divider | 分割线颜色 #F2F3F5 |
阴影
| 变量 | 说明 |
|---|---|
Theme.shadowColor | 主阴影色 #15000000 |
Theme.shadowLight | 轻阴影 #08000000 |
Theme.shadowMedium | 中阴影 #06000000 |
Theme.shadowSubtle | 微阴影 #04000000 |
字体
| 变量 | 说明 |
|---|---|
Theme.fontSizeXs | 超小字体 (10px) |
Theme.fontSizeSm | 小号字体 (12px) |
Theme.fontSizeMd | 中号字体 (14px) |
Theme.fontSizeLg | 大号字体 (16px) |
Theme.fontSizeXl | 超大字体 (20px) |
Theme.fontSizeXxl | 巨大字体 (24px) |
Theme.fontFamily | 系统字体 |
Theme.fontFamilyMono | 等宽字体 |
间距
| 变量 | 说明 |
|---|---|
Theme.spacingXs | 超小间距 (4px) |
Theme.spacingSm | 小间距 (8px) |
Theme.spacingMd | 中间距 (12px) |
Theme.spacingLg | 大间距 (16px) |
Theme.spacingXl | 超大间距 (20px) |
Theme.spacingXxl | 巨大间距 (24px) |
圆角
| 变量 | 说明 |
|---|---|
Theme.radiusSm | 小圆角 (4px) |
Theme.radiusMd | 中圆角 (8px) |
Theme.radiusLg | 大圆角 (12px) |
Theme.radiusXl | 超大圆角 (16px) |
动画
| 变量 | 说明 |
|---|---|
Theme.animationFast | 快速动画 (150ms) |
Theme.animationNormal | 普通动画 (200ms) |
Theme.animationSlow | 慢速动画 (300ms) |
事件总线通信
EventBus 是组件间解耦通信的核心机制。
基本用法
from src.core.event_bus import Events, EventBus
# 获取 EventBus 实例(通过插件上下文)
event_bus = self.ctx.event_bus
# 订阅事件
async def my_handler(data):
print(f"收到数据: {data}")
event_bus.on(Events.DEVICE_STATE_CHANGED, my_handler)
# 触发事件(并行执行所有处理器)
await event_bus.emit(Events.DEVICE_STATE_CHANGED, new_state)
# 顺序触发事件
await event_bus.emit_sequential(Events.DEVICE_STATE_CHANGED, new_state)
# 取消订阅
event_bus.off(Events.DEVICE_STATE_CHANGED, my_handler)
# 清除所有处理器
event_bus.clear(Events.DEVICE_STATE_CHANGED)预定义事件
class Events:
# 设备状态
DEVICE_STATE_CHANGED = "device_state_changed"
# 协议相关
PROTOCOL_CONNECTED = "protocol_connected"
PROTOCOL_DISCONNECTED = "protocol_disconnected"
INCOMING_JSON = "incoming_json"
INCOMING_AUDIO = "incoming_audio"
# 网络
NETWORK_ERROR = "network_error"
# 音频通道
AUDIO_CHANNEL_OPENED = "audio_channel_opened"
AUDIO_CHANNEL_CLOSED = "audio_channel_closed"
# 应用生命周期
APP_SHUTDOWN = "app_shutdown"
# 音乐播放器
MUSIC_STATE_CHANGED = "music_state_changed"
MUSIC_LYRICS_UPDATE = "music_lyrics_update"
MUSIC_PROGRESS_UPDATE = "music_progress_update"
MUSIC_PAUSE_REQUEST = "music_pause_request"
MUSIC_RESUME_REQUEST = "music_resume_request"
# UI 用户操作(View → Plugin)
UI_BUTTON_PRESS = "ui_button_press"
UI_BUTTON_RELEASE = "ui_button_release"
UI_MANUAL_TOGGLE = "ui_manual_toggle"
UI_AUTO_TOGGLE = "ui_auto_toggle"
UI_AUTO_START = "ui_auto_start"
UI_ABORT_REQUEST = "ui_abort_request"
UI_SEND_TEXT = "ui_send_text"
UI_QUIT_REQUEST = "ui_quit_request"
UI_OPEN_SETTINGS = "ui_open_settings"
# UI 更新(Plugin → View)
UI_UPDATE_TEXT = "ui_update_text"
UI_UPDATE_EMOTION = "ui_update_emotion"
UI_UPDATE_STATUS = "ui_update_status"
UI_TOGGLE_MODE = "ui_toggle_mode"
UI_TOGGLE_WINDOW = "ui_toggle_window"
# 配置
CONFIG_CHANGED = "config_changed"添加自定义事件
# src/core/event_bus.py
class Events:
# ... 现有事件 ...
# 添加自定义事件
MY_CUSTOM_EVENT = "my_custom_event"
MY_DATA_UPDATED = "my_data_updated"事件数据类
建议为复杂事件数据创建数据类:
# src/my_module/events.py
from dataclasses import dataclass
@dataclass
class MyEventData:
"""我的事件数据"""
id: str
name: str
value: int
timestamp: float
# 使用
await event_bus.emit(Events.MY_CUSTOM_EVENT, MyEventData(
id="123",
name="test",
value=42,
timestamp=time.time(),
))配置管理
使用 ConfigManager 管理应用配置。
读取配置
from src.utils.config_manager import ConfigManager
config = ConfigManager.get_instance()
# 读取配置项
value = config.get_config("SECTION.KEY", default_value)
# 示例
aec_enabled = config.get_config("AEC_OPTIONS.ENABLED", True)
server_url = config.get_config("SERVER.URL", "ws://localhost:8080")写入配置
# 设置配置项
config.set_config("SECTION.KEY", value)
# 保存到文件
config.save()配置热重载
当配置变更需要通知其他组件时:
from src.core.event_bus import Events
# 保存配置后触发事件
config.save()
await self.ctx.event_bus.emit(Events.CONFIG_CHANGED, {
"key": "SECTION.KEY",
"value": new_value,
})调试与测试
调试模式启动
# 跳过激活流程
python main.py --mode gui --skip-activation
# CLI 模式(更容易看日志)
python main.py --mode cli --skip-activation日志系统
from src.logging import get_logger
logger = get_logger()
logger.debug("调试信息")
logger.info("一般信息")
logger.warning("警告信息")
logger.error("错误信息", exc_info=True)单元测试
# tests/test_my_plugin.py
import pytest
from unittest.mock import AsyncMock, MagicMock
from src.plugins.my_plugin import MyPlugin
class TestMyPlugin:
@pytest.fixture
def plugin(self):
return MyPlugin()
@pytest.fixture
def mock_ctx(self):
ctx = MagicMock()
ctx.event_bus = MagicMock()
ctx.get_config.return_value = MagicMock()
return ctx
@pytest.fixture
def mock_cmd(self):
return MagicMock()
@pytest.mark.asyncio
async def test_setup(self, plugin, mock_ctx, mock_cmd):
await plugin.setup(mock_ctx, mock_cmd)
assert plugin._ctx == mock_ctx
assert plugin._cmd == mock_cmd
@pytest.mark.asyncio
async def test_on_incoming_json(self, plugin, mock_ctx, mock_cmd):
await plugin.setup(mock_ctx, mock_cmd)
message = {"type": "test", "data": "value"}
await plugin.on_incoming_json(message)
# 验证逻辑最佳实践
1. 插件开发
- ✅ 使用
requires声明依赖,而非直接导入其他插件 - ✅ 在
shutdown()中清理资源和取消订阅 - ✅ 使用
self.cmd.spawn()派生异步任务 - ❌ 不要在插件间直接引用,使用 EventBus 通信
2. MCP 工具
- ✅ 提供清晰的
description,说明触发场景 - ✅ 返回有意义的结果字符串
- ✅ 捕获并处理异常,返回友好的错误消息
- ❌ 不要在工具中执行长时间阻塞操作
3. UI 开发
- ✅ 使用 Theme 变量保持视觉一致性
- ✅ Model 中只存储状态,View 中只处理显示
- ✅ 通过 EventBridge 进行 QML ↔ Python 通信
- ❌ 不要在 QML 中直接调用 Python 业务逻辑
4. 事件总线
- ✅ 使用预定义的 Events 常量
- ✅ 为复杂数据创建数据类
- ✅ 在组件销毁时取消订阅
- ❌ 不要创建循环事件触发
5. 异步编程
- ✅ 使用
async/await处理 I/O 操作 - ✅ 使用
asyncio.gather()并行执行 - ✅ 设置合理的超时时间
- ❌ 不要在异步函数中使用同步阻塞调用
参考资料
常见问题
Q: 插件加载顺序不对怎么办?
调整 priority 值,数值越小越优先。也可以使用 requires 声明依赖关系,PluginManager 会自动拓扑排序。
Q: MCP 工具没有被发现?
确保:
- 工具文件在
src/mcp/tools/目录下 - 目录包含
__init__.py文件 - 工具文件名为
_tools.py或在根目录下 - 使用了
@mcp_tool装饰器
Q: QML 中无法访问 Python 对象?
确保在 ViewManager._inject_context() 中正确注入:
ctx.setContextProperty("myModel", self._my_model)Q: 事件处理器没有被调用?
- 检查是否正确订阅:
event_bus.on(Events.xxx, handler) - 确保处理器是异步函数
- 检查是否在销毁前取消订阅了