302 lines
11 KiB
Python
302 lines
11 KiB
Python
"""SkillRegistry - Skill 注册中心（v2: 版本管理、能力查询、依赖检查）"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
from typing import TYPE_CHECKING
|
||
|
||
from agentkit.core.exceptions import SkillNotFoundError
|
||
from agentkit.skills.base import Skill, SkillConfig
|
||
from agentkit.skills.schema import DependencyDecl, HealthCheckResult
|
||
|
||
if TYPE_CHECKING:
|
||
from agentkit.skills.pipeline import SkillPipeline
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
class SkillRegistry:
|
||
"""Skill 注册中心，管理 Skill 的注册、发现、更新
|
||
|
||
v2 增强：
|
||
- 版本管理：同名 Skill 可注册多个版本，默认使用最新版
|
||
- 能力查询：按 capability 标签查询匹配的 Skill
|
||
- 依赖检查：health_check() 验证所有声明依赖是否已注册
|
||
"""
|
||
|
||
def __init__(self):
|
||
self._skills: dict[str, Skill] = {}
|
||
# 版本历史：name → {version → Skill}
|
||
self._skill_versions: dict[str, dict[str, Skill]] = {}
|
||
self._pipelines: dict[str, SkillPipeline] = {}
|
||
|
||
def register(self, skill: Skill) -> None:
|
||
"""注册 Skill，支持多版本共存
|
||
|
||
同名 Skill 注册时保留版本历史，默认指向最新注册的版本。
|
||
"""
|
||
name = skill.name
|
||
version = skill.version
|
||
|
||
# 维护版本历史
|
||
if name not in self._skill_versions:
|
||
self._skill_versions[name] = {}
|
||
self._skill_versions[name][version] = skill
|
||
|
||
# 默认指向最新注册的版本
|
||
self._skills[name] = skill
|
||
logger.info(f"Skill '{name}' v{version} registered")
|
||
|
||
def unregister(self, name: str, version: str | None = None) -> None:
|
||
"""注销 Skill
|
||
|
||
Args:
|
||
name: Skill 名称
|
||
version: 可选版本号。若指定则仅注销该版本；
|
||
若不指定则注销所有版本。
|
||
"""
|
||
if version is not None:
|
||
# 仅注销指定版本
|
||
if name in self._skill_versions and version in self._skill_versions[name]:
|
||
del self._skill_versions[name][version]
|
||
logger.info(f"Skill '{name}' v{version} unregistered")
|
||
# 如果删除的是当前默认版本，切换到最新版本
|
||
if name in self._skills and self._skills[name].version == version:
|
||
remaining = self._skill_versions[name]
|
||
if remaining:
|
||
latest = max(remaining.keys())
|
||
self._skills[name] = remaining[latest]
|
||
logger.info(
|
||
f"Skill '{name}' default switched to v{latest}"
|
||
)
|
||
else:
|
||
del self._skills[name]
|
||
del self._skill_versions[name]
|
||
else:
|
||
# 注销所有版本
|
||
if name in self._skills:
|
||
del self._skills[name]
|
||
if name in self._skill_versions:
|
||
del self._skill_versions[name]
|
||
logger.info(f"Skill '{name}' unregistered (all versions)")
|
||
|
||
def get(self, name: str, version: str | None = None) -> Skill:
|
||
"""获取 Skill
|
||
|
||
Args:
|
||
name: Skill 名称
|
||
version: 可选版本号。若指定则返回特定版本，否则返回默认（最新）版本。
|
||
|
||
Raises:
|
||
SkillNotFoundError: Skill 或指定版本不存在
|
||
"""
|
||
if version is not None:
|
||
if name not in self._skill_versions:
|
||
raise SkillNotFoundError(name)
|
||
if version not in self._skill_versions[name]:
|
||
raise SkillNotFoundError(f"{name}@{version}")
|
||
return self._skill_versions[name][version]
|
||
|
||
if name not in self._skills:
|
||
raise SkillNotFoundError(name)
|
||
return self._skills[name]
|
||
|
||
def list_skills(self) -> list[Skill]:
|
||
"""列出所有已注册的 Skill（每个名称返回默认版本）"""
|
||
return list(self._skills.values())
|
||
|
||
def update_skill(self, name: str, config: SkillConfig) -> Skill:
|
||
"""更新已注册 Skill 的配置，返回更新后的 Skill"""
|
||
if name not in self._skills:
|
||
raise SkillNotFoundError(name)
|
||
old_skill = self._skills[name]
|
||
new_skill = Skill(config, tools=old_skill.tools)
|
||
self._skills[name] = new_skill
|
||
# 同时更新版本历史
|
||
version = config.version
|
||
if name not in self._skill_versions:
|
||
self._skill_versions[name] = {}
|
||
self._skill_versions[name][version] = new_skill
|
||
logger.info(f"Skill '{name}' updated to v{version}")
|
||
return new_skill
|
||
|
||
def has_skill(self, name: str, version: str | None = None) -> bool:
|
||
"""检查 Skill 是否已注册
|
||
|
||
Args:
|
||
name: Skill 名称
|
||
version: 可选版本号
|
||
"""
|
||
if version is not None:
|
||
return (
|
||
name in self._skill_versions
|
||
and version in self._skill_versions[name]
|
||
)
|
||
return name in self._skills
|
||
|
||
# ---- 版本管理 ----
|
||
|
||
def get_versions(self, name: str) -> list[str]:
|
||
"""获取指定 Skill 的所有已注册版本号
|
||
|
||
Args:
|
||
name: Skill 名称
|
||
|
||
Returns:
|
||
版本号列表（按注册顺序）
|
||
|
||
Raises:
|
||
SkillNotFoundError: Skill 不存在
|
||
"""
|
||
if name not in self._skill_versions:
|
||
raise SkillNotFoundError(name)
|
||
return list(self._skill_versions[name].keys())
|
||
|
||
# ---- 能力查询 ----
|
||
|
||
def query_by_capability(self, tag: str) -> list[Skill]:
|
||
"""按能力标签查询 Skill
|
||
|
||
Args:
|
||
tag: 能力标签名（如 "rag", "terminal", "computer_use"）
|
||
|
||
Returns:
|
||
匹配的 Skill 列表
|
||
"""
|
||
result = []
|
||
for skill in self._skills.values():
|
||
capability_tags = [
|
||
cap.tag for cap in skill.capabilities
|
||
]
|
||
if tag in capability_tags:
|
||
result.append(skill)
|
||
return result
|
||
|
||
# ---- 依赖检查 ----
|
||
|
||
def health_check(self, name: str | None = None) -> list[HealthCheckResult]:
|
||
"""依赖健康检查
|
||
|
||
验证所有声明依赖是否已注册，以及版本约束是否满足。
|
||
|
||
Args:
|
||
name: 可选，指定检查某个 Skill。若不指定则检查所有 Skill。
|
||
|
||
Returns:
|
||
健康检查结果列表
|
||
"""
|
||
if name is not None:
|
||
if name not in self._skills:
|
||
raise SkillNotFoundError(name)
|
||
skills_to_check = [self._skills[name]]
|
||
else:
|
||
skills_to_check = list(self._skills.values())
|
||
|
||
results = []
|
||
for skill in skills_to_check:
|
||
result = self._check_skill_dependencies(skill)
|
||
results.append(result)
|
||
|
||
return results
|
||
|
||
def _check_skill_dependencies(self, skill: Skill) -> HealthCheckResult:
|
||
"""检查单个 Skill 的依赖是否满足"""
|
||
result = HealthCheckResult(
|
||
skill_name=skill.name,
|
||
skill_version=skill.version,
|
||
healthy=True,
|
||
)
|
||
|
||
for dep in skill.dependencies:
|
||
if dep.type == "skill":
|
||
if not self.has_skill(dep.name):
|
||
if dep.required:
|
||
result.healthy = False
|
||
result.missing_dependencies.append(dep.name)
|
||
else:
|
||
result.warnings.append(
|
||
f"Optional skill dependency '{dep.name}' not registered"
|
||
)
|
||
elif dep.version_constraint:
|
||
# 简化版本约束检查：仅检查已注册版本是否满足
|
||
dep_skill = self.get(dep.name)
|
||
if not self._check_version_constraint(
|
||
dep_skill.version, dep.version_constraint
|
||
):
|
||
result.version_mismatches.append(
|
||
f"{dep.name}: need {dep.version_constraint}, "
|
||
f"got {dep_skill.version}"
|
||
)
|
||
if dep.required:
|
||
result.healthy = False
|
||
elif dep.type == "tool":
|
||
# Tool 依赖检查需要 ToolRegistry，此处仅记录
|
||
# 实际检查在运行时由 SkillLoader 配合 ToolRegistry 完成
|
||
pass
|
||
|
||
return result
|
||
|
||
@staticmethod
|
||
def _check_version_constraint(
|
||
actual_version: str, constraint: str
|
||
) -> bool:
|
||
"""简化的版本约束检查
|
||
|
||
支持基本的约束格式：
|
||
- ">=x.y.z" — 大于等于
|
||
- "<=x.y.z" — 小于等于
|
||
- "==x.y.z" — 精确匹配
|
||
- ">=x.y.z,<a.b.c" — 范围约束
|
||
|
||
使用简单的元组比较进行 semver 检查。
|
||
"""
|
||
try:
|
||
actual = tuple(int(x) for x in actual_version.split("."))
|
||
except (ValueError, AttributeError):
|
||
return True # 无法解析时默认通过
|
||
|
||
parts = [p.strip() for p in constraint.split(",")]
|
||
for part in parts:
|
||
if part.startswith(">="):
|
||
target = tuple(int(x) for x in part[2:].split("."))
|
||
if actual < target:
|
||
return False
|
||
elif part.startswith("<="):
|
||
target = tuple(int(x) for x in part[2:].split("."))
|
||
if actual > target:
|
||
return False
|
||
elif part.startswith("=="):
|
||
target = tuple(int(x) for x in part[2:].split("."))
|
||
if actual != target:
|
||
return False
|
||
elif part.startswith(">"):
|
||
target = tuple(int(x) for x in part[1:].split("."))
|
||
if actual <= target:
|
||
return False
|
||
elif part.startswith("<"):
|
||
target = tuple(int(x) for x in part[1:].split("."))
|
||
if actual >= target:
|
||
return False
|
||
return True
|
||
|
||
# ---- Pipeline 管理 ----
|
||
|
||
def register_pipeline(self, pipeline: SkillPipeline) -> None:
|
||
"""注册 SkillPipeline，同名覆盖"""
|
||
self._pipelines[pipeline.name] = pipeline
|
||
logger.info(f"SkillPipeline '{pipeline.name}' registered")
|
||
|
||
def get_pipeline(self, name: str) -> SkillPipeline | None:
|
||
"""获取已注册的 SkillPipeline，不存在返回 None"""
|
||
return self._pipelines.get(name)
|
||
|
||
def list_pipelines(self) -> list[str]:
|
||
"""列出所有已注册的 Pipeline 名称"""
|
||
return list(self._pipelines.keys())
|
||
|
||
def unregister_pipeline(self, name: str) -> None:
|
||
"""注销 SkillPipeline"""
|
||
if name in self._pipelines:
|
||
del self._pipelines[name]
|
||
logger.info(f"SkillPipeline '{name}' unregistered")
|