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 工具系统
│   │   ├── mcp_server.py   # MCP 服务器
│   │   ├── tooling.py      # McpTool / Property
│   │   └── tools/          # 工具实现(register_* 显式注册)
│   ├── 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 模型交互的工具协议。内置工具通过 register_*_tools(add_tool, …) 显式注册,由 McpServer.add_common_tools / McpPlugin.setup 显式挂载;不再使用 @mcp_tool 装饰器与目录自动发现。

完整步骤与示例见 MCP 工具开发指南。

外挂插件(用户目录、自带依赖)见 MCP 外挂扩展。

创建 MCP 工具(摘要) ​

在 src/mcp/tools/<name>/ 中实现:

python
# src/mcp/tools/my_tools/register.py

from collections.abc import Callable
from typing import Any

from src.logging import get_logger
from src.mcp.tooling import McpTool, Property, PropertyList, PropertyType

logger = get_logger()


def register_my_tools(add_tool: Callable[[McpTool], None]) -> None:
    async def greet(args: dict[str, Any]) -> str:
        name = args.get("name", "朋友")
        return f"你好,{name}!很高兴认识你!"

    add_tool(
        McpTool(
            "my_tools.greet",
            "向用户打招呼。当用户说'你好'、'Hi'时调用。",
            PropertyList([Property("name", PropertyType.STRING)]),
            greet,
        )
    )
    logger.info("已注册 my_tools")
python
# src/mcp/tools/my_tools/__init__.py
from .register import register_my_tools
__all__ = ["register_my_tools"]

在 mcp_server.add_common_tools(或 McpPlugin.setup)中调用:

python
from src.mcp.tools.my_tools import register_my_tools
register_my_tools(self.add_tool)

需要 MusicPlayer / 摄像头等运行时对象时,由容器创建后作为参数传入 register_*,闭包持有实例,不要写模块级单例。

属性类型 ​

PropertyType说明
STRING字符串
INTEGER整数,可选 min_value / max_value
BOOLEAN布尔,可选 default_value

工具描述最佳实践 ​

description 应包含:简短说明、触发话术、行为与注意点,便于模型选择调用时机。


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"
    
    # GUI 窗口
    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

from src.utils.config_manager import get_config

config = get_config()  # 入口须先 initialize_config()

# 读取配置项
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. 包导出 register_*_tools(实现于 register.py),并在 add_common_tools / McpPlugin.setup 中显式调用
  2. 目录含 __init__.py,且 from .register import register_*_tools
  3. 工具名全局唯一;重启应用后查看日志中的「已注册 … MCP 工具」
  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. 检查是否在销毁前取消订阅了