477 lines
12 KiB
Markdown
477 lines
12 KiB
Markdown
# Phase T.2：工具注册中心
|
||
|
||
日期：2026-04-04
|
||
状态：待开始
|
||
依赖：T.1（待完成）
|
||
|
||
---
|
||
|
||
## 1. 本阶段目的
|
||
|
||
建立 Jarvis 的工具注册中心：
|
||
|
||
- 工具动态发现
|
||
- 工具注册与管理
|
||
- 工具描述生成
|
||
- 调用统计与监控
|
||
|
||
---
|
||
|
||
## 2. 工具注册中心架构
|
||
|
||
### 2.1 核心类
|
||
|
||
```python
|
||
# 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 自动发现
|
||
|
||
```python
|
||
# 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 启动时注册
|
||
|
||
```python
|
||
# 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 友好的工具描述
|
||
|
||
```python
|
||
# 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 工具权限
|
||
|
||
```python
|
||
# 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 集成
|
||
|
||
```python
|
||
# 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 可使用注册的工具
|
||
- [ ] 单元测试通过
|