JARVIS/development-doc/plan/tool-update/phase-t-2-tool-registry.md

Phase T.2：工具注册中心

日期：2026-04-04 状态：待开始 依赖：T.1（待完成）

---

1. 本阶段目的

建立 Jarvis 的工具注册中心：

• 工具动态发现
• 工具注册与管理
• 工具描述生成
• 调用统计与监控

---

2. 工具注册中心架构

2.1 核心类

# tools/registry.py
from typing import Dict, List, Optional, Callable
from dataclasses import dataclass, field
from datetime import datetime
import asyncio


@dataclass
class ToolMetadata:
    """工具元数据"""
    name: str
    display_name: str
    description: str
    version: str
    author: Optional[str] = None
    tags: List[str] = field(default_factory=list)
    dependencies: List[str] = field(default_factory=list)
    enabled: bool = True
    registered_at: datetime = field(default_factory=datetime.utcnow)
    
    # 统计
    call_count: int = 0
    error_count: int = 0
    total_duration_ms: int = 0
    
    @property
    def avg_duration_ms(self) -> int:
        if self.call_count == 0:
            return 0
        return self.total_duration_ms // self.call_count
    
    @property
    def error_rate(self) -> float:
        if self.call_count == 0:
            return 0.0
        return self.error_count / self.call_count


class ToolRegistry:
    """工具注册中心"""
    
    def __init__(self):
        self._tools: Dict[str, ToolMetadata] = {}
        self._executors: Dict[str, Callable] = {}
        self._configs: Dict[str, dict] = {}
        self._lock = asyncio.Lock()
    
    # === 注册方法 ===
    
    async def register(
        self,
        manifest_path: str,
        executor: Callable,
        config: Optional[dict] = None,
    ) -> ToolMetadata:
        """注册工具"""
        from tools.schemas.validator import validate_manifest
        import yaml
        
        with open(manifest_path) as f:
            data = yaml.safe_load(f)
        
        manifest = validate_manifest(data)
        
        metadata = ToolMetadata(
            name=manifest.name,
            display_name=manifest.display_name,
            description=manifest.description,
            version=manifest.version,
            author=manifest.author,
            tags=manifest.tags or [],
            dependencies=manifest.dependencies or [],
            enabled=manifest.enabled,
        )
        
        async with self._lock:
            self._tools[manifest.name] = metadata
            self._executors[manifest.name] = executor
            if config:
                self._configs[manifest.name] = config
        
        return metadata
    
    async def unregister(self, name: str) -> bool:
        """注销工具"""
        async with self._lock:
            if name in self._tools:
                del self._tools[name]
                del self._executors[name]
                self._configs.pop(name, None)
                return True
            return False
    
    async def enable(self, name: str) -> None:
        """启用工具"""
        async with self._lock:
            if name in self._tools:
                self._tools[name].enabled = True
    
    async def disable(self, name: str) -> None:
        """禁用工具"""
        async with self._lock:
            if name in self._tools:
                self._tools[name].enabled = False
    
    # === 查询方法 ===
    
    async def get(self, name: str) -> Optional[ToolMetadata]:
        """获取工具元数据"""
        return self._tools.get(name)
    
    async def get_executor(self, name: str) -> Optional[Callable]:
        """获取工具执行器"""
        return self._executors.get(name)
    
    async def get_config(self, name: str) -> dict:
        """获取工具配置"""
        return self._configs.get(name, {})
    
    async def list_all(self) -> List[ToolMetadata]:
        """列出所有工具"""
        return list(self._tools.values())
    
    async def list_enabled(self) -> List[ToolMetadata]:
        """列出已启用的工具"""
        return [t for t in self._tools.values() if t.enabled]
    
    async def list_by_tag(self, tag: str) -> List[ToolMetadata]:
        """按标签筛选工具"""
        return [t for t in self._tools.values() if tag in t.tags]
    
    async def search(self, query: str) -> List[ToolMetadata]:
        """搜索工具"""
        query = query.lower()
        return [
            t for t in self._tools.values()
            if query in t.name.lower()
            or query in t.description.lower()
            or query in t.display_name.lower()
        ]
    
    # === 统计方法 ===
    
    async def record_call(
        self,
        name: str,
        duration_ms: int,
        error: bool = False,
    ) -> None:
        """记录调用"""
        async with self._lock:
            if name in self._tools:
                tool = self._tools[name]
                tool.call_count += 1
                tool.total_duration_ms += duration_ms
                if error:
                    tool.error_count += 1
    
    async def get_stats(self) -> dict:
        """获取统计信息"""
        tools = list(self._tools.values())
        return {
            "total_tools": len(tools),
            "enabled_tools": sum(1 for t in tools if t.enabled),
            "total_calls": sum(t.call_count for t in tools),
            "total_errors": sum(t.error_count for t in tools),
            "avg_error_rate": sum(t.error_rate for t in tools) / len(tools) if tools else 0,
        }

---

3. 工具发现机制

3.1 自动发现

# tools/discovery.py
from pathlib import Path
from typing import List


class ToolDiscovery:
    """工具自动发现"""
    
    def __init__(self, manifest_dir: Path):
        self.manifest_dir = manifest_dir
    
    def discover(self) -> List[Path]:
        """发现所有 Manifest 文件"""
        manifests = list(self.manifest_dir.glob("**/*.yaml"))
        manifests.extend(self.manifest_dir.glob("**/*.yml"))
        manifests.extend(self.manifest_dir.glob("**/*.json"))
        return manifests
    
    def discover_by_tag(self, tag: str) -> List[Path]:
        """按标签发现"""
        # 读取所有 manifest，筛选标签
        pass
    
    async def hot_reload(self, registry: ToolRegistry) -> None:
        """热重载工具"""
        for manifest_path in self.discover():
            # 重新注册
            pass

3.2 启动时注册

# tools/loader.py
from pathlib import Path


async def load_all_tools(registry: ToolRegistry) -> None:
    """加载所有工具"""
    manifest_dir = Path(__file__).parent / "manifests"
    
    discovery = ToolDiscovery(manifest_dir)
    
    for manifest_path in discovery.discover():
        tool_name = manifest_path.stem
        
        # 加载 executor
        executor = load_executor(manifest_path)
        
        # 加载配置
        config = load_config(tool_name)
        
        # 注册
        await registry.register(manifest_path, executor, config)


def load_executor(manifest_path: Path) -> Callable:
    """加载工具执行器"""
    import yaml
    
    with open(manifest_path) as f:
        manifest = yaml.safe_load(f)
    
    # 根据运行时类型加载
    runtime = manifest.get("runtime", "python")
    
    if runtime == "python":
        return load_python_executor(manifest)
    elif runtime == "javascript":
        return load_js_executor(manifest)
    else:
        return load_native_executor(manifest)

---

4. 工具描述生成

4.1 AI 友好的工具描述

# tools/description.py
from typing import Dict, List


def generate_tool_description(manifest: dict) -> str:
    """生成 AI 友好的工具描述"""
    lines = [
        f"## {manifest['display_name']}",
        f"{manifest['description']}",
        "",
        "### 可用命令:",
    ]
    
    for cmd in manifest.get("commands", []):
        lines.append(f"#### {cmd['name']}")
        lines.append(cmd["description"])
        lines.append("")
        
        if cmd.get("example"):
            lines.append("**示例:**")
            lines.append(f"```\n{cmd['example']}\n```")
            lines.append("")
    
    return "\n".join(lines)


def generate_tools_for_llm(registry: ToolRegistry) -> str:
    """生成给 LLM 的工具列表"""
    tools = registry.list_enabled()
    
    sections = ["## 可用工具\n"]
    
    for tool in tools:
        manifest = load_manifest(tool.name)
        sections.append(generate_tool_description(manifest))
        sections.append("\n---\n")
    
    return "\n".join(sections)

---

5. 权限控制

5.1 工具权限

# tools/permissions.py
from enum import Enum
from typing import Set


class ToolPermission(str, Enum):
    """工具权限"""
    EXECUTE = "tool:execute"
    CONFIGURE = "tool:configure"
    ENABLE = "tool:enable"
    DISABLE = "tool:disable"


class ToolPermissionChecker:
    """工具权限检查"""
    
    def __init__(self):
        self._user_permissions: Dict[str, Set[ToolPermission]] = {}
        self._tool_roles: Dict[str, Set[str]] = {}  # tool_name -> required_roles
    
    def set_user_permissions(
        self,
        user_id: str,
        permissions: Set[ToolPermission],
    ) -> None:
        """设置用户权限"""
        self._user_permissions[user_id] = permissions
    
    def set_tool_roles(
        self,
        tool_name: str,
        required_roles: Set[str],
    ) -> None:
        """设置工具所需角色"""
        self._tool_roles[tool_name] = required_roles
    
    def can_execute(self, user_id: str, tool_name: str) -> bool:
        """检查用户是否可以执行工具"""
        # 检查全局权限
        if ToolPermission.EXECUTE in self._user_permissions.get(user_id, set()):
            return True
        
        # 检查工具特定角色
        required_roles = self._tool_roles.get(tool_name, set())
        if not required_roles:
            return True
        
        # TODO: 检查用户是否有所需角色
        return False

---

6. 集成到 Agent

6.1 LangChain 集成

# tools/langchain_adapter.py
from typing import List, Optional
from langchain.tools import BaseTool
from pydantic import BaseModel


class LangChainToolAdapter:
    """LangChain 工具适配器"""
    
    def __init__(self, registry: ToolRegistry):
        self.registry = registry
    
    def to_langchain_tools(self) -> List[BaseTool]:
        """转换为 LangChain 工具"""
        tools = []
        
        for metadata in self.registry.list_enabled():
            executor = self.registry.get_executor(metadata.name)
            config = self.registry.get_config(metadata.name)
            
            tool = self._create_tool(metadata, executor, config)
            tools.append(tool)
        
        return tools
    
    def _create_tool(
        self,
        metadata: ToolMetadata,
        executor: Callable,
        config: dict,
    ) -> BaseTool:
        """创建单个 LangChain 工具"""
        # 根据 manifest 创建工具
        pass

---

7. 实现步骤

步骤	任务	优先级
1	实现 ToolRegistry 类	🟢 高
2	实现 ToolDiscovery	🟢 高
3	实现工具描述生成	🟡 中
4	实现权限检查	🟡 中
5	实现 LangChain 适配器	🟡 中
6	集成到 Agent	🟢 高
7	单元测试	🟡 中

---

8. 核心文件变更

文件	变更
tools/registry.py	新增
tools/discovery.py	新增
tools/description.py	新增
tools/permissions.py	新增
tools/langchain_adapter.py	新增
tools/__init__.py	更新导出

---

9. 工作量估算

任务	工作量
ToolRegistry	0.5 天
ToolDiscovery	0.5 天
描述生成	0.3 天
权限检查	0.3 天
LangChain 适配	0.3 天
集成	0.5 天
单元测试	0.5 天
总计	2.5 天

---

10. 验收标准

• ToolRegistry 可正确注册/注销工具
• ToolDiscovery 可发现所有 Manifest
• 工具描述生成正确
• 权限检查正常工作
• LangChain 工具可正常转换
• Agent 可使用注册的工具
• 单元测试通过