插件化 Agent 可扩展架构设计

from abc import ABC, abstractmethod
from typing import Dict, List, Type, Any
from enum import Enum

class PluginState(Enum):
    INSTALLED = "installed"
    ACTIVE = "active"
    INACTIVE = "inactive"
    ERROR = "error"

class IPlugin(ABC):
    """插件接口"""
    
    @property
    @abstractmethod
    def id(self) -> str:
        pass
    
    @property
    @abstractmethod
    def name(self) -> str:
        pass
    
    @property
    @abstractmethod
    def version(self) -> str:
        pass
    
    @abstractmethod
    def activate(self):
        """激活插件"""
        pass
    
    @abstractmethod
    def deactivate(self):
        """停用插件"""
        pass

class ExtensionPoint(ABC):
    """扩展点基类"""
    
    def __init__(self, name: str):
        self.name = name
        self.extensions: List[Any] = []
    
    def register(self, extension: Any):
        """注册扩展"""
        self.extensions.append(extension)
    
    def unregister(self, extension: Any):
        """注销扩展"""
        self.extensions.remove(extension)
    
    def get_extensions(self) -> List[Any]:
        """获取所有扩展"""
        return self.extensions.copy()

class MicroKernel:
    """微内核实现"""
    
    def __init__(self):
        self.extension_points: Dict[str, ExtensionPoint] = {}
        self.plugins: Dict[str, IPlugin] = {}
        self.services: Dict[str, Any] = {}
    
    def register_extension_point(self, point: ExtensionPoint):
        """注册扩展点"""
        self.extension_points[point.name] = point
    
    def get_extension_point(self, name: str) -> ExtensionPoint:
        """获取扩展点"""
        if name not in self.extension_points:
            raise ValueError(f"Extension point {name} not found")
        return self.extension_points[name]
    
    def register_plugin(self, plugin: IPlugin):
        """注册插件"""
        self.plugins[plugin.id] = plugin
    
    def activate_plugin(self, plugin_id: str):
        """激活插件"""
        plugin = self.plugins.get(plugin_id)
        if not plugin:
            raise ValueError(f"Plugin {plugin_id} not found")
        plugin.activate()
    
    def deactivate_plugin(self, plugin_id: str):
        """停用插件"""
        plugin = self.plugins.get(plugin_id)
        if not plugin:
            raise ValueError(f"Plugin {plugin_id} not found")
        plugin.deactivate()
    
    def register_service(self, name: str, service: Any):
        """注册服务"""
        self.services[name] = service
    
    def get_service(self, name: str) -> Any:
        """获取服务"""
        return self.services.get(name)

# 使用示例
kernel = MicroKernel()

# 注册扩展点
tool_extension_point = ExtensionPoint("tools")
kernel.register_extension_point(tool_extension_point)

# 注册插件
class SearchPlugin(IPlugin):
    @property
    def id(self): return "search-plugin"
    @property
    def name(self): return "Search Plugin"
    @property
    def version(self): return "1.0.0"
    
    def activate(self):
        print("Search plugin activated")
    
    def deactivate(self):
        print("Search plugin deactivated")

plugin = SearchPlugin()
kernel.register_plugin(plugin)
kernel.activate_plugin("search-plugin")
                        

┌─────────────────────────────────────────────────────────────┐
│                  插件生命周期状态机                          │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  ┌──────────┐                                              │
│  │  CREATED │ (创建)                                       │
│  └────┬─────┘                                              │
│       │ install()                                          │
│       ↓                                                    │
│  ┌──────────┐                                              │
│  │INSTALLED │ (已安装)                                     │
│  └────┬─────┘                                              │
│       │ activate()                                         │
│       ↓                                                    │
│  ┌──────────┐    deactivate()    ┌──────────┐             │
│  │  ACTIVE  │ ────────────────→  │ INACTIVE │             │
│  │ (已激活) │                    │ (已停用) │             │
│  └────┬─────┘ ←────────────────  └────┬─────┘             │
│       │ activate()                    │                    │
│       │                               │ uninstall()        │
│       │                               ↓                    │
│       │                        ┌──────────┐                │
│       │                        │  ERROR   │                │
│       │                        │  (错误)  │                │
│       │                        └──────────┘                │
│       │                                                    │
│       ↓ uninstall()                                        │
│  ┌──────────┐                                              │
│  │UNINSTALLED│ (已卸载)                                    │
│  └──────────┘                                              │
│                                                             │
│  状态转换：                                                 │
│    CREATED → INSTALLED: install()                          │
│    INSTALLED → ACTIVE: activate()                          │
│    ACTIVE → INACTIVE: deactivate()                         │
│    INACTIVE → ACTIVE: activate()                           │
│    ANY → ERROR: 发生异常                                    │
│    ANY → UNINSTALLED: uninstall()                          │
│                                                             │
└─────────────────────────────────────────────────────────────┘
                        

my-plugin/
├── plugin.json              # 插件描述符
├── README.md                # 插件文档
├── LICENSE                  # 许可证
├── src/                     # 源代码
│   ├── __init__.py
│   ├── main.py              # 插件入口
│   ├── tools/               # 工具实现
│   │   ├── __init__.py
│   │   └── search_tool.py
│   └── utils/               # 工具函数
│       └── helpers.py
├── tests/                   # 测试代码
│   ├── __init__.py
│   └── test_search.py
├── assets/                  # 资源文件
│   └── icon.png
├── requirements.txt         # Python 依赖
└── setup.py                 # 打包配置

# plugin.json 示例
{
  "id": "my-search-plugin",
  "name": "My Search Plugin",
  "version": "1.0.0",
  "description": "A powerful search plugin",
  "author": "Your Name",
  "license": "MIT",
  "engines": {
    "agent-core": ">=2.0.0"
  },
  "extensionPoints": [
    {
      "type": "tool",
      "name": "search",
      "entry": "src/main.py:SearchTool"
    }
  ],
  "permissions": [
    "network:search",
    "storage:read"
  ],
  "dependencies": [
    "requests>=2.28.0"
  ]
}
                        

# 创建新插件项目
$ agent-plugin create my-plugin
✓ Created plugin project: my-plugin
✓ Initialized git repository
✓ Installed dependencies

# 开发模式运行
$ cd my-plugin
$ agent-plugin dev
🔌 Starting plugin development server...
✓ Plugin loaded: my-search-plugin@1.0.0
✓ Watching for file changes...
✓ Hot reload enabled

# 运行测试
$ agent-plugin test
Running tests...
✓ test_search_basic
✓ test_search_advanced
✓ test_search_error_handling
3 passed in 0.52s

# 打包插件
$ agent-plugin package
✓ Building plugin...
✓ Created my-search-plugin-1.0.0.zip

# 发布到市场
$ agent-plugin publish
✓ Uploading to marketplace...
✓ Plugin published: my-search-plugin@1.0.0
✓ URL: https://marketplace.example.com/plugins/my-search-plugin
                        

# 权限定义
permissions = {
    "network": {
        "search": "允许网络搜索",
        "api_call": "允许调用外部 API",
        "websocket": "允许 WebSocket 连接"
    },
    "storage": {
        "read": "允许读取存储",
        "write": "允许写入存储",
        "delete": "允许删除存储"
    },
    "system": {
        "execute": "允许执行系统命令",
        "file_access": "允许访问文件系统",
        "env_read": "允许读取环境变量"
    },
    "agent": {
        "tool_call": "允许调用其他工具",
        "memory_read": "允许读取记忆",
        "memory_write": "允许写入记忆"
    }
}

# 插件权限声明
{
  "id": "my-plugin",
  "permissions": [
    "network:search",
    "storage:read",
    "agent:tool_call"
  ]
}

# 权限检查
class PermissionChecker:
    def check(self, plugin_id: str, permission: str) -> bool:
        plugin = self.get_plugin(plugin_id)
        return permission in plugin.permissions
    
    def enforce(self, plugin_id: str, permission: str):
        if not self.check(plugin_id, permission):
            raise PermissionError(
                f"Plugin {plugin_id} lacks permission {permission}"
            )

# 使用示例
checker = PermissionChecker()
try:
    checker.enforce("my-plugin", "network:search")
    # 执行需要权限的操作
except PermissionError as e:
    print(f"Permission denied: {e}")