Skip to content

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   │  │  │
│  │  └────────┘ └────────┘ └────────┘ └────────────┘  │  │
│  └───────────────────────────────────────────────────┘  │
└─────────────────────────────────────────────────────────┘

快速开始

环境准备

bash
# 克隆项目
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

运行应用

bash
# 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

python
# 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 参数:

python
from src.utils.activation_announcer import announce_activation_code

# 使用英语播报
announce_activation_code(code, locale="en-US")

# 使用日语播报
announce_activation_code(code, locale="ja-JP")

替换音频文件

  1. 准备新的 OGG 格式音频文件(建议采样率 24kHz)
  2. 替换对应语言目录下的文件
  3. 文件名必须保持一致(0.ogg ~ 9.ogg, activation.ogg

添加新语言

  1. assets/sounds/ 下创建新的语言目录(如 my-lang/
  2. 添加 11 个必需的 OGG 音频文件
  3. 调用时指定新的 locale:announce_activation_code(code, locale="my-lang")

自定义 GPIO 引脚

GPIO 模式仅支持 Linux 系统(树莓派),用于通过物理按键控制设备。

推荐硬件:四按钮独立按键模块

四按钮独立按键模块

按钮模块原理图

硬件接线说明

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

树莓派5引脚图

电源接线:

模块引脚树莓派引脚说明
V3.3V(Pin 1 或 Pin 17)电源正极
GGND(Pin 6/9/14/20/25/30/34/39)电源负极,推荐 Pin 6

⚠️ 注意:不要接 5V(Pin 2/4),会损坏模块!

按键接线:

模块引脚GPIO(BCM)物理引脚功能
1GPIO 17Pin 11开始/停止对话
2GPIO 27Pin 13中断当前语音
3GPIO 22Pin 15切换手动/自动模式
4GPIO 23Pin 16退出程序

模块特性:按下输出低电平,代码使用 BCM 编号:17/27/22/23

修改引脚配置

编辑 src/ui/gpio/input.py 中的 DEFAULT_PINS

python
# 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]

依赖安装

bash
# 安装 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 模式

bash
python main.py --mode gpio --protocol websocket

购买按钮模块

按钮模块产品二维码

插件开发

插件是扩展 py-xiaozhi 功能的核心方式。通过继承 Plugin 基类,你可以:

  • 响应系统事件(音频数据、JSON 消息、状态变更等)
  • 调用核心命令(发送音频、启动监听等)
  • 与其他插件交互

创建插件

src/plugins/ 目录下创建新文件:

python
# 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 中注册插件:

python
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/ 目录下创建工具:

python
# 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        # 工具定义文件(会被自动发现)

方式二:手动注册

python
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 会根据描述决定何时调用工具:

python
@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 文件

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

python
# 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:
        """刷新数据"""
        # 实现刷新逻辑
        pass

3. 扩展 EventBridge

python
# 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 中注册

python
# 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 是组件间解耦通信的核心机制。

基本用法

python
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)

预定义事件

python
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"

添加自定义事件

python
# src/core/event_bus.py

class Events:
    # ... 现有事件 ...
    
    # 添加自定义事件
    MY_CUSTOM_EVENT = "my_custom_event"
    MY_DATA_UPDATED = "my_data_updated"

事件数据类

建议为复杂事件数据创建数据类:

python
# 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 管理应用配置。

读取配置

python
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")

写入配置

python
# 设置配置项
config.set_config("SECTION.KEY", value)

# 保存到文件
config.save()

配置热重载

当配置变更需要通知其他组件时:

python
from src.core.event_bus import Events

# 保存配置后触发事件
config.save()
await self.ctx.event_bus.emit(Events.CONFIG_CHANGED, {
    "key": "SECTION.KEY",
    "value": new_value,
})

调试与测试

调试模式启动

bash
# 跳过激活流程
python main.py --mode gui --skip-activation

# CLI 模式(更容易看日志)
python main.py --mode cli --skip-activation

日志系统

python
from src.logging import get_logger

logger = get_logger()

logger.debug("调试信息")
logger.info("一般信息")
logger.warning("警告信息")
logger.error("错误信息", exc_info=True)

单元测试

python
# 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 工具没有被发现?

确保:

  1. 工具文件在 src/mcp/tools/ 目录下
  2. 目录包含 __init__.py 文件
  3. 工具文件名为 _tools.py 或在根目录下
  4. 使用了 @mcp_tool 装饰器

Q: QML 中无法访问 Python 对象?

确保在 ViewManager._inject_context() 中正确注入:

python
ctx.setContextProperty("myModel", self._my_model)

Q: 事件处理器没有被调用?

  1. 检查是否正确订阅:event_bus.on(Events.xxx, handler)
  2. 确保处理器是异步函数
  3. 检查是否在销毁前取消订阅了