User Message
    │
    ▼
┌─────────────────────────────────┐
│   Agent Loop                    │
│   (run_conversation)            │
│                                 │
│   while budget_remaining:       │
│     response = LLM.call(...)    │
│     if tool_calls:              │
│       execute tools             │
│       append results            │
│     else:                       │
│       return response           │
│                                 │
│   ┌─────────────────────────┐   │
│   │ Self-Evaluation         │   │
│   │ Checkpoint              │   │
│   │ (每 15 次 tool call)     │   │
│   └─────────┬───────────────┘   │
│             ▼                   │
│   ┌─────────────────────────┐   │
│   │ Skill Creation / Update │   │
│   │ Memory Nudge            │   │
│   └─────────────────────────┘   │
└─────────────────────────────────┘

# run_agent.py - AIAgent.run_conversation() 核心循环（简化）
def run_conversation(self):
    while (api_call_count < self.max_iterations
           and self.iteration_budget.remaining > 0) \
           or self._budget_grace_call:

        if self._interrupt_requested:
            break

        # 1. 构建系统提示（含 skills 索引、记忆、上下文文件）
        system_prompt = self._build_system_prompt()

        # 2. 调用 LLM（支持多种 provider 和 API mode）
        response = client.chat.completions.create(
            model=model,
            messages=messages,
            tools=tool_schemas
        )

        # 3. 解析响应
        if response.tool_calls:
            for tool_call in response.tool_calls:
                # 执行工具调用（支持 ThreadPoolExecutor 并发）
                result = handle_function_call(
                    tool_call.name,
                    tool_call.args,
                    task_id
                )
                messages.append(tool_result_message(result))

                # ★ 自评价检查点：每 15 次 tool call 触发
                if self._should_evaluate():
                    self._learning_checkpoint(messages)

            api_call_count += 1
        else:
            # 纯文本响应 → 持久化记忆后返回
            self._flush_memory()
            return response.content

---
name: code-review
description: 执行标准代码审查工作流，检查 CI 状态、生成 diff 摘要、标记风格违规
version: 1.2.0
platforms: [macos, linux]
metadata:
  hermes:
    tags: [code-review, github, ci]
    related_skills: [github-pr-workflow, github-issues]
    config:
      - key: review.strictness
        description: 审查严格程度 (low/medium/high)
        default: "medium"
        prompt: 代码审查严格等级
---

# 代码审查技能

## 触发条件
当用户说"审查这个 PR"或"review PR #xxx"时自动加载。

## 步骤
1. 读取 PR 的变更文件列表
2. 对每个文件运行 linter 检查
3. 生成变更摘要报告
4. 发布审查评论到 PR

## 已知陷阱
- 超过 500 行变更的 PR，先要求作者拆分
- 确保仅审查变更行，不要审查整个文件

# agent/prompt_builder.py
def build_skills_system_prompt(
    available_tools=None,
    available_toolsets=None,
) -> str:
    """构建紧凑的技能索引，注入 system prompt。

    三层缓存：
      1. 进程内 LRU 字典（keyed by skills_dir + tools + toolsets）
      2. 磁盘快照 .skills_prompt_snapshot.json（跨进程重启）
      3. 完整文件系统扫描（缓存未命中时的回退）
    """

    # 索引行：仅包含技能名称 + 简短描述
    # 每行约 60-80 字符，100 个技能仅约 3K tokens
    index_lines = []
    for category in sorted(skills_by_category):
        for name, desc in sorted(...):
            if desc:
                index_lines.append(f"    - {name}: {desc}")
            else:
                index_lines.append(f"    - {name}")

    result = (
        "## Skills (mandatory)\n"
        "Before replying, scan the skills below. "
        "If a skill matches your task, you MUST load it with skill_view(name) "
        "and follow its instructions.\n"
        "<available_skills>\n"
        + "\n".join(index_lines) + "\n"
        "</available_skills>\n"
    )
    return result

# tools/skill_manager_tool.py
def skill_manage(action, name, content=None, category=None,
                 old_string=None, new_string=None, replace_all=False,
                 file_path=None, file_content=None):
    """
    Agent 管理自己的技能（过程性记忆）。

    Actions:
      create     - 创建新技能（完整 SKILL.md）
      patch      - 定向修补（old_string → new_string，首选方式）
      edit       - 整体重写（替换整个 SKILL.md）
      delete     - 删除技能
      write_file - 添加/更新附属文件
      remove_file - 删除附属文件
    """
    if action == "create":
        return _create_skill(name, content, category)
    elif action == "patch":
        return _patch_skill(name, old_string, new_string,
                            replace_all, file_path)
    elif action == "edit":
        return _edit_skill(name, content)
    elif action == "delete":
        return _delete_skill(name)
    elif action == "write_file":
        return _write_file(name, file_path, file_content)
    elif action == "remove_file":
        return _remove_file(name, file_path)

# tools/skill_manager_tool.py
def _patch_skill(name, old_string, new_string,
                 replace_all=False, file_path=None):
    # ... 定位技能文件 ...

    content = target.read_text(encoding="utf-8")

    # 使用 8 策略模糊匹配引擎
    # 处理：空白标准化、缩进差异、转义序列、块锚点匹配
    from tools.fuzzy_match import fuzzy_find_and_replace

    new_content, match_count, _strategy, match_error = \
        fuzzy_find_and_replace(content, old_string, new_string, replace_all)

    if match_error:
        # 返回文件预览以便模型自纠正
        preview = content[:500] + ("..." if len(content) > 500 else "")
        return {
            "success": False,
            "error": match_error,
            "file_preview": preview,
        }

    # 安全检查：注入检测 + 大小限制
    err = _validate_content_size(new_content)
    if err:
        return {"success": False, "error": err}

    # 原子写入（失败时回滚）
    _atomic_write_text(target, new_content)

    return {
        "success": True,
        "message": f"Patched skill '{name}' ({match_count} replacements).",
    }

def _create_skill(name, content, category=None):
    # 1. 从 content 解析 YAML frontmatter
    frontmatter, body = _parse_frontmatter(content)
    skill_name = frontmatter.get("name", name)

    # 2. 安全检查
    #    - 注入模式检测
    #    - 内容大小限制（默认 100K chars / 1 MiB）
    #    - 名称冲突检测
    if _detect_injection(content):
        return {"success": False, "error": "Potential prompt injection detected"}

    err = _validate_content_size(content)
    if err:
        return {"success": False, "error": err}

    # 3. 确定目标路径
    if category:
        skill_dir = SKILLS_DIR / category / skill_name
    else:
        skill_dir = SKILLS_DIR / skill_name
    skill_dir.mkdir(parents=True, exist_ok=True)

    # 4. 写入 SKILL.md
    target = skill_dir / "SKILL.md"
    _atomic_write_text(target, content)

    return {
        "success": True,
        "message": f"Skill '{skill_name}' created.",
        "path": str(target),
    }

_CONTENT_SIZE_LIMIT = 100_000       # 100K 字符
_CONTENT_BYTE_LIMIT = 1_048_576     # 1 MiB

def _validate_content_size(content: str, label="SKILL.md"):
    if len(content) > _CONTENT_SIZE_LIMIT:
        return (
            f"{label} content is {len(content):,} characters "
            f"(limit: {_CONTENT_SIZE_LIMIT:,}). Consider splitting "
            f"into a smaller SKILL.md with supporting files "
            f"in references/ or templates/."
        )
    if len(content.encode("utf-8")) > _CONTENT_BYTE_LIMIT:
        return f"{label} exceeds 1 MiB."
    return None  # OK

# agent/prompt_builder.py
SKILLS_GUIDANCE = (
    "After completing a complex task (5+ tool calls), fixing a tricky error, "
    "or discovering a non-trivial workflow, save the approach as a "
    "skill with skill_manage so you can reuse it next time.\n"
    "When using a skill and finding it outdated, incomplete, or wrong, "
    "patch it immediately with skill_manage(action='patch') — "
    "don't wait to be asked. "
    "Skills that aren't maintained become liabilities."
)

# agent/curator.py（核心逻辑）
class SkillCurator:
    def run(self):
        """Curator 运行循环"""
        # 1. 加载使用统计数据
        usage = self._load_usage_stats()  # ~/.hermes/skills/.usage.json

        for skill in self._get_agent_skills():
            state = usage.get(skill.name, {})

            # 2. 根据使用频率判断状态转移
            if self._is_stale(skill, state):
                # 超过 30 天未使用 → 归档
                self._archive_skill(skill)
            elif self._is_frequently_used(skill, state):
                # 高频使用 → 固定（pin）
                self._pin_skill(skill)

            # 3. 对活跃技能进行 LLM 质量审查
            if self._needs_review(skill, state):
                review = self._llm_review_skill(skill)
                if review.suggested_improvements:
                    self._apply_review(skill, review)

    def _archive_skill(self, skill):
        """将技能移到 .archive/ 目录，永不删除"""
        archive_dir = SKILLS_DIR / ".archive"
        shutil.move(str(skill.dir), str(archive_dir / skill.name))

# agent/memory_manager.py
class MemoryManager:
    def flush(self, conversation_history):
        """将当前会话的关键信息持久化到记忆文件"""
        memory_path = HERMES_HOME / "MEMORY.md"
        user_path = HERMES_HOME / "USER.md"

        # 提取关键事实、用户偏好、工作约定
        insights = self._extract_insights(conversation_history)

        # MEMORY.md：工作相关的记忆（API 端点、项目结构、密码等）
        with open(memory_path, "a") as f:
            for insight in insights.work_memories:
                f.write(f"- {insight}\n")

        # USER.md：用户画像（偏好、风格、约束）
        with open(user_path, "a") as f:
            for insight in insights.user_insights:
                f.write(f"- {insight}\n")

-- 会话历史索引表（简化）
CREATE VIRTUAL TABLE session_history USING fts5(
    content,          -- 会话内容
    metadata,         -- JSON 元数据（时间、平台、模型等）
    tokenize='porter' -- 词干分析器
);

-- 查询：10ms 内跨 10,000+ 文档检索
SELECT snippet(session_history, 1, '<b>', '</b>', '...', 32)
FROM session_history
WHERE session_history MATCH ?
ORDER BY rank
LIMIT 5;

# Honcho 用户建模（概念简化）
class HonchoModel:
    def evolve(self, interaction):
        """基于新交互更新用户模型"""

        # 1. 更新用户画像
        self.profile.update({
            "preferences": self._extract_preferences(interaction),
            "style": self._detect_communication_style(interaction),
            "constraints": self._extract_constraints(interaction),
        })

        # 2. 模式检测
        if self._detect_pattern(self.history[-3:]):
            # 发现重复模式 → 创建模式洞察
            self.insights.append(
                self._synthesize_pattern(self.history[-3:])
            )

        # 3. 洞察综合
        self.history.append(interaction)

1. Agent 收到请求
2. 扫描 skills_list（从 system prompt 中读取技能索引）
3. 没有匹配的 code-review 技能 → 从头推理
4. 执行 7 次 tool call：
   - gh pr view #42
   - gh pr diff
   - 对每个文件运行 lint
   - 总结变更
   - 发布评论
5. 成功完成 ✓
6. ★ 自评价检查点触发（7 > 5）
7. Agent 判断这是可复用的工作流
8. 调用 skill_manage(action='create', name='code-review', content=...)
9. 文件写入 ~/.hermes/skills/code-review/SKILL.md

1. Agent 收到请求
2. 扫描 skills_list → 发现 code-review 技能匹配
3. ★ 使用触发条件匹配！调用 skill_view("code-review")
4. 加载完整 SKILL.md → 按步骤执行
5. 遇到新问题：PR 包含二进制文件
6. Agent 发现技能中没有处理二进制文件的步骤
7. 完成审查后：
   → skill_manage(action='patch', name='code-review',
                  old_string='## 步骤\n1. 读取 PR 的变更文件列表',
                  new_string='## 步骤\n1. 读取 PR 的变更文件列表\n'
                             '2. 过滤二进制文件（.png, .ico 等），仅审查文本文件')
8. 技能从 5 步变成了 6 步，增加了边界条件处理

# rl_cli.py - Atropos RL 训练入口
class AtroposRLPipeline:
    def train_from_trajectories(self, trajectories_dir):
        """从交互轨迹进行强化学习训练"""

        # 1. 批量轨迹生成
        trajectories = self._load_trajectories(trajectories_dir)

        # 2. 轨迹压缩（减少 token 开销）
        compressed = trajectory_compressor.compress(trajectories)

        # 3. 支持 RLHF / DPO 训练
        #    用户评分、纠正标记、自动评估
        for trajectory in compressed:
            reward = self._compute_reward(trajectory)
            # RLHF: 用户反馈作为奖励信号
            # DPO: 偏好对比训练
            self._training_step(trajectory, reward)

        # 4. 导出为 ShareGPT 格式
        self._export_for_finetuning(compressed)

"""极简学习循环演示"""
import json
from pathlib import Path

SKILLS_DIR = Path.home() / ".demo-skills"
TOOL_CALL_THRESHOLD = 3  # 3+ 次 tool call 后触发学习

class LearningAgent:
    def __init__(self):
        self.tool_call_count = 0
        self.conversation_history = []
        SKILLS_DIR.mkdir(exist_ok=True)

    def run(self, user_input):
        self.conversation_history.append({"role": "user", "content": user_input})

        # 1. 构建提示词（含技能索引）
        prompt = self._build_prompt()

        # 2. 调用 LLM
        response = self._call_llm(prompt)

        # 3. 执行工具调用
        if response.get("tool_calls"):
            for tc in response["tool_calls"]:
                result = self._execute_tool(tc)
                self.conversation_history.append({
                    "role": "tool",
                    "content": result
                })
                self.tool_call_count += 1

            # ★ 自评价检查点
            if self.tool_call_count >= TOOL_CALL_THRESHOLD:
                self._learning_checkpoint()

            # 继续循环...
            return self.run(user_input)
        else:
            return response["content"]

    def _learning_checkpoint(self):
        """自评价：判断是否需要创建/更新技能"""
        # 提取本次任务的步骤
        steps = self._extract_steps()

        if not steps:
            return

        # 检测任务类型
        task_type = self._classify_task(steps)

        existing_skill = self._find_skill(task_type)

        if existing_skill:
            # 比较现有技能和实际执行步骤的差异
            new_steps = self._find_new_steps(existing_skill, steps)
            if new_steps:
                # Patch 技能：添加新发现的步骤
                self._patch_skill(existing_skill, new_steps)
                print(f"  → 技能 '{task_type}' 已更新 (+{len(new_steps)} steps)")
        else:
            # 创建新技能
            self._create_skill(task_type, steps)
            print(f"  → 新技能 '{task_type}' 已创建 ({len(steps)} steps)")

    def _create_skill(self, name, steps):
        """创建 SKILL.md 文件"""
        content = f"""---
name: {name}
description: 自动创建的技能
created_by: agent
---

# {name}

## 步骤
"""
        for i, step in enumerate(steps, 1):
            content += f"{i}. {step}\n"

        skill_dir = SKILLS_DIR / name
        skill_dir.mkdir(exist_ok=True)
        (skill_dir / "SKILL.md").write_text(content)

    def _patch_skill(self, name, new_steps):
        """修补技能：添加新步骤"""
        skill_file = SKILLS_DIR / name / "SKILL.md"
        content = skill_file.read_text()

        # 在 ## 步骤 部分追加新步骤
        old = "## 步骤\n"
        new = old + "\n".join(
            f"{i}. {step}" for i, step in enumerate(new_steps, 1)
        ) + "\n"
        content = content.replace(old, new, 1)

        skill_file.write_text(content)

agent = LearningAgent()
agent.run("帮我设置 CI/CD 流水线")
# 执行了 5 次 tool call
# → 自评价触发
# → 创建技能 'ci-cd-pipeline' (8 steps)

agent.run("再设置一个前端项目的 CI/CD")
# 技能匹配！加载 ci-cd-pipeline 技能
# 执行中发现缺少 npm install 步骤
# → 自动 patch 技能
# → 技能更新为 9 steps