Skill 系统设计与实现

Skill 是可复用的 AI 能力单元，通过组合和复用构建复杂的 AI 应用。

一、核心概念

1.1 什么是 Skill

Skill = 能力单元

特点：
- 单一职责（做好一件事）
- 可复用（多处使用）
- 可组合（多个 Skill 组合）
- 可发现（注册和查找）

1.2 Skill vs Tool

Tool（工具）:
- 外部 API 调用
- 固定输入输出
- 被动执行

Skill（能力）:
- 可包含多个 Tool
- 智能决策
- 主动规划

二、Skill 定义

2.1 基本结构

class Skill:
    def __init__(
        self,
        name: str,
        description: str,
        version: str = "1.0.0"
    ):
        self.name = name
        self.description = description
        self.version = version
        self.tags = []
        self.dependencies = []
    
    async def execute(self, **kwargs) -> any:
        """执行 Skill"""
        raise NotImplementedError
    
    def validate_input(self, **kwargs) -> bool:
        """验证输入"""
        raise NotImplementedError

2.2 Skill 示例

class SearchSkill(Skill):
    def __init__(self):
        super().__init__(
            name="web_search",
            description="搜索互联网获取信息",
            version="1.0.0"
        )
        self.tags = ["search", "internet"]
    
    async def execute(self, query: str, num_results: int = 5) -> list:
        # 调用搜索 API
        results = await search_api(query, num_results)
        return results
    
    def validate_input(self, **kwargs) -> bool:
        return "query" in kwargs and isinstance(kwargs["query"], str)

三、Skill 组合

3.1 顺序组合

class SequentialSkill(Skill):
    def __init__(self, name: str, skills: list[Skill]):
        super().__init__(name=name, description="顺序执行多个 Skill")
        self.skills = skills
    
    async def execute(self, **kwargs) -> any:
        result = kwargs
        for skill in self.skills:
            result = await skill.execute(**result)
        return result

3.2 并行组合

class ParallelSkill(Skill):
    def __init__(self, name: str, skills: list[Skill]):
        super().__init__(name=name, description="并行执行多个 Skill")
        self.skills = skills
    
    async def execute(self, **kwargs) -> list:
        tasks = [skill.execute(**kwargs) for skill in self.skills]
        results = await asyncio.gather(*tasks)
        return results

3.3 条件组合

class ConditionalSkill(Skill):
    def __init__(self, name: str, condition: callable, 
                 true_skill: Skill, false_skill: Skill = None):
        super().__init__(name=name, description="条件执行")
        self.condition = condition
        self.true_skill = true_skill
        self.false_skill = false_skill
    
    async def execute(self, **kwargs) -> any:
        if await self.condition(**kwargs):
            return await self.true_skill.execute(**kwargs)
        elif self.false_skill:
            return await self.false_skill.execute(**kwargs)

四、注册与发现

4.1 Skill 注册表

class SkillRegistry:
    def __init__(self):
        self.skills = {}
        self.tags_index = defaultdict(list)
    
    def register(self, skill: Skill):
        self.skills[skill.name] = skill
        
        # 建立标签索引
        for tag in skill.tags:
            self.tags_index[tag].append(skill.name)
    
    def get(self, name: str) -> Skill:
        return self.skills.get(name)
    
    def search_by_tag(self, tag: str) -> list[Skill]:
        names = self.tags_index.get(tag, [])
        return [self.skills[name] for name in names]

4.2 Skill 发现

class SkillDiscovery:
    def __init__(self, registry: SkillRegistry):
        self.registry = registry
    
    def find_skills(self, query: str) -> list[Skill]:
        # 基于语义搜索
        results = []
        for skill in self.registry.skills.values():
            score = self.calculate_relevance(skill, query)
            if score > 0.7:
                results.append((skill, score))
        
        return sorted(results, key=lambda x: x[1], reverse=True)
    
    def calculate_relevance(self, skill: Skill, query: str) -> float:
        # 计算相关性分数
        pass

五、实战案例

5.1 数据分析 Skill

class DataAnalysisSkill(Skill):
    def __init__(self):
        super().__init__(
            name="data_analysis",
            description="分析数据并生成报告"
        )
    
    async def execute(self, data: str, analysis_type: str) -> dict:
        # 1. 数据清洗
        cleaned_data = await self.clean_data(data)
        
        # 2. 统计分析
        stats = await self.calculate_stats(cleaned_data, analysis_type)
        
        # 3. 生成报告
        report = await self.generate_report(stats)
        
        return report

5.2 多 Skill 协作

class ResearchAssistant:
    def __init__(self, registry: SkillRegistry):
        self.search = registry.get("web_search")
        self.summarize = registry.get("summarize")
        self.translate = registry.get("translate")
    
    async def research(self, topic: str, language: str = "zh") -> str:
        # 1. 搜索信息
        search_results = await self.search.execute(query=topic)
        
        # 2. 总结内容
        summary = await self.summarize.execute(
            text=search_results,
            max_length=500
        )
        
        # 3. 翻译（如果需要）
        if language != "en":
            summary = await self.translate.execute(
                text=summary,
                target_lang=language
            )
        
        return summary

六、最佳实践

6.1 Skill 设计原则

1. 单一职责：一个 Skill 只做一件事
2. 接口清晰：输入输出明确定义
3. 无状态：不依赖外部状态
4. 可测试：易于单元测试
5. 可组合：支持与其他 Skill 组合

6.2 版本管理

# 语义化版本
class SkillVersion:
    def __init__(self, major: int, minor: int, patch: int):
        self.major = major  # 破坏性变更
        self.minor = minor  # 新功能
        self.patch = patch  # Bug 修复
    
    def is_compatible(self, other: 'SkillVersion') -> bool:
        return self.major == other.major

七、总结

Skill 系统核心要点：

Skill 系统是构建可复用 AI 能力的基础，通过组合和复用快速构建复杂应用。