Hermes Agent Skills 自生成机制解析

┌─────────────────────────────────────┐
│        触发器（自生成核心）          │
│     迭代计数器 + 后台审查线程        │
└──────────────┬──────────────────────┘
               │
    ┌──────────┼──────────┐
    ▼          ▼          ▼
┌───────┐  ┌───────┐  ┌───────┐
│skill_ │  │skills_│  │本地存储 │
│manage │  │guard  │  │skills/ │
└───────┘  └───────┘  └───┬───┘
                          │
                          ▼
                    ┌───────────┐
                    │索引 → sp  │
                    └───────────┘

# Track tool-calling iterations for skill nudge.
# Counter resets whenever skill_manage is actually used.
if (self._skill_nudge_interval > 0
        and "skill_manage" in self.valid_tool_names):
    self._iters_since_skill += 1

_should_review_skills = False
if (self._skill_nudge_interval > 0
        and self._iters_since_skill >= self._skill_nudge_interval
        and "skill_manage" in self.valid_tool_names):
    _should_review_skills = True
    self._iters_since_skill = 0

def _spawn_background_review(self, messages_snapshot, review_memory=False, review_skills=False):
    import threading
    def _run_review():
        review_agent = AIAgent(
            model=self.model,
            max_iterations=8,          # 限制审查 agent 的迭代次数
            quiet_mode=True,           # 不输出到终端
            platform=self.platform,
            provider=self.provider,
        )
        review_agent._memory_store = self._memory_store
        review_agent._skill_nudge_interval = 0   # 禁止嵌套审查
        review_agent.run_conversation(
            user_message=prompt,
            conversation_history=messages_snapshot,
        )
    threading.Thread(target=_run_review, daemon=True).start()

_SKILL_REVIEW_PROMPT = (
    "Review the conversation above and consider saving or updating a skill if appropriate.\n\n"
    "Focus on: was a non-trivial approach used to complete a task that required trial "
    "and error, or changing course due to experiential findings along the way, or did "
    "the user expect or desire a different method or outcome?\n\n"
    "If a relevant skill already exists, update it with what you learned. "
    "Otherwise, create a new skill if the approach is reusable.\n"
    "If nothing is worth saving, just say 'Nothing to save.' and stop."
)

def _create_skill(name: str, content: str, category: str = None) -> Dict[str, Any]:
    # 1. 校验名称格式
    err = _validate_name(name)
    if err: return {"success": False, "error": err}

    # 2. 校验 frontmatter（必须包含 name 和 description）
    err = _validate_frontmatter(content)
    if err: return {"success": False, "error": err}

    # 3. 校验内容大小（上限 100,000 字符）
    err = _validate_content_size(content)
    if err: return {"success": False, "error": err}

    # 4. 检查名称冲突
    existing = _find_skill(name)
    if existing:
        return {"success": False, "error": f"A skill named '{name}' already exists..."}

    # 5. 原子写入 SKILL.md
    skill_dir = _resolve_skill_dir(name, category)
    skill_dir.mkdir(parents=True, exist_ok=True)
    _atomic_write_text(skill_md, content)

    # 6. 安全扫描 —— 不通过则回滚
    scan_error = _security_scan_skill(skill_dir)
    if scan_error:
        shutil.rmtree(skill_dir, ignore_errors=True)
        return {"success": False, "error": scan_error}

    return {"success": True, "message": f"Skill '{name}' created."}

def _security_scan_skill(skill_dir: Path) -> Optional[str]:
    if not _GUARD_AVAILABLE:
        return None
    result = scan_skill(skill_dir, source="agent-created")
    allowed, reason = should_allow_install(result)
    if allowed is False:
        report = format_scan_report(result)
        return f"Security scan blocked this skill ({reason}):\n{report}"

来源 ──→ 安装管道 ──→ 本地存储 ──→ 索引构建 ──→ 运行时加载
         (隔离区+安检)           (system prompt)   (/skill 命令)

远程下载 / skill_manage 写入
        │
        ▼
   隔离区（Quarantine）
        │
        ▼
   skills_guard 扫描
        │
   ┌────┴────┐
   ▼         ▼
 通过      失败
   │         │
   ▼         ▼
本地目录   回滚删除

def quarantine_bundle(bundle: SkillBundle) -> Path:
    """远程技能先写入隔离目录"""
    dest = QUARANTINE_DIR / skill_name
    # ... 写入文件
    return dest

def install_from_quarantine(quarantine_path, skill_name, category, bundle, scan_result) -> Path:
    """扫描通过后移入 skills 目录"""
    install_dir = SKILLS_DIR / category / skill_name
    shutil.move(str(quarantine_path), str(install_dir))
    lock.record_install(...)   # 记录到 lock.json

skill-name/
├── SKILL.md           # 主指令文件（必需）
├── references/        # 参考文档
├── templates/         # 输出模板
├── scripts/           # 可执行脚本
└── assets/            # 辅助资源

## Skills (mandatory)

Before replying, scan the skills below. If a skill matches or is even partially relevant
to your task, you MUST load it with skill_view(name) and follow its instructions.

productivity: - maps: Location intelligence — geocode, reverse-geocode, POI search, routing - linear: Manage Linear issues via GraphQL API
devops: - docker-deploy: Container build and deploy procedures

metadata:
  hermes:
    requires_toolsets: [terminal] # 仅在 terminal 工具集可用时显示
    fallback_for_toolsets: [mcp] # 当 mcp 工具集不可用时作为 fallback 显示

def _build_skill_message(loaded_skill, skill_dir, activation_note, ...):
    content = loaded_skill.get("content", "")

    # 1. 模板变量替换（${HERMES_SKILL_DIR}, ${HERMES_SESSION_ID}）
    content = _substitute_template_vars(content, skill_dir, session_id)

    # 2. 内联 shell 执行（!`date +%Y-%m-%d`）
    content = _expand_inline_shell(content, skill_dir, timeout)

    parts = [activation_note, "", content.strip()]

    # 3. 注入技能目录绝对路径
    if skill_dir:
        parts.append(f"[Skill directory: {skill_dir}]")

    # 4. 注入技能配置值
    _inject_skill_config(loaded_skill, parts)

    # 5. 自动发现 supporting files
    for subdir in ("references", "templates", "scripts", "assets"):
        # ... 收集并列出

    return "\n".join(parts)