Skills:工具的组合与复用

7/6/2026 AI AgentSkills技能编排

# Skills:工具的组合与复用

第二篇:Agent 的手脚

📌 本章目标

第5章讲了单个工具的调用,第6章讲了工具的标准化协议(MCP)。本章讲工具的组合——把多个工具打包成一个"技能",让 Agent 拥有更高层次的能力。

# 8.1 从工具到技能

打个比方:工具是一把锤子,技能是"钉钉子"——锤子只是一个工具,但"钉钉子"需要锤子 + 钉子 + 判断角度 + 控制力度。技能是工具的组合 + 使用知识

维度 Tool(工具) Skill(技能)
粒度 单一操作(get_weather) 复合能力(天气查询+行程规划+穿衣建议)
工具数量 1 个 1~N 个
包含知识 使用策略、调用顺序、参数约束
可复用性 低(每个项目重新定义) 高(打包成包,跨项目共享)
类比 函数 类/模块

# 8.2 Skills 的三层结构

一个完整的 Skill 包含三层:

流程图: 元数据层(name/description/version/触发条件/权限)→ 工具层(工具列表+Schema/工具间的依赖关系)→ 知识层(使用策略/调用顺序/参数约束/异常处理)

# 一个 Skill 的 YAML 定义示例
name: weather-advisor
version: 1.0.0
description: 天气顾问技能,查询天气并提供穿衣/出行建议
triggers:
  - "天气"
  - "下雨"
  - "穿什么"
  - "带伞"
permissions:
  - network: true    # 需要网络访问
  - location: false  # 不需要定位

tools:
  - name: get_weather
    description: 查询天气
    parameters:
      city: { type: string, required: true }
      date: { type: string, default: today }

  - name: get_clothing_advice
    description: 根据天气获取穿衣建议
    parameters:
      temperature: { type: number, required: true }
      weather: { type: string, required: true }

knowledge: |
  ## 使用策略
  1. 先调用 get_weather 获取天气信息
  2. 如果用户问穿衣建议,再调用 get_clothing_advice
  3. 如果是雨天,主动提醒带伞
  4. 温度低于 10°C 提醒保暖,高于 30°C 提醒防暑

  ## 参数约束
  - city 必须是中文城市名
  - date 支持 today/tomorrow/YYYY-MM-DD
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

# 8.3 Skills 的发现与加载

Agent 怎么知道有哪些 Skills 可用?和 CLI 工具类似,Skills 也需要自动发现机制:

# 8.3.1 文件系统发现

Skill 以文件形式存储在 skills/ 目录下,Agent 启动时扫描目录:

import os
import yaml

class SkillLoader:
    def __init__(self, skills_dir: str = "skills"):
        self.skills_dir = skills_dir
        self.registry: dict = {}  # name -> skill_definition

    def scan(self):
        """扫描 skills 目录,加载所有技能"""
        if not os.path.exists(self.skills_dir):
            return

        for entry in os.scandir(self.skills_dir):
            if not entry.is_dir():
                continue

            skill_file = os.path.join(entry.path, "skill.yml")
            if not os.path.exists(skill_file):
                continue

            with open(skill_file, "r") as f:
                skill_def = yaml.safe_load(f)
                skill_def["path"] = entry.path
                self.registry[skill_def["name"]] = skill_def

        print(f"加载了 {len(self.registry)} 个技能: {list(self.registry.keys())}")

    def get_skill_prompt(self) -> str:
        """生成技能描述,注入到 system prompt"""
        if not self.registry:
            return ""

        lines = ["你可以使用以下技能:"]
        for name, skill in self.registry.items():
            triggers = ", ".join(skill.get("triggers", []))
            lines.append(f"- {name}: {skill['description']} (触发词: {triggers})")

        return "\n".join(lines)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39

# 8.3.2 热重载

生产环境需要热重载——修改 Skill 文件后无需重启 Agent:

from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler

class SkillHotReloader(FileSystemEventHandler):
    def __init__(self, loader: SkillLoader):
        self.loader = loader

    def on_modified(self, event):
        if event.src_path.endswith("skill.yml"):
            print(f"检测到 Skill 变更,重新加载...")
            self.loader.scan()

# 启动热重载监控
observer = Observer()
observer.schedule(
    SkillHotReloader(loader),
    path="skills/",
    recursive=True
)
observer.start()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

# 8.4 Skills 的懒加载

和 CLI 工具一样,Skills 太多会导致 Prompt 过长。解决方案是三层懒加载

层级 策略 注入到 Prompt 示例
第一层 始终加载 Skill 名称 + 描述 weather-advisor, file-reader
第二层 触发词匹配时加载 完整 Schema + 使用知识 用户提到"天气"时加载 weather-advisor 的完整定义
第三层 AI 主动搜索 按需 罕见 Skill,靠 AI 用 ToolSearch 查找
class LazySkillManager:
    def __init__(self, loader: SkillLoader):
        self.loader = loader
        self.activated: set = set()  # 已激活的 Skill

    def get_prompt(self, user_message: str = "") -> str:
        """根据用户消息生成 Prompt"""
        lines = []

        # 第一层:始终列出所有 Skill 名称(简短)
        for name, skill in self.loader.registry.items():
            triggers = skill.get("triggers", [])
            # 第二层:触发词匹配 → 激活完整定义
            if any(t in user_message for t in triggers):
                self.activated.add(name)
                lines.append(self._format_full_skill(skill))
            else:
                # 第一层:只注入名称和描述
                lines.append(f"- {name}: {skill['description']}")

        # 第三层:已激活但用户消息不匹配的,保持激活
        for name in self.activated:
            if name not in [l.split(":")[0].strip("- ") for l in lines]:
                skill = self.loader.registry.get(name)
                if skill:
                    lines.append(self._format_full_skill(skill))

        return "\n你可以使用以下技能:\n" + "\n".join(lines)

    def _format_full_skill(self, skill: dict) -> str:
        """格式化完整 Skill 定义"""
        tools_desc = "\n".join(
            f"  - {t['name']}: {t['description']}"
            for t in skill.get("tools", [])
        )
        return f"- {skill['name']} (已激活)\n  工具:\n{tools_desc}\n  策略:\n{skill.get('knowledge', '')}"
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

# 8.5 Skill 匹配与调度

当用户说"北京明天天气怎么样,穿什么合适?",Agent 需要:

  1. 匹配到 weather-advisor Skill(触发词"天气")
  2. 激活该 Skill 的完整定义
  3. 按知识层策略执行:先调 get_weather,再调 get_clothing_advice
  4. 综合两个工具的结果,生成自然语言回答

流程图: 用户("北京明天天气穿什么合适?")→ Skill 匹配(触发词"天气" → weather-advisor)→ 激活 Skill(加载完整 Schema + 知识)→ AI 规划(1.调 get_weather 2.调 get_clothing_advice)→ 执行 get_weather(→ 晴 25°C)→ 执行 get_clothing_advice(→ 薄外套+长裤)→ 综合回答(北京明天晴 25°C,建议穿薄外套+长裤)

# 8.6 Skills vs MCP vs Function Calling

维度 Function Calling MCP Skills
本质 单个工具调用 工具的标准协议 工具的组合 + 知识
粒度 函数级 Server 级 能力级
包含知识 ✅ 使用策略、调用顺序
跨项目复用 ❌ 需重新定义 ✅ Server 共享 ✅ Skill 包共享
关系 底层机制 传输层 上层组织

三者的关系是层级递进的:

  • Function Calling 是底层机制——AI 调用函数的能力
  • MCP 是传输层——标准化工具的发现和调用协议
  • Skills 是上层组织——把多个工具(无论 Function Calling 还是 MCP)打包成可复用的能力单元

💡 一个 Skill 可以包含 MCP 工具

Skills 不排斥 MCP。一个 database-advisor Skill 可以包含一个 MCP database server 提供的查询工具,加上自定义的 SQL 优化建议工具,以及"先分析表结构再优化查询"的使用策略。

# 8.7 实战:构建一个 Skill

把第2章的天气查询升级为一个完整的 Skill:

# skills/weather-advisor/skill.yml
name: weather-advisor
version: 1.0.0
description: 天气顾问,查询天气并提供穿衣和出行建议
triggers:
  - "天气"
  - "下雨"
  - "穿什么"
  - "带伞"
  - "气温"

tools:
  - name: get_weather
    description: 查询指定城市指定日期的天气
    parameters:
      city: { type: string, required: true, description: "城市名" }
      date: { type: string, default: "today", description: "日期" }

  - name: get_clothing_advice
    description: 根据天气获取穿衣建议
    parameters:
      temperature: { type: number, required: true, description: "温度(°C)" }
      weather: { type: string, required: true, description: "天气状况" }

knowledge: |
  ## 使用策略
  1. 总是先调用 get_weather 获取天气
  2. 如果用户问穿衣/出行建议,调用 get_clothing_advice
  3. 主动建议:
     - 雨天 → 带伞
     - <10°C → 保暖
     - >30°C → 防暑
     - 雾霾 → 口罩
  4. 如果查多天天气,逐天调用 get_weather
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

对应的 Python 实现:

# skills/weather-advisor/main.py
def get_weather(city: str, date: str = "today") -> dict:
    # ... 同第2章实现 ...

def get_clothing_advice(temperature: float, weather: str) -> dict:
    advice = []
    if temperature < 10:
        advice.append("厚外套、围巾、手套")
    elif temperature < 20:
        advice.append("薄外套或毛衣")
    elif temperature < 28:
        advice.append("长袖衬衫")
    else:
        advice.append("短袖、短裤")

    if "雨" in weather:
        advice.append("带伞")
    if "雷" in weather:
        advice.append("避免户外活动")

    return {"advice": "、".join(advice)}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

# 8.8 Skill 市场:未来的应用商店

Skills 的终极目标是形成市场——开发者像发布 App 一样发布 Skill 包,用户像装 App 一样安装 Skill:

# 安装 Skill(类似 npm install)
skill install weather-advisor
skill install code-reviewer
skill install data-analyst

# 查看已安装的 Skills
skill list
# weather-advisor    1.0.0  天气顾问
# code-reviewer      2.1.0  代码审查
# data-analyst       0.9.0  数据分析

# 卸载
skill remove weather-advisor
1
2
3
4
5
6
7
8
9
10
11
12
13

这和 MCP 的愿景类似,但层次不同:MCP 是工具级的市场,Skills 是能力级的市场。未来可能两者融合——MCP Server 提供工具,Skill 包提供组合策略。

# 8.9 Skill 的分层体系

就像企业架构有基础设施层、业务逻辑层、应用层一样,Skill 也存在三层金字塔——从底层的基础 Skill 到顶层的领域 Skill,每一层封装不同复杂度的能力:

流程图: 基础 Skill(Layer 1:单工具封装,get_weather/read_file/send_email,类比:函数)→ 复合 Skill(Layer 2:多工具组合+调用策略,weather-advisor/code-reviewer/data-analyst,类比:类/模块)→ 领域 Skill(Layer 3:行业知识包+多复合 Skill,finance-advisor/legal-assistant/medical-consult,类比:框架/SDK)

维度 基础 Skill 复合 Skill 领域 Skill
工具数量 1 2~5 5~20+(含子 Skill)
知识含量 无(纯工具映射) 中等(调用策略) 高(行业规则 + 合规约束)
触发方式 AI 自主调用 触发词匹配 领域关键词 + 上下文推理
复用范围 跨项目通用 跨团队复用 跨行业复用(需定制)
类比 WaLiCode ToolDefinition 单工具 WorkflowChain 多步骤编排 DomainPackage 行业配置包
验证方式 单元测试 集成测试 + Prompt 测试 端到端场景测试 + 合规审计

# 8.9.1 基础 Skill:单工具的标准化封装

基础 Skill 就是把一个原始工具加上元数据、Schema 校验、异常处理,变成一个标准化的工具单元。它不包含调用策略,但包含防御性编程

# skills/base/read_file/skill.yml — 基础 Skill 示例
name: read_file
version: 1.0.0
description: 读取文件内容,支持文本和图片
layer: base

tools:
  - name: read_file
    description: 读取指定路径的文件
    parameters:
      path: { type: string, required: true, description: "文件路径" }
      offset: { type: number, default: null, description: "起始行号" }
      limit: { type: number, default: null, description: "最大行数" }

guards:
  - path_traversal: true    # 防止路径穿越攻击
  - max_file_size: 10MB      # 限制读取大小
  - allowed_extensions: [.txt, .md, .py, .json, .yaml, .yml, .csv, .jpg, .png]

error_handling:
  file_not_found: "文件不存在,请检查路径是否正确"
  permission_denied: "无权限读取该文件,请确认访问权限"
  file_too_large: "文件超过 10MB 限制,请使用 offset/limit 分段读取"
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
# skills/base/read_file/main.py
import os
import pathlib

class ReadFileSkill:
    """基础 Skill:单工具封装,侧重防御性编程"""

    ALLOWED_EXTENSIONS = {'.txt', '.md', '.py', '.json', '.yaml', '.yml', '.csv', '.jpg', '.png'}
    MAX_FILE_SIZE = 10 * 1024 * 1024  # 10MB

    def execute(self, path: str, offset: int = None, limit: int = None) -> dict:
        # 1. 路径穿越防御(WaLiCode 的 permissionGuard 模式)
        resolved = pathlib.Path(path).resolve()
        if ".." in str(resolved) or not str(resolved).startswith(os.getcwd()):
            return {"error": "路径穿越攻击被拦截,禁止访问"}

        # 2. 扩展名白名单
        if resolved.suffix not in self.ALLOWED_EXTENSIONS:
            return {"error": f"不支持的文件类型: {resolved.suffix}"}

        # 3. 文件大小限制
        if resolved.stat().st_size > self.MAX_FILE_SIZE:
            return {"error": f"文件超过 {self.MAX_FILE_SIZE//1024//1024}MB 限制"}

        # 4. 执行核心逻辑
        if resolved.suffix in {'.jpg', '.png'}:
            return {"type": "image", "path": str(resolved)}

        with open(resolved, "r", encoding="utf-8") as f:
            lines = f.readlines()
            if offset:
                lines = lines[offset-1:]  # 1-indexed
            if limit:
                lines = lines[:limit]
            return {"type": "text", "content": "".join(lines), "lines": len(lines)}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

# 8.9.2 复合 Skill:多工具组合 + 调用策略

复合 Skill 是基础 Skill 的组合,核心价值是调用策略——告诉 AI "先做什么、后做什么、什么条件下做什么":

# skills/compound/code-reviewer/skill.yml — 复合 Skill 示例
name: code-reviewer
version: 2.1.0
description: 代码审查技能,自动检查代码质量、安全漏洞和最佳实践
layer: compound
triggers: ["代码审查", "review", "检查代码", "code review", "PR"]

tools:
  - name: read_file           # 基础 Skill
    description: 读取代码文件
  - name: lint_check          # 基础 Skill
    description: 运行 lint 工具检查代码风格
    parameters:
      file_path: { type: string, required: true }
      linter: { type: string, default: "auto" }
  - name: security_scan       # 基础 Skill
    description: 安全漏洞扫描
    parameters:
      file_path: { type: string, required: true }
      severity: { type: string, default: "medium" }
  - name: generate_report     # 基础 Skill
    description: 生成审查报告
    parameters:
      findings: { type: array, required: true }
      format: { type: string, default: "markdown" }

knowledge: |
  ## 调用策略
  1. 先 read_file 读取目标代码
  2. 并行调用 lint_check + security_scan
  3. 汇总结果 → generate_report
  4. 严重安全问题 → 优先报告(severity=high)
  5. 如果是 Python 文件,优先用 pylint;JS 用 eslint

  ## WaLiCode 对应
  - WorkflowChain 模式:read → [lint + security] → report
  - ParallelGateway 模式:lint 和 security 可并行
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

流程图: 用户("审查这个 PR")→ read_file(读取代码文件)→ 并行执行(ParallelGateway)→ lint_check(代码风格检查)+ security_scan(安全漏洞扫描)→ 汇总结果(合并 findings)→ generate_report(生成审查报告)

# 8.9.3 领域 Skill:行业知识包

领域 Skill 是金字塔顶端——它不是几个工具的组合,而是一个行业的完整知识框架。一个金融领域 Skill 可能包含合规检查、风控计算、报表生成等 10+ 个复合 Skill,以及金融法规的领域知识:

# skills/domain/finance-advisor/skill.yml — 领域 Skill 示例
name: finance-advisor
version: 1.0.0
description: 金融顾问技能包,包含合规检查、风控分析和报表生成
layer: domain
triggers: ["投资", "风险评估", "合规", "财报", "K线", "理财"]

sub_skills:
  - compliance_checker     # 合规检查(复合 Skill)
  - risk_analyzer          # 风控分析(复合 Skill)
  - report_generator       # 报表生成(复合 Skill)

domain_knowledge: |
  ## 金融合规规则
  - 所有投资建议必须附带风险提示
  - 不得推荐具体个股(合规红线)
  - 用户资产配置需遵循"适当性原则"
  - 引用数据需标注来源和时效性

  ## 风控计算规则
  - 夏普比率 > 1.0 才算优质策略
  - 最大回撤 > 30% 需特别警示
  - 杠杆倍数 > 3x 需风险提示

  ## WaLiCode 对应模式
  - DomainPackage:行业配置包
  - ComplianceGuard:合规红线拦截
  - RiskThresholdGuard:风控阈值告警

permissions:
  - requires_approval: true    # 领域 Skill 需审批才能启用
  - audit_log: true            # 所有调用需审计日志
  - data_classification: financial_confidential
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

💡 金字塔原则

底层越通用越好(read_file 跨所有项目可用),顶层越专业越好(finance-advisor 只在金融场景有用)。企业 Agent 应自下而上构建——先把基础 Skill 打磨好,再组装复合 Skill,最后才封装领域 Skill。不要跳层——没有稳固的基础 Skill,复合 Skill 就是空中楼阁。

# 8.10 Skill 的沉淀与演进

Skill 不是一次设计就完成的,它像代码一样有版本迭代的生命周期。好的 Skill 来自项目实践的反复提炼——从临时脚本 → 内部最佳实践 → 可复用 Skill 包。

流程图: 项目实践(发现重复模式)→ 提取模式(抽象共性逻辑)→ Skill v0.1(内部验证版)→ 项目反馈(Bug/缺失功能/新需求)→ 版本迭代(v0.2 → v1.0 → v2.0)→ 稳定版 Skill(跨项目复用)

# 8.10.1 从项目中提炼 Skill 的四步流程

步骤 动作 产出 WaLiCode 对应
1. 模式识别 发现项目中重复出现的工具组合模式 模式列表(如:每次都先读文件再 lint) PatternMining
2. 抽象提炼 把模式抽象为通用 Skill,去除项目特有逻辑 Skill YAML + 知识层 AbstractionLayer
3. 验证打磨 在 2~3 个项目中试用,收集反馈 反馈清单 + Bug 修复 IterativeRefinement
4. 发布复用 稳定后发布到 Skill 市场 / 内部仓库 版本号 + Changelog ReleasePipeline
# 步骤 1:模式识别 — 从项目日志中提取重复模式
import re
from collections import Counter

def extract_patterns(execution_logs: list[str]) -> list[dict]:
    """从 Agent 执行日志中识别重复的工具调用模式"""
    # 解析工具调用序列
    sequences = []
    current_seq = []
    for log in execution_logs:
        if "tool_call:" in log:
            tool_name = re.search(r"tool_call: (\w+)", log).group(1)
            current_seq.append(tool_name)
        elif "response:" in log:
            if current_seq:
                sequences.append(tuple(current_seq))
                current_seq = []

    # 统计高频模式(长度 >= 2)
    pattern_counter = Counter(
        seq for seq in sequences if len(seq) >= 2
    )
    return [
        {"pattern": list(p), "count": c}
        for p, c in pattern_counter.most_common(5)
    ]

# 示例输出
patterns = extract_patterns(logs)
# [
#   {"pattern": ["read_file", "lint_check", "security_scan"], "count": 23},
#   {"pattern": ["read_file", "generate_report"], "count": 15},
#   {"pattern": ["get_weather", "get_clothing_advice"], "count": 12},
# ]
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
# 步骤 2:抽象提炼 — 把模式转化为 Skill YAML
def pattern_to_skill_yaml(pattern: list[str], count: int) -> str:
    """将高频工具调用模式转化为 Skill 定义模板"""
    skill_name = f"{pattern[0]}-workflow"
    tools_yaml = []
    for i, tool in enumerate(pattern):
        tools_yaml.append(f"""  - name: {tool}
    description: 第{i+1}步:{tool}
    parameters:
      input_from_previous: {{ type: string, default: null }}""")

    knowledge = "## 调用策略\n" + "\n".join(
        f"{i+1}. 调用 {tool}" for i, tool in enumerate(pattern)
    )

    return f"""name: {skill_name}
version: 0.1.0
description: 从 {count} 次项目实践中提炼的工具编排模式
layer: compound
triggers: ["{pattern[0]}"]

tools:
{chr(10).join(tools_yaml)}

knowledge: |
{knowledge}"""

# 输出:code-reviewer 的初始 YAML 定义
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

# 8.10.2 Skill 的版本迭代策略

Skill 版本号遵循语义化版本(SemVer):MAJOR.MINOR.PATCH

版本变化 含义 示例 兼容性
PATCH (x.x.Z) Bug 修复、文档更新 1.0.0 → 1.0.1 完全兼容,用户无感
MINOR (x.Y.z) 新增工具/功能,不破坏现有调用 1.0.1 → 1.1.0 向后兼容,新功能可选
MAJOR (X.y.z) 重构调用策略、删除工具、改变行为 1.1.0 → 2.0.0 不兼容,需迁移
# skills/weather-advisor/CHANGELOG.yml — Skill 版本变更记录
versions:
  - version: 2.0.0
    date: 2025-03-15
    changes:
      - "MAJOR: 重构调用策略,支持多天天气批量查询"
      - "MAJOR: 新增 get_travel_advice 工具"
      - "MINOR: 穿衣建议增加雾霾场景"
    migration_note: "多天查询需要传 date_list 参数,旧 date 参数仍兼容"

  - version: 1.1.0
    date: 2025-01-20
    changes:
      - "MINOR: 新增紫外线指数提醒"
      - "PATCH: 修复 date='tomorrow' 解析错误"

  - version: 1.0.0
    date: 2024-11-01
    changes:
      - "初始稳定版本发布"
      - "包含 get_weather + get_clothing_advice"
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
# 版本管理器:自动处理兼容性
class SkillVersionManager:
    def __init__(self):
        self.installed: dict = {}  # skill_name -> version

    def check_compatibility(self, skill_name: str, required_version: str) -> bool:
        """检查已安装版本是否满足需求"""
        installed = self.installed.get(skill_name)
        if not installed:
            return False
        # SemVer 兼容性检查
        req_major, req_minor, req_patch = map(int, required_version.split("."))
        inst_major, inst_minor, inst_patch = map(int, installed.split("."))
        # MAJOR 必须一致,MINOR >= 要求
        return inst_major == req_major and inst_minor >= req_minor

    def upgrade(self, skill_name: str, target_version: str) -> dict:
        """升级 Skill,返回迁移指南"""
        current = self.installed.get(skill_name)
        if not current:
            return {"error": f"Skill {skill_name} 未安装"}

        # 加载 CHANGELOG,检查是否需要迁移
        changelog = self._load_changelog(skill_name)
        migrations = []
        for entry in changelog:
            if self._version_between(entry["version"], current, target_version):
                if any(c.startswith("MAJOR") for c in entry["changes"]):
                    migrations.append(entry)

        self.installed[skill_name] = target_version
        return {
            "upgraded": f"{current}{target_version}",
            "migration_steps": migrations
        }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

💡 沉淀优先级

不要一开始就追求完美的 Skill 设计。WaLiCode 的 IterativeRefinement 模式告诉我们:先让 Agent 在项目中跑起来,观察它反复使用哪些工具组合,再把这些组合提炼成 Skill。项目实践是最好的 Skill 设计师——你只是把最佳实践固化下来。

# 8.11 Skill 的权限与安全

Skill 越强大,潜在风险越大。一个可以删除文件的 Skill 如果不加限制,Agent 可能误删关键数据。本章引入五层安全架构——参考 WaLiCode 的 permissionGuard 设计模式,给每一层加上护栏:

流程图: Layer 1:声明式权限(YAML 中声明所需权限 network/filesystem/email/payment)→ Layer 2:运行时拦截(permissionGuard 检查每次调用,白名单+黑名单)→ Layer 3:审批机制(危险操作需用户确认 rm/send_email/payment)→ Layer 4:审计日志(记录所有 Skill 调用 who/when/what/result)→ Layer 5:沙箱隔离(危险 Skill 在沙箱中执行,限制网络+文件系统访问)

安全层 机制 拦截时机 示例 WaLiCode 对应
Layer 1 声明式 YAML 中声明 permissions Skill 注册时 network: true, filesystem: write PermissionDeclaration
Layer 2 运行时拦截 permissionGuard 中间件 每次工具调用前 拦截 path_traversal、SSRF RuntimeGuard
Layer 3 审批机制 危险操作暂停,等用户确认 高危操作执行前 删除文件、发送邮件需 /approve ApprovalGate
Layer 4 审计日志 结构化记录每次调用 调用完成后 skill_audit_log 表 AuditTrail
Layer 5 沙箱隔离 Docker/VM 限制资源访问 Skill 执行期间 payment Skill 在沙箱中运行 SandboxExecutor

# 8.11.1 Layer 1:声明式权限

每个 Skill 在 YAML 中声明自己需要哪些权限——这是最小权限原则的起点:

# 声明式权限定义
# skills/file-manager/skill.yml
name: file-manager
version: 1.0.0
description: 文件管理技能,支持读写删除操作
permissions:
  filesystem:
    read: true       # 需要读取文件
    write: true      # 需要写入文件
    delete: false    # 不允许删除(即使有工具也不给权限)
  network: false     # 不需要网络访问
  email: false       # 不需要发送邮件

danger_level: medium  # low / medium / high / critical
requires_approval:    # 哪些操作需要审批
  - filesystem.write  # 写文件需审批
  # filesystem.delete 不在列表中——因为权限已禁止
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

# 8.11.2 Layer 2:运行时拦截(permissionGuard)

声明只是声明,运行时拦截才是真正的防线。permissionGuard 是一个中间件,在每次工具调用前检查权限:

# permissionGuard 中间件实现
class PermissionGuard:
    """WaLiCode 运行时拦截模式——在工具调用前检查权限"""

    # 黑名单:永远禁止的操作
    BLOCKED_OPERATIONS = {
        "rm_rf": "禁止递归删除",
        "eval": "禁止动态代码执行",
        "exec_shell": "禁止直接执行 shell 命令(除非白名单)",
    }

    # 白名单:特定 Skill 的允许操作
    ALLOWED_PER_SKILL = {
        "file-manager": {
            "filesystem.read": True,
            "filesystem.write": True,
            "filesystem.delete": False,
        },
        "weather-advisor": {
            "network.http_get": True,
            "network.http_post": False,
        },
    }

    def check(self, skill_name: str, operation: str, params: dict) -> dict:
        """检查权限,返回 allow/deny/need_approval"""
        # 1. 黑名单检查(最高优先级)
        if operation in self.BLOCKED_OPERATIONS:
            return {
                "status": "denied",
                "reason": self.BLOCKED_OPERATIONS[operation]
            }

        # 2. 白名单检查
        skill_perms = self.ALLOWED_PER_SKILL.get(skill_name, {})
        if not skill_perms.get(operation, False):
            return {
                "status": "denied",
                "reason": f"Skill {skill_name} 未获得 {operation} 权限"
            }

        # 3. 参数级安全检查(如路径穿越、SSRF)
        if "path" in params:
            if ".." in params["path"] or params["path"].startswith("/etc"):
                return {"status": "denied", "reason": "路径穿越攻击被拦截"}

        if "url" in params:
            blocked_hosts = ["localhost", "127.0.0.1", "0.0.0.0", "internal."]
            if any(h in params["url"] for h in blocked_hosts):
                return {"status": "denied", "reason": "SSRF 攻击被拦截"}

        # 4. 审批检查
        if operation in self.NEEDS_APPROVAL.get(skill_name, []):
            return {"status": "need_approval", "operation": operation}

        return {"status": "allow"}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56

# 8.11.3 Layer 3:审批机制(ApprovalGate)

对于高危操作(删除文件、发送邮件、支付),Agent 必须暂停并等待用户确认

# 审批机制实现
class ApprovalGate:
    """WaLiCode ApprovalGate 模式——高危操作需用户确认"""

    def __init__(self):
        self.pending: dict = {}  # approval_id → operation details

    def request_approval(self, skill_name: str, operation: str, params: dict) -> str:
        """请求审批,返回 approval_id"""
        approval_id = f"apr_{skill_name}_{operation}_{int(time.time())}"
        self.pending[approval_id] = {
            "skill": skill_name,
            "operation": operation,
            "params": params,
            "status": "pending",
            "created_at": time.time()
        }
        # 暂停执行,通知用户
        print(f"⚠️ 需要审批: {skill_name} 想执行 {operation}")
        print(f"   参数: {params}")
        print(f"   请回复 /approve {approval_id} 或 /reject {approval_id}")
        return approval_id

    def handle_response(self, approval_id: str, approved: bool) -> dict:
        """处理用户的审批响应"""
        request = self.pending.get(approval_id)
        if not request:
            return {"error": "审批 ID 不存在"}

        if approved:
            request["status"] = "approved"
            # 执行被暂停的操作
            return {"status": "approved", "proceed": True}
        else:
            request["status"] = "rejected"
            return {"status": "rejected", "proceed": False}

# 使用示例
guard_result = permission_guard.check("file-manager", "filesystem.write", {"path": "/data/report.csv"})
if guard_result["status"] == "need_approval":
    approval_id = approval_gate.request_approval("file-manager", "filesystem.write", {"path": "/data/report.csv"})
    # Agent 等待用户响应...
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42

# 8.11.4 Layer 4:审计日志

所有 Skill 调用必须可追溯——谁在什么时候调用了什么、参数是什么、结果是什么:

# 审计日志实现
import sqlite3
from datetime import datetime

class SkillAuditLog:
    """WaLiCode AuditTrail 模式——所有调用可追溯"""

    def __init__(self, db_path: str = "skill_audit.db"):
        self.conn = sqlite3.connect(db_path)
        self.conn.execute("""
            CREATE TABLE IF NOT EXISTS audit_log (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                timestamp TEXT NOT NULL,
                session_id TEXT NOT NULL,
                skill_name TEXT NOT NULL,
                operation TEXT NOT NULL,
                params TEXT NOT NULL,       -- JSON 格式
                result TEXT NOT NULL,        -- JSON 格式
                approved TEXT DEFAULT NULL,  -- approval_id 或 NULL
                risk_level TEXT NOT NULL     -- low/medium/high/critical
            )
        """)

    def log(self, session_id: str, skill_name: str, operation: str,
            params: dict, result: dict, approved: str = None,
            risk_level: str = "low"):
        """记录一次 Skill 调用"""
        self.conn.execute(
            "INSERT INTO audit_log VALUES (NULL, ?, ?, ?, ?, ?, ?, ?, ?)",
            (datetime.now().isoformat(), session_id, skill_name, operation,
             json.dumps(params), json.dumps(result), approved, risk_level)
        )
        self.conn.commit()

    def query(self, skill_name: str = None, risk_level: str = None,
              since: str = None) -> list[dict]:
        """查询审计日志(合规审查用)"""
        conditions = []
        if skill_name:
            conditions.append(f"skill_name = '{skill_name}'")
        if risk_level:
            conditions.append(f"risk_level = '{risk_level}'")
        if since:
            conditions.append(f"timestamp >= '{since}'")

        where = " AND ".join(conditions) if conditions else "1=1"
        rows = self.conn.execute(
            f"SELECT * FROM audit_log WHERE {where} ORDER BY timestamp DESC LIMIT 100"
        ).fetchall()
        return [dict(zip(["id","ts","sid","skill","op","params","result","approved","risk"], r)) for r in rows]

# 查询所有高风险调用(合规审查场景)
high_risk_calls = audit_log.query(risk_level="critical")
# → 返回所有 critical 级别的调用记录,供安全团队审查
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54

# 8.11.5 Layer 5:沙箱隔离

最危险的操作(如支付、访问生产数据库)应该在沙箱中执行——限制网络和文件系统访问:

# 沙箱执行器实现(基于 subprocess + 资源限制)
import subprocess
import tempfile

class SandboxExecutor:
    """WaLiCode SandboxExecutor 模式——危险 Skill 在沙箱中执行"""

    SANDBOX_CONFIG = {
        "payment-skill": {
            "network": "restricted",       # 只允许访问支付 API 域名
            "filesystem": "read_only",     # 只读文件系统
            "max_memory": "256MB",         # 内存限制
            "max_cpu_time": 30,            # CPU 时间限制(秒)
            "allowed_domains": ["api.payment.com"],
        },
        "database-query": {
            "network": "restricted",
            "filesystem": "none",          # 无文件系统访问
            "max_memory": "512MB",
            "max_cpu_time": 60,
            "allowed_domains": ["db.internal.com"],
        }
    }

    def execute_in_sandbox(self, skill_name: str, tool_name: str,
                           params: dict) -> dict:
        """在沙箱中执行 Skill 工具"""
        config = self.SANDBOX_CONFIG.get(skill_name)
        if not config:
            # 无沙箱配置 → 正常执行
            return self._execute_normal(skill_name, tool_name, params)

        # 将调用序列化为临时脚本
        script = self._serialize_call(skill_name, tool_name, params)
        script_path = tempfile.mktemp(suffix=".py")
        with open(script_path, "w") as f:
            f.write(script)

        # 在受限环境中执行
        result = subprocess.run(
            ["python3", script_path],
            capture_output=True,
            timeout=config["max_cpu_time"],
            env={
                "SKILL_SANDBOX": "true",
                "ALLOWED_DOMAINS": ",".join(config["allowed_domains"]),
                "MAX_MEMORY": config["max_memory"],
            }
        )

        return {
            "stdout": result.stdout.decode(),
            "stderr": result.stderr.decode(),
            "exit_code": result.returncode,
            "sandbox": True
        }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56

💡 安全分层原则

不是所有 Skill 都需要五层防护。基础 Skill(如 get_weather)只需 Layer 1(声明)+ Layer 2(运行时拦截)。复合 Skill 加上 Layer 3(审批)+ Layer 4(审计)。领域 Skill 才需要完整的五层——包括沙箱隔离。安全不是越厚越好,而是按风险等级分层

# 8.12 实战:构建企业级 Skill 包

前面讲了 Skill 的分层体系和安全架构,现在把它们串起来——从零构建一个完整的企业级 Skill 包,包括 YAML 定义、Python 实现、测试和发布流程:

流程图: Step 1:YAML 定义(元数据+工具+知识+权限)→ Step 2:Python 实现(工具函数+permissionGuard)→ Step 3:测试(单元测试+Prompt 测试+安全测试)→ Step 4:发布(版本号+Changelog+注册)

# 8.12.1 Step 1:YAML 定义

report-generator(报表生成器)为例——这是一个复合 Skill,需要读取数据、分析数据、生成报表三个工具:

# skills/compound/report-generator/skill.yml
name: report-generator
version: 1.0.0
description: 企业报表生成器,从数据源读取数据并生成格式化报表
layer: compound
triggers: ["报表", "报告", "生成报告", "generate report", "周报", "月报"]

permissions:
  filesystem:
    read: true
    write: true         # 需要写入报表文件
  network:
    http_get: true      # 需要从 API 获取数据
  email:
    send: false         # 不直接发邮件(由上层 Skill 处理)

danger_level: medium
requires_approval:
  - filesystem.write    # 写文件需审批

tools:
  - name: fetch_data
    description: 从指定数据源获取原始数据
    parameters:
      source: { type: string, required: true, description: "数据源(api/file/db)" }
      query: { type: string, default: null, description: "查询条件" }
      format: { type: string, default: "json", description: "返回格式" }

  - name: analyze_data
    description: 分析数据,计算统计指标
    parameters:
      data: { type: array, required: true, description: "原始数据" }
      metrics: { type: array, default: ["sum", "avg", "max", "min"], description: "统计指标" }
      group_by: { type: string, default: null, description: "分组字段" }

  - name: render_report
    description: 将分析结果渲染为格式化报表
    parameters:
      analysis: { type: object, required: true, description: "分析结果" }
      template: { type: string, default: "default", description: "报表模板" }
      output_format: { type: string, default: "markdown", description: "输出格式" }

knowledge: |
  ## 调用策略
  1. 先 fetch_data 从数据源获取数据
  2. 再 analyze_data 计算统计指标
  3. 最后 render_report 生成报表
  4. 如果数据量 > 10万条,建议分批处理
  5. 如果源是 API,注意分页和超时处理

  ## WaLiCode 对应
  - WorkflowChain: fetch → analyze → render
  - ErrorRecovery: fetch 失败时尝试备用数据源
  - PermissionGuard: 写文件前需审批

  ## 异常处理
  - fetch_data 失败 → 提示用户检查数据源配置
  - analyze_data 数据为空 → 返回"数据不足,无法生成报表"
  - render_report 模板缺失 → 使用默认模板
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59

# 8.12.2 Step 2:Python 实现

# skills/compound/report-generator/main.py
import json
import statistics
from pathlib import Path
from datetime import datetime

class ReportGeneratorSkill:
    """企业级报表生成 Skill——带 permissionGuard"""

    def __init__(self, permission_guard: PermissionGuard = None):
        self.guard = permission_guard

    def fetch_data(self, source: str, query: str = None, format: str = "json") -> dict:
        """从数据源获取原始数据"""
        if source.startswith("http"):
            # API 数据源(带超时和分页)
            import requests
            params = {"query": query} if query else {}
            response = requests.get(source, params=params, timeout=30)
            return {"data": response.json(), "source": source}
        elif source.endswith(".csv") or source.endswith(".json"):
            # 文件数据源(带安全检查)
            if self.guard:
                check = self.guard.check("report-generator", "filesystem.read", {"path": source})
                if check["status"] == "denied":
                    return {"error": check["reason"]}
            with open(source, "r", encoding="utf-8") as f:
                if source.endswith(".json"):
                    data = json.load(f)
                else:
                    import csv
                    reader = csv.DictReader(f)
                    data = list(reader)
            return {"data": data, "source": source}
        else:
            return {"error": f"不支持的数据源类型: {source}"}

    def analyze_data(self, data: list, metrics: list = None, group_by: str = None) -> dict:
        """分析数据,计算统计指标"""
        if not data:
            return {"error": "数据为空,无法分析"}

        metrics = metrics or ["sum", "avg", "max", "min"]
        numeric_fields = self._find_numeric_fields(data)
        results = {}

        for field in numeric_fields:
            values = [row.get(field, 0) for row in data if row.get(field) is not None]
            field_stats = {}
            for m in metrics:
                if m == "sum":
                    field_stats["sum"] = sum(values)
                elif m == "avg":
                    field_stats["avg"] = statistics.mean(values) if values else 0
                elif m == "max":
                    field_stats["max"] = max(values) if values else 0
                elif m == "min":
                    field_stats["min"] = min(values) if values else 0
            results[field] = field_stats

        # 分组统计(如果指定 group_by)
        if group_by:
            groups = {}
            for row in data:
                key = row.get(group_by, "unknown")
                groups.setdefault(key, []).append(row)
            results["groups"] = {k: len(v) for k, v in groups.items()}

        return {"analysis": results, "data_count": len(data), "fields": numeric_fields}

    def render_report(self, analysis: dict, template: str = "default",
                      output_format: str = "markdown") -> dict:
        """渲染报表"""
        stats = analysis.get("analysis", {})
        count = analysis.get("data_count", 0)

        # 生成 Markdown 报表
        report = f"# 数据报表\n\n"
        report += f"> 生成时间: {datetime.now().strftime('%Y-%m-%d %H:%M')}\n"
        report += f"> 数据量: {count} 条\n\n"

        report += "## 统计指标\n\n"
        report += "| 字段 | 求和 | 平均值 | 最大值 | 最小值 |\n"
        report += "|------|------|--------|--------|--------|\n"
        for field, field_stats in stats.items():
            if field == "groups":
                continue
            report += f"| {field} | {field_stats.get('sum', '-')} | "
            report += f"{field_stats.get('avg', '-')} | {field_stats.get('max', '-')} | "
            report += f"{field_stats.get('min', '-')} |\n"

        if "groups" in stats:
            report += "\n## 分组统计\n\n"
            for group, group_count in stats["groups"].items():
                report += f"- **{group}**: {group_count} 条\n"

        # 写入文件(需审批)
        output_path = f"reports/report_{datetime.now().strftime('%Y%m%d_%H%M%S')}.md"
        if self.guard:
            check = self.guard.check("report-generator", "filesystem.write", {"path": output_path})
            if check["status"] == "need_approval":
                return {
                    "report": report,
                    "approval_needed": True,
                    "message": f"需要审批才能写入文件: {output_path}"
                }
            elif check["status"] == "denied":
                return {"error": check["reason"]}

        Path(output_path).parent.mkdir(parents=True, exist_ok=True)
        with open(output_path, "w", encoding="utf-8") as f:
            f.write(report)

        return {"report": report, "output_path": output_path, "written": True}

    def _find_numeric_fields(self, data: list) -> list[str]:
        """自动检测数值型字段"""
        numeric = []
        if data:
            for key, val in data[0].items():
                if isinstance(val, (int, float)):
                    numeric.append(key)
        return numeric
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123

# 8.12.3 Step 3:测试

企业级 Skill 需要三层测试:单元测试(工具函数)、Prompt 测试(AI 调用策略)、安全测试(权限拦截):

# tests/test_report_generator.py
import pytest
from skills.compound.report_generator.main import ReportGeneratorSkill
from skills.compound.report_generator.permission_guard import PermissionGuard

class TestReportGenerator:
    """单元测试——验证工具函数逻辑"""

    def setup_method(self):
        self.skill = ReportGeneratorSkill()

    def test_analyze_data_basic(self):
        """测试基本统计分析"""
        data = [
            {"name": "A", "value": 10},
            {"name": "B", "value": 20},
            {"name": "C", "value": 30},
        ]
        result = self.skill.analyze_data(data)
        assert result["data_count"] == 3
        assert result["analysis"]["value"]["sum"] == 60
        assert result["analysis"]["value"]["avg"] == 20

    def test_analyze_data_empty(self):
        """测试空数据返回错误"""
        result = self.skill.analyze_data([])
        assert "error" in result
        assert "数据为空" in result["error"]

    def test_analyze_data_with_groups(self):
        """测试分组统计"""
        data = [
            {"department": "工程", "salary": 15000},
            {"department": "工程", "salary": 18000},
            {"department": "市场", "salary": 12000},
        ]
        result = self.skill.analyze_data(data, group_by="department")
        assert result["analysis"]["groups"]["工程"] == 2
        assert result["analysis"]["groups"]["市场"] == 1

    def test_render_report_markdown(self):
        """测试 Markdown 报表生成"""
        analysis = {
            "analysis": {"salary": {"sum": 45000, "avg": 15000}},
            "data_count": 3,
        }
        result = self.skill.render_report(analysis)
        assert "# 数据报表" in result["report"]
        assert "salary" in result["report"]


class TestPromptIntegration:
    """Prompt 测试——验证 AI 能正确调用 Skill"""

    PROMPT_TEST_CASES = [
        {
            "user_message": "帮我生成上周的销售报表",
            "expected_skill": "report-generator",
            "expected_call_order": ["fetch_data", "analyze_data", "render_report"],
        },
        {
            "user_message": "分析一下各部门的工资分布",
            "expected_skill": "report-generator",
            "expected_call_order": ["fetch_data", "analyze_data"],
        },
    ]

    def test_skill_activation(self):
        """验证触发词能正确激活 Skill"""
        manager = LazySkillManager(loader)
        for case in self.PROMPT_TEST_CASES:
            prompt = manager.get_prompt(case["user_message"])
            assert "report-generator" in prompt
            assert "(已激活)" in prompt


class TestSecurity:
    """安全测试——验证 permissionGuard 拦截"""

    def setup_method(self):
        self.guard = PermissionGuard()
        self.skill = ReportGeneratorSkill(permission_guard=self.guard)

    def test_path_traversal_blocked(self):
        """测试路径穿越被拦截"""
        result = self.skill.fetch_data(source="/etc/passwd")
        assert "error" in result
        assert "拦截" in result["error"]

    def test_write_needs_approval(self):
        """测试写文件需要审批"""
        result = self.skill.render_report({"analysis": {}, "data_count": 0})
        # 在审批机制下,render_report 应返回 need_approval
        assert result.get("approval_needed") == True or "审批" in str(result)

    def test_ssrf_blocked(self):
        """测试 SSRF 被拦截"""
        result = self.skill.fetch_data(source="http://127.0.0.1:8080/internal")
        assert "error" in result
        assert "SSRF" in result["error"] or "拦截" in result["error"]
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100

# 8.12.4 Step 4:发布流程

# 发布流程脚本 — publish_skill.sh
#!/bin/bash
# 企业级 Skill 发布流程(WaLiCode ReleasePipeline 模式)

SKILL_NAME="report-generator"
SKILL_DIR="skills/compound/${SKILL_NAME}"

echo "=== Step 1: 版本检查 ==="
CURRENT_VERSION=$(grep "version:" ${SKILL_DIR}/skill.yml | head -1 | cut -d' ' -f2)
echo "当前版本: ${CURRENT_VERSION}"

echo "=== Step 2: 运行测试 ==="
cd tests && python -m pytest test_report_generator.py -v
if [ $? -ne 0 ]; then
    echo "❌ 测试失败,中止发布"
    exit 1
fi
echo "✅ 所有测试通过"

echo "=== Step 3: 安全审计 ==="
python security_audit.py ${SKILL_NAME}
# 检查: 权限声明完整性、danger_level 评估、是否有 known vulnerabilities

echo "=== Step 4: 生成 Changelog ==="
# 从 CHANGELOG.yml 中提取当前版本的变更记录
python generate_changelog.py ${SKILL_NAME} ${CURRENT_VERSION}

echo "=== Step 5: 打包 ==="
tar -czf ${SKILL_NAME}-${CURRENT_VERSION}.tar.gz \
    ${SKILL_DIR}/skill.yml \
    ${SKILL_DIR}/main.py \
    ${SKILL_DIR}/CHANGELOG.yml \
    tests/test_report_generator.py

echo "=== Step 6: 注册到 Skill 仓库 ==="
# 内部仓库(类似 npm registry)
curl -X POST http://skill-registry.internal.com/api/register \
    -F "name=${SKILL_NAME}" \
    -F "version=${CURRENT_VERSION}" \
    -F "file=@${SKILL_NAME}-${CURRENT_VERSION}.tar.gz"

echo "✅ 发布完成: ${SKILL_NAME} ${CURRENT_VERSION}"

# 安装验证
skill install ${SKILL_NAME}
skill list | grep ${SKILL_NAME}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
发布阶段 检查项 失败处理 WaLiCode 对应
版本检查 版本号是否符合 SemVer 格式错误 → 修正后重试 VersionValidation
运行测试 单元 + Prompt + 安全三层测试 测试失败 → 中止发布 QualityGate
安全审计 权限声明、danger_level、已知漏洞 高危漏洞 → 修复后重新审计 SecurityAudit
打包 YAML + 代码 + 测试 + Changelog 文件缺失 → 补充后重新打包 PackageBuilder
注册 推送到 Skill 仓库 / 市场 网络失败 → 重试 3 次 RegistryPublish
安装验证 skill install → skill list 能看到 安装失败 → 回滚发布 SmokeTest

# 8.13 Skill 与 OpenClaw 的关系

OpenClaw 是一个实际落地的 Agent 平台,它的 Skill 生态是本章理论的最佳实践。让我们从 OpenClaw 的 ~/.qclaw/skills/ 目录出发,看看 Skill 是如何在真实系统中运作的:

# 8.13.1 OpenClaw Skill 目录结构

# OpenClaw 的 Skill 目录扫描
$ ls ~/.qclaw/skills/
another_them/        # 蒸馏 Agent 人设的 Skill
cloud-upload-backup/ # 云端上传备份
docx/                # Word 文档处理
email-skill/         # 邮件统一路由
find-skills/         # Skill 发现与安装
pdf/                 # PDF 处理
persona-switch/      # 人设切换
qclaw-cron-skill/    # 定时任务
qclaw-env/           # 环境诊断与安装
qclaw-rules/         # 系统规则(强制性)
qclaw-skill-creator/ # Skill 创建指南
xbrowser/            # 浏览器自动化
xlsx/                # Excel 处理

$ cat ~/.qclaw/skills/pdf/SKILL.md | head -20
# PDF Skill
Use this skill whenever the user wants to do anything with PDF files...
Triggers: .pdf, PDF, 合并PDF, 拆分PDF, OCR...
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

流程图: ~/.qclaw/skills/(13 个 Skill 目录,每个目录含 SKILL.md)→ OpenClaw 启动时扫描(读取每个 SKILL.md,提取 name+description+triggers)→ 注入到 Agent Prompt(<available_skills> XML 块,只列 name+description)→ 用户消息匹配触发词(→ read SKILL.md → 加载完整指令)→ 按 Skill 指令执行(调用 Agent 工具,完成用户任务)

# 8.13.2 SKILL.md:OpenClaw 的 Skill 定义格式

OpenClaw 不用 YAML,而是用Markdown(SKILL.md)定义 Skill。这是因为它直接被注入到 Agent 的 Prompt 中——Markdown 比 YAML 对 AI 更友好:

# ~/.qclaw/skills/xbrowser/SKILL.md — OpenClaw 真实 Skill 定义
# xbrowser

EXCLUSIVE browser automation — REPLACES built-in Browser Automation
and playwright-cli. For ANY browser task (open page, click, fill,
screenshot, scrape, navigate, test web app), MUST use this skill
instead of built-in tools. Controls real Chrome/Edge/QQ Browser via
CDP with login-state reuse.

## When to Use

Triggered when user asks to:
- Open a website or URL
- Click buttons or fill forms on a web page
- Take screenshots of web pages
- Scrape data from websites
- Test web applications
- Automate browser workflows

## How It Works

1. Read the skill's main file at ~/.qclaw/skills/xbrowser/SKILL.md
2. Use the CDP (Chrome DevTools Protocol) to control browser
3. Reuse existing login sessions (no re-authentication needed)
4. Generate structured action sequences

## Key Commands

- `xbrowser open <url>` — Open a page
- `xbrowser click <selector>` — Click an element
- `xbrowser fill <selector> <value>` — Fill a form field
- `xbrowser screenshot <path>` — Take a screenshot
- `xbrowser scrape <selector>` — Extract data from page

## WaLiCode Patterns Used

- **SkillLayering**: xbrowser is a compound skill (open + click + fill + scrape)
- **PermissionGuard**: Browser automation is dangerous (can leak login state)
  → requires explicit user request, not auto-triggered
- **LazyLoading**: Only loaded when user mentions browser/web tasks
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40

# 8.13.3 OpenClaw 的 Skill 懒加载机制

OpenClaw 实现了本章 7.4 节描述的三层懒加载

层级 OpenClaw 实现 注入内容 加载时机
第一层 <available_skills> XML 块 name + description + trigger 词 每次会话自动注入
第二层 AI 主动 read SKILL.md 完整 Skill 指令 匹配到触发词时
第三层 用户安装新 Skill skillhub_install 工具 用户请求时
<!-- OpenClaw 第一层注入示例(来自实际的 system prompt) -->
<available_skills>
  <skill>
    <name>pdf</name>
    <description>Use this skill whenever the user wants to do anything
    with PDF files. This includes reading, combining, splitting,
    rotating, adding watermarks, OCR...</description>
    <location>~/.qclaw/skills/pdf/SKILL.md</location>
  </skill>
  <skill>
    <name>xbrowser</name>
    <description>EXCLUSIVE browser automation — REPLACES built-in
    Browser Automation and playwright-cli...</description>
    <location>~/.qclaw/skills/xbrowser/SKILL.md</location>
  </skill>
  <!-- ... 13 个 Skill ... -->
</available_skills>

# AI 的决策流程
# 1. 用户说"帮我合并这两个PDF"
# 2. AI 匹配到 pdf Skill(触发词 "PDF")
# 3. AI 执行: read ~/.qclaw/skills/pdf/SKILL.md(第二层加载)
# 4. 按照 SKILL.md 中的指令调用 pdf 工具
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

# 8.13.4 OpenClaw Skill 的分层映射

将 OpenClaw 的 13 个 Skill 映射到本章的三层金字塔:

层级 OpenClaw Skill 原因
基础 Skill qclaw-rules、qclaw-env、qclaw-cron-skill 单功能封装:规则加载、环境安装、定时任务
复合 Skill pdf、docx、xlsx、xbrowser、email-skill 多工具组合:如 pdf = read + merge + split + OCR
领域 Skill another_them、qclaw-skill-creator 领域知识:蒸馏人设、创建 Skill 的方法论

# 8.13.5 OpenClaw Skill 的安全实践

OpenClaw 的 Skill 安全设计直接体现了本章 7.11 节的五层架构:

# OpenClaw 安全实践对照

# Layer 1:声明式权限 — qclaw-rules Skill 的强制性声明
# (SKILL.md 中写道:[SYSTEM RULES - MANDATORY - ALWAYS LOAD - DO NOT SKIP])
# → 这是一种"强制加载"声明,比普通权限声明更严格

# Layer 2:运行时拦截 — elevated 权限机制
# exec 工具参数中有 elevated: true
# → 需要用户 /approve 才能执行提权命令

# Layer 3:审批机制 — /approve 命令
# Agent 请求审批 → 用户回复 /approve &lt;command>
# → 体现了 ApprovalGate 模式

# Layer 4:审计日志 — 会话记录
# 所有工具调用记录在会话历史中
# → 用户可以随时回溯 Agent 做了什么

# Layer 5:沙箱隔离 — sandbox 执行
# exec 工具支持 host: sandbox 参数
# → 危险命令在沙箱中执行

# 具体示例:qclaw-cron-skill 的安全设计
# ~/.qclaw/skills/qclaw-cron-skill/SKILL.md:
# "[MANDATORY - MUST LOAD] 凡是涉及定时/提醒/闹钟/周期执行/
#  打卡/签到/cron/schedule/remind 等需求...必须读取本 skill,
#  严禁凭记忆猜测参数。"
# → Layer 1(声明式权限)+ Layer 3(审批式执行)的混合
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

💡 从 OpenClaw 学什么

OpenClaw 的 Skill 生态证明了本章的核心论点:Skill = 工具 + 知识。每个 SKILL.md 不仅是工具列表,更是使用策略——告诉 AI 什么时候用、怎么用、什么不能用。这是 Skill 区别于普通工具的关键。

🔗 第8章 核心要点

Skill = 工具 + 知识:不只是工具的集合,还包含使用策略、调用顺序、参数约束。

三层结构:元数据层(what)+ 工具层(how)+ 知识层(when/why)。

自动发现:文件系统扫描 + 触发词匹配 + 热重载。

三层懒加载:名称列表 → 触发词激活 → AI 主动搜索,控制 Prompt 长度。

三者关系:Function Calling(底层)→ MCP(传输层)→ Skills(组织层)。

分层金字塔:基础 Skill(单工具)→ 复合 Skill(多工具组合)→ 领域 Skill(行业知识包),自下而上构建。

沉淀与演进:从项目实践中提炼 Skill,遵循 SemVer 版本迭代,模式识别 → 抽象提炼 → 验证打磨 → 发布复用。

五层安全架构:声明式权限 → 运行时拦截 → 审批机制 → 审计日志 → 沙箱隔离,按风险等级分层。

OpenClaw 实践:SKILL.md 格式 + 三层懒加载 + 五层安全,是本章理论的落地验证。

# 面试八股

Q1:Skill 和 Tool 的区别?

A: Tool 是单一操作,Skill 是工具的组合 + 使用知识。Tool 是函数,Skill 是模块。Skill 不仅包含工具,还包含调用策略、参数约束和使用时机。

Q2:如何让 Agent 发现 Skills?

A: 文件系统扫描 + YAML 元数据 + 触发词匹配 + 热重载。

Q3:Skills 太多导致 Prompt 过长怎么办?

A: 三层懒加载——始终加载名称、触发词匹配时加载完整定义、AI 主动搜索罕见 Skill。这样可以在 Skill 数量很多时控制 Prompt 长度。

Q4:Skills 和 MCP 的关系?

A: 层次不同。MCP 是工具级标准协议,Skills 是能力级组织单元。一个 Skill 可以包含 MCP 工具。MCP 连接 Agent 和工具,Skills 组织工具为能力包。

Q5:Skill 的三层金字塔是什么?

A: 基础 Skill(单工具封装,如 read_file)→ 复合 Skill(多工具组合 + 调用策略,如 code-reviewer)→ 领域 Skill(行业知识包,如 finance-advisor)。底层越通用、顶层越专业,应自下而上构建。

Q6:Skill 如何从项目实践中沉淀?

A: 四步流程——模式识别(从日志提取重复模式)→ 抽象提炼(转化为 Skill YAML)→ 验证打磨(在多项目试用)→ 发布复用(版本号 + Changelog)。版本遵循 SemVer:PATCH 修复、MINOR 新增、MAJOR 重构。

Q7:Skill 的五层安全架构是什么?

A: Layer 1 声明式权限(YAML 中声明)→ Layer 2 运行时拦截(permissionGuard 中间件)→ Layer 3 审批机制(高危操作需 /approve)→ Layer 4 审计日志(所有调用可追溯)→ Layer 5 沙箱隔离(危险 Skill 限制资源)。不是所有 Skill 都需要五层,按风险等级分层。

# 课后练习

# 第 1 题(单选)

Skill 和 Tool 的核心区别是?

A. Skill 就是 Tool 的另一个名字

B. Tool 是单一操作,Skill 是工具的组合 + 使用知识。Tool 是函数,Skill 是模块

C. Skill 比 Tool 速度更快

D. Tool 只能用在 Agent 中,Skill 只能用在 ChatBot 中

答案与解析

答案: B

解析: Tool 是单一操作(函数级),Skill 是工具的组合 + 使用知识(模块级)。Skill 不仅包含工具,还包含调用策略、参数约束和使用时机。

# 第 2 题(单选)

Skills 三层懒加载策略中,始终加载的是?

A. Skill 的完整定义

B. Skill 的名称

C. Skill 的所有工具实现

D. Skill 的知识库

答案与解析

答案: B

解析: 三层懒加载:始终加载名称 → 触发词匹配时加载完整定义 → AI 主动搜索罕见 Skill。这样可以在 Skill 数量很多时控制 Prompt 长度。

# 第 3 题(单选)

Skills 和 MCP 的关系是?

A. Skills 和 MCP 是同一回事

B. MCP 是工具级标准协议,Skills 是能力级组织单元,一个 Skill 可以包含 MCP 工具

C. MCP 替代了 Skills

D. Skills 替代了 MCP

答案与解析

答案: B

解析: 层次不同:MCP 是工具级标准协议(连接 Agent 和工具),Skills 是能力级组织单元(工具组合 + 知识)。一个 Skill 可以包含多个 MCP 工具。

# 第 4 题(单选)

Skill 的三层金字塔从底到顶是?

A. 领域 Skill → 复合 Skill → 基础 Skill

B. 基础 Skill → 复合 Skill → 领域 Skill

C. 复合 Skill → 基础 Skill → 领域 Skill

D. 领域 Skill → 基础 Skill → 复合 Skill

答案与解析

答案: B

解析: 基础 Skill(单工具封装,如 read_file)→ 复合 Skill(多工具组合 + 调用策略,如 code-reviewer)→ 领域 Skill(行业知识包,如 finance-advisor)。底层越通用、顶层越专业。

# 第 5 题(单选)

Skill 版本号遵循 SemVer 规范,修复 Bug 应递增哪个版本号?

A. MAJOR(主版本号)

B. MINOR(次版本号)

C. PATCH(补丁号)

D. BUILD(构建号)

答案与解析

答案: C

解析: SemVer 规范:PATCH 修复(向后兼容的 Bug 修复)、MINOR 新增(向后兼容的新功能)、MAJOR 重构(不兼容的 API 变更)。

# 第 6 题(多选)

Skill 从项目实践中沉淀的四步流程是?(多选)

A. 模式识别(从日志提取重复模式)

B. 抽象提炼(转化为 Skill YAML)

C. 验证打磨(在多项目试用)

D. 发布复用(版本号 + Changelog)

答案与解析

答案: ABCD

解析: 四步流程:模式识别 → 抽象提炼 → 验证打磨 → 发布复用。每一步都不可省略,确保 Skill 的质量和可复用性。

# 第 7 题(多选)

Skill 的五层安全架构包括哪些?(多选)

A. 声明式权限(YAML 中声明)

B. 运行时拦截(permissionGuard 中间件)

C. 审批机制(高危操作需 /approve)

D. 审计日志(所有调用可追溯)

E. 沙箱隔离(危险 Skill 限制资源)

答案与解析

答案: ABCDE

解析: 五层安全架构:Layer 1 声明式权限 → Layer 2 运行时拦截 → Layer 3 审批机制 → Layer 4 审计日志 → Layer 5 沙箱隔离。不是所有 Skill 都需要五层,按风险等级分层。

# 第 8 题(多选)

以下哪些是 Skill 发现与加载的机制?(多选)

A. 文件系统扫描

B. YAML 元数据解析

C. 触发词匹配

D. 热重载

E. AI 自动生成 Skill

答案与解析

答案: ABCD

解析: Skill 发现与加载机制:文件系统扫描 + YAML 元数据 + 触发词匹配 + 热重载。选项 E 不属于发现机制,AI 不会自动生成 Skill(需要人工创建)。