react_plugin_loader.py

"""
ReAct Agent 插件加载器
支持动态加载外部工具，无需修改核心代码
"""

import os
import json
import importlib.util
from pathlib import Path
from typing import Dict, Any, List, Callable
import traceback


class ToolPlugin:
    """工具插件基类"""
    
    def __init__(self):
        self.name = ""
        self.description = ""
        self.parameters = {}
        self.func = None
    
    def execute(self, **kwargs):
        """执行工具函数"""
        if self.func:
            return self.func(**kwargs)
        raise NotImplementedError("工具函数未实现")


class PluginLoader:
    """插件加载器"""
    
    def __init__(self, plugin_dir: str = "react_plugins"):
        self.plugin_dir = Path(plugin_dir)
        self.plugins: Dict[str, ToolPlugin] = {}
        self.plugin_dir.mkdir(exist_ok=True)
    

    def load_plugins(self) -> int:
        """
        加载所有插件
        
        Returns:
            int: 成功加载的插件数量
        """
        loaded_count = 0
        
        # 遍历插件目录
        for plugin_file in self.plugin_dir.glob("*.py"):
            # 跳过 __init__.py 和示例文件
            if plugin_file.name.startswith('__') or plugin_file.name.endswith('.example'):
                continue
            
            try:
                # 动态导入模块
                spec = importlib.util.spec_from_file_location(
                    plugin_file.stem, 
                    plugin_file
                )
                if spec and spec.loader:
                    module = importlib.util.module_from_spec(spec)
                    spec.loader.exec_module(module)
                    
                    # 检查是否有 TOOL_CONFIG
                    if not hasattr(module, 'TOOL_CONFIG'):
                        print(f"[WARNING] {plugin_file.name} 缺少 TOOL_CONFIG 配置")
                        continue
                    
                    config = module.TOOL_CONFIG
                    
                    # 检查是否启用
                    if not config.get('enabled', True):
                        print(f"[INFO] 插件 {config.get('name', plugin_file.name)} 已禁用")
                        continue
                    
                    # 验证必需字段
                    required_fields = ['name', 'description', 'function', 'parameters']
                    missing_fields = [f for f in required_fields if f not in config]
                    if missing_fields:
                        print(f"[ERROR] {plugin_file.name} 缺少字段: {missing_fields}")
                        continue
                    
                    # 创建插件实例
                    plugin = ToolPlugin()
                    plugin.name = config['name']
                    plugin.description = config['description']
                    plugin.func = config['function']
                    plugin.parameters = config['parameters']
                    
                    # 检查名称冲突
                    if plugin.name in self.plugins:
                        print(f"[WARNING] 插件名称冲突: {plugin.name}，跳过加载")
                        continue
                    
                    # 注册插件
                    self.plugins[plugin.name] = plugin
                    loaded_count += 1
                    print(f"[SUCCESS] 加载插件: {plugin.name}")
                    
            except Exception as e:
                print(f"[ERROR] 加载插件 {plugin_file.name} 失败: {str(e)}")
                traceback.print_exc()
        
        return loaded_count
    
    def get_plugin(self, name: str) -> ToolPlugin:
        """获取指定插件"""
        return self.plugins.get(name)
    
    def get_all_plugins(self) -> Dict[str, ToolPlugin]:
        """获取所有插件"""
        return self.plugins
    
    def register_to_agent(self, agent):
        """
        将所有插件注册到 ReAct Agent
        
        Args:
            agent: ReActAgent 实例
        """
        for name, plugin in self.plugins.items():
            try:
                agent.register_tool(
                    name=plugin.name,
                    func=plugin.func,
                    description=plugin.description,
                    parameters=plugin.parameters
                )
                print(f"[SUCCESS] 插件 {name} 已注册到 Agent")
            except Exception as e:
                print(f"[ERROR] 注册插件 {name} 失败: {str(e)}")


# 便捷函数
def load_and_register_plugins(agent, plugin_dir: str = "react_plugins") -> int:
    """
    加载并注册所有插件到 Agent
    
    Args:
        agent: ReActAgent 实例
        plugin_dir: 插件目录路径
    
    Returns:
        int: 成功加载的插件数量
    """
    print("\n" + "=" * 50)
    print("开始加载 ReAct 插件...")
    print("=" * 50)
    
    loader = PluginLoader(plugin_dir)
    count = loader.load_plugins()
    
    if count > 0:
        loader.register_to_agent(agent)
        print(f"\n[SUCCESS] 共加载 {count} 个插件工具")
    else:
        print(f"\n[INFO] 未发现可用插件")
        print(f"[INFO] 请在 {plugin_dir} 目录下创建工具插件")
    
    print("=" * 50 + "\n")
    
    return count