21188 字
约 70 分钟
1
第12章 Skills-工具的组合与复用
2026-09-18
第12章 Skills-工具的组合与复用
来源：https://ai-agent-guide.xiaofuge.cn/chapters/ch08-skills.html 所属：第三篇-Agent的手脚
第三篇：Agent 的手脚 📌 本章目标
第10章讲了 Function Calling 工具调用，第11章讲了工具的标准化协议（MCP）。本章讲工具的组合——把多个工具打包成一个"技能"，让 Agent 拥有更高层次的能力。
12.1 从工具到技能
打个比方：工具是一把锤子，技能是"钉钉子"——锤子只是一个工具，但"钉钉子"需要锤子 + 钉子 + 判断角度 + 控制力度。技能是工具的组合 + 使用知识。 | 维度 | Tool（工具） | Skill（技能） | | --- | --- | --- | | 粒度 | 单一操作（get_weather） | 复合能力（天气查询+行程规划+穿衣建议） | | 工具数量 | 1 个 | 1~N 个 | | 包含知识 | 无 | 使用策略、调用顺序、参数约束 | | 可复用性 | 低（每个项目重新定义） | 高（打包成包，跨项目共享） | | 类比 | 函数 | 类/模块 | ## 12.2 Skills 的三层结构
一个完整的 Skill 包含三层：
# 一个 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
12.3 Skills 的发现与加载
Agent 怎么知道有哪些 Skills 可用？和 CLI 工具类似，Skills 也需要自动发现机制：
12.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)
import * as fs from 'fs';
import * as path from 'path';
import * as yaml from 'js-yaml';
class SkillLoader {
private registry: Map = new Map();
constructor(private skillsDir: string = 'skills') {}
scan(): void {
if (!fs.existsSync(this.skillsDir)) return;
for (const entry of fs.readdirSync(this.skillsDir, { withFileTypes: true })) {
if (!entry.isDirectory()) continue;
const skillFile = path.join(this.skillsDir, entry.name, 'skill.yml');
if (!fs.existsSync(skillFile)) continue;
const skillDef = yaml.load(fs.readFileSync(skillFile, 'utf-8')) as any;
skillDef.path = path.join(this.skillsDir, entry.name);
this.registry.set(skillDef.name, skillDef);
}
console.log(`加载了 ${this.registry.size} 个技能: ${[...this.registry.keys()]}`);
}
getSkillPrompt(): string {
if (this.registry.size === 0) return '';
const lines: string[] = ['你可以使用以下技能：'];
for (const [name, skill] of this.registry) {
const triggers = (skill.triggers || []).join(', ');
lines.push(`- ${name}: ${skill.description} (触发词: ${triggers})`);
}
return lines.join('\n');
}
}
package main
import (
"fmt"
"os"
"path/filepath"
"gopkg.in/yaml.v3"
)
type SkillLoader struct {
SkillsDir string
Registry map[string]map[string]interface{}
}
func NewSkillLoader(skillsDir string) *SkillLoader {
return &SkillLoader{
SkillsDir: skillsDir,
Registry: make(map[string]map[string]interface{}),
}
}
func (sl *SkillLoader) Scan() error {
entries, err := os.ReadDir(sl.SkillsDir)
if err != nil {
if os.IsNotExist(err) {
return nil
}
return err
}
for _, entry := range entries {
if !entry.IsDir() {
continue
}
skillFile := filepath.Join(sl.SkillsDir, entry.Name(), "skill.yml")
data, err := os.ReadFile(skillFile)
if err != nil {
continue
}
var skillDef map[string]interface{}
if err := yaml.Unmarshal(data, &skillDef); err != nil {
continue
}
skillDef["path"] = filepath.Join(sl.SkillsDir, entry.Name())
if name, ok := skillDef["name"].(string); ok {
sl.Registry[name] = skillDef
}
}
keys := make([]string, 0, len(sl.Registry))
for k := range sl.Registry {
keys = append(keys, k)
}
fmt.Printf("加载了 %d 个技能: %v\n", len(sl.Registry), keys)
return nil
}
func (sl *SkillLoader) GetSkillPrompt() string {
if len(sl.Registry) == 0 {
return ""
}
lines := []string{"你可以使用以下技能："}
for name, skill := range sl.Registry {
triggers, _ := skill["triggers"].([]interface{})
triggerStrs := make([]string, len(triggers))
for i, t := range triggers {
triggerStrs[i] = fmt.Sprintf("%v", t)
}
desc, _ := skill["description"].(string)
lines = append(lines, fmt.Sprintf("- %s: %s (触发词: %s)", name, desc, joinStrings(triggerStrs, ", ")))
}
return joinStrings(lines, "\n")
}
func joinStrings(strs []string, sep string) string {
result := ""
for i, s := range strs {
if i > 0 {
result += sep
}
result += s
}
return result
}
import java.io.*;
import java.nio.file.*;
import java.util.*;
import org.yaml.snakeyaml.Yaml;
public class SkillLoader {
private String skillsDir;
private Map> registry = new HashMap<>();
public SkillLoader(String skillsDir) {
this.skillsDir = skillsDir;
}
public void scan() {
Path dirPath = Paths.get(skillsDir);
if (!Files.exists(dirPath)) return;
try (var stream = Files.newDirectoryStream(dirPath)) {
for (var entry : stream) {
if (!Files.isDirectory(entry)) continue;
Path skillFile = entry.resolve("skill.yml");
if (!Files.exists(skillFile)) continue;
try (var reader = Files.newBufferedReader(skillFile)) {
Yaml yaml = new Yaml();
Map skillDef = yaml.load(reader);
skillDef.put("path", entry.toString());
String name = (String) skillDef.get("name");
registry.put(name, skillDef);
}
}
System.out.printf("加载了 %d 个技能: %s%n", registry.size(), registry.keySet());
} catch (IOException e) {
e.printStackTrace();
}
}
public String getSkillPrompt() {
if (registry.isEmpty()) return "";
StringBuilder sb = new StringBuilder("你可以使用以下技能：\n");
for (var entry : registry.entrySet()) {
String name = entry.getKey();
Map skill = entry.getValue();
List triggers = (List) skill.getOrDefault("triggers", Collections.emptyList());
String desc = (String) skill.get("description");
sb.append(String.format("- %s: %s (触发词: %s)%n", name, desc, String.join(", ", triggers)));
}
return sb.toString();
}
}
12.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()
import * as chokidar from 'chokidar';
class SkillHotReloader {
constructor(private loader: SkillLoader) {}
start(): void {
chokidar.watch('skills/', {
ignored: /(^|[/\\])\./,
persistent: true
}).on('change', (filePath) => {
if (filePath.endsWith('skill.yml')) {
console.log('检测到 Skill 变更，重新加载...');
this.loader.scan();
}
});
console.log('热重载监控已启动');
}
}
// 启动热重载监控
const reloader = new SkillHotReloader(loader);
reloader.start();
package main
import (
"fmt"
"path/filepath"
"time"
"github.com/fsnotify/fsnotify"
)
type SkillHotReloader struct {
Loader *SkillLoader
}
func NewSkillHotReloader(loader *SkillLoader) *SkillHotReloader {
return &SkillHotReloader{Loader: loader}
}
func (r *SkillHotReloader) Start() error {
watcher, err := fsnotify.NewWatcher()
if err != nil {
return err
}
defer watcher.Close()
err = watcher.Add("skills/")
if err != nil {
return err
}
fmt.Println("热重载监控已启动")
for {
select {
case event := {
while (true) {
try {
WatchKey key = watchService.take();
for (WatchEvent event : key.pollEvents()) {
Path changedFile = (Path) event.context();
if (changedFile.toString().endsWith("skill.yml")) {
System.out.println("检测到 Skill 变更，重新加载...");
loader.scan();
}
}
key.reset();
} catch (InterruptedException e) {
break;
}
}
}).start();
}
}
// 启动热重载监控
SkillHotReloader reloader = new SkillHotReloader(loader);
reloader.start();
12.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', '')}"
class LazySkillManager { private activated: Set = new Set();
constructor(private loader: SkillLoader) {}
getPrompt(userMessage: string = ''): string { const lines: string[] = [];
// 第一层：始终列出所有 Skill 名称（简短）
for (const [name, skill] of this.loader.registry) {
const triggers: string[] = skill.triggers || [];
// 第二层：触发词匹配 → 激活完整定义
if (triggers.some(t => userMessage.includes(t))) {
this.activated.add(name);
lines.push(this.formatFullSkill(skill));
} else {
// 第一层：只注入名称和描述
lines.push(- ${name}: ${skill.description});
}
}
// 第三层：已激活但用户消息不匹配的，保持激活
for (const name of this.activated) {
const alreadyListed = lines.some(l => l.startsWith(- ${name}:));
if (!alreadyListed) {
const skill = this.loader.registry.get(name);
if (skill) lines.push(this.formatFullSkill(skill));
}
}
return '\n你可以使用以下技能：\n' + lines.join('\n'); }
private formatFullSkill(skill: any): string {
const toolsDesc = (skill.tools || [])
.map((t: any) => - ${t.name}: ${t.description})
.join('\n');
return - ${skill.name} (已激活)\n 工具:\n${toolsDesc}\n 策略:\n${skill.knowledge || ''};
}
}
package main
import ( "fmt" "strings" )
type LazySkillManager struct { Loader *SkillLoader Activated map[string]bool }
func NewLazySkillManager(loader *SkillLoader) *LazySkillManager { return &LazySkillManager{ Loader: loader, Activated: make(map[string]bool), } }
func (m *LazySkillManager) GetPrompt(userMessage string) string { var lines []string
// 第一层：始终列出所有 Skill 名称（简短）
for name, skill := range m.Loader.Registry {
triggers, _ := skill["triggers"].([]interface{}) matched := false for _, t := range triggers { if strings.Contains(userMessage, fmt.Sprintf("%v", t)) { matched = true break } } if matched { m.Activated[name] = true lines = append(lines, m.formatFullSkill(skill)) } else { desc, _ := skill["description"].(string) lines = append(lines, fmt.Sprintf("- %s: %s", name, desc)) } }
// 第三层：已激活但用户消息不匹配的，保持激活
for name := range m.Activated {
alreadyListed := false for _, l := range lines { if strings.HasPrefix(l, fmt.Sprintf("- %s:", name)) { alreadyListed = true break } } if !alreadyListed { skill, ok := m.Loader.Registry[name] if ok { lines = append(lines, m.formatFullSkill(skill)) } } }
return "\n你可以使用以下技能：\n" + strings.Join(lines, "\n")
}
func (m *LazySkillManager) formatFullSkill(skill map[string]interface{}) string { tools, _ := skill["tools"].([]interface{}) var toolLines []string for _, t := range tools { tool, _ := t.(map[string]interface{}) toolLines = append(toolLines, fmt.Sprintf(" - %v: %v", tool["name"], tool["description"])) } knowledge, _ := skill["knowledge"].(string) return fmt.Sprintf("- %v (已激活)\n 工具:\n%s\n 策略:\n%s", skill["name"], strings.Join(toolLines, "\n"), knowledge) }
import java.util.*;
public class LazySkillManager { private SkillLoader loader; private Set activated = new HashSet<>();
public LazySkillManager(SkillLoader loader) { this.loader = loader; }
public String getPrompt(String userMessage) { List lines = new ArrayList<>();
// 第一层：始终列出所有 Skill 名称（简短） for (var entry : loader.getRegistry().entrySet()) { String name = entry.getKey(); Map skill = entry.getValue(); List triggers = (List) skill.getOrDefault("triggers", Collections.emptyList()); boolean matched = triggers.stream().anyMatch(userMessage::contains); if (matched) { activated.add(name); lines.add(formatFullSkill(skill)); } else { lines.add(String.format("- %s: %s", name, skill.get("description"))); } }
// 第三层：已激活但用户消息不匹配的，保持激活 for (String name : activated) { boolean alreadyListed = lines.stream().anyMatch(l -> l.startsWith("- " + name + ":")); if (!alreadyListed) { Map skill = loader.getRegistry().get(name); if (skill != null) lines.add(formatFullSkill(skill)); } }
return "\n你可以使用以下技能：\n" + String.join("\n", lines); }
private String formatFullSkill(Map skill) { List> tools = (List>) skill.getOrDefault("tools", Collections.emptyList()); String toolsDesc = tools.stream() .map(t -> String.format(" - %s: %s", t.get("name"), t.get("description"))) .reduce((a, b) -> a + "\n" + b) .orElse(""); return String.format("- %s (已激活)\n 工具:\n%s\n 策略:\n%s", skill.get("name"), toolsDesc, skill.getOrDefault("knowledge", "")); } }
## 12.5 Skill 匹配与调度
当用户说"北京明天天气怎么样，穿什么合适？"，Agent 需要：
- 匹配到 `weather-advisor` Skill（触发词"天气"）
- 激活该 Skill 的完整定义
- 按知识层策略执行：先调 get_weather，再调 get_clothing_advice
- 综合两个工具的结果，生成自然语言回答
## 12.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 优化建议工具，以及"先分析表结构再优化查询"的使用策略。
## 12.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: | ## 使用策略
- 总是先调用 get_weather 获取天气
- 如果用户问穿衣/出行建议，调用 get_clothing_advice
- 主动建议：
- 雨天 → 带伞
- 30°C → 防暑
- 雾霾 → 口罩
- 如果查多天天气，逐天调用 get_weather
对应的 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 { // ... 同第2章实现 ... }
function getClothingAdvice(temperature: number, weather: string): { advice: string } { const advice: string[] = []; if (temperature getWeather(String city, String date) { // ... 同第2章实现 ... return null; }
public static Map getClothingAdvice(double temperature, String weather) { List advice = new ArrayList<>(); if (temperature result = new HashMap<>(); result.put("advice", String.join("、", advice)); return result; } }
## 12.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
这和 MCP 的愿景类似，但层次不同：MCP 是工具级的市场，Skills 是能力级的市场。未来可能两者融合——MCP Server 提供工具，Skill 包提供组合策略。
## 12.9 Skill 的分层体系
就像企业架构有基础设施层、业务逻辑层、应用层一样，Skill 也存在**三层金字塔**——从底层的基础 Skill 到顶层的领域 Skill，每一层封装不同复杂度的能力： | 维度 | 基础 Skill | 复合 Skill | 领域 Skill | | --- | --- | --- | --- | | 工具数量 | 1 | 2~5 | 5~20+（含子 Skill） | | 知识含量 | 无（纯工具映射） | 中等（调用策略） | 高（行业规则 + 合规约束） | | 触发方式 | AI 自主调用 | 触发词匹配 | 领域关键词 + 上下文推理 | | 复用范围 | 跨项目通用 | 跨团队复用 | 跨行业复用（需定制） | | 类比 WaLiCode | ToolDefinition 单工具 | WorkflowChain 多步骤编排 | DomainPackage 行业配置包 | | 验证方式 | 单元测试 | 集成测试 + Prompt 测试 | 端到端场景测试 + 合规审计 | ### 12.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 分段读取"
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)}
// skills/base/read_file/main.ts import * as fs from 'fs'; import * as path from 'path';
class ReadFileSkill { /** 基础 Skill：单工具封装，侧重防御性编程 */
private readonly ALLOWED_EXTENSIONS = new Set([ '.txt', '.md', '.py', '.json', '.yaml', '.yml', '.csv', '.jpg', '.png' ]); private readonly MAX_FILE_SIZE = 10 * 1024 * 1024; // 10MB
execute(filePath: string, offset?: number, limit?: number): Record { // 1. 路径穿越防御 const resolved = path.resolve(filePath); if (resolved.includes('..') || !resolved.startsWith(process.cwd())) { return { error: '路径穿越攻击被拦截，禁止访问' }; }
// 2. 扩展名白名单
const ext = path.extname(resolved).toLowerCase();
if (!this.ALLOWED_EXTENSIONS.has(ext)) {
return { error: 不支持的文件类型: ${ext} };
}
// 3. 文件大小限制
const stats = fs.statSync(resolved);
if (stats.size > this.MAX_FILE_SIZE) {
return { error: 文件超过 ${this.MAX_FILE_SIZE / 1024 / 1024}MB 限制 };
}
// 4. 执行核心逻辑 if (['.jpg', '.png'].includes(ext)) { return { type: 'image', path: resolved }; }
const content = fs.readFileSync(resolved, 'utf-8'); let lines = content.split('\n'); if (offset) { lines = lines.slice(offset - 1); // 1-indexed } if (limit) { lines = lines.slice(0, limit); } return { type: 'text', content: lines.join('\n'), lines: lines.length }; } }
package main
import ( "fmt" "os" "path/filepath" "strings" )
type ReadFileSkill struct { AllowedExtensions map[string]bool MaxFileSize int64 }
func NewReadFileSkill() *ReadFileSkill { return &ReadFileSkill{ AllowedExtensions: map[string]bool{ ".txt": true, ".md": true, ".py": true, ".json": true, ".yaml": true, ".yml": true, ".csv": true, ".jpg": true, ".png": true, }, MaxFileSize: 10 * 1024 * 1024, // 10MB } }
func (s *ReadFileSkill) Execute(pathStr string, offset, limit int) map[string]interface{} { // 1. 路径穿越防御 resolved, _ := filepath.Abs(pathStr) cwd, _ := os.Getwd() if strings.Contains(resolved, "..") || !strings.HasPrefix(resolved, cwd) { return map[string]interface{}{"error": "路径穿越攻击被拦截，禁止访问"} }
// 2. 扩展名白名单
ext := filepath.Ext(resolved)
if !s.AllowedExtensions[ext] {
return map[string]interface{}{"error": fmt.Sprintf("不支持的文件类型: %s", ext)} }
// 3. 文件大小限制
info, err := os.Stat(resolved)
if err != nil {
return map[string]interface{}{"error": err.Error()} } if info.Size() > s.MaxFileSize { return map[string]interface{}{"error": fmt.Sprintf("文件超过 %dMB 限制", s.MaxFileSize/1024/1024)} }
// 4. 执行核心逻辑
if ext == ".jpg" || ext == ".png" {
return map[string]interface{}{"type": "image", "path": resolved} }
data, err := os.ReadFile(resolved)
if err != nil {
return map[string]interface{}{"error": err.Error()} } lines := strings.Split(string(data), "\n") if offset > 0 { lines = lines[offset-1:] } if limit > 0 && limit ALLOWED_EXTENSIONS = Set.of( ".txt", ".md", ".py", ".json", ".yaml", ".yml", ".csv", ".jpg", ".png" ); private static final long MAX_FILE_SIZE = 10 * 1024 * 1024; // 10MB
public Map execute(String path, Integer offset, Integer limit) { // 1. 路径穿越防御 Path resolved = Paths.get(path).toAbsolutePath().normalize(); String cwd = System.getProperty("user.dir"); if (resolved.toString().contains("..") || !resolved.startsWith(cwd)) { return Map.of("error", "路径穿越攻击被拦截，禁止访问"); }
// 2. 扩展名白名单 String ext = resolved.toString().substring(resolved.toString().lastIndexOf(".")); if (!ALLOWED_EXTENSIONS.contains(ext)) { return Map.of("error", "不支持的文件类型: " + ext); }
try { // 3. 文件大小限制 long size = Files.size(resolved); if (size > MAX_FILE_SIZE) { return Map.of("error", "文件超过 " + (MAX_FILE_SIZE / 1024 / 1024) + "MB 限制"); }
// 4. 执行核心逻辑 if (ext.equals(".jpg") || ext.equals(".png")) { return Map.of("type", "image", "path", resolved.toString()); }
List allLines = Files.readAllLines(resolved); int start = offset != null ? offset - 1 : 0; int end = limit != null ? Math.min(start + limit, allLines.size()) : allLines.size(); List lines = allLines.subList(start, end); return Map.of("type", "text", "content", String.join("\n", lines), "lines", lines.size()); } catch (IOException e) { return Map.of("error", e.getMessage()); } } }
### 12.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: | ## 调用策略
- 先 read_file 读取目标代码
- 并行调用 lint_check + security_scan
- 汇总结果 → generate_report
- 严重安全问题 → 优先报告（severity=high）
- 如果是 Python 文件，优先用 pylint；JS 用 eslint
WaLiCode 对应
- WorkflowChain 模式：read → [lint + security] → report
- ParallelGateway 模式：lint 和 security 可并行
### 12.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
**💡 金字塔原则**
底层越通用越好（read_file 跨所有项目可用），顶层越专业越好（finance-advisor 只在金融场景有用）。企业 Agent 应**自下而上构建**——先把基础 Skill 打磨好，再组装复合 Skill，最后才封装领域 Skill。不要跳层——没有稳固的基础 Skill，复合 Skill 就是空中楼阁。
## 12.10 Skill 的沉淀与演进
Skill 不是一次设计就完成的，它像代码一样有**版本迭代**的生命周期。好的 Skill 来自项目实践的**反复提炼**——从临时脚本 → 内部最佳实践 → 可复用 Skill 包。
### 12.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：模式识别 — 从项目日志中提取重复模式
import * as re from 'regex';
interface PatternResult {
pattern: string[];
count: number;
}
function extractPatterns(executionLogs: string[]): PatternResult[] {
/** 从 Agent 执行日志中识别重复的工具调用模式 */
// 解析工具调用序列
const sequences: string[][] = [];
let currentSeq: string[] = [];
for (const log of executionLogs) {
if (log.includes('tool_call:')) {
const match = log.match(/tool_call: (\w+)/);
if (match) currentSeq.push(match[1]);
} else if (log.includes('response:')) {
if (currentSeq.length > 0) {
sequences.push([...currentSeq]);
currentSeq = [];
}
}
}
// 统计高频模式（长度 >= 2）
const counter = new Map();
for (const seq of sequences) {
if (seq.length >= 2) {
const key = seq.join(' → ');
counter.set(key, (counter.get(key) || 0) + 1);
}
}
return Array.from(counter.entries())
.map(([key, count]) => ({ pattern: key.split(' → '), count }))
.sort((a, b) => b.count - a.count)
.slice(0, 5);
}
// 示例输出
const patterns = extractPatterns(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 },
// ]
package main
import (
"fmt"
"regexp"
"sort"
"strings"
)
// 步骤 1：模式识别 — 从项目日志中提取重复模式
func ExtractPatterns(executionLogs []string) []map[string]interface{} {
/** 从 Agent 执行日志中识别重复的工具调用模式 */
// 解析工具调用序列
var sequences [][]string
var currentSeq []string
toolCallRegex := regexp.MustCompile(`tool_call: (\w+)`)
for _, log := range executionLogs {
if strings.Contains(log, "tool_call:") {
match := toolCallRegex.FindStringSubmatch(log)
if len(match) > 1 {
currentSeq = append(currentSeq, match[1])
}
} else if strings.Contains(log, "response:") {
if len(currentSeq) > 0 {
sequences = append(sequences, currentSeq)
currentSeq = nil
}
}
}
// 统计高频模式（长度 >= 2）
counter := make(map[string]int)
patternMap := make(map[string][]string)
for _, seq := range sequences {
if len(seq) >= 2 {
key := strings.Join(seq, " → ")
counter[key]++
patternMap[key] = seq
}
}
type kv struct {
key string
count int
}
var sorted []kv
for k, v := range counter {
sorted = append(sorted, kv{k, v})
}
sort.Slice(sorted, func(i, j int) bool {
return sorted[i].count > sorted[j].count
})
var result []map[string]interface{}
for i, kv := range sorted {
if i >= 5 {
break
}
result = append(result, map[string]interface{}{
"pattern": patternMap[kv.key],
"count": kv.count,
})
}
return result
}
// 示例输出
// patterns := ExtractPatterns(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},
// ]
import java.util.*;
import java.util.regex.*;
import java.util.stream.*;
// 步骤 1：模式识别 — 从项目日志中提取重复模式
public class PatternExtractor {
public static List> extractPatterns(List executionLogs) {
/** 从 Agent 执行日志中识别重复的工具调用模式 */
// 解析工具调用序列
List> sequences = new ArrayList<>();
List currentSeq = new ArrayList<>();
Pattern toolCallPattern = Pattern.compile("tool_call: (\\\\w+)");
for (String log : executionLogs) {
if (log.contains("tool_call:")) {
Matcher matcher = toolCallPattern.matcher(log);
if (matcher.find()) {
currentSeq.add(matcher.group(1));
}
} else if (log.contains("response:")) {
if (!currentSeq.isEmpty()) {
sequences.add(new ArrayList<>(currentSeq));
currentSeq.clear();
}
}
}
// 统计高频模式（长度 >= 2）
Map counter = new HashMap<>();
Map> patternMap = new HashMap<>();
for (List seq : sequences) {
if (seq.size() >= 2) {
String key = String.join(" → ", seq);
counter.merge(key, 1, Integer::sum);
patternMap.put(key, seq);
}
}
return counter.entrySet().stream()
.sorted((a, b) -> b.getValue().compareTo(a.getValue()))
.limit(5)
.map(e -> {
Map m = new LinkedHashMap<>();
m.put("pattern", patternMap.get(e.getKey()));
m.put("count", e.getValue());
return m;
})
.collect(Collectors.toList());
}
}
// 示例输出
// List> patterns = PatternExtractor.extractPatterns(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},
// ]
# 步骤 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 定义
// 步骤 2：抽象提炼 — 把模式转化为 Skill YAML
function patternToSkillYaml(pattern: string[], count: number): string {
/** 将高频工具调用模式转化为 Skill 定义模板 */
const skillName = `${pattern[0]}-workflow`;
const toolsYaml: string[] = [];
for (let i = 0; i `${i + 1}. 调用 ${tool}`).join('\n');
return `name: ${skillName}
version: 0.1.0
description: 从 ${count} 次项目实践中提炼的工具编排模式
layer: compound
triggers: ["${pattern[0]}"]
tools:
${toolsYaml.join('\n')}
knowledge: | ${knowledge}`;
}
// 输出：code-reviewer 的初始 YAML 定义
package main
import (
"fmt"
"strings"
)
// 步骤 2：抽象提炼 — 把模式转化为 Skill YAML
func PatternToSkillYaml(pattern []string, count int) string {
/** 将高频工具调用模式转化为 Skill 定义模板 */
skillName := fmt.Sprintf("%s-workflow", pattern[0])
var toolsYaml []string
for i, tool := range pattern {
toolsYaml = append(toolsYaml, fmt.Sprintf(
" - name: %s\n description: 第%d步：%s\n parameters:\n input_from_previous: { type: string, default: null }",
tool, i+1, tool,
))
}
var steps []string
for i, tool := range pattern {
steps = append(steps, fmt.Sprintf("%d. 调用 %s", i+1, tool))
}
knowledge := "## 调用策略\n" + strings.Join(steps, "\n")
return fmt.Sprintf(`name: %s
version: 0.1.0
description: 从 %d 次项目实践中提炼的工具编排模式
layer: compound
triggers: ["%s"]
tools:
%s
knowledge: | %s`, skillName, count, pattern[0], strings.Join(toolsYaml, "\n"), knowledge)
}
// 输出：code-reviewer 的初始 YAML 定义
import java.util.*;
// 步骤 2：抽象提炼 — 把模式转化为 Skill YAML
public class PatternToYaml {
public static String patternToSkillYaml(List pattern, int count) {
/** 将高频工具调用模式转化为 Skill 定义模板 */
String skillName = pattern.get(0) + "-workflow";
StringBuilder toolsYaml = new StringBuilder();
for (int i = 0; i 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
}
// 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"
*/
// 版本管理器：自动处理兼容性
class SkillVersionManager {
private installed: Map = new Map();
checkCompatibility(skillName: string, requiredVersion: string): boolean {
/** 检查已安装版本是否满足需求 */
const installed = this.installed.get(skillName);
if (!installed) return false;
// SemVer 兼容性检查
const [reqMajor, reqMinor, reqPatch] = requiredVersion.split('.').map(Number);
const [instMajor, instMinor, instPatch] = installed.split('.').map(Number);
// MAJOR 必须一致，MINOR >= 要求
return instMajor === reqMajor && instMinor >= reqMinor;
}
upgrade(skillName: string, targetVersion: string): Record {
/** 升级 Skill，返回迁移指南 */
const current = this.installed.get(skillName);
if (!current) return { error: `Skill ${skillName} 未安装` };
// 加载 CHANGELOG，检查是否需要迁移
const changelog = this.loadChangelog(skillName);
const migrations = changelog.filter(entry =>
this.versionBetween(entry.version, current, targetVersion) &&
entry.changes.some((c: string) => c.startsWith('MAJOR'))
);
this.installed.set(skillName, targetVersion);
return {
upgraded: `${current} → ${targetVersion}`,
migration_steps: migrations
};
}
private loadChangelog(skillName: string): any[] { /* ... */ return []; }
private versionBetween(v: string, from: string, to: string): boolean { /* ... */ return true; }
}
package main
import (
"fmt"
"strconv"
"strings"
)
// 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"
// 版本管理器：自动处理兼容性
type SkillVersionManager struct {
Installed map[string]string
}
func (m *SkillVersionManager) CheckCompatibility(skillName, requiredVersion string) bool {
installed, ok := m.Installed[skillName]
if !ok {
return false
}
// SemVer 兼容性检查
reqParts := strings.Split(requiredVersion, ".")
instParts := strings.Split(installed, ".")
reqMajor, _ := strconv.Atoi(reqParts[0])
reqMinor, _ := strconv.Atoi(reqParts[1])
instMajor, _ := strconv.Atoi(instParts[0])
instMinor, _ := strconv.Atoi(instParts[1])
// MAJOR 必须一致，MINOR >= 要求
return instMajor == reqMajor && instMinor >= reqMinor
}
func (m *SkillVersionManager) Upgrade(skillName, targetVersion string) map[string]interface{} {
current, ok := m.Installed[skillName]
if !ok {
return map[string]interface{}{"error": fmt.Sprintf("Skill %s 未安装", skillName)}
}
// 加载 CHANGELOG，检查是否需要迁移
changelog := m.loadChangelog(skillName)
var migrations []map[string]interface{}
for _, entry := range changelog {
if m.versionBetween(entry["version"].(string), current, targetVersion) {
changes, _ := entry["changes"].([]interface{})
for _, c := range changes {
if strings.HasPrefix(c.(string), "MAJOR") {
migrations = append(migrations, entry)
break
}
}
}
}
m.Installed[skillName] = targetVersion
return map[string]interface{}{
"upgraded": fmt.Sprintf("%s → %s", current, targetVersion),
"migration_steps": migrations,
}
}
func (m *SkillVersionManager) loadChangelog(skillName string) []map[string]interface{} { /* ... */ return nil }
func (m *SkillVersionManager) versionBetween(v, from, to string) bool { /* ... */ return true }
import java.util.*;
// 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"
*/
// 版本管理器：自动处理兼容性
public class SkillVersionManager {
private Map installed = new HashMap<>();
public boolean checkCompatibility(String skillName, String requiredVersion) {
/** 检查已安装版本是否满足需求 */
String installedVersion = installed.get(skillName);
if (installedVersion == null) return false;
// SemVer 兼容性检查
String[] reqParts = requiredVersion.split("\\.");
String[] instParts = installedVersion.split("\\.");
int reqMajor = Integer.parseInt(reqParts[0]);
int reqMinor = Integer.parseInt(reqParts[1]);
int instMajor = Integer.parseInt(instParts[0]);
int instMinor = Integer.parseInt(instParts[1]);
// MAJOR 必须一致，MINOR >= 要求
return instMajor == reqMajor && instMinor >= reqMinor;
}
public Map upgrade(String skillName, String targetVersion) {
/** 升级 Skill，返回迁移指南 */
String current = installed.get(skillName);
if (current == null) {
return Map.of("error", "Skill " + skillName + " 未安装");
}
// 加载 CHANGELOG，检查是否需要迁移
List> changelog = loadChangelog(skillName);
List> migrations = new ArrayList<>();
for (var entry : changelog) {
if (versionBetween((String) entry.get("version"), current, targetVersion)) {
List changes = (List) entry.get("changes");
if (changes.stream().anyMatch(c -> c.startsWith("MAJOR"))) {
migrations.add(entry);
}
}
}
installed.put(skillName, targetVersion);
Map result = new HashMap<>();
result.put("upgraded", current + " → " + targetVersion);
result.put("migration_steps", migrations);
return result;
}
private List> loadChangelog(String skillName) { /* ... */ return Collections.emptyList(); }
private boolean versionBetween(String v, String from, String to) { /* ... */ return true; }
}
💡 沉淀优先级
不要一开始就追求完美的 Skill 设计。WaLiCode 的 IterativeRefinement 模式告诉我们：先让 Agent 在项目中跑起来，观察它反复使用哪些工具组合，再把这些组合提炼成 Skill。项目实践是最好的 Skill 设计师——你只是把最佳实践固化下来。
12.11 Skill 的权限与安全
Skill 越强大，潜在风险越大。一个可以删除文件的 Skill 如果不加限制，Agent 可能误删关键数据。本章引入五层安全架构——参考 WaLiCode 的 permissionGuard 设计模式，给每一层加上护栏： | 安全层 | 机制 | 拦截时机 | 示例 | 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 | ### 12.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 不在列表中——因为权限已禁止
12.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"}
// permissionGuard 中间件实现
class PermissionGuard {
/** WaLiCode 运行时拦截模式——在工具调用前检查权限 */
// 黑名单：永远禁止的操作
private readonly BLOCKED_OPERATIONS: Record = {
rm_rf: '禁止递归删除',
eval: '禁止动态代码执行',
exec_shell: '禁止直接执行 shell 命令（除非白名单）',
};
// 白名单：特定 Skill 的允许操作
private readonly ALLOWED_PER_SKILL: Record> = {
'file-manager': {
'filesystem.read': true,
'filesystem.write': true,
'filesystem.delete': false,
},
'weather-advisor': {
'network.http_get': true,
'network.http_post': false,
},
};
private readonly NEEDS_APPROVAL: Record = {};
check(skillName: string, operation: string, params: Record): {
status: string; reason?: string; operation?: string;
} {
/** 检查权限，返回 allow/deny/need_approval */
// 1. 黑名单检查（最高优先级）
if (this.BLOCKED_OPERATIONS[operation]) {
return { status: 'denied', reason: this.BLOCKED_OPERATIONS[operation] };
}
// 2. 白名单检查
const skillPerms = this.ALLOWED_PER_SKILL[skillName] || {};
if (!skillPerms[operation]) {
return {
status: 'denied',
reason: `Skill ${skillName} 未获得 ${operation} 权限`,
};
}
// 3. 参数级安全检查（如路径穿越、SSRF）
if (params.path) {
const path = params.path as string;
if (path.includes('..') || path.startsWith('/etc')) {
return { status: 'denied', reason: '路径穿越攻击被拦截' };
}
}
if (params.url) {
const url = params.url as string;
const blockedHosts = ['localhost', '127.0.0.1', '0.0.0.0', 'internal.'];
if (blockedHosts.some(h => url.includes(h))) {
return { status: 'denied', reason: 'SSRF 攻击被拦截' };
}
}
// 4. 审批检查
const needsApproval = this.NEEDS_APPROVAL[skillName] || [];
if (needsApproval.includes(operation)) {
return { status: 'need_approval', operation };
}
return { status: 'allow' };
}
}
package main
import (
"fmt"
"strings"
)
// permissionGuard 中间件实现
type PermissionGuard struct {
// 黑名单：永远禁止的操作
BlockedOperations map[string]string
// 白名单：特定 Skill 的允许操作
AllowedPerSkill map[string]map[string]bool
// 需审批的操作
NeedsApproval map[string][]string
}
func NewPermissionGuard() *PermissionGuard {
return &PermissionGuard{
BlockedOperations: map[string]string{
"rm_rf": "禁止递归删除",
"eval": "禁止动态代码执行",
"exec_shell": "禁止直接执行 shell 命令（除非白名单）",
},
AllowedPerSkill: map[string]map[string]bool{
"file-manager": {
"filesystem.read": true,
"filesystem.write": true,
"filesystem.delete": false,
},
"weather-advisor": {
"network.http_get": true,
"network.http_post": false,
},
},
NeedsApproval: map[string][]string{},
}
}
func (g *PermissionGuard) Check(skillName, operation string, params map[string]interface{}) map[string]interface{} {
/** 检查权限，返回 allow/deny/need_approval */
// 1. 黑名单检查（最高优先级）
if reason, ok := g.BlockedOperations[operation]; ok {
return map[string]interface{}{"status": "denied", "reason": reason}
}
// 2. 白名单检查
skillPerms, ok := g.AllowedPerSkill[skillName]
if !ok || !skillPerms[operation] {
return map[string]interface{}{
"status": "denied",
"reason": fmt.Sprintf("Skill %s 未获得 %s 权限", skillName, operation),
}
}
// 3. 参数级安全检查（如路径穿越、SSRF）
if path, ok := params["path"].(string); ok {
if strings.Contains(path, "..") || strings.HasPrefix(path, "/etc") {
return map[string]interface{}{"status": "denied", "reason": "路径穿越攻击被拦截"}
}
}
if url, ok := params["url"].(string); ok {
blockedHosts := []string{"localhost", "127.0.0.1", "0.0.0.0", "internal."}
for _, h := range blockedHosts {
if strings.Contains(url, h) {
return map[string]interface{}{"status": "denied", "reason": "SSRF 攻击被拦截"}
}
}
}
// 4. 审批检查
for _, op := range g.NeedsApproval[skillName] {
if op == operation {
return map[string]interface{}{"status": "need_approval", "operation": operation}
}
}
return map[string]interface{}{"status": "allow"}
}
import java.util.*;
// permissionGuard 中间件实现
public class PermissionGuard {
/** WaLiCode 运行时拦截模式——在工具调用前检查权限 */
// 黑名单：永远禁止的操作
private static final Map BLOCKED_OPERATIONS = Map.of(
"rm_rf", "禁止递归删除",
"eval", "禁止动态代码执行",
"exec_shell", "禁止直接执行 shell 命令（除非白名单）"
);
// 白名单：特定 Skill 的允许操作
private static final Map> ALLOWED_PER_SKILL = Map.of(
"file-manager", Map.of(
"filesystem.read", true,
"filesystem.write", true,
"filesystem.delete", false
),
"weather-advisor", Map.of(
"network.http_get", true,
"network.http_post", false
)
);
private final Map> needsApproval = new HashMap<>();
public Map check(String skillName, String operation, Map params) {
/** 检查权限，返回 allow/deny/need_approval */
// 1. 黑名单检查（最高优先级）
if (BLOCKED_OPERATIONS.containsKey(operation)) {
return Map.of("status", "denied", "reason", BLOCKED_OPERATIONS.get(operation));
}
// 2. 白名单检查
Map skillPerms = ALLOWED_PER_SKILL.getOrDefault(skillName, Collections.emptyMap());
if (!skillPerms.getOrDefault(operation, false)) {
return Map.of(
"status", "denied",
"reason", "Skill " + skillName + " 未获得 " + operation + " 权限"
);
}
// 3. 参数级安全检查（如路径穿越、SSRF）
if (params.containsKey("path")) {
String path = (String) params.get("path");
if (path.contains("..") || path.startsWith("/etc")) {
return Map.of("status", "denied", "reason", "路径穿越攻击被拦截");
}
}
if (params.containsKey("url")) {
String url = (String) params.get("url");
List blockedHosts = List.of("localhost", "127.0.0.1", "0.0.0.0", "internal.");
if (blockedHosts.stream().anyMatch(url::contains)) {
return Map.of("status", "denied", "reason", "SSRF 攻击被拦截");
}
}
// 4. 审批检查
List approvalList = needsApproval.getOrDefault(skillName, Collections.emptyList());
if (approvalList.contains(operation)) {
return Map.of("status", "need_approval", "operation", operation);
}
return Map.of("status", "allow");
}
}
12.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 等待用户响应...
// 审批机制实现
class ApprovalGate {
/** WaLiCode ApprovalGate 模式——高危操作需用户确认 */
private pending: Map> = new Map();
requestApproval(skillName: string, operation: string, params: Record): string {
/** 请求审批，返回 approval_id */
const approvalId = `apr_${skillName}_${operation}_${Date.now()}`;
this.pending.set(approvalId, {
skill: skillName,
operation,
params,
status: 'pending',
created_at: Date.now(),
});
// 暂停执行，通知用户
console.log(`⚠️ 需要审批: ${skillName} 想执行 ${operation}`);
console.log(` 参数: ${JSON.stringify(params)}`);
console.log(` 请回复 /approve ${approvalId} 或 /reject ${approvalId}`);
return approvalId;
}
handleResponse(approvalId: string, approved: boolean): Record {
/** 处理用户的审批响应 */
const request = this.pending.get(approvalId);
if (!request) return { error: '审批 ID 不存在' };
if (approved) {
request.status = 'approved';
return { status: 'approved', proceed: true };
} else {
request.status = 'rejected';
return { status: 'rejected', proceed: false };
}
}
}
// 使用示例
const guardResult = permissionGuard.check('file-manager', 'filesystem.write', { path: '/data/report.csv' });
if (guardResult.status === 'need_approval') {
const approvalId = approvalGate.requestApproval('file-manager', 'filesystem.write', { path: '/data/report.csv' });
// Agent 等待用户响应...
}
package main
import (
"fmt"
"time"
)
// 审批机制实现
type ApprovalGate struct {
Pending map[string]map[string]interface{}
}
func NewApprovalGate() *ApprovalGate {
return &ApprovalGate{
Pending: make(map[string]map[string]interface{}),
}
}
func (g *ApprovalGate) RequestApproval(skillName, operation string, params map[string]interface{}) string {
/** 请求审批，返回 approval_id */
approvalId := fmt.Sprintf("apr_%s_%s_%d", skillName, operation, time.Now().Unix())
g.Pending[approvalId] = map[string]interface{}{
"skill": skillName,
"operation": operation,
"params": params,
"status": "pending",
"created_at": time.Now().Unix(),
}
// 暂停执行，通知用户
fmt.Printf("⚠️ 需要审批: %s 想执行 %s\n", skillName, operation)
fmt.Printf(" 参数: %v\n", params)
fmt.Printf(" 请回复 /approve %s 或 /reject %s\n", approvalId, approvalId)
return approvalId
}
func (g *ApprovalGate) HandleResponse(approvalId string, approved bool) map[string]interface{} {
/** 处理用户的审批响应 */
request, ok := g.Pending[approvalId]
if !ok {
return map[string]interface{}{"error": "审批 ID 不存在"}
}
if approved {
request["status"] = "approved"
return map[string]interface{}{"status": "approved", "proceed": true}
}
request["status"] = "rejected"
return map[string]interface{}{"status": "rejected", "proceed": false}
}
// 使用示例
// guardResult := permissionGuard.Check("file-manager", "filesystem.write", map[string]interface{}{"path": "/data/report.csv"})
// if guardResult["status"] == "need_approval" {
// approvalId := approvalGate.RequestApproval("file-manager", "filesystem.write", map[string]interface{}{"path": "/data/report.csv"})
// // Agent 等待用户响应...
// }
import java.util.*;
import java.time.Instant;
// 审批机制实现
public class ApprovalGate {
/** WaLiCode ApprovalGate 模式——高危操作需用户确认 */
private Map> pending = new HashMap<>();
public String requestApproval(String skillName, String operation, Map params) {
/** 请求审批，返回 approval_id */
String approvalId = String.format("apr_%s_%s_%d", skillName, operation, Instant.now().getEpochSecond());
Map request = new HashMap<>();
request.put("skill", skillName);
request.put("operation", operation);
request.put("params", params);
request.put("status", "pending");
request.put("created_at", Instant.now().getEpochSecond());
pending.put(approvalId, request);
// 暂停执行，通知用户
System.out.printf("⚠️ 需要审批: %s 想执行 %s%n", skillName, operation);
System.out.printf(" 参数: %s%n", params);
System.out.printf(" 请回复 /approve %s 或 /reject %s%n", approvalId, approvalId);
return approvalId;
}
public Map handleResponse(String approvalId, boolean approved) {
/** 处理用户的审批响应 */
Map request = pending.get(approvalId);
if (request == null) {
return Map.of("error", "审批 ID 不存在");
}
if (approved) {
request.put("status", "approved");
return Map.of("status", "approved", "proceed", true);
} else {
request.put("status", "rejected");
return Map.of("status", "rejected", "proceed", false);
}
}
}
// 使用示例
// Map guardResult = permissionGuard.check("file-manager", "filesystem.write", Map.of("path", "/data/report.csv"));
// if (guardResult.get("status").equals("need_approval")) {
// String approvalId = approvalGate.requestApproval("file-manager", "filesystem.write", Map.of("path", "/data/report.csv"));
// // Agent 等待用户响应...
// }
12.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 级别的调用记录，供安全团队审查
// 审计日志实现
import Database from 'better-sqlite3';
class SkillAuditLog {
/** WaLiCode AuditTrail 模式——所有调用可追溯 */
private db: Database.Database;
constructor(dbPath: string = 'skill_audit.db') {
this.db = new Database(dbPath);
this.db.exec(`
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,
result TEXT NOT NULL,
approved TEXT DEFAULT NULL,
risk_level TEXT NOT NULL
)
`);
}
log(sessionId: string, skillName: string, operation: string,
params: Record, result: Record,
approved: string | null = null, riskLevel: string = 'low'): void {
/** 记录一次 Skill 调用 */
this.db.prepare(
'INSERT INTO audit_log VALUES (NULL, ?, ?, ?, ?, ?, ?, ?, ?)'
).run(
new Date().toISOString(), sessionId, skillName, operation,
JSON.stringify(params), JSON.stringify(result), approved, riskLevel
);
}
query(opts?: { skillName?: string; riskLevel?: string; since?: string }): Record[] {
/** 查询审计日志（合规审查用） */
const conditions: string[] = [];
const values: string[] = [];
if (opts?.skillName) {
conditions.push('skill_name = ?');
values.push(opts.skillName);
}
if (opts?.riskLevel) {
conditions.push('risk_level = ?');
values.push(opts.riskLevel);
}
if (opts?.since) {
conditions.push('timestamp >= ?');
values.push(opts.since);
}
const where = conditions.length > 0 ? conditions.join(' AND ') : '1=1';
const rows = this.db.prepare(
`SELECT * FROM audit_log WHERE ${where} ORDER BY timestamp DESC LIMIT 100`
).all(...values);
return rows as Record[];
}
}
// 查询所有高风险调用（合规审查场景）
const highRiskCalls = auditLog.query({ riskLevel: 'critical' });
// → 返回所有 critical 级别的调用记录，供安全团队审查
package main
import (
"database/sql"
"encoding/json"
"fmt"
"time"
_ "github.com/mattn/go-sqlite3"
)
// 审计日志实现
type SkillAuditLog struct {
Db *sql.DB
}
func NewSkillAuditLog(dbPath string) *SkillAuditLog {
db, _ := sql.Open("sqlite3", dbPath)
db.Exec(`
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,
result TEXT NOT NULL,
approved TEXT DEFAULT NULL,
risk_level TEXT NOT NULL
)
`)
return &SkillAuditLog{Db: db}
}
func (l *SkillAuditLog) Log(sessionId, skillName, operation string,
params, result map[string]interface{}, approved, riskLevel string) {
/** 记录一次 Skill 调用 */
paramsJSON, _ := json.Marshal(params)
resultJSON, _ := json.Marshal(result)
l.Db.Exec(
"INSERT INTO audit_log VALUES (NULL, ?, ?, ?, ?, ?, ?, ?, ?)",
time.Now().Format(time.RFC3339), sessionId, skillName, operation,
string(paramsJSON), string(resultJSON), approved, riskLevel,
)
}
func (l *SkillAuditLog) Query(skillName, riskLevel, since string) []map[string]interface{} {
/** 查询审计日志（合规审查用） */
conditions := []string{}
args := []interface{}{}
if skillName != "" {
conditions = append(conditions, "skill_name = ?")
args = append(args, skillName)
}
if riskLevel != "" {
conditions = append(conditions, "risk_level = ?")
args = append(args, riskLevel)
}
if since != "" {
conditions = append(conditions, "timestamp >= ?")
args = append(args, since)
}
where := "1=1"
if len(conditions) > 0 {
where = fmt.Sprintf("%s", joinStrings(conditions, " AND "))
}
rows, _ := l.Db.Query(
fmt.Sprintf("SELECT * FROM audit_log WHERE %s ORDER BY timestamp DESC LIMIT 100", where),
args...,
)
defer rows.Close()
var results []map[string]interface{}
for rows.Next() {
var id int
var ts, sid, skill, op, paramsStr, resultStr, approvedStr, risk sql.NullString
rows.Scan(&id, &ts, &sid, &skill, &op, ¶msStr, &resultStr, &approvedStr, &risk)
results = append(results, map[string]interface{}{
"id": id, "ts": ts.String, "sid": sid.String,
"skill": skill.String, "op": op.String,
"params": paramsStr.String, "result": resultStr.String,
"approved": approvedStr.String, "risk": risk.String,
})
}
return results
}
// 查询所有高风险调用（合规审查场景）
// highRiskCalls := auditLog.Query("", "critical", "")
// → 返回所有 critical 级别的调用记录，供安全团队审查
import java.sql.*;
import java.time.Instant;
import java.util.*;
import com.google.gson.Gson;
// 审计日志实现
public class SkillAuditLog {
/** WaLiCode AuditTrail 模式——所有调用可追溯 */
private Connection conn;
private Gson gson = new Gson();
public SkillAuditLog(String dbPath) throws SQLException {
conn = DriverManager.getConnection("jdbc:sqlite:" + dbPath);
conn.createStatement().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," +
" result TEXT NOT NULL," +
" approved TEXT DEFAULT NULL," +
" risk_level TEXT NOT NULL" +
")"
);
}
public void log(String sessionId, String skillName, String operation,
Map params, Map result,
String approved, String riskLevel) throws SQLException {
/** 记录一次 Skill 调用 */
PreparedStatement ps = conn.prepareStatement(
"INSERT INTO audit_log VALUES (NULL, ?, ?, ?, ?, ?, ?, ?, ?)"
);
ps.setString(1, Instant.now().toString());
ps.setString(2, sessionId);
ps.setString(3, skillName);
ps.setString(4, operation);
ps.setString(5, gson.toJson(params));
ps.setString(6, gson.toJson(result));
ps.setString(7, approved);
ps.setString(8, riskLevel);
ps.executeUpdate();
}
public List> query(String skillName, String riskLevel, String since) throws SQLException {
/** 查询审计日志（合规审查用） */
List conditions = new ArrayList<>();
List values = new ArrayList<>();
if (skillName != null) {
conditions.add("skill_name = ?");
values.add(skillName);
}
if (riskLevel != null) {
conditions.add("risk_level = ?");
values.add(riskLevel);
}
if (since != null) {
conditions.add("timestamp >= ?");
values.add(since);
}
String where = conditions.isEmpty() ? "1=1" : String.join(" AND ", conditions);
PreparedStatement ps = conn.prepareStatement(
"SELECT * FROM audit_log WHERE " + where + " ORDER BY timestamp DESC LIMIT 100"
);
for (int i = 0; i > results = new ArrayList<>();
while (rs.next()) {
Map row = new LinkedHashMap<>();
row.put("id", rs.getInt("id"));
row.put("ts", rs.getString("timestamp"));
row.put("sid", rs.getString("session_id"));
row.put("skill", rs.getString("skill_name"));
row.put("op", rs.getString("operation"));
row.put("params", rs.getString("params"));
row.put("result", rs.getString("result"));
row.put("approved", rs.getString("approved"));
row.put("risk", rs.getString("risk_level"));
results.add(row);
}
return results;
}
}
// 查询所有高风险调用（合规审查场景）
// List> highRiskCalls = auditLog.query(null, "critical", null);
// → 返回所有 critical 级别的调用记录，供安全团队审查
12.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
}
// 沙箱执行器实现（基于 child_process + 资源限制）
import { execFile } from 'child_process';
import * as fs from 'fs';
import * as path from 'path';
import * as os from 'os';
class SandboxExecutor {
/** WaLiCode SandboxExecutor 模式——危险 Skill 在沙箱中执行 */
private readonly SANDBOX_CONFIG: Record = {
'payment-skill': {
network: 'restricted',
filesystem: 'read_only',
maxMemory: '256MB',
maxCpuTime: 30,
allowedDomains: ['api.payment.com'],
},
'database-query': {
network: 'restricted',
filesystem: 'none',
maxMemory: '512MB',
maxCpuTime: 60,
allowedDomains: ['db.internal.com'],
},
};
executeInSandbox(skillName: string, toolName: string, params: Record): Promise> {
/** 在沙箱中执行 Skill 工具 */
const config = this.SANDBOX_CONFIG[skillName];
if (!config) {
// 无沙箱配置 → 正常执行
return this.executeNormal(skillName, toolName, params);
}
return new Promise((resolve) => {
// 将调用序列化为临时脚本
const script = this.serializeCall(skillName, toolName, params);
const scriptPath = path.join(os.tmpdir(), `skill_${Date.now()}.js`);
fs.writeFileSync(scriptPath, script);
// 在受限环境中执行
execFile('node', [scriptPath], {
timeout: config.maxCpuTime * 1000,
env: {
...process.env,
SKILL_SANDBOX: 'true',
ALLOWED_DOMAINS: config.allowedDomains.join(','),
MAX_MEMORY: config.maxMemory,
},
}, (err, stdout, stderr) => {
resolve({
stdout: stdout || '',
stderr: stderr || (err?.message || ''),
exitCode: err ? 1 : 0,
sandbox: true,
});
});
});
}
private serializeCall(skillName: string, toolName: string, params: Record): string {
return `// Auto-generated sandbox script\n` +
`const skill = require('./skills/${skillName}');\n` +
`skill.${toolName}(${JSON.stringify(params)});`;
}
private async executeNormal(skillName: string, toolName: string, params: Record): Promise> {
return { error: 'No sandbox config found, executing normally' };
}
}
package main
import (
"bytes"
"fmt"
"os"
"os/exec"
"path/filepath"
)
// 沙箱执行器实现（基于 subprocess + 资源限制）
type SandboxExecutor struct {
SandboxConfig map[string]map[string]interface{}
}
func NewSandboxExecutor() *SandboxExecutor {
return &SandboxExecutor{
SandboxConfig: map[string]map[string]interface{}{
"payment-skill": {
"network": "restricted",
"filesystem": "read_only",
"max_memory": "256MB",
"max_cpu_time": 30,
"allowed_domains": []string{"api.payment.com"},
},
"database-query": {
"network": "restricted",
"filesystem": "none",
"max_memory": "512MB",
"max_cpu_time": 60,
"allowed_domains": []string{"db.internal.com"},
},
},
}
}
func (s *SandboxExecutor) ExecuteInSandbox(skillName, toolName string, params map[string]interface{}) map[string]interface{} {
/** 在沙箱中执行 Skill 工具 */
config, ok := s.SandboxConfig[skillName]
if !ok {
// 无沙箱配置 → 正常执行
return s.executeNormal(skillName, toolName, params)
}
// 将调用序列化为临时脚本
script := s.serializeCall(skillName, toolName, params)
tmpDir, _ := os.MkdirTemp("", "skill_*")
scriptPath := filepath.Join(tmpDir, "run.py")
os.WriteFile(scriptPath, []byte(script), 0644)
// 在受限环境中执行
maxCpuTime := int(config["max_cpu_time"].(int))
domains := config["allowed_domains"].([]string)
domainStrs := make([]string, len(domains))
for i, d := range domains {
domainStrs[i] = fmt.Sprintf("%v", d)
}
cmd := exec.Command("python3", scriptPath)
cmd.Env = append(os.Environ(),
"SKILL_SANDBOX=true",
fmt.Sprintf("ALLOWED_DOMAINS=%s", joinStrings(domainStrs, ",")),
fmt.Sprintf("MAX_MEMORY=%v", config["max_memory"]),
)
var stdout, stderr bytes.Buffer
cmd.Stdout = &stdout
cmd.Stderr = &stderr
err := cmd.Run()
exitCode := 0
if err != nil {
exitCode = 1
}
return map[string]interface{}{
"stdout": stdout.String(),
"stderr": stderr.String(),
"exit_code": exitCode,
"sandbox": true,
}
}
func (s *SandboxExecutor) serializeCall(skillName, toolName string, params map[string]interface{}) string {
return fmt.Sprintf("# Auto-generated sandbox script\nimport %s\n", skillName)
}
func (s *SandboxExecutor) executeNormal(skillName, toolName string, params map[string]interface{}) map[string]interface{} {
return map[string]interface{}{"error": "No sandbox config found"}
}
import java.io.*;
import java.nio.file.*;
import java.util.*;
// 沙箱执行器实现（基于 ProcessBuilder + 资源限制）
public class SandboxExecutor {
/** WaLiCode SandboxExecutor 模式——危险 Skill 在沙箱中执行 */
private static final Map> SANDBOX_CONFIG = new HashMap<>();
static {
Map paymentConfig = new HashMap<>();
paymentConfig.put("network", "restricted");
paymentConfig.put("filesystem", "read_only");
paymentConfig.put("max_memory", "256MB");
paymentConfig.put("max_cpu_time", 30);
paymentConfig.put("allowed_domains", List.of("api.payment.com"));
SANDBOX_CONFIG.put("payment-skill", paymentConfig);
Map dbConfig = new HashMap<>();
dbConfig.put("network", "restricted");
dbConfig.put("filesystem", "none");
dbConfig.put("max_memory", "512MB");
dbConfig.put("max_cpu_time", 60);
dbConfig.put("allowed_domains", List.of("db.internal.com"));
SANDBOX_CONFIG.put("database-query", dbConfig);
}
public Map executeInSandbox(String skillName, String toolName, Map params) throws Exception {
/** 在沙箱中执行 Skill 工具 */
Map config = SANDBOX_CONFIG.get(skillName);
if (config == null) {
// 无沙箱配置 → 正常执行
return executeNormal(skillName, toolName, params);
}
// 将调用序列化为临时脚本
String script = serializeCall(skillName, toolName, params);
Path scriptPath = Files.createTempFile("skill_", ".py");
Files.writeString(scriptPath, script);
// 在受限环境中执行
int maxCpuTime = (int) config.get("max_cpu_time");
List allowedDomains = (List) config.get("allowed_domains");
ProcessBuilder pb = new ProcessBuilder("python3", scriptPath.toString());
pb.environment().put("SKILL_SANDBOX", "true");
pb.environment().put("ALLOWED_DOMAINS", String.join(",", allowedDomains));
pb.environment().put("MAX_MEMORY", (String) config.get("max_memory"));
pb.redirectErrorStream(false);
Process proc = pb.start();
// 设置超时
if (!proc.waitFor(maxCpuTime, TimeUnit.SECONDS)) {
proc.destroyForcibly();
}
String stdout = new String(proc.getInputStream().readAllBytes());
String stderr = new String(proc.getErrorStream().readAllBytes());
int exitCode = proc.exitValue();
Map result = new HashMap<>();
result.put("stdout", stdout);
result.put("stderr", stderr);
result.put("exit_code", exitCode);
result.put("sandbox", true);
return result;
}
private String serializeCall(String skillName, String toolName, Map params) {
return "# Auto-generated sandbox script\nimport " + skillName + "\n";
}
private Map executeNormal(String skillName, String toolName, Map params) {
return Map.of("error", "No sandbox config found");
}
}
💡 安全分层原则
不是所有 Skill 都需要五层防护。基础 Skill（如 get_weather）只需 Layer 1（声明）+ Layer 2（运行时拦截）。复合 Skill 加上 Layer 3（审批）+ Layer 4（审计）。领域 Skill 才需要完整的五层——包括沙箱隔离。安全不是越厚越好，而是按风险等级分层。
12.12 实战：构建企业级 Skill 包
前面讲了 Skill 的分层体系和安全架构，现在把它们串起来——从零构建一个完整的企业级 Skill 包，包括 YAML 定义、Python 实现、测试和发布流程：
12.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 模板缺失 → 使用默认模板
12.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
// skills/compound/report-generator/main.ts
import * as fs from 'fs';
import * as path from 'path';
class ReportGeneratorSkill {
/** 企业级报表生成 Skill——带 permissionGuard */
constructor(private guard?: PermissionGuard) {}
fetchData(source: string, query?: string, format: string = 'json'): Record {
/** 从数据源获取原始数据 */
if (source.startsWith('http')) {
// API 数据源（带超时和分页）
// 使用 fetch API
const params = query ? `?query=${query}` : '';
// In production: const response = await fetch(source + params);
return { data: [], source };
} else if (source.endsWith('.csv') || source.endsWith('.json')) {
// 文件数据源（带安全检查）
if (this.guard) {
const check = this.guard.check('report-generator', 'filesystem.read', { path: source });
if (check.status === 'denied') {
return { error: check.reason };
}
}
const content = fs.readFileSync(source, 'utf-8');
if (source.endsWith('.json')) {
return { data: JSON.parse(content), source };
} else {
// Parse CSV
const lines = content.split('\n');
const headers = lines[0].split(',');
const data = lines.slice(1).map(line => {
const values = line.split(',');
const row: Record = {};
headers.forEach((h, i) => { row[h.trim()] = values[i]?.trim() || ''; });
return row;
});
return { data, source };
}
}
return { error: `不支持的数据源类型: ${source}` };
}
analyzeData(data: Record[], metrics?: string[], groupBy?: string): Record {
/** 分析数据，计算统计指标 */
if (!data || data.length === 0) return { error: '数据为空，无法分析' };
metrics = metrics || ['sum', 'avg', 'max', 'min'];
const numericFields = this.findNumericFields(data);
const results: Record = {};
for (const field of numericFields) {
const values = data.map(row => Number(row[field] || 0)).filter(v => !isNaN(v));
const fieldStats: Record = {};
for (const m of metrics) {
if (m === 'sum') fieldStats.sum = values.reduce((a, b) => a + b, 0);
else if (m === 'avg') fieldStats.avg = values.length ? values.reduce((a, b) => a + b, 0) / values.length : 0;
else if (m === 'max') fieldStats.max = values.length ? Math.max(...values) : 0;
else if (m === 'min') fieldStats.min = values.length ? Math.min(...values) : 0;
}
results[field] = fieldStats;
}
if (groupBy) {
const groups: Record = {};
for (const row of data) {
const key = String(row[groupBy] || 'unknown');
groups[key] = (groups[key] || 0) + 1;
}
results.groups = groups;
}
return { analysis: results, data_count: data.length, fields: numericFields };
}
renderReport(analysis: Record, template: string = 'default',
outputFormat: string = 'markdown'): Record {
/** 渲染报表 */
const stats = analysis.analysis as Record;
const count = analysis.data_count as number;
let report = '# 数据报表\n\n';
report += `> 生成时间: ${new Date().toLocaleString()}\n`;
report += `> 数据量: ${count} 条\n\n`;
report += '## 统计指标\n\n';
report += '| 字段 | 求和 | 平均值 | 最大值 | 最小值 |\n';
report += '|------|------|--------|--------|--------|\n';
for (const [field, fieldStats] of Object.entries(stats)) {
if (field === 'groups') continue;
const fs2 = fieldStats as Record;
report += `| ${field} | ${fs2.sum ?? '-'} | ${fs2.avg ?? '-'} | ${fs2.max ?? '-'} | ${fs2.min ?? '-'} |\n`;
}
if (stats.groups) {
report += '\n## 分组统计\n\n';
for (const [group, groupCount] of Object.entries(stats.groups as Record)) {
report += `- **${group}**: ${groupCount} 条\n`;
}
}
const outputPath = `reports/report_${Date.now()}.md`;
if (this.guard) {
const check = this.guard.check('report-generator', 'filesystem.write', { path: outputPath });
if (check.status === 'need_approval') {
return { report, approval_needed: true, message: `需要审批才能写入文件: ${outputPath}` };
} else if (check.status === 'denied') {
return { error: check.reason };
}
}
fs.mkdirSync(path.dirname(outputPath), { recursive: true });
fs.writeFileSync(outputPath, report, 'utf-8');
return { report, output_path: outputPath, written: true };
}
private findNumericFields(data: Record[]): string[] {
/** 自动检测数值型字段 */
const numeric: string[] = [];
if (data.length > 0) {
for (const [key, val] of Object.entries(data[0])) {
if (typeof val === 'number') numeric.push(key);
}
}
return numeric;
}
}
package main
import (
"encoding/json"
"fmt"
"os"
"path/filepath"
"sort"
"strconv"
"strings"
"time"
)
// skills/compound/report-generator/main.go
type ReportGeneratorSkill struct {
Guard *PermissionGuard
}
func NewReportGeneratorSkill(guard *PermissionGuard) *ReportGeneratorSkill {
return &ReportGeneratorSkill{Guard: guard}
}
func (s *ReportGeneratorSkill) FetchData(source, query, format string) map[string]interface{} {
/** 从数据源获取原始数据 */
if strings.HasPrefix(source, "http") {
// API 数据源（带超时和分页）
return map[string]interface{}{"data": nil, "source": source}
} else if strings.HasSuffix(source, ".csv") || strings.HasSuffix(source, ".json") {
// 文件数据源（带安全检查）
if s.Guard != nil {
check := s.Guard.Check("report-generator", "filesystem.read", map[string]interface{}{"path": source})
if check["status"] == "denied" {
return map[string]interface{}{"error": check["reason"]}
}
}
data, err := os.ReadFile(source)
if err != nil {
return map[string]interface{}{"error": err.Error()}
}
if strings.HasSuffix(source, ".json") {
var result interface{}
json.Unmarshal(data, &result)
return map[string]interface{}{"data": result, "source": source}
}
// Parse CSV
lines := strings.Split(string(data), "\n")
if len(lines) 0 {
sum := 0.0
for _, v := range values { sum += v }
fieldStats["avg"] = sum / float64(len(values))
} else { fieldStats["avg"] = 0 }
case "max":
if len(values) > 0 {
maxVal := values[0]
for _, v := range values { if v > maxVal { maxVal = v } }
fieldStats["max"] = maxVal
} else { fieldStats["max"] = 0 }
case "min":
if len(values) > 0 {
minVal := values[0]
for _, v := range values { if v 生成时间: %s\n", time.Now().Format("2006-01-02 15:04"))
report += fmt.Sprintf("> 数据量: %d 条\n\n", count)
report += "## 统计指标\n\n"
report += "| 字段 | 求和 | 平均值 | 最大值 | 最小值 |\n"
report += "|------|------|--------|--------|--------|\n"
for field, fsRaw := range stats {
if field == "groups" { continue }
fs, _ := fsRaw.(map[string]interface{})
report += fmt.Sprintf("| %s | %v | %v | %v | %v |\n",
field, fs["sum"], fs["avg"], fs["max"], fs["min"])
}
if groups, ok := stats["groups"].(map[string]int); ok {
report += "\n## 分组统计\n\n"
for group, groupCount := range groups {
report += fmt.Sprintf("- **%s**: %d 条\n", group, groupCount)
}
}
outputPath := fmt.Sprintf("reports/report_%s.md", time.Now().Format("20060102_150405"))
if s.Guard != nil {
check := s.Guard.Check("report-generator", "filesystem.write", map[string]interface{}{"path": outputPath})
if check["status"] == "need_approval" {
return map[string]interface{}{"report": report, "approval_needed": true, "message": fmt.Sprintf("需要审批才能写入文件: %s", outputPath)}
} else if check["status"] == "denied" {
return map[string]interface{}{"error": check["reason"]}
}
}
os.MkdirAll(filepath.Dir(outputPath), 0755)
os.WriteFile(outputPath, []byte(report), 0644)
return map[string]interface{}{"report": report, "output_path": outputPath, "written": true}
}
func (s *ReportGeneratorSkill) findNumericFields(data []map[string]interface{}) []string {
var numeric []string
if len(data) > 0 {
for key, val := range data[0] {
switch val.(type) {
case int, int64, float64, float32:
numeric = append(numeric, key)
}
}
}
sort.Strings(numeric)
return numeric
}
func toFloat64(val interface{}) (float64, error) {
switch v := val.(type) {
case int:
return float64(v), nil
case float64:
return v, nil
case string:
return strconv.ParseFloat(v, 64)
}
return 0, fmt.Errorf("not numeric")
}
import java.io.*;
import java.nio.file.*;
import java.time.*;
import java.util.*;
import com.google.gson.Gson;
// skills/compound/report-generator/ReportGeneratorSkill.java
public class ReportGeneratorSkill {
/** 企业级报表生成 Skill——带 permissionGuard */
private PermissionGuard guard;
private Gson gson = new Gson();
public ReportGeneratorSkill(PermissionGuard guard) {
this.guard = guard;
}
public Map fetchData(String source, String query, String format) {
/** 从数据源获取原始数据 */
if (source.startsWith("http")) {
// API 数据源（带超时和分页）
// In production: use HttpClient with timeout
return Map.of("data", List.of(), "source", source);
} else if (source.endsWith(".csv") || source.endsWith(".json")) {
// 文件数据源（带安全检查）
if (guard != null) {
Map check = guard.check("report-generator", "filesystem.read", Map.of("path", source));
if (check.get("status").equals("denied")) {
return Map.of("error", check.get("reason"));
}
}
try {
String content = Files.readString(Paths.get(source));
if (source.endsWith(".json")) {
Object data = gson.fromJson(content, Object.class);
return Map.of("data", data, "source", source);
} else {
// Parse CSV
String[] lines = content.split("\n");
String[] headers = lines[0].split(",");
List> data = new ArrayList<>();
for (int i = 1; i row = new HashMap<>();
for (int j = 0; j analyzeData(List> data, List metrics, String groupBy) {
/** 分析数据，计算统计指标 */
if (data == null || data.isEmpty()) {
return Map.of("error", "数据为空，无法分析");
}
if (metrics == null) metrics = List.of("sum", "avg", "max", "min");
List numericFields = findNumericFields(data);
Map results = new LinkedHashMap<>();
for (String field : numericFields) {
List values = new ArrayList<>();
for (Map row : data) {
Object val = row.get(field);
if (val instanceof Number) {
values.add(((Number) val).doubleValue());
}
}
Map fieldStats = new LinkedHashMap<>();
for (String m : metrics) {
switch (m) {
case "sum":
fieldStats.put("sum", values.stream().mapToDouble(d -> d).sum());
break;
case "avg":
fieldStats.put("avg", values.isEmpty() ? 0 : values.stream().mapToDouble(d -> d).average().orElse(0));
break;
case "max":
fieldStats.put("max", values.isEmpty() ? 0 : values.stream().mapToDouble(d -> d).max().orElse(0));
break;
case "min":
fieldStats.put("min", values.isEmpty() ? 0 : values.stream().mapToDouble(d -> d).min().orElse(0));
break;
}
}
results.put(field, fieldStats);
}
if (groupBy != null) {
Map groups = new LinkedHashMap<>();
for (Map row : data) {
String key = String.valueOf(row.getOrDefault(groupBy, "unknown"));
groups.merge(key, 1, Integer::sum);
}
results.put("groups", groups);
}
Map result = new LinkedHashMap<>();
result.put("analysis", results);
result.put("data_count", data.size());
result.put("fields", numericFields);
return result;
}
public Map renderReport(Map analysis, String template, String outputFormat) {
/** 渲染报表 */
Map stats = (Map) analysis.get("analysis");
int count = (int) analysis.get("data_count");
StringBuilder report = new StringBuilder();
report.append("# 数据报表\n\n");
report.append("> 生成时间: ").append(LocalDateTime.now()).append("\n");
report.append("> 数据量: ").append(count).append(" 条\n\n");
report.append("## 统计指标\n\n");
report.append("| 字段 | 求和 | 平均值 | 最大值 | 最小值 |\n");
report.append("|------|------|--------|--------|--------|\n");
for (var entry : stats.entrySet()) {
if (entry.getKey().equals("groups")) continue;
Map fs = (Map) entry.getValue();
report.append(String.format("| %s | %s | %s | %s | %s |\n",
entry.getKey(), fs.getOrDefault("sum", "-"), fs.getOrDefault("avg", "-"),
fs.getOrDefault("max", "-"), fs.getOrDefault("min", "-")));
}
if (stats.containsKey("groups")) {
report.append("\n## 分组统计\n\n");
Map groups = (Map) stats.get("groups");
for (var entry : groups.entrySet()) {
report.append(String.format("- **%s**: %d 条\n", entry.getKey(), entry.getValue()));
}
}
String outputPath = "reports/report_" + System.currentTimeMillis() + ".md";
if (guard != null) {
Map check = guard.check("report-generator", "filesystem.write", Map.of("path", outputPath));
if (check.get("status").equals("need_approval")) {
return Map.of("report", report.toString(), "approval_needed", true,
"message", "需要审批才能写入文件: " + outputPath);
} else if (check.get("status").equals("denied")) {
return Map.of("error", check.get("reason"));
}
}
try {
Files.createDirectories(Paths.get("reports"));
Files.writeString(Paths.get(outputPath), report.toString());
} catch (IOException e) {
return Map.of("error", e.getMessage());
}
return Map.of("report", report.toString(), "output_path", outputPath, "written", true);
}
private List findNumericFields(List> data) {
List numeric = new ArrayList<>();
if (!data.isEmpty()) {
for (var entry : data.get(0).entrySet()) {
if (entry.getValue() instanceof Number) {
numeric.add(entry.getKey());
}
}
}
return numeric;
}
}
12.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"]
// tests/test_report_generator.ts
import { describe, it, expect, beforeEach } from 'vitest';
import { ReportGeneratorSkill } from '../skills/compound/report-generator/main';
import { PermissionGuard } from '../skills/compound/report-generator/permission_guard';
describe('TestReportGenerator', () => {
/** 单元测试——验证工具函数逻辑 */
let skill: ReportGeneratorSkill;
beforeEach(() => {
skill = new ReportGeneratorSkill();
});
it('test_analyze_data_basic', () => {
/** 测试基本统计分析 */
const data = [
{ name: 'A', value: 10 },
{ name: 'B', value: 20 },
{ name: 'C', value: 30 },
];
const result = skill.analyzeData(data);
expect(result.data_count).toBe(3);
expect((result.analysis as any).value.sum).toBe(60);
expect((result.analysis as any).value.avg).toBe(20);
});
it('test_analyze_data_empty', () => {
/** 测试空数据返回错误 */
const result = skill.analyzeData([]);
expect(result.error).toBeDefined();
expect(result.error).toContain('数据为空');
});
it('test_analyze_data_with_groups', () => {
/** 测试分组统计 */
const data = [
{ department: '工程', salary: 15000 },
{ department: '工程', salary: 18000 },
{ department: '市场', salary: 12000 },
];
const result = skill.analyzeData(data, undefined, 'department');
expect((result.analysis as any).groups['工程']).toBe(2);
expect((result.analysis as any).groups['市场']).toBe(1);
});
it('test_render_report_markdown', () => {
/** 测试 Markdown 报表生成 */
const analysis = {
analysis: { salary: { sum: 45000, avg: 15000 } },
data_count: 3,
};
const result = skill.renderReport(analysis);
expect(result.report).toContain('# 数据报表');
expect(result.report).toContain('salary');
});
});
describe('TestPromptIntegration', () => {
/** Prompt 测试——验证 AI 能正确调用 Skill */
const 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'],
},
];
it('test_skill_activation', () => {
/** 验证触发词能正确激活 Skill */
const manager = new LazySkillManager(loader);
for (const testCase of PROMPT_TEST_CASES) {
const prompt = manager.getPrompt(testCase.user_message);
expect(prompt).toContain('report-generator');
expect(prompt).toContain('(已激活)');
}
});
});
describe('TestSecurity', () => {
/** 安全测试——验证 permissionGuard 拦截 */
let guard: PermissionGuard;
let skill: ReportGeneratorSkill;
beforeEach(() => {
guard = new PermissionGuard();
skill = new ReportGeneratorSkill(guard);
});
it('test_path_traversal_blocked', () => {
/** 测试路径穿越被拦截 */
const result = skill.fetchData('/etc/passwd');
expect(result.error).toBeDefined();
expect(result.error).toContain('拦截');
});
it('test_write_needs_approval', () => {
/** 测试写文件需要审批 */
const result = skill.renderReport({ analysis: {}, data_count: 0 });
expect(result.approval_needed === true || JSON.stringify(result).includes('审批')).toBeTruthy();
});
it('test_ssrf_blocked', () => {
/** 测试 SSRF 被拦截 */
const result = skill.fetchData('http://127.0.0.1:8080/internal');
expect(result.error).toBeDefined();
expect(result.error.includes('SSRF') || result.error.includes('拦截')).toBeTruthy();
});
});
package main
import (
"testing"
)
// tests/test_report_generator_test.go
func TestAnalyzeDataBasic(t *testing.T) {
/** 测试基本统计分析 */
skill := NewReportGeneratorSkill(nil)
data := []map[string]interface{}{
{"name": "A", "value": 10},
{"name": "B", "value": 20},
{"name": "C", "value": 30},
}
result := skill.AnalyzeData(data, nil, "")
if result["data_count"].(int) != 3 {
t.Error("Expected data_count=3")
}
analysis := result["analysis"].(map[string]interface{})
valueStats := analysis["value"].(map[string]interface{})
if valueStats["sum"].(float64) != 60 {
t.Error("Expected sum=60")
}
}
func TestAnalyzeDataEmpty(t *testing.T) {
/** 测试空数据返回错误 */
skill := NewReportGeneratorSkill(nil)
result := skill.AnalyzeData([]map[string]interface{}{}, nil, "")
if _, ok := result["error"]; !ok {
t.Error("Expected error for empty data")
}
}
func TestAnalyzeDataWithGroups(t *testing.T) {
/** 测试分组统计 */
skill := NewReportGeneratorSkill(nil)
data := []map[string]interface{}{
{"department": "工程", "salary": 15000},
{"department": "工程", "salary": 18000},
{"department": "市场", "salary": 12000},
}
result := skill.AnalyzeData(data, nil, "department")
analysis := result["analysis"].(map[string]interface{})
groups := analysis["groups"].(map[string]int)
if groups["工程"] != 2 {
t.Error("Expected 工程=2")
}
if groups["市场"] != 1 {
t.Error("Expected 市场=1")
}
}
func TestPathTraversalBlocked(t *testing.T) {
/** 测试路径穿越被拦截 */
guard := NewPermissionGuard()
skill := NewReportGeneratorSkill(guard)
result := skill.FetchData("/etc/passwd", "", "json")
if _, ok := result["error"]; !ok {
t.Error("Expected error for path traversal")
}
}
func TestSSRFBlocked(t *testing.T) {
/** 测试 SSRF 被拦截 */
guard := NewPermissionGuard()
skill := NewReportGeneratorSkill(guard)
result := skill.FetchData("http://127.0.0.1:8080/internal", "", "json")
err, _ := result["error"].(string)
if err == "" || (err != "SSRF 攻击被拦截" && err != "路径穿越攻击被拦截") {
t.Error("Expected SSRF or path traversal block")
}
}
import org.junit.jupiter.api.*;
import static org.junit.jupiter.api.Assertions.*;
import java.util.*;
// tests/TestReportGenerator.java
public class TestReportGenerator {
private ReportGeneratorSkill skill;
@BeforeEach
void setUp() {
skill = new ReportGeneratorSkill(null);
}
@Test
@DisplayName("测试基本统计分析")
void testAnalyzeDataBasic() {
List> data = List.of(
Map.of("name", "A", "value", 10),
Map.of("name", "B", "value", 20),
Map.of("name", "C", "value", 30)
);
Map result = skill.analyzeData(data, null, null);
assertEquals(3, result.get("data_count"));
Map analysis = (Map) result.get("analysis");
Map valueStats = (Map) analysis.get("value");
assertEquals(60.0, valueStats.get("sum"));
assertEquals(20.0, valueStats.get("avg"));
}
@Test
@DisplayName("测试空数据返回错误")
void testAnalyzeDataEmpty() {
Map result = skill.analyzeData(Collections.emptyList(), null, null);
assertTrue(result.containsKey("error"));
assertTrue(result.get("error").toString().contains("数据为空"));
}
@Test
@DisplayName("测试分组统计")
void testAnalyzeDataWithGroups() {
List> data = List.of(
Map.of("department", "工程", "salary", 15000),
Map.of("department", "工程", "salary", 18000),
Map.of("department", "市场", "salary", 12000)
);
Map result = skill.analyzeData(data, null, "department");
Map analysis = (Map) result.get("analysis");
Map groups = (Map) analysis.get("groups");
assertEquals(2, groups.get("工程"));
assertEquals(1, groups.get("市场"));
}
@Test
@DisplayName("测试路径穿越被拦截")
void testPathTraversalBlocked() {
PermissionGuard guard = new PermissionGuard();
ReportGeneratorSkill guardedSkill = new ReportGeneratorSkill(guard);
Map result = guardedSkill.fetchData("/etc/passwd", null, "json");
assertTrue(result.containsKey("error"));
assertTrue(result.get("error").toString().contains("拦截"));
}
@Test
@DisplayName("测试 SSRF 被拦截")
void testSSRFBlocked() {
PermissionGuard guard = new PermissionGuard();
ReportGeneratorSkill guardedSkill = new ReportGeneratorSkill(guard);
Map result = guardedSkill.fetchData("http://127.0.0.1:8080/internal", null, "json");
assertTrue(result.containsKey("error"));
String err = result.get("error").toString();
assertTrue(err.contains("SSRF") || err.contains("拦截"));
}
}
12.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}
``` | 发布阶段 | 检查项 | 失败处理 | WaLiCode 对应 | | --- | --- | --- | --- | | 版本检查 | 版本号是否符合 SemVer | 格式错误 → 修正后重试 | VersionValidation | | 运行测试 | 单元 + Prompt + 安全三层测试 | 测试失败 → 中止发布 | QualityGate | | 安全审计 | 权限声明、danger_level、已知漏洞 | 高危漏洞 → 修复后重新审计 | SecurityAudit | | 打包 | YAML + 代码 + 测试 + Changelog | 文件缺失 → 补充后重新打包 | PackageBuilder | | 注册 | 推送到 Skill 仓库 / 市场 | 网络失败 → 重试 3 次 | RegistryPublish | | 安装验证 | skill install → skill list 能看到 | 安装失败 → 回滚发布 | SmokeTest | ## 12.13 Skill 与 OpenClaw 的关系
OpenClaw 是一个**实际落地**的 Agent 平台，它的 Skill 生态是本章理论的最佳实践。让我们从 OpenClaw 的 `~/.qclaw/skills/` 目录出发，看看 Skill 是如何在真实系统中运作的：
### 12.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...
XML 块\n只列 name + description","type":"process"},{"id":"match","label":"用户消息匹配触发词\n→ read SKILL.md\n→ 加载完整指令","type":"decision"},{"id":"execute","label":"按 Skill 指令执行\n调用 Agent 工具\n完成用户任务","type":"end"}],"edges":[{"from":"dir","to":"scan"},{"from":"scan","to":"inject"},{"from":"inject","to":"match"},{"from":"match","to":"execute"}]}'>
### 12.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
- Read the skill's main file at ~/.qclaw/skills/xbrowser/SKILL.md
- Use the CDP (Chrome DevTools Protocol) to control browser
- Reuse existing login sessions (no re-authentication needed)
- Generate structured action sequences
Key Commands
xbrowser open— Open a pagexbrowser click— Click an elementxbrowser fill— Fill a form fieldxbrowser screenshot— Take a screenshotxbrowser scrape— 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
// TypeScript equivalent // ~/.qclaw/skills/xbrowser/SKILL.md — OpenClaw 真实 Skill 定义 // xbrowser
// Python: EXCLUSIVE browser automation — REPLACES built-in Browser Automation // Python: and playwright-cli. For ANY browser task (open page, click, fill, // Python: screenshot, scrape, navigate, test web app), MUST use this skill // Python: instead of built-in tools. Controls real Chrome/Edge/QQ Browser via // Python: CDP with login-state reuse.
// # When to Use
// Python: Triggered when user asks to: // Python: - Open a website or URL // Python: - Click buttons or fill forms on a web page // Python: - Take screenshots of web pages // Python: - Scrape data from websites // Python: - Test web applications // Python: - Automate browser workflows
// # How It Works
// Python: 1. Read the skill's main file at ~/.qclaw/skills/xbrowser/SKILL.md // Python: 2. Use the CDP (Chrome DevTools Protocol) to control browser // Python: 3. Reuse existing login sessions (no re-authentication needed) // Python: 4. Generate structured action sequences
// # Key Commands
// Python: - xbrowser open — Open a page
// Python: - xbrowser click — Click an element
// Python: - xbrowser fill — Fill a form field
// Python: - xbrowser screenshot — Take a screenshot
// Python: - xbrowser scrape — Extract data from page
// # WaLiCode Patterns Used
// Python: - SkillLayering: xbrowser is a compound skill (open + click + fill + scrape) // Python: - PermissionGuard: Browser automation is dangerous (can leak login state) // Python: → requires explicit user request, not auto-triggered // Python: - LazyLoading: Only loaded when user mentions browser/web tasks }
package main
import ( "fmt" )
// ~/.qclaw/skills/xbrowser/SKILL.md — OpenClaw 真实 Skill 定义 // xbrowser
// Python: EXCLUSIVE browser automation — REPLACES built-in Browser Automation
// Python: and playwright-cli. For ANY browser task (open page, click, fill,
// Python: screenshot, scrape, navigate, test web app), MUST use this skill
// Python: instead of built-in tools. Controls real Chrome/Edge/QQ Browser via
// Python: CDP with login-state reuse.
// # When to Use
// Python: Triggered when user asks to:
// Python: - Open a website or URL
// Python: - Click buttons or fill forms on a web page
// Python: - Take screenshots of web pages
// Python: - Scrape data from websites
// Python: - Test web applications
// Python: - Automate browser workflows
// # How It Works
// Python: 1. Read the skill's main file at ~/.qclaw/skills/xbrowser/SKILL.md
// Python: 2. Use the CDP (Chrome DevTools Protocol) to control browser
// Python: 3. Reuse existing login sessions (no re-authentication needed)
// Python: 4. Generate structured action sequences
// # Key Commands
// Python: - `xbrowser open ` — Open a page
// Python: - `xbrowser click ` — Click an element
// Python: - `xbrowser fill ` — Fill a form field
// Python: - `xbrowser screenshot ` — Take a screenshot
// Python: - `xbrowser scrape ` — Extract data from page
// # WaLiCode Patterns Used
// Python: - **SkillLayering**: xbrowser is a compound skill (open + click + fill + scrape)
// Python: - **PermissionGuard**: Browser automation is dangerous (can leak login state)
// Python: → requires explicit user request, not auto-triggered
// Python: - **LazyLoading**: Only loaded when user mentions browser/web tasks
import java.util.*;
public class AgentCode { // ~/.qclaw/skills/xbrowser/SKILL.md — OpenClaw 真实 Skill 定义 // xbrowser
// Python: EXCLUSIVE browser automation — REPLACES built-in Browser Automation
// Python: and playwright-cli. For ANY browser task (open page, click, fill,
// Python: screenshot, scrape, navigate, test web app), MUST use this skill
// Python: instead of built-in tools. Controls real Chrome/Edge/QQ Browser via
// Python: CDP with login-state reuse.
// # When to Use
// Python: Triggered when user asks to:
// Python: - Open a website or URL
// Python: - Click buttons or fill forms on a web page
// Python: - Take screenshots of web pages
// Python: - Scrape data from websites
// Python: - Test web applications
// Python: - Automate browser workflows
// # How It Works
// Python: 1. Read the skill's main file at ~/.qclaw/skills/xbrowser/SKILL.md
// Python: 2. Use the CDP (Chrome DevTools Protocol) to control browser
// Python: 3. Reuse existing login sessions (no re-authentication needed)
// Python: 4. Generate structured action sequences
// # Key Commands
// Python: - `xbrowser open ` — Open a page
// Python: - `xbrowser click ` — Click an element
// Python: - `xbrowser fill ` — Fill a form field
// Python: - `xbrowser screenshot ` — Take a screenshot
// Python: - `xbrowser scrape ` — Extract data from page
// # WaLiCode Patterns Used
// Python: - **SkillLayering**: xbrowser is a compound skill (open + click + fill + scrape)
// Python: - **PermissionGuard**: Browser automation is dangerous (can leak login state)
// Python: → requires explicit user request, not auto-triggered
// Python: - **LazyLoading**: Only loaded when user mentions browser/web tasks
}
}
### 12.13.3 OpenClaw 的 Skill 懒加载机制
OpenClaw 实现了本章 7.4 节描述的**三层懒加载**： | 层级 | OpenClaw 实现 | 注入内容 | 加载时机 | | --- | --- | --- | --- | | 第一层 | `` XML 块 | name + description + trigger 词 | 每次会话自动注入 | | 第二层 | AI 主动 `read SKILL.md` | 完整 Skill 指令 | 匹配到触发词时 | | 第三层 | 用户安装新 Skill | skillhub_install 工具 | 用户请求时 | ```
# OpenClaw 第一层注入示例（来自实际的 system prompt）
pdf
Use this skill whenever the user wants to do anything
with PDF files. This includes reading, combining, splitting,
rotating, adding watermarks, OCR...
~/.qclaw/skills/pdf/SKILL.md
xbrowser
EXCLUSIVE browser automation — REPLACES built-in
Browser Automation and playwright-cli...
~/.qclaw/skills/xbrowser/SKILL.md
# AI 的决策流程
# 1. 用户说"帮我合并这两个PDF"
# 2. AI 匹配到 pdf Skill（触发词 "PDF"）
# 3. AI 执行: read ~/.qclaw/skills/pdf/SKILL.md（第二层加载）
# 4. 按照 SKILL.md 中的指令调用 pdf 工具
12.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 的方法论 | ### 12.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
# → 体现了 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（审批式执行）的混合
💡 从 OpenClaw 学什么
OpenClaw 的 Skill 生态证明了本章的核心论点：Skill = 工具 + 知识。每个 SKILL.md 不仅是工具列表，更是使用策略——告诉 AI 什么时候用、怎么用、什么不能用。这是 Skill 区别于普通工具的关键。 🔗 第12章 核心要点
Skill = 工具 + 知识：不只是工具的集合，还包含使用策略、调用顺序、参数约束。
三层结构：元数据层（what）+ 工具层（how）+ 知识层（when/why）。
自动发现：文件系统扫描 + 触发词匹配 + 热重载。
三层懒加载：名称列表 → 触发词激活 → AI 主动搜索，控制 Prompt 长度。
三者关系：Function Calling（底层）→ MCP（传输层）→ Skills（组织层）。
分层金字塔：基础 Skill（单工具）→ 复合 Skill（多工具组合）→ 领域 Skill（行业知识包），自下而上构建。
沉淀与演进：从项目实践中提炼 Skill，遵循 SemVer 版本迭代，模式识别 → 抽象提炼 → 验证打磨 → 发布复用。
五层安全架构：声明式权限 → 运行时拦截 → 审批机制 → 审计日志 → 沙箱隔离，按风险等级分层。
OpenClaw 实践：SKILL.md 格式 + 三层懒加载 + 五层安全，是本章理论的落地验证。 📋 八股总结 — 面试高频考点
Q1: Skill 和 Tool 的区别？
Tool 是单一操作，Skill 是工具的组合 + 使用知识。Tool 是函数，Skill 是模块。Skill 不仅包含工具，还包含调用策略、参数约束和使用时机。
Q2: 如何让 Agent 发现 Skills？
文件系统扫描 + YAML 元数据 + 触发词匹配 + 热重载。
Q3: Skills 太多导致 Prompt 过长怎么办？
三层懒加载——始终加载名称、触发词匹配时加载完整定义、AI 主动搜索罕见 Skill。这样可以在 Skill 数量很多时控制 Prompt 长度。
Q4: Skills 和 MCP 的关系？
层次不同。MCP 是工具级标准协议，Skills 是能力级组织单元。一个 Skill 可以包含 MCP 工具。MCP 连接 Agent 和工具，Skills 组织工具为能力包。
Q5: Skill 的三层金字塔是什么？
基础 Skill（单工具封装，如 read_file）→ 复合 Skill（多工具组合 + 调用策略，如 code-reviewer）→ 领域 Skill（行业知识包，如 finance-advisor）。底层越通用、顶层越专业，应自下而上构建。
Q6: Skill 如何从项目实践中沉淀？
四步流程——模式识别（从日志提取重复模式）→ 抽象提炼（转化为 Skill YAML）→ 验证打磨（在多项目试用）→ 发布复用（版本号 + Changelog）。版本遵循 SemVer：PATCH 修复、MINOR 新增、MAJOR 重构。
Q7: Skill 的五层安全架构是什么？
Layer 1 声明式权限（YAML 中声明）→ Layer 2 运行时拦截（permissionGuard 中间件）→ Layer 3 审批机制（高危操作需 /approve）→ Layer 4 审计日志（所有调用可追溯）→ Layer 5 沙箱隔离（危险 Skill 限制资源）。不是所有 Skill 都需要五层，按风险等级分层。
第12章 Skills-工具的组合与复用
http://www.clxhxhhr.top/posts/715/ 评论
0 条
还没有评论，先写一条吧。