第 17 章

Skill 注册与执行机制

第17章:Skill 注册与执行机制

在 Hermes Agent 的工具体系中,Skill(技能)是比单一工具调用更高层次的抽象。一个 Skill 可以封装一段完整的业务逻辑——它有自己的元数据描述、参数约束、执行步骤,以及失败后的回退策略。理解 Skill 的生命周期,是构建可维护、可扩展 Agent 系统的核心前提。


17.1 什么是 Skill?

Skill 是 Hermes Agent 对"可复用能力单元"的正式抽象。与直接调用工具(Tool)不同,Skill 具备以下特征:

维度 Tool Skill
粒度 单一原子操作(如 web_search) 多步组合逻辑
描述方式 函数签名 + docstring SKILL.md 文件
可发现性 运行时枚举 语义检索索引
状态管理 无状态 可携带执行上下文
错误处理 抛出异常 内置回退策略
复用方式 代码引用 元数据驱动

Skill 的核心价值在于声明式描述:你告诉 Agent "这个技能能做什么、怎么调用",而非"如何一步步实现"。Agent 在推理时自行决定何时、如何组合这些 Skill。


17.2 Skill 生命周期:从注册到反馈

Hermes 的 Skill 生命周期分为五个阶段,形成一个完整的闭环:

注册(Register)
     ↓
索引(Index)
     ↓
检索(Retrieve)
     ↓
执行(Execute)
     ↓
反馈(Feedback)
     ↓(更新索引权重)
注册(下一轮迭代)

17.2.1 阶段一:注册(Register)

注册是将 Skill 引入系统的入口。每个 Skill 以一个目录形式存在,目录中必须包含 SKILL.md 文件作为主描述文件。

# Skill 目录结构示例
skills/
├── web_research/
│   ├── SKILL.md          # 必需:技能描述文件
│   ├── skill.py          # 可选:Python 执行入口
│   ├── examples/         # 可选:调用示例
│   │   ├── example1.json
│   │   └── example2.json
│   └── tests/            # 可选:测试用例
│       └── test_web_research.py
├── code_review/
│   ├── SKILL.md
│   └── skill.py
└── data_analysis/
    ├── SKILL.md
    └── skill.py

注册过程通过 Hermes CLI 触发:

# 注册单个 Skill
hermes skill register ./skills/web_research

# 批量注册目录下所有 Skill
hermes skill register ./skills/ --recursive

# 注册并立即验证
hermes skill register ./skills/web_research --validate

# 查看已注册的 Skill 列表
hermes skill list

注册时,系统会:

  1. 解析 SKILL.md 文件,提取元数据
  2. 验证必填字段的完整性
  3. 检查依赖工具是否可用
  4. 分配唯一的 Skill ID

17.2.2 阶段二:索引(Index)

注册完成后,Skill 的描述文本会被向量化,写入语义检索索引。Hermes 默认使用本地嵌入模型(如 nomic-embed-text)生成向量:

# Hermes 内部索引逻辑(简化版)
from hermes.skills import SkillRegistry
from hermes.embeddings import LocalEmbedder

registry = SkillRegistry()
embedder = LocalEmbedder(model="nomic-embed-text")

def index_skill(skill_path: str):
    skill = registry.load(skill_path)
    
    # 构造索引文本:name + description + examples
    index_text = f"""
    技能名称: {skill.name}
    功能描述: {skill.description}
    使用场景: {skill.use_cases}
    示例查询: {' | '.join(skill.example_queries)}
    """
    
    # 生成嵌入向量
    vector = embedder.encode(index_text)
    
    # 写入向量数据库
    registry.vector_db.upsert(
        id=skill.id,
        vector=vector,
        metadata={
            "name": skill.name,
            "version": skill.version,
            "tags": skill.tags,
            "priority": skill.priority,
        }
    )
    
    return skill.id

17.2.3 阶段三:检索(Retrieve)

当 Agent 收到用户任务后,会对任务描述进行向量检索,找出最相关的 Skill 候选集:

# Skill 检索示例
def retrieve_skills(query: str, top_k: int = 5) -> list[Skill]:
    query_vector = embedder.encode(query)
    
    results = registry.vector_db.search(
        vector=query_vector,
        top_k=top_k,
        filter={"tags": {"$in": ["active", "stable"]}},
    )
    
    skills = []
    for result in results:
        skill = registry.get(result.id)
        skill.relevance_score = result.score
        skills.append(skill)
    
    return skills

# 在 Agent 推理阶段调用
relevant_skills = retrieve_skills(
    query="帮我分析这个 Python 文件的代码质量",
    top_k=3,
)
# 返回: [code_review (0.94), static_analysis (0.87), test_generation (0.71)]

17.2.4 阶段四:执行(Execute)

Agent 选定 Skill 后,进入执行阶段。执行遵循严格的参数绑定→预检查→运行→后处理流程:

# Skill 执行引擎(简化版)
class SkillExecutor:
    def execute(self, skill: Skill, params: dict, context: AgentContext) -> SkillResult:
        # 1. 参数绑定与验证
        bound_params = self._bind_params(skill.schema, params, context)
        
        # 2. 前置检查(权限、资源、依赖)
        self._pre_check(skill, bound_params, context)
        
        # 3. 执行核心逻辑
        try:
            result = skill.run(bound_params, context)
        except SkillTimeoutError:
            result = self._handle_timeout(skill, context)
        except SkillPermissionError as e:
            result = self._handle_permission(e, context)
        except Exception as e:
            result = self._handle_generic_error(skill, e, context)
        
        # 4. 后处理(格式化输出、日志记录)
        return self._post_process(skill, result, context)
    
    def _bind_params(self, schema: dict, params: dict, context: AgentContext) -> dict:
        bound = {}
        for field_name, field_schema in schema["properties"].items():
            if field_name in params:
                bound[field_name] = params[field_name]
            elif field_name in context.variables:
                # 从上下文变量自动绑定
                bound[field_name] = context.variables[field_name]
            elif "default" in field_schema:
                bound[field_name] = field_schema["default"]
            elif field_name in schema.get("required", []):
                raise MissingParameterError(f"必填参数缺失: {field_name}")
        return bound

17.2.5 阶段五:反馈(Feedback)

执行结果会被评分并反馈至索引系统,影响后续检索的排序权重:

# 反馈机制
class SkillFeedback:
    def record(
        self, 
        skill_id: str, 
        query: str, 
        result: SkillResult,
        user_rating: int | None = None,
    ):
        score = self._compute_score(result, user_rating)
        
        # 更新 Skill 的使用统计
        self.registry.update_stats(skill_id, {
            "execution_count": +1,
            "success_count": +1 if result.success else 0,
            "avg_latency_ms": result.latency_ms,
            "avg_score": score,
        })
        
        # 调整索引权重(使用率高的 Skill 检索优先级略微提升)
        self.registry.update_priority(skill_id, score)

17.3 SKILL.md 文件格式规范

SKILL.md 是 Skill 的"身份证",采用带 YAML Front Matter 的 Markdown 格式:

---
name: web_research
version: "1.2.0"
description: "通过多引擎网络搜索和内容提取,对指定主题进行深度调研,返回结构化的调研报告"
author: "NousResearch Team"
tags: [search, research, web, information-retrieval]
priority: 80
min_context_tokens: 8000
timeout_seconds: 120
requires_tools:
  - web_search
  - browser_navigate
  - text_extract
permissions:
  - network_access
  - no_file_write
use_cases:
  - "调研某个技术主题的最新进展"
  - "收集竞品信息并对比分析"
  - "验证某个事实声明的可信度"
example_queries:
  - "帮我研究一下 Rust 异步编程的最佳实践"
  - "查找关于量子计算商业化的最新新闻"
  - "调研 Hermes 模型与 GPT-4 的性能对比"
---

# Web Research Skill

## 功能描述

本技能通过以下步骤执行深度网络调研:

1. 对查询进行多角度分解,构造 3-5 个搜索子查询
2. 并发执行搜索,收集前 10 条结果
3. 对高质量来源进行全文提取
4. 交叉验证关键事实
5. 生成带引用的结构化报告

## 参数规范

| 参数名 | 类型 | 必填 | 说明 |
|--------|------|------|------|
| query | string | 是 | 调研主题或问题 |
| depth | enum | 否 | quick/normal/deep,默认 normal |
| language | string | 否 | 输出语言,默认 zh-CN |
| max_sources | integer | 否 | 最大来源数,默认 10 |

## 输出格式

```json
{
  "summary": "执行摘要(200字以内)",
  "key_findings": ["发现1", "发现2"],
  "sources": [{"title": "...", "url": "...", "relevance": 0.95}],
  "confidence": 0.87,
  "report": "完整报告正文(Markdown 格式)"
}

注意事项


### 关键字段说明

| 字段 | 类型 | 说明 |
|------|------|------|
| `name` | string | Skill 的唯一标识符,使用下划线命名 |
| `version` | semver | 语义化版本号,用于兼容性检查 |
| `priority` | 0-100 | 检索优先级基础分,越高越容易被选中 |
| `min_context_tokens` | integer | 执行所需最小上下文窗口 |
| `timeout_seconds` | integer | 执行超时时间 |
| `requires_tools` | list | 依赖的内置工具列表 |
| `permissions` | list | 所需权限声明 |

---

## 17.4 Skill 元数据结构(Python 类定义)

```python
from dataclasses import dataclass, field
from typing import Literal
from datetime import datetime

@dataclass
class SkillMetadata:
    """Skill 元数据的完整结构定义"""
    
    # 基础标识
    id: str                          # 系统分配的唯一 ID(UUID4)
    name: str                        # 技能名称(小写+下划线)
    version: str                     # 语义化版本
    description: str                 # 功能描述(用于 LLM 理解)
    
    # 分类与检索
    tags: list[str]                  # 标签列表
    priority: int = 50               # 基础优先级(0-100)
    use_cases: list[str] = field(default_factory=list)
    example_queries: list[str] = field(default_factory=list)
    
    # 技术约束
    min_context_tokens: int = 4000
    timeout_seconds: int = 60
    max_retries: int = 3
    requires_tools: list[str] = field(default_factory=list)
    
    # 权限控制
    permissions: list[str] = field(default_factory=list)
    sandbox: bool = False            # 是否在沙箱中执行
    
    # 参数 Schema(JSON Schema 格式)
    input_schema: dict = field(default_factory=dict)
    output_schema: dict = field(default_factory=dict)
    
    # 运行时统计(动态更新)
    execution_count: int = 0
    success_rate: float = 1.0
    avg_latency_ms: float = 0.0
    last_used: datetime | None = None
    
    # 作者信息
    author: str = ""
    source_path: str = ""


@dataclass  
class SkillResult:
    """Skill 执行结果"""
    skill_id: str
    success: bool
    output: dict | str | None
    error: str | None = None
    latency_ms: float = 0.0
    tool_calls_made: list[str] = field(default_factory=list)
    tokens_consumed: int = 0
    metadata: dict = field(default_factory=dict)

17.5 执行时的参数绑定机制

参数绑定(Parameter Binding)是 Skill 执行最关键的环节之一。Hermes 支持三层绑定策略,优先级从高到低:

显式参数(用户/Agent 直接提供)
    > 上下文变量(Context Variables)
        > Schema 默认值

17.5.1 显式参数绑定

Agent 在调用 Skill 时,会将从用户意图提取的参数直接传入:

# Agent 内部调用示例
skill_call = {
    "skill": "web_research",
    "params": {
        "query": "Hermes 4 与 GPT-4o 的 Function Calling 性能对比",
        "depth": "deep",
        "language": "zh-CN"
    }
}

17.5.2 上下文变量绑定

上下文变量来自当前对话的共享状态,避免重复传参:

# 上下文变量池示例
context = AgentContext(
    variables={
        "user_language": "zh-CN",      # 自动绑定到 language 参数
        "current_project": "hermes-guide",
        "working_directory": "/workspace",
        "session_id": "sess_abc123",
    },
    history=[...],  # 对话历史
    active_tools=[...],
)

17.5.3 类型强制转换

绑定时会自动进行类型转换和验证:

# 参数绑定时的类型处理
COERCE_MAP = {
    ("string", int): str,
    ("integer", str): int,
    ("boolean", str): lambda v: v.lower() in ("true", "1", "yes"),
    ("array", str): lambda v: [v],  # 字符串包装为单元素列表
}

def coerce_param(value, target_type: str):
    if isinstance(value, type_map[target_type]):
        return value
    
    coercer = COERCE_MAP.get((target_type, type(value)))
    if coercer:
        return coercer(value)
    
    raise ParameterTypeError(
        f"无法将 {type(value).__name__} 转换为 {target_type}"
    )

17.6 Skill 调用链与异常处理

复杂任务往往需要多个 Skill 串联执行,形成 Skill 调用链(Skill Chain)。

17.6.1 调用链定义

# 调用链配置示例:竞品调研 + 分析报告
skill_chain = SkillChain(
    name="competitive_analysis",
    steps=[
        SkillStep(
            skill="web_research",
            params={"query": "{product_name} 竞品分析"},
            output_var="raw_research",
        ),
        SkillStep(
            skill="data_analysis",
            params={
                "data": "{raw_research.key_findings}",
                "analysis_type": "comparison",
            },
            output_var="analysis_result",
            depends_on=["web_research"],  # 声明依赖关系
        ),
        SkillStep(
            skill="report_generation",
            params={
                "data": "{analysis_result}",
                "format": "markdown",
                "language": "zh-CN",
            },
            output_var="final_report",
            depends_on=["data_analysis"],
        ),
    ]
)

17.6.2 异常处理策略

Hermes 提供四种异常处理策略,在 Skill 定义中声明:

class ErrorStrategy(Enum):
    FAIL_FAST = "fail_fast"        # 立即失败,向上传播
    RETRY = "retry"                # 自动重试(使用 max_retries)
    FALLBACK = "fallback"          # 切换到备用 Skill
    CONTINUE = "continue"          # 记录错误但继续执行链

# 带错误处理的 SkillStep
SkillStep(
    skill="web_research",
    params={"query": "{query}"},
    error_strategy=ErrorStrategy.FALLBACK,
    fallback_skill="cached_search",      # 降级到缓存搜索
    fallback_params={"query": "{query}", "max_age_days": 7},
    retry_on=[NetworkTimeoutError, RateLimitError],
    max_retries=3,
    retry_backoff_seconds=2.0,
)

17.6.3 执行日志与追踪

# Skill 执行的追踪日志格式
{
  "trace_id": "trace_7f3a9b2c",
  "chain_name": "competitive_analysis",
  "steps": [
    {
      "step": 1,
      "skill": "web_research",
      "status": "success",
      "started_at": "2024-01-15T10:23:45.123Z",
      "duration_ms": 8432,
      "tokens_used": 2341,
      "output_size_bytes": 15240
    },
    {
      "step": 2,
      "skill": "data_analysis",
      "status": "retried_success",
      "attempt": 2,
      "error_on_attempt_1": "TimeoutError: exceeded 30s",
      "duration_ms": 4218,
      "tokens_used": 1876
    }
  ],
  "total_duration_ms": 12650,
  "total_tokens": 4217
}

17.7 实战:创建一个自定义 Skill

以下是一个完整示例:创建"代码安全扫描"Skill。

目录结构:

skills/code_security_scan/
├── SKILL.md
└── skill.py

SKILL.md:

---
name: code_security_scan
version: "1.0.0"
description: "对给定的代码文件或代码片段执行安全漏洞扫描,识别 OWASP Top 10 常见安全问题"
tags: [security, code-analysis, vulnerability, static-analysis]
priority: 75
timeout_seconds: 90
requires_tools:
  - terminal
  - file_read
permissions:
  - file_read
  - terminal_execute
---

## 参数

| 参数 | 类型 | 必填 | 说明 |
|------|------|------|------|
| target | string | 是 | 文件路径或代码片段 |
| language | string | 否 | 编程语言,自动检测 |
| severity_filter | string | 否 | high/medium/low,默认 all |

skill.py:

from hermes.skills import BaseSkill, SkillResult
from hermes.tools import terminal, file_read

class CodeSecurityScanSkill(BaseSkill):
    name = "code_security_scan"
    
    async def run(self, params: dict, context) -> SkillResult:
        target = params["target"]
        language = params.get("language", "auto")
        severity = params.get("severity_filter", "all")
        
        # 读取目标文件(如果是路径)
        if target.endswith((".py", ".js", ".go", ".java")):
            code = await file_read(target)
            language = self._detect_language(target) if language == "auto" else language
        else:
            code = target
        
        # 调用安全扫描工具
        scan_result = await terminal(
            f"bandit -r {target} --format json" 
            if language == "python" 
            else f"semgrep --config=auto {target} --json"
        )
        
        # 解析并过滤结果
        findings = self._parse_findings(scan_result, severity)
        
        return SkillResult(
            skill_id=self.id,
            success=True,
            output={
                "total_issues": len(findings),
                "findings": findings,
                "risk_score": self._compute_risk(findings),
                "recommendations": self._generate_recommendations(findings),
            }
        )

注册并使用:

# 注册
hermes skill register ./skills/code_security_scan

# 验证
hermes skill test code_security_scan \
  --params '{"target": "./app.py", "severity_filter": "high"}'

# 在 Agent 会话中使用
hermes chat
> 请帮我扫描 ./src/auth.py 的安全漏洞,只显示高危问题
# Agent 自动检索并调用 code_security_scan Skill

17.8 小结

本章系统讲解了 Hermes Agent 的 Skill 机制:

Skill 机制让 Hermes Agent 从"工具调用者"升级为"能力编排者",是构建企业级 Agent 应用的关键抽象层。

思考题

  1. 如果一个 Skill 执行失败率超过 20%,Hermes 的反馈机制会如何影响其检索权重?应该如何设计告警机制?

  2. 在一个 Skill 调用链中,如果中间步骤的输出格式不符合下一步骤的输入 Schema,你会如何设计一个"格式转换适配器"Skill?

  3. Skill 的 priority 字段是静态配置的,但实际业务中优先级可能随时间或用户角色变化。如何设计一个动态优先级系统?

本章评分
4.9  / 5  (20 评分)

💬 留言讨论