第四篇：【插件篇】插件系统深度解析：设计哲学与实现机制

4.1 章节概述

在前面的文章中，我们已经了解了QwenPaw的整体架构和Agent核心实现。本章将深入剖析QwenPaw的插件系统设计。QwenPaw的插件系统是其扩展性的核心所在，也是近年来GitHub上讨论最热烈的话题之一。本章将系统解析插件机制的设计哲学、整体架构、核心模块源码、当前支持的插件类型、从零开发实践，以及未来的扩展方向。

通过阅读本章，您将掌握：

4.2 设计哲学：四大核心原则

QwenPaw的插件系统设计遵循四大核心原则：Duck Typing（鸭子类型）、Convention over Configuration（约定优于配置）、Isolation（隔离性）、Safety First（安全优先）。这四大原则相互配合，共同构成了QwenPaw插件系统的设计基础。

4.2.1 Duck Typing：灵活的对象识别

在传统的面向对象设计中，通常需要通过继承某个基类或实现某个接口来表明身份。但QwenPaw采用了Duck Typing的设计哲学——不要求插件必须继承严格的PluginBase抽象基类。只要一个对象实现了特定的方法（对于QwenPaw来说是register(api)方法），它就可以被视为合法插件。

这种设计的优势在于：

# 传统模式：需要继承基类
class OldStylePlugin(PluginBase):
    def register(self, api):
        pass

# QwenPaw模式：只需实现所需方法
class MyPlugin:
    def register(self, api: PluginApi):
        # 在这里注册你的能力
        pass

# 两者都可以被QwenPaw识别为有效插件
plugin = MyPlugin()

这种设计带来了极大的灵活性：

4.2.2 Convention over Configuration：约定优于配置

QwenPaw插件系统遵循"约定优于配置"的原则，通过目录结构和文件命名来自动识别和加载插件，而不是依赖复杂的配置文件：

// 插件目录结构
~/.qwenpaw/plugins/
├── my-provider/           # 插件目录
│   ├── plugin.json        # ✅ 包含plugin.json即被识别为插件
│   └── plugin.py          # ✅ 模块级导出的plugin对象即为入口
└── my-hook/
    ├── plugin.json        # ✅ 发现入口
    └── plugin.py          # ✅ 注册点

核心约定规则：

plugin.json示例：

{
  "id": "my-first-plugin",
  "name": "My First Plugin",
  "version": "0.1.0",
  "description": "A minimal QwenPaw plugin",
  "author": "Your Name",
  "entry_point": "plugin.py",
  "dependencies": [],
  "min_version": "0.1.0"
}

4.2.3 Isolation：模块隔离性

每个QwenPaw插件都被作为独立的Python模块加载，底层使用importlib.util.spec_from_file_location实现动态导入。这种设计带来了以下保障：

# 插件隔离加载的实现示意
def load_plugin(plugin_path: Path) -> PluginRecord:
    # 1. 创建模块规范
    spec = importlib.util.spec_from_file_location(
        f"qwenpaw_plugin_{plugin_path.name}",
        plugin_path / "plugin.py"
    )

    # 2. 创建模块
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)

    # 3. 设置包信息（支持相对导入）
    module.__package__ = f"qwenpaw_plugins.{plugin_path.name}"
    module.__path__ = [str(plugin_path)]

    # 4. 获取插件入口
    plugin = module.plugin

    return PluginRecord(
        manifest=manifest,
        module=module,
        instance=plugin
    )

隔离性的核心价值：

4.2.4 Safety First：安全第一

插件的安装和卸载必须在QwenPaw离线状态下进行，这是出于安全考虑的强制性要求：

class PluginManager:
    """插件管理器 - 安全优先原则"""

    async def install_plugin(self, plugin_path: Path) -> None:
        # 1. 检查QwenPaw是否处于离线状态
        if not self.is_qwenpaw_offline():
            raise SafetyException(
                "插件安装必须在QwenPaw离线状态下进行。"
                "请先停止所有运行中的QwenPaw实例。"
            )

        # 2. 安全扫描
        scanner = SecurityScanner()
        if scanner.has_prompt_injection(plugin_path):
            raise SafetyException("检测到提示词注入风险")
        if scanner.has_command_injection(plugin_path):
            raise SafetyException("检测到命令注入风险")
        if scanner.has_hardcoded_secrets(plugin_path):
            raise SafetyException("检测到硬编码密钥")

        # 3. 验证依赖
        manifest = self._load_manifest(plugin_path)
        self._verify_dependencies(manifest.dependencies)

        # 4. 执行安装
        await self._do_install(plugin_path, manifest)

    async def uninstall_plugin(self, plugin_id: str) -> None:
        # 同样需要离线状态检查
        if not self.is_qwenpaw_offline():
            raise SafetyException(
                "插件卸载必须在QwenPaw离线状态下进行。"
            )
        # 执行卸载逻辑...

离线操作的原因：

4.3 整体架构：分层设计

4.3.1 架构图解

QwenPaw插件系统的架构可以表示为以下层次：

┌─────────────────────────────────────────────────────────────┐
│                     QwenPaw Application                      │
│  ┌─────────────┐   ┌──────────────┐   ┌───────────────┐    │
│  │   FastAPI   │   │  AgentRunner  │   │  ProviderMgr │    │
│  │  (lifespan) │   │              │   │              │    │
│  └──────┬──────┘   └──────┬───────┘   └───────┬───────┘    │
│         │                  │                   │            │
│  ┌──────▼──────────────────▼───────────────────▼───────┐   │
│  │              Plugin System Layer                      │   │
│  │                                                        │   │
│  │   PluginLoader ──► PluginRegistry ◄── PluginApi       │   │
│  │        │                │    │    │                    │   │
│  │        │           Providers │  Hooks    Commands      │   │
│  │        │                │    │    │                    │   │
│  │  ┌────▼────┐    ┌───▼────▼────▼──┐  ┌───▼──────┐     │   │
│  │  │discover │    │RegistrationDB │  │PluginApi │     │   │
│  │  │& load   │    │ (Singleton)    │  │(per-plug)│     │   │
│  │  └─────────┘    └────────────────┘  └──────────┘     │   │
│  └───────────────────────────────────────────────────────┘   │
└─────────────────────────────────────────────────────────────┘

4.3.2 核心组件说明

4.3.3 关键数据流

QwenPaw插件系统的数据流遵循以下模式：

1. 启动发现 (Startup Discovery)
   PluginLoader.discover_plugins()
       └─ scan ~/.qwenpaw/plugins/ 目录
           └─ locate subfolders containing plugin.json
               └─ parse them into PluginManifest

2. 动态加载 (Dynamic Loading)
   PluginLoader.load_plugin(plugin_id)
       ├─ dynamically import module via importlib
       ├─ configure __package__ / __path__ for relative imports
       ├─ fetch module.plugin
       ├─ create PluginApi(plugin_id, config, manifest)
       ├─ invoke plugin.register(api)   # sync or async
       └─ return PluginRecord

3. 注册 (Registration)
   PluginApi.register_provider(...)
   PluginApi.register_startup_hook(...)
   PluginApi.register_shutdown_hook(...)
   PluginApi.register_control_command(...)
       └─ all registrations stored in PluginRegistry

4. 集成 (Integration)
   Application reads from PluginRegistry
       └─ injects providers into ProviderManager
       └─ registers hooks into lifecycle
       └─ registers commands into CLI

5. 执行 (Execution)
   Startup hooks execute in priority order (0 → 200)
       └─ Environment setup
       └─ SDK initialization
       └─ Telemetry startup

4.4 核心模块源码解析

4.4.1 PluginManifest - 插件清单

from dataclasses import dataclass, field
from typing import List, Dict, Any

@dataclass
class PluginManifest:
    """插件清单数据类"""

    id: str                                          # 插件唯一标识
    name: str                                        # 插件显示名称
    version: str                                     # 插件版本号
    description: str = ""                            # 插件描述
    author: str = ""                                 # 作者信息
    entry_point: str = "plugin.py"                   # 入口文件（默认plugin.py）
    dependencies: List[str] = field(default_factory=list)  # 依赖列表
    min_version: str = "0.1.0"                       # 最低QwenPaw版本要求
    meta: Dict[str, Any] = field(default_factory=dict)    # 元数据（扩展字段）

    def to_json(self) -> str:
        """序列化为JSON"""
        import json
        return json.dumps(asdict(self), indent=2)

    @classmethod
    def from_json(cls, json_str: str) -> 'PluginManifest':
        """从JSON反序列化"""
        import json
        data = json.loads(json_str)
        return cls(**data)

    def validate(self) -> List[str]:
        """验证清单完整性，返回错误列表"""
        errors = []
        if not self.id:
            errors.append("插件ID不能为空")
        if not self.name:
            errors.append("插件名称不能为空")
        if not self.version:
            errors.append("版本号不能为空")
        if not self._is_valid_version(self.version):
            errors.append(f"无效的版本号格式: {self.version}")
        return errors

    def _is_valid_version(self, version: str) -> bool:
        """验证版本号格式 (x.y.z)"""
        import re
        return bool(re.match(r'^\d+\.\d+\.\d+$', version))

4.4.2 PluginLoader - 发现与动态加载

class PluginLoader:
    """插件加载器 - 负责发现和动态加载插件"""

    def __init__(self, plugins_dir: Path):
        self.plugins_dir = plugins_dir
        self._loaded_plugins: Dict[str, PluginRecord] = {}

    def discover_plugins(self) -> List[PluginManifest]:
        """发现所有可用的插件"""
        discovered = []

        if not self.plugins_dir.exists():
            logger.warning(f"插件目录不存在: {self.plugins_dir}")
            return discovered

        for entry in self.plugins_dir.iterdir():
            if not entry.is_dir():
                continue

            plugin_json = entry / "plugin.json"
            if not plugin_json.exists():
                continue  # 没有plugin.json，跳过

            try:
                manifest = self._parse_manifest(plugin_json)
                # 检查版本兼容性
                if self._is_version_compatible(manifest.min_version):
                    discovered.append(manifest)
                    logger.info(f"发现插件: {manifest.id} v{manifest.version}")
            except Exception as e:
                logger.error(f"解析插件清单失败 {plugin_json}: {e}")

        return discovered

    async def load_plugin(
        self,
        manifest: PluginManifest,
        config: Dict[str, Any]
    ) -> PluginRecord:
        """加载单个插件"""

        plugin_path = self.plugins_dir / manifest.id
        plugin_py = plugin_path / manifest.entry_point

        if not plugin_py.exists():
            raise PluginLoadError(
                f"插件入口文件不存在: {plugin_py}"
            )

        # 动态导入模块
        spec = importlib.util.spec_from_file_location(
            f"qwenpaw_plugins.{manifest.id}",
            plugin_py
        )

        module = importlib.util.module_from_spec(spec)
        spec.loader.exec_module(module)

        # 设置包信息以支持相对导入
        module.__package__ = f"qwenpaw_plugins.{manifest.id}"
        module.__path__ = [str(plugin_path)]
        module.__file__ = str(plugin_py)

        # 获取插件入口
        if not hasattr(module, 'plugin'):
            raise PluginLoadError(
                f"插件模块缺少 'plugin' 属性: {manifest.id}"
            )

        plugin_instance = module.plugin

        # 创建插件API
        api = PluginApi(
            plugin_id=manifest.id,
            config=config.get(manifest.id, {}),
            manifest=manifest
        )

        # 调用注册方法（支持异步）
        if asyncio.iscoroutinefunction(plugin_instance.register):
            await plugin_instance.register(api)
        else:
            plugin_instance.register(api)

        # 创建并存储插件记录
        record = PluginRecord(
            manifest=manifest,
            module=module,
            instance=plugin_instance,
            api=api,
            loaded_at=datetime.now()
        )

        self._loaded_plugins[manifest.id] = record
        logger.info(f"插件加载成功: {manifest.id}")

        return record

    def _parse_manifest(self, json_path: Path) -> PluginManifest:
        """解析插件清单JSON"""
        with open(json_path, 'r', encoding='utf-8') as f:
            data = json.load(f)
        return PluginManifest(**data)

4.4.3 PluginRegistry - 中央注册表

PluginRegistry采用单例模式实现，是整个插件系统的中央协调器：

from threading import Lock

class PluginRegistry:
    """插件注册表 - 单例模式"""

    _instance = None
    _lock = Lock()

    def __new__(cls):
        if cls._instance is None:
            with cls._lock:
                if cls._instance is None:
                    cls._instance = super().__new__(cls)
                    cls._instance._initialized = False
        return cls._instance

    def __init__(self):
        if self._initialized:
            return

        # 存储四类注册信息
        self._providers: Dict[str, ProviderRegistration] = {}
        self._startup_hooks: List[HookRegistration] = []
        self._shutdown_hooks: List[HookRegistration] = []
        self._control_commands: List[ControlCommandRegistration] = []

        self._initialized = True

    # ============ Provider 注册 ============

    def register_provider(
        self,
        provider_id: str,
        provider_class: Type[ProviderBase],
        label: str,
        base_url: str,
        require_api_key: bool = True,
        config_schema: Optional[Dict] = None
    ) -> None:
        """注册模型提供商"""
        registration = ProviderRegistration(
            id=provider_id,
            provider_class=provider_class,
            label=label,
            base_url=base_url,
            require_api_key=require_api_key,
            config_schema=config_schema or {}
        )
        self._providers[provider_id] = registration
        logger.info(f"提供商注册: {provider_id}")

    def get_provider(self, provider_id: str) -> Optional[ProviderRegistration]:
        """获取提供商注册信息"""
        return self._providers.get(provider_id)

    def list_providers(self) -> List[ProviderRegistration]:
        """列出所有已注册的提供商"""
        return list(self._providers.values())

    # ============ Hook 注册 ============

    def register_startup_hook(
        self,
        hook_name: str,
        callback: Callable,
        priority: int = 100,
        description: str = ""
    ) -> None:
        """注册启动钩子"""
        registration = HookRegistration(
            name=hook_name,
            callback=callback,
            priority=priority,
            description=description,
            hook_type="startup"
        )
        self._startup_hooks.append(registration)
        # 按优先级排序（数字越小越先执行）
        self._startup_hooks.sort(key=lambda h: h.priority)

    def register_shutdown_hook(
        self,
        hook_name: str,
        callback: Callable,
        priority: int = 100,
        description: str = ""
    ) -> None:
        """注册关闭钩子"""
        registration = HookRegistration(
            name=hook_name,
            callback=callback,
            priority=priority,
            description=description,
            hook_type="shutdown"
        )
        self._shutdown_hooks.append(registration)
        self._shutdown_hooks.sort(key=lambda h: h.priority)

    async def execute_startup_hooks(self) -> None:
        """执行所有启动钩子"""
        for hook in self._startup_hooks:
            try:
                logger.info(f"执行启动钩子: {hook.name}")
                if asyncio.iscoroutinefunction(hook.callback):
                    await hook.callback()
                else:
                    hook.callback()
            except Exception as e:
                logger.error(f"启动钩子执行失败 {hook.name}: {e}")

    async def execute_shutdown_hooks(self) -> None:
        """执行所有关闭钩子"""
        for hook in self._shutdown_hooks:
            try:
                logger.info(f"执行关闭钩子: {hook.name}")
                if asyncio.iscoroutinefunction(hook.callback):
                    await hook.callback()
                else:
                    hook.callback()
            except Exception as e:
                logger.error(f"关闭钩子执行失败 {hook.name}: {e}")

    # ============ Command 注册 ============

    def register_control_command(
        self,
        command_name: str,
        handler: Type[BaseControlCommandHandler],
        description: str = "",
        aliases: List[str] = None
    ) -> None:
        """注册控制命令"""
        registration = ControlCommandRegistration(
            name=command_name,
            handler=handler,
            description=description,
            aliases=aliases or []
        )
        self._control_commands.append(registration)

    def get_command_handler(self, command_name: str) -> Optional[Type]:
        """获取命令处理器"""
        for cmd in self._control_commands:
            if cmd.name == command_name or command_name in cmd.aliases:
                return cmd.handler
        return None

4.4.4 PluginApi - 开发者接口

PluginApi是插件开发者与QwenPaw交互的主要接口，每个插件在注册时都会获得一个独享的PluginApi实例：

@dataclass
class PluginApi:
    """插件API - 插件开发者的接口"""

    plugin_id: str
    config: Dict[str, Any]
    manifest: PluginManifest

    def register_provider(
        self,
        provider_id: str,
        provider_class: Type[ProviderBase],
        label: str,
        base_url: str,
        require_api_key: bool = True,
        config_schema: Optional[Dict] = None
    ) -> None:
        """注册自定义模型提供商

        适用场景：
        - 企业内部模型服务
        - 私有化部署模型
        - 自研推理服务

        Example:
            api.register_provider(
                provider_id="my-llm",
                provider_class=MyLLMProvider,
                label="My LLM Service",
                base_url="https://api.internal.company.com/v1",
                require_api_key=True,
            )
        """
        registry = PluginRegistry()
        registry.register_provider(
            provider_id=provider_id,
            provider_class=provider_class,
            label=label,
            base_url=base_url,
            require_api_key=require_api_key,
            config_schema=config_schema
        )

    def register_startup_hook(
        self,
        hook_name: str,
        callback: Callable,
        priority: int = 100,
        description: str = ""
    ) -> None:
        """注册启动钩子

        适用场景：
        - 监控初始化
        - 环境检查
        - 第三方SDK启动
        - 日志与上报集成

        优先级说明：
        - priority=0   → 最早执行
        - priority=100 → 默认值
        - priority=200 → 最晚执行
        """
        registry = PluginRegistry()
        registry.register_startup_hook(
            hook_name=hook_name,
            callback=callback,
            priority=priority,
            description=description
        )

    def register_shutdown_hook(
        self,
        hook_name: str,
        callback: Callable,
        priority: int = 100,
        description: str = ""
    ) -> None:
        """注册关闭钩子"""
        registry = PluginRegistry()
        registry.register_shutdown_hook(
            hook_name=hook_name,
            callback=callback,
            priority=priority,
            description=description
        )

    def register_control_command(
        self,
        command_name: str,
        handler: Type[BaseControlCommandHandler],
        description: str = "",
        aliases: Optional[List[str]] = None
    ) -> None:
        """注册斜杠命令

        两种实现路径：
        - Path A: BaseControlCommandHandler（推荐）
        - Path B: Monkey-patch查询重写
        """
        registry = PluginRegistry()
        registry.register_control_command(
            command_name=command_name,
            handler=handler,
            description=description,
            aliases=aliases
        )

    @property
    def runtime(self) -> 'RuntimeHelpers':
        """访问运行时辅助能力"""
        return RuntimeHelpers(plugin_id=self.plugin_id)


@dataclass
class RuntimeHelpers:
    """运行时辅助能力"""

    plugin_id: str

    def get_provider(self, provider_id: str) -> Optional[ProviderBase]:
        """获取某个Provider实例"""
        registry = PluginRegistry()
        reg = registry.get_provider(provider_id)
        if reg:
            return reg.provider_class()
        return None

    def list_providers(self) -> List[str]:
        """列出所有可用Provider ID"""
        registry = PluginRegistry()
        return [p.id for p in registry.list_providers()]

    def log_info(self, message: str) -> None:
        """统一日志接口 - INFO级别"""
        logger.info(f"[{self.plugin_id}] {message}")

    def log_error(self, message: str, exc: Optional[Exception] = None) -> None:
        """统一日志接口 - ERROR级别"""
        if exc:
            logger.error(f"[{self.plugin_id}] {message}", exc_info=exc)
        else:
            logger.error(f"[{self.plugin_id}] {message}")

    def log_debug(self, message: str) -> None:
        """统一日志接口 - DEBUG级别"""
        logger.debug(f"[{self.plugin_id}] {message}")

4.5 当前三大插件类型

4.5.1 Provider Plugins - 提供商插件

Provider插件用于注册自定义的大语言模型提供商：

# my-provider/plugin.py

from qwenpaw.plugins.api import PluginApi
from qwenpaw.providers.base import ProviderBase
from my_provider.chat_model import MyChatModel

class MyLLMProvider(ProviderBase):
    """自定义LLM提供商"""

    def __init__(self, config: Dict):
        self.api_key = config.get('api_key')
        self.base_url = config.get('base_url', 'https://api.example.com')

    async def create_chat_model(self, model_config: ModelConfig) -> ChatModelBase:
        return MyChatModel(
            api_key=self.api_key,
            base_url=self.base_url,
            model_name=model_config.model_name,
            temperature=model_config.temperature,
        )

    def get_available_models(self) -> List[str]:
        return ["my-model-v1", "my-model-v2", "my-model-gpt"]

def register(api: PluginApi):
    api.register_provider(
        provider_id="my-llm",
        provider_class=MyLLMProvider,
        label="My LLM Service",
        base_url="https://api.internal.company.com/v1",
        require_api_key=True,
        config_schema={
            "api_key": {"type": "string", "required": True},
            "base_url": {"type": "string", "required": False},
        }
    )

# 模块级入口
plugin = type('Plugin', (), {'register': register})()

4.5.2 Hook Plugins - 钩子插件

Hook插件用于在应用启动或关闭时执行自定义逻辑：

# my-hook/plugin.py

from qwenpaw.plugins.api import PluginApi
import logging

logger = logging.getLogger(__name__)

class MyHookPlugin:
    def register(self, api: PluginApi):
        logger.info(f"注册插件: {api.plugin_id}")

        # 注册启动钩子
        api.register_startup_hook(
            hook_name="my_init",
            callback=self._on_startup,
            priority=100,
            description="初始化自定义服务"
        )

        # 注册关闭钩子
        api.register_shutdown_hook(
            hook_name="my_cleanup",
            callback=self._on_shutdown,
            priority=100,
            description="清理自定义资源"
        )

    async def _on_startup(self) -> None:
        """启动时的初始化逻辑"""
        logger.info("执行启动初始化...")

        # 初始化自定义服务
        await self._init_my_service()

        # 连接外部系统
        await self._connect_external_systems()

        logger.info("启动初始化完成")

    async def _on_shutdown(self) -> None:
        """关闭时的清理逻辑"""
        logger.info("执行关闭清理...")

        # 断开外部连接
        await self._disconnect_external_systems()

        # 保存状态
        await self._save_state()

        logger.info("关闭清理完成")

    async def _init_my_service(self) -> None:
        pass  # 实现自定义初始化逻辑

    async def _connect_external_systems(self) -> None:
        pass  # 实现外部系统连接

    async def _disconnect_external_systems(self) -> None:
        pass  # 实现外部系统断开

    async def _save_state(self) -> None:
        pass  # 实现状态保存

plugin = MyHookPlugin()

4.5.3 Command Plugins - 命令插件

Command插件用于注册斜杠命令（slash commands）：

# my-command/plugin.py

from qwenpaw.plugins.api import PluginApi
from qwenpaw.agents.command_handler import BaseControlCommandHandler

class MyCommandHandler(BaseControlCommandHandler):
    """自定义命令处理器"""

    command_name = "mycommand"
    description = "执行自定义命令"

    async def handle(self, args: str, context: CommandContext) -> str:
        """处理命令"""
        # 解析参数
        params = self._parse_args(args)

        #执行业务逻辑
        result = await self._do_something(params)

        return result

    def _parse_args(self, args: str) -> Dict:
        """解析命令参数"""
        parts = args.split()
        return {
            'action': parts[0] if parts else None,
            'params': parts[1:]
        }

    async def _do_something(self, params: Dict) -> str:
        """执行业务逻辑"""
        return f"执行完成: {params}"

def register(api: PluginApi):
    api.register_control_command(
        command_name="mycommand",
        handler=MyCommandHandler,
        description="我的自定义命令",
        aliases=["mc", "my"]
    )

plugin = type('Plugin', (), {'register': register})()

4.6 动手实践：从零开发一个插件

4.6.1 创建最小可行插件

让我们创建一个最简单的"Hello World"插件：

目录结构：

~/.qwenpaw/plugins/
└── my-first-plugin/
    ├── plugin.json
    └── plugin.py

plugin.json：

{
  "id": "my-first-plugin",
  "name": "My First Plugin",
  "version": "0.1.0",
  "description": "A minimal QwenPaw plugin demonstrating the basics",
  "author": "Your Name",
  "entry_point": "plugin.py",
  "dependencies": [],
  "min_version": "1.0.0"
}

plugin.py：

"""
QwenPaw插件示例：Hello World

这个插件演示了QwenPaw插件的基本结构。
"""

from qwenpaw.plugins.api import PluginApi
import logging

logger = logging.getLogger(__name__)

class MyFirstPlugin:
    """我的第一个QwenPaw插件"""

    def register(self, api: PluginApi) -> None:
        """插件注册方法

        当QwenPaw加载此插件时会调用此方法。
        在这里注册插件的各种能力。
        """
        logger.info(f"Hello from {api.plugin_id}!")

        # 注册一个启动钩子
        api.register_startup_hook(
            hook_name="hello_world",
            callback=self._on_startup,
            priority=100,
            description="打印启动消息"
        )

        # 注册一个关闭钩子
        api.register_shutdown_hook(
            hook_name="goodbye_world",
            callback=self._on_shutdown,
            priority=100,
            description="打印关闭消息"
        )

        # 访问配置
        custom_setting = api.config.get('custom_setting', 'default_value')
        logger.info(f"Custom setting: {custom_setting}")

    def _on_startup(self) -> None:
        """启动时的回调"""
        logger.info("🎉 QwenPaw已启动！我的插件正在运行。")

    def _on_shutdown(self) -> None:
        """关闭时的回调"""
        logger.info("👋 QwenPaw正在关闭。我的插件完成工作。")


# 模块级入口对象 - QwenPaw会查找这个
plugin = MyFirstPlugin()

4.6.2 配置并使用插件

在~/.qwenpaw/config.json中配置插件：

{
  "plugins": {
    "my-first-plugin": {
      "custom_setting": "custom_value"
    }
  }
}

4.6.3 异步注册支持

QwenPaw的插件支持异步注册，这对于需要执行异步初始化操作的插件非常有用：

class AsyncPlugin:
    """异步注册示例"""

    async def register(self, api: PluginApi) -> None:
        """异步注册方法

        如果插件的注册逻辑需要执行异步操作，
        可以将register方法定义为async。
        """
        # 异步获取远程配置
        remote_config = await self._fetch_remote_config()

        # 根据远程配置注册提供商
        api.register_provider(
            provider_id="my-async-provider",
            provider_class=MyAsyncProvider,
            label="Async Provider",
            base_url=remote_config['base_url'],
        )

        # 注册启动钩子
        api.register_startup_hook(
            hook_name="async_init",
            callback=self._async_startup,
            priority=50,  # 较高优先级，提前执行
        )

    async def _fetch_remote_config(self) -> Dict:
        """模拟从远程获取配置"""
        import asyncio
        await asyncio.sleep(0.1)  # 模拟网络延迟
        return {
            'base_url': 'https://api.async-provider.example.com'
        }

    async def _async_startup(self) -> None:
        """异步启动逻辑"""
        import asyncio
        await asyncio.sleep(0.1)  # 模拟初始化
        logging.getLogger(__name__).info("异步启动完成")

plugin = AsyncPlugin()

4.6.4 完整的Provider插件示例

以下是创建一个完整的自定义模型提供商的完整示例：

# custom-provider/plugin.py

import logging
from typing import List, Dict, Any
from qwenpaw.plugins.api import PluginApi
from qwenpaw.providers.base import ProviderBase
from qwenpaw.agents.model import ChatModelBase

logger = logging.getLogger(__name__)

# ============ 自定义模型类 ============

class CustomChatModel(ChatModelBase):
    """自定义聊天模型实现"""

    def __init__(
        self,
        api_key: str,
        base_url: str,
        model_name: str,
        temperature: float = 0.7
    ):
        self.api_key = api_key
        self.base_url = base_url
        self.model_name = model_name
        self.temperature = temperature

    async def chat(
        self,
        messages: List[Dict],
        **kwargs
    ) -> Dict:
        """调用自定义API"""
        import aiohttp

        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }

        payload = {
            "model": self.model_name,
            "messages": messages,
            "temperature": kwargs.get('temperature', self.temperature)
        }

        async with aiohttp.ClientSession() as session:
            async with session.post(
                f"{self.base_url}/chat",
                json=payload,
                headers=headers
            ) as response:
                if response.status != 200:
                    error = await response.text()
                    raise Exception(f"API调用失败: {error}")

                return await response.json()

# ============ 自定义提供商类 ============

class CustomLLMProvider(ProviderBase):
    """自定义LLM提供商"""

    def __init__(self, config: Dict[str, Any]):
        self.api_key = config.get('api_key')
        self.base_url = config.get('base_url', 'https://api.example.com/v1')

    async def create_chat_model(
        self,
        model_config: 'ModelConfig'
    ) -> ChatModelBase:
        """创建聊天模型实例"""
        return CustomChatModel(
            api_key=self.api_key,
            base_url=self.base_url,
            model_name=model_config.model_name,
            temperature=model_config.temperature
        )

    def get_available_models(self) -> List[str]:
        """获取可用模型列表"""
        return [
            "custom-model-v1",
            "custom-model-v2",
            "custom-model-instruct"
        ]

# ============ 插件入口 ============

class CustomProviderPlugin:
    """自定义提供商插件"""

    def register(self, api: PluginApi) -> None:
        """注册插件"""
        logger.info("注册自定义模型提供商...")

        # 注册提供商
        api.register_provider(
            provider_id="custom-llm",
            provider_class=CustomLLMProvider,
            label="Custom LLM",
            base_url="https://api.example.com/v1",
            require_api_key=True,
            config_schema={
                "api_key": {
                    "type": "string",
                    "required": True,
                    "description": "API密钥"
                },
                "base_url": {
                    "type": "string",
                    "required": False,
                    "description": "API基础URL"
                }
            }
        )

        # 注册启动钩子（用于日志记录）
        api.register_startup_hook(
            hook_name="log_provider_init",
            callback=self._log_init,
            priority=100
        )

    def _log_init(self) -> None:
        """记录初始化"""
        logger.info("Custom LLM Provider已初始化")


plugin = CustomProviderPlugin()

4.7 未来插件点规划

QwenPaw的插件系统仍在不断演进。根据社区讨论，未来将支持更多扩展点：

4.7.1 路线图概览

4.7.2 Tool Plugins - 工具插件

允许插件直接向Agent注册工具，打通插件系统与Agent工具生态之间的壁垒：

# 未来API设计（预览）
class FuturePlugin:
    def register(self, api: PluginApi) -> None:
        # 注册自定义工具
        api.register_tool(
            tool_id="my_custom_tool",
            tool_class=MyCustomTool,
            description="执行自定义操作",
            parameters=[
                {"name": "input", "type": "string", "required": True}
            ]
        )

4.7.3 Middleware / Pipeline Hooks - 中间件与处理链

提供请求/响应处理链的钩子：

# 未来API设计（预览）
class MiddlewarePlugin:
    def register(self, api: PluginApi) -> None:
        # 注册前置钩子
        api.register_middleware_hook(
            stage="pre_query",
            callback=self._preprocess_query,
            priority=100
        )

        # 注册后置钩子
        api.register_middleware_hook(
            stage="post_response",
            callback=self._postprocess_response,
            priority=100
        )

# 钩子阶段说明
MIDDLEWARE_STAGES = {
    "pre_query": "用户输入到达Agent之前",
    "post_query": "Agent生成输出之后",
    "pre_tool_call": "工具执行之前",
    "post_tool_call": "工具执行之后",
    "on_error": "发生错误时",
    "on_memory_update": "记忆/上下文更新时"
}

4.7.4 Event Bus - 事件总线

统一的事件发布/订阅机制，让插件之间产生更自然的协作：

# 未来API设计（预览）
class EventBusPlugin:
    def register(self, api: PluginApi) -> None:
        # 订阅事件
        api.subscribe_event(
            event_type="tool_executed",
            handler=self._on_tool_executed
        )

        # 发布事件
        api.publish_event(
            event_type="custom_event",
            data={"key": "value"}
        )

4.7.5 UI Extensions - UI扩展点

允许插件向Web UI注入扩展能力：

# 未来API设计（预览）
class UIExtensionPlugin:
    def register(self, api: PluginApi) -> None:
        api.register_ui_extension(
            location="sidebar",
            component=MySidebarComponent,
            title="My Extension"
        )

4.8 本章小结

本章深入分析了QwenPaw的插件系统设计，主要涵盖以下内容：

核心要点回顾

三大插件类型

开发实践要点

源码位置索引

预告：下一篇文章将进入【安全篇】，我们将详细分析QwenPaw的多层安全防护体系，包括工具守卫、文件访问控制、注入防护等核心机制。敬请期待！

往期回顾：

参考文献：