编程 Hermes Agent 自进化架构全拆解:从 Learning Loop 到工程落地的深度实战

2026-04-24 20:03:22 +0800 CST views 7

Hermes Agent 自进化架构全拆解:从 Learning Loop 到工程落地的深度实战

引言:为什么 Agent 需要"进化"

如果你用过任何 AI Agent 工具,一定遇到过这样的痛点:每次对话都像失忆,昨天教它的事今天全忘了;同一个任务反复做,每次都要从头推理;技能全靠手动装,复杂一点的工作流就抓瞎。

这不是个别产品的问题,而是当前 Agent 架构的根本缺陷——它们不会从经验中学习

2026 年 2 月,Nous Research 开源的 Hermes Agent 用一个设计给出了不同的答案:Learning Loop(自学习循环)。截至 4 月下旬,这个项目已经突破 10 万 Star,持续霸榜 GitHub Trending。

但 Star 数不能说明一切。本文不聊"它有多强",而是把它的自进化架构拆开来看:Learning Loop 到底怎么运作?Skill 自动创建的技术实现是什么?分层记忆系统如何设计?自训练飞轮的工程挑战在哪里? 每一层我都配代码讲清楚,看完你不仅能理解 Hermes,还能把这些设计搬到自己的项目里。


一、整体架构:五层分离,自进化是核心

Hermes Agent 的架构可以用五层来概括:

┌─────────────────────────────────────────────┐
│            CLI / Gateway 接入层              │  ← 用户交互入口
├─────────────────────────────────────────────┤
│            Agent 核心调度层                  │  ← run_conversation 循环
├─────────────────────────────────────────────┤
│            LLM 对接层                       │  ← 多模型适配
├─────────────────────────────────────────────┤
│            Tool 执行层                      │  ← 工具调用系统
├─────────────────────────────────────────────┤
│            Memory + Skill 持久层            │  ← 自进化的根基
└─────────────────────────────────────────────┘

大部分 Agent 框架都有前四层,差异在第五层。Hermes 的 Memory + Skill 持久层不是一个简单的存储模块,而是三个子系统的组合:

  1. Skill 系统:自动创建、迭代优化的可复用技能文档
  2. 分层记忆:模拟人类记忆的工作/情景/语义/程序四层架构
  3. 自训练飞轮:对话轨迹回流为微调数据,模型越跑越好

这三者形成的闭环,才是 Hermes "越用越聪明"的真正引擎。


二、Learning Loop:从经验到能力的完整回路

2.1 核心循环拆解

Learning Loop 的本质是一个 OODA 循环(观察-导向-决策-行动)在 Agent 上的工程实现:

完成任务 → 触发复盘 → 提取模式 → 固化为 Skill
    ↑                                    │
    └────── 下次调用 Skill ←────────────┘

这不是"达到了阈值就写个文件"那么简单。Hermes 把它拆成了前台执行和后台复盘两个阶段。

2.2 前台执行:系统提示中的经验引导

agent/prompt_builder.py 中,Hermes 向 LLM 注入了一段 SKILLS_GUIDANCE:

# hermes/agent/prompt_builder.py (核心逻辑简化)

SKILLS_GUIDANCE = """
After completing a complex task (5+ tool calls), fixing a tricky error, 
or when the user corrects your approach, consider creating a Skill.

Skills are Markdown files that capture:
- What the task was
- The steps that worked (and why)
- Pitfalls to avoid next time
- Key decisions and their rationale

To create: skill_manage(action='create', name='descriptive-name', ...)
To update: skill_manage(action='patch', name='existing-skill', ...)
"""

def build_system_prompt(self, user_id: str, session_id: str) -> str:
    parts = [
        self.base_prompt,
        SKILLS_GUIDANCE,
        self._load_relevant_skills(user_id),
        self._load_memory_context(user_id, session_id),
        self._load_user_preferences(user_id),
    ]
    return "\n\n".join(p for p in parts if p)

注意这里的触发条件设计得很精妙:

  • 5+ 工具调用:排除简单的一次性查询,只对复杂任务触发
  • 错误恢复后:从错误中学习比从成功中学习更有价值
  • 用户纠正后:人工反馈是最高质量的训练信号

这三个条件不是随便定的,而是基于一个工程洞察:不是所有经验都值得沉淀。低价值经验(比如问天气、写个 hello world)如果都存下来,Skill 库会快速膨胀,检索噪声变大,反而降低后续任务的效率。

2.3 后台复盘:Skill 质量的二次保障

前台执行只是"提议创建 Skill",真正的质量把控在后台。Hermes 在对话循环结束后,会异步触发一个 Review 流程:

# hermes/agent/skill_reviewer.py (核心逻辑简化)

class SkillReviewer:
    """后台异步复盘,验证和优化前台创建的 Skill"""
    
    def __init__(self, llm_client, skill_store):
        self.llm = llm_client
        self.store = skill_store
    
    async def review_skill(self, skill_name: str, conversation_trace: list):
        """对一个新创建或更新的 Skill 进行复盘"""
        skill_content = self.store.read(skill_name)
        
        review_prompt = f"""
        Review this Skill that was auto-generated during a conversation:
        
        {skill_content}
        
        Conversation trace (last 10 turns):
        {self._format_trace(conversation_trace[-10:])}
        
        Check for:
        1. Are the steps accurate and complete?
        2. Are there any hallucinated commands or APIs?
        3. Is the skill too specific (overfitting) or too generic (useless)?
        4. Can it be merged with an existing skill?
        
        Output a revised version if needed, or 'KEEP_AS_IS' if the skill is good.
        """
        
        result = await self.llm.generate(review_prompt)
        
        if result.strip() != "KEEP_AS_IS":
            self.store.patch(skill_name, result)
            logger.info(f"Skill '{skill_name}' reviewed and updated")
        else:
            logger.info(f"Skill '{skill_name}' passed review")

这个前台提议 + 后台验证的分离架构,是 Hermes 的 Skill 系统区别于简单"自动保存对话"的关键。它解决了两个问题:

  1. 质量下限:LLM 在对话中创建的 Skill 可能有幻觉(编造不存在的命令),Review 机制可以捕获
  2. 去重合并:多个相似 Skill 可以在 Review 阶段被合并

2.4 Skill 的生命周期管理

一个 Skill 从创建到成熟,经历三个阶段:

# ~/.hermes/skills/clean-docker-images.md

---
name: clean-docker-images
description: 安全清理 Docker 无用镜像和容器
trigger: 
  - 用户要求清理 Docker
  - 磁盘空间不足且存在 Docker 镜像
category: devops
fallback_for_toolsets: [web]
requires_toolsets: [terminal]
version: 3  # ← 每次 patch 递增
created_at: 2026-03-15
last_used: 2026-04-20
use_count: 12
success_rate: 0.91  # ← 后台统计
---

# 清理 Docker 镜像

## 步骤

1. 先检查磁盘占用:
   ```bash
   docker system df
  1. 查看正在运行的容器(不要误删):

    docker ps --format '{{.Names}} {{.Image}}'
    
  2. 停止已退出的容器:

    docker container prune -f
    
  3. 删除悬空镜像(标签):

    docker image prune -f
    
  4. 如需深度清理,按大小排序列出镜像:

    docker images --format '{{.Repository}}:{{.Tag}} {{.Size}}' | sort -k2 -h -r
    

踩坑记录

  • ⚠️ docker system prune -a 会删除所有未使用的镜像,包括你可能还需要的
  • ⚠️ 某些 K8s 节点上的镜像不能随意删除,先检查是否有 Pod 在用
  • 💡 加 --filter "until=168h" 可以只删除7天前的

触发条件

  • 用户说"清理 Docker" / "Docker 占空间"
  • docker system df 显示 Reclaimable > 50%

注意这个 Skill 文件的设计有几个工程细节:

1. **YAML Front Matter**:结构化元数据,用于检索匹配和版本管理
2. **version 字段**:每次 patch 自增,可以回溯 Skill 的演化历史
3. **success_rate**:后台统计的成功率,低于阈值会触发自动 Review
4. **踩坑记录**:这是最有价值的部分,不是"怎么做",而是"什么会出错"

---

## 三、四层记忆架构:模拟人类认知

Hermes 的记忆系统不是简单的向量数据库 + RAG,而是模拟了认知科学中的人类记忆模型:

┌──────────────────────────────────────────┐
│ 工作记忆 Working Memory │ ← 当前对话上下文
│ 存储:内存 (dict/list) │ 生命周期:单次会话
├──────────────────────────────────────────┤
│ 情景记忆 Episodic Memory │ ← 历史对话记录
│ 存储:SQLite + FTS5 全文索引 │ 生命周期:永久
├──────────────────────────────────────────┤
│ 语义记忆 Semantic Memory │ ← 提取的知识和概念
│ 存储:SQLite + 向量嵌入 │ 生命周期:永久
├──────────────────────────────────────────┤
│ 程序记忆 Procedural Memory │ ← 技能和方法
│ 存储:Skill 文件 (Markdown) │ 生命周期:永久 + 迭代
└──────────────────────────────────────────┘


### 3.1 工作记忆:轻量但关键

工作记忆是当前对话的上下文窗口,它的核心挑战不是"记住",而是"别溢出"。

```python
# hermes/memory/working_memory.py (核心逻辑简化)

class WorkingMemory:
    """管理当前会话的上下文窗口"""
    
    def __init__(self, max_tokens: int = 8000):
        self.messages: list[dict] = []
        self.max_tokens = max_tokens
        self.compressed = False
    
    def add(self, role: str, content: str):
        self.messages.append({"role": role, "content": content})
        
        if self._estimate_tokens() > self.max_tokens:
            self._compress()
    
    def _compress(self):
        """当上下文过长时,压缩早期消息"""
        if len(self.messages) <= 4:
            return  # 至少保留最近2轮对话
        
        # 保留系统消息和最近2轮,中间的压缩成摘要
        system_msgs = [m for m in self.messages if m["role"] == "system"]
        recent = self.messages[-4:]  # 最近2轮
        middle = self.messages[len(system_msgs):-4]
        
        if middle:
            summary = self._summarize(middle)
            self.messages = system_msgs + [
                {"role": "system", "content": f"[Earlier context summary]\n{summary}"}
            ] + recent
            self.compressed = True
    
    def _summarize(self, messages: list[dict]) -> str:
        """用 LLM 生成中间对话的摘要"""
        # 实际实现中使用轻量模型来降低成本
        formatted = "\n".join(f"{m['role']}: {m['content'][:200]}" for m in messages)
        prompt = f"Summarize this conversation segment concisely, preserving key facts and decisions:\n{formatted}"
        return self._call_llm(prompt)
    
    def _estimate_tokens(self) -> int:
        """粗略估算 token 数(中文约1.5字/token,英文约4字符/token)"""
        total = 0
        for msg in self.messages:
            content = msg["content"]
            # 简化估算:实际项目中应使用 tiktoken
            total += len(content) // 3
        return total

这个压缩策略的关键点:

  1. 永远不压缩最近2轮:当前任务最相关的信息一定在最近的对话里
  2. 压缩而非丢弃:中间消息变成摘要,保留关键事实
  3. 标记压缩状态:后续检索可以知道当前上下文是否完整

3.2 情景记忆:SQLite + FTS5 的实战选择

Hermes 用 SQLite 做持久化存储,而不是向量数据库。这个选择背后有工程考量:

# hermes/memory/episodic_memory.py (核心逻辑简化)

import sqlite3
import json
from datetime import datetime

class EpisodicMemory:
    """情景记忆:存储所有历史交互数据"""
    
    def __init__(self, db_path: str = "~/.hermes/memory.db"):
        self.db_path = db_path
        self._init_db()
    
    def _init_db(self):
        conn = sqlite3.connect(self.db_path)
        conn.executescript("""
            CREATE TABLE IF NOT EXISTS conversations (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                user_id TEXT NOT NULL,
                session_id TEXT NOT NULL,
                role TEXT NOT NULL,
                content TEXT NOT NULL,
                timestamp DATETIME DEFAULT CURRENT_TIMESTAMP,
                metadata JSON
            );
            
            -- FTS5 全文索引,支持中文分词
            CREATE VIRTUAL TABLE IF NOT EXISTS conversations_fts 
            USING fts5(content, content='conversations', content_rowid='id',
                       tokenize='unicode61');
            
            -- 触发器:自动同步到 FTS 索引
            CREATE TRIGGER IF NOT EXISTS conversations_ai AFTER INSERT ON conversations
            BEGIN
                INSERT INTO conversations_fts(rowid, content) 
                VALUES (new.id, new.content);
            END;
            
            CREATE INDEX IF NOT EXISTS idx_conv_user_session 
            ON conversations(user_id, session_id, timestamp);
        """)
        conn.close()
    
    def store(self, user_id: str, session_id: str, role: str, 
              content: str, metadata: dict = None):
        """存储一条对话记录"""
        conn = sqlite3.connect(self.db_path)
        conn.execute(
            "INSERT INTO conversations (user_id, session_id, role, content, metadata) "
            "VALUES (?, ?, ?, ?, ?)",
            (user_id, session_id, role, content, json.dumps(metadata or {}))
        )
        conn.commit()
        conn.close()
    
    def search(self, user_id: str, query: str, limit: int = 10) -> list[dict]:
        """全文搜索历史对话"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.execute("""
            SELECT c.id, c.role, c.content, c.timestamp, c.metadata,
                   rank
            FROM conversations c
            JOIN conversations_fts f ON c.id = f.rowid
            WHERE c.user_id = ? AND conversations_fts MATCH ?
            ORDER BY rank
            LIMIT ?
        """, (user_id, query, limit))
        
        results = []
        for row in cursor:
            results.append({
                "id": row[0],
                "role": row[1],
                "content": row[2],
                "timestamp": row[3],
                "metadata": json.loads(row[4]) if row[4] else {},
                "relevance": row[5]
            })
        conn.close()
        return results
    
    def get_recent(self, user_id: str, limit: int = 20) -> list[dict]:
        """获取用户最近的对话记录"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.execute(
            "SELECT id, role, content, timestamp FROM conversations "
            "WHERE user_id = ? ORDER BY timestamp DESC LIMIT ?",
            (user_id, limit)
        )
        results = [{"id": r[0], "role": r[1], "content": r[2], "timestamp": r[3]} 
                   for r in cursor]
        conn.close()
        return results

为什么用 SQLite 而不是 Chroma/Pinecone/Weaviate?

  1. 零依赖部署:一行 curl 安装,不需要额外的数据库服务
  2. FTS5 足够好用:对于历史对话这种文本检索场景,全文索引的精度不比向量搜索差,而且没有嵌入模型的额外开销
  3. 单用户场景:Hermes 是个人 Agent,不是 SaaS,不需要分布式存储
  4. 事务安全:SQLite 的 ACID 特性保证对话记录不会丢失

3.3 语义记忆:向量嵌入层

对于需要语义理解(而非关键词匹配)的场景,Hermes 在 SQLite 之上加了一层向量嵌入:

# hermes/memory/semantic_memory.py (核心逻辑简化)

import numpy as np
import sqlite3
import json

class SemanticMemory:
    """语义记忆:通过向量嵌入实现语义检索"""
    
    def __init__(self, db_path: str, embed_dim: int = 1536):
        self.db_path = db_path
        self.embed_dim = embed_dim
        self._init_db()
    
    def _init_db(self):
        conn = sqlite3.connect(self.db_path)
        # SQLite 的 vec0 扩展或存储为 JSON blob
        conn.execute("""
            CREATE TABLE IF NOT EXISTS semantic_store (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                user_id TEXT NOT NULL,
                concept TEXT NOT NULL,
                embedding BLOB,
                source_ids JSON,
                created_at DATETIME DEFAULT CURRENT_TIMESTAMP
            )
        """)
        conn.close()
    
    def store_concept(self, user_id: str, concept: str, 
                      embedding: list[float], source_ids: list[int]):
        """存储一个提取的概念/知识点"""
        embedding_blob = np.array(embedding, dtype=np.float32).tobytes()
        conn = sqlite3.connect(self.db_path)
        conn.execute(
            "INSERT INTO semantic_store (user_id, concept, embedding, source_ids) "
            "VALUES (?, ?, ?, ?)",
            (user_id, concept, embedding_blob, json.dumps(source_ids))
        )
        conn.commit()
        conn.close()
    
    def search_similar(self, user_id: str, query_embedding: list[float], 
                       top_k: int = 5, threshold: float = 0.7) -> list[dict]:
        """向量相似度搜索"""
        query_vec = np.array(query_embedding, dtype=np.float32)
        
        conn = sqlite3.connect(self.db_path)
        cursor = conn.execute(
            "SELECT id, concept, embedding, source_ids FROM semantic_store "
            "WHERE user_id = ?",
            (user_id,)
        )
        
        results = []
        for row in cursor:
            stored_vec = np.frombuffer(row[2], dtype=np.float32)
            similarity = np.dot(query_vec, stored_vec) / (
                np.linalg.norm(query_vec) * np.linalg.norm(stored_vec) + 1e-8
            )
            if similarity >= threshold:
                results.append({
                    "id": row[0],
                    "concept": row[1],
                    "similarity": float(similarity),
                    "source_ids": json.loads(row[3])
                })
        
        results.sort(key=lambda x: x["similarity"], reverse=True)
        conn.close()
        return results[:top_k]

这里的工程取舍很明确:向量搜索在 SQLite 里做暴力扫描。对于个人 Agent 的数据量(几万条以内),这个方案完全够用,而且避免了引入额外的向量数据库依赖。

3.4 程序记忆:Skill 的存储与检索

程序记忆就是 Skill 系统,前面已经详细讲了 Skill 的结构。这里补充检索逻辑:

# hermes/memory/skill_retrieval.py (核心逻辑简化)

class SkillRetrieval:
    """从 Skill 库中检索匹配的技能"""
    
    def __init__(self, skills_dir: str = "~/.hermes/skills"):
        self.skills_dir = skills_dir
    
    def find_relevant(self, task_description: str, top_k: int = 3) -> list[dict]:
        """根据任务描述找到最相关的 Skill"""
        all_skills = self._load_all_skills()
        scored = []
        
        for skill in all_skills:
            score = self._compute_relevance(task_description, skill)
            if score > 0.3:  # 最低相关性阈值
                scored.append((skill, score))
        
        scored.sort(key=lambda x: x[1], reverse=True)
        return [s[0] for s in scored[:top_k]]
    
    def _compute_relevance(self, task: str, skill: dict) -> float:
        """计算任务与 Skill 的相关性"""
        score = 0.0
        
        # 1. 关键词匹配(权重 0.3)
        task_lower = task.lower()
        keywords = skill.get("keywords", []) + skill["name"].split("-")
        keyword_hits = sum(1 for kw in keywords if kw in task_lower)
        score += 0.3 * min(keyword_hits / max(len(keywords), 1), 1.0)
        
        # 2. 触发条件匹配(权重 0.4)
        triggers = skill.get("trigger", [])
        trigger_hits = sum(1 for t in triggers if t.lower() in task_lower)
        score += 0.4 * min(trigger_hits / max(len(triggers), 1), 1.0)
        
        # 3. 历史成功率加权(权重 0.3)
        success_rate = skill.get("success_rate", 0.5)
        use_count = skill.get("use_count", 0)
        # 使用贝叶斯平滑:成功率 + 先验
        smoothed = (success_rate * use_count + 0.5 * 10) / (use_count + 10)
        score += 0.3 * smoothed
        
        return score

注意这里的贝叶斯平滑——一个只用了 1 次且成功率 100% 的 Skill,不应该比用了 100 次且成功率 90% 的 Skill 排名更高。这是一个容易被忽略但很关键的工程细节。


四、run_conversation:一切逻辑的交汇点

Hermes 的核心是一个叫 run_conversation 的循环,它是 Agent 层、Tool 层、Memory 层、Skill 层的交汇点。理解这个循环,就理解了 Hermes 的全部运行逻辑。

# hermes/agent/conversation.py (核心逻辑简化)

class ConversationRunner:
    """Agent 的主循环"""
    
    def __init__(self, agent_config, llm_client, tool_executor, 
                 memory_system, skill_manager):
        self.config = agent_config
        self.llm = llm_client
        self.tools = tool_executor
        self.memory = memory_system
        self.skills = skill_manager
        self.max_turns = 25  # 单次对话最大轮次
    
    async def run(self, user_input: str, user_id: str, 
                  session_id: str) -> str:
        """执行一次完整的对话循环"""
        
        # 1. 构建系统提示(注入 Skill + Memory)
        system_prompt = self._build_system_prompt(user_id, session_id)
        
        # 2. 初始化消息列表
        messages = [
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": user_input}
        ]
        
        # 3. 持续对话直到完成
        turn_count = 0
        final_response = ""
        
        while turn_count < self.max_turns:
            turn_count += 1
            
            # 3a. 调用 LLM
            response = await self.llm.chat(
                messages=messages,
                tools=self.tools.get_tool_definitions(),
                temperature=0.1  # 低温度,追求确定性
            )
            
            # 3b. 处理响应
            if response.finish_reason == "stop":
                # 普通文本回复,对话结束
                final_response = response.content
                break
            
            elif response.finish_reason == "tool_calls":
                # 需要调用工具
                messages.append(response.to_message())
                
                for tool_call in response.tool_calls:
                    # 执行工具
                    result = await self.tools.execute(
                        tool_call.name, 
                        tool_call.arguments
                    )
                    messages.append({
                        "role": "tool",
                        "tool_call_id": tool_call.id,
                        "content": result
                    })
                    
                    # 关键:Skill 创建在这里触发
                    if tool_call.name == "skill_manage":
                        await self._on_skill_event(
                            tool_call.arguments, user_id, messages
                        )
        
        # 4. 对话结束后,异步触发复盘
        await self._post_conversation_review(
            user_id, session_id, messages
        )
        
        # 5. 持久化对话记录
        await self.memory.episodic.store_conversation(
            user_id, session_id, messages
        )
        
        return final_response
    
    def _build_system_prompt(self, user_id: str, session_id: str) -> str:
        """构建包含 Skill 和 Memory 上下文的系统提示"""
        parts = [self.config.base_prompt]
        
        # 加载相关 Skill
        relevant_skills = self.skills.find_relevant(
            self._get_last_user_message(),
            top_k=3
        )
        if relevant_skills:
            skill_section = "## Relevant Skills\n\n"
            for skill in relevant_skills:
                skill_section += f"### {skill['name']}\n{skill['content']}\n\n"
            parts.append(skill_section)
        
        # 加载工作记忆
        working_ctx = self.memory.working.get_context(session_id)
        if working_ctx:
            parts.append(f"## Current Context\n{working_ctx}")
        
        # 加载用户偏好
        prefs = self.memory.semantic.get_user_preferences(user_id)
        if prefs:
            parts.append(f"## User Preferences\n{prefs}")
        
        return "\n\n".join(parts)

这个循环的核心设计决策:

  1. Skill 和 Memory 在构建提示时注入,而不是作为独立模块在循环中轮转。这意味着 Skill 不是"另一个工具",而是 LLM 决策的先验知识。

  2. Skill 创建是工具调用,不是后台自动的。LLM 主动决定是否创建 Skill,后台只负责 Review。

  3. 对话结束后异步复盘,不阻塞用户等待。


五、自训练飞轮:从轨迹到数据

Hermes 最让我惊讶的设计是:它把对话轨迹回流为微调数据。这不是简单的日志记录,而是一套完整的数据生产线。

5.1 轨迹录制

每次 Agent 执行任务时,都会录制完整的轨迹(trajectory),包括每一步的思考、工具调用、结果:

{"prompt": "请帮我搜索AI领域的最新进展", "turns": [
  {"role": "assistant", "thinking": "用户想要搜索AI新闻,我需要使用搜索工具", "tool_calls": [{"name": "web_search", "arguments": {"query": "AI 最新进展 2026"}}]},
  {"role": "tool", "content": "搜索结果:1. OpenAI发布GPT-5..."},
  {"role": "assistant", "thinking": "搜索结果很丰富,我需要整理关键信息", "content": "根据搜索结果,以下是AI领域最新进展..."}
], "completed": true, "model": "anthropic/claude-4.6-opus"}

5.2 数据清洗:batch_runner.py

batch_runner.py 是 Hermes 自进化的数据工厂,能并行处理大量提示词:

# hermes/training/batch_runner.py (核心逻辑简化)

import asyncio
import json
from concurrent.futures import ThreadPoolExecutor

class BatchRunner:
    """批量生成训练数据"""
    
    def __init__(self, agent_factory, max_workers: int = 4):
        self.agent_factory = agent_factory
        self.max_workers = max_workers
    
    async def run(self, prompts_file: str, output_dir: str):
        """
        prompts_file: JSONL格式的提示词文件
        每行格式: {"prompt": "...", "category": "..."}
        """
        prompts = self._load_prompts(prompts_file)
        
        results = {"success": [], "failed": []}
        
        # 并行执行:每个提示词创建一个独立的 Agent 实例
        with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
            loop = asyncio.get_event_loop()
            futures = []
            
            for i, prompt_data in enumerate(prompts):
                future = loop.run_in_executor(
                    executor,
                    self._run_single,
                    prompt_data,
                    f"trajectory_{i}"
                )
                futures.append(future)
            
            for future in asyncio.as_completed(futures):
                try:
                    result = await future
                    if result["completed"]:
                        results["success"].append(result)
                    else:
                        results["failed"].append(result)
                except Exception as e:
                    logger.error(f"Trajectory generation failed: {e}")
        
        # 保存结果
        self._save_results(results, output_dir)
    
    def _run_single(self, prompt_data: dict, trajectory_id: str) -> dict:
        """运行单个 Agent 对话,录制轨迹"""
        agent = self.agent_factory.create()
        
        # 使用 Teacher 模型(默认 claude-opus-4.6)
        trajectory = agent.run_with_recording(
            prompt=prompt_data["prompt"],
            record_trajectory=True
        )
        
        return {
            "id": trajectory_id,
            "prompt": prompt_data["prompt"],
            "category": prompt_data.get("category", "general"),
            "trajectory": trajectory.to_dict(),
            "completed": trajectory.completed,
            "total_turns": len(trajectory.turns),
            "total_tool_calls": trajectory.tool_call_count
        }

5.3 数据质检:过滤残缺轨迹

不是所有轨迹都能用来训练。Hermes 用两个质检步骤过滤低质量数据:

# hermes/training/data_quality.py (核心逻辑简化)

class DataQualityChecker:
    """训练数据质量检查"""
    
    def check_trajectory(self, trajectory: dict) -> dict:
        """检查单条轨迹质量"""
        issues = []
        
        # 1. 检查推理标签完整性
        if self._has_incomplete_scratchpad(trajectory):
            issues.append("incomplete_scratchpad")
            # 因截断导致的数据残缺,直接丢弃
            return {"valid": False, "issues": issues}
        
        # 2. 检查是否有幻觉工具调用
        if self._has_hallucinated_tools(trajectory):
            issues.append("hallucinated_tools")
        
        # 3. 检查逻辑连贯性
        if not self._is_logically_coherent(trajectory):
            issues.append("incoherent_logic")
        
        # 4. 检查是否完成任务
        if not trajectory.get("completed", False):
            issues.append("incomplete_task")
            return {"valid": False, "issues": issues}
        
        return {
            "valid": len([i for i in issues if i in [
                "hallucinated_tools", "incoherent_logic"
            ]]) == 0,
            "issues": issues
        }
    
    def _has_incomplete_scratchpad(self, trajectory: dict) -> bool:
        """检测推理标签是否被截断"""
        for turn in trajectory.get("turns", []):
            thinking = turn.get("thinking", "")
            # 检查是否有未闭合的标签
            if thinking.count("<think>") != thinking.count("</think>"):
                return True
            if thinking.count("```") % 2 != 0:
                return True
        return False
    
    def _has_hallucinated_tools(self, trajectory: dict) -> bool:
        """检测是否有不存在的工具调用"""
        valid_tools = {"web_search", "file_read", "file_write", 
                       "terminal", "skill_manage", "browser_navigate"}
        for turn in trajectory.get("turns", []):
            for tc in turn.get("tool_calls", []):
                if tc["name"] not in valid_tools:
                    return True
        return False

5.4 格式转换:适配训练管线

最后一步是将轨迹转换为模型训练通用的格式,适配主流大模型的思维链训练要求:

# hermes/training/format_converter.py (核心逻辑简化)

class FormatConverter:
    """将轨迹转换为训练格式"""
    
    def to_chatml(self, trajectory: dict) -> list[dict]:
        """转换为 ChatML 格式"""
        messages = []
        
        for turn in trajectory["turns"]:
            if turn["role"] == "assistant":
                # 将 thinking 和 content 合并
                content = ""
                if turn.get("thinking"):
                    content += f"<think>\n{turn['thinking']}\n</think>\n"
                if turn.get("content"):
                    content += turn["content"]
                if turn.get("tool_calls"):
                    content += "\n" + self._format_tool_calls(turn["tool_calls"])
                messages.append({"role": "assistant", "content": content})
            else:
                messages.append({
                    "role": turn["role"],
                    "content": turn.get("content", "")
                })
        
        return messages
    
    def to_jsonl(self, trajectories: list[dict], output_path: str):
        """批量导出为 JSONL 文件"""
        with open(output_path, 'w', encoding='utf-8') as f:
            for traj in trajectories:
                messages = self.to_chatml(traj)
                record = {
                    "messages": messages,
                    "metadata": {
                        "category": traj.get("category", "general"),
                        "turn_count": len(messages),
                        "source": "hermes_trajectory"
                    }
                }
                f.write(json.dumps(record, ensure_ascii=False) + "\n")

这套自训练飞轮形成了一个完整的闭环:

用户使用 → 录制轨迹 → 质检过滤 → 格式转换 → 微调模型 → 更强的 Agent → 更好的轨迹

六、性能优化:长期运行的关键

一个需要 7×24 运行的 Agent,性能衰减是最大敌人。Hermes 用了几个关键优化:

6.1 动态上下文压缩

不是把所有历史都塞进提示,而是按需加载:

# hermes/agent/context_manager.py (核心逻辑简化)

class ContextManager:
    """动态管理上下文窗口"""
    
    def __init__(self, max_context_tokens: int = 32000, 
                 reserved_for_response: int = 4000):
        self.max_tokens = max_context_tokens
        self.reserved = reserved_for_response
        self.available = max_context_tokens - reserved_for_response
    
    def build_context(self, user_id: str, session_id: str, 
                      current_query: str) -> dict:
        """构建最优上下文窗口"""
        budget = self.available
        context = {}
        
        # 1. 系统提示(固定,约2000 tokens)
        system = self._get_system_prompt()
        budget -= self._count_tokens(system)
        
        # 2. 当前对话(最高优先级)
        conversation = self._get_current_conversation(session_id)
        conv_tokens = self._count_tokens(conversation)
        if conv_tokens <= budget * 0.5:
            context["conversation"] = conversation
            budget -= conv_tokens
        else:
            # 压缩早期对话
            compressed = self._compress_conversation(conversation, int(budget * 0.5))
            context["conversation"] = compressed
            budget -= self._count_tokens(compressed)
        
        # 3. 相关 Skill(次优先级)
        if budget > 1000:
            skills = self._get_relevant_skills(user_id, current_query)
            skills_text = self._format_skills(skills)
            skills_tokens = self._count_tokens(skills_text)
            if skills_tokens <= budget * 0.3:
                context["skills"] = skills_text
                budget -= skills_tokens
            else:
                # 只加载 Skill 的摘要
                summaries = self._get_skill_summaries(skills)
                context["skills"] = summaries
                budget -= self._count_tokens(summaries)
        
        # 4. 记忆片段(最低优先级,用剩余空间)
        if budget > 500:
            memories = self._get_relevant_memories(
                user_id, current_query, max_tokens=budget
            )
            context["memories"] = memories
        
        return context

6.2 Skill 去重与合并

长期运行后,Skill 库可能积累大量相似内容。Hermes 的 Review 机制会在后台做去重:

# hermes/agent/skill_dedup.py (核心逻辑简化)

class SkillDeduplicator:
    """检测和合并相似的 Skill"""
    
    def find_duplicates(self, user_id: str) -> list[list[str]]:
        """找出所有可能的重复 Skill 组"""
        skills = self._load_all_skills(user_id)
        groups = []
        visited = set()
        
        for i, s1 in enumerate(skills):
            if i in visited:
                continue
            group = [s1["name"]]
            visited.add(i)
            
            for j, s2 in enumerate(skills):
                if j in visited:
                    continue
                similarity = self._compute_similarity(s1, s2)
                if similarity > 0.75:  # 高度相似
                    group.append(s2["name"])
                    visited.add(j)
            
            if len(group) > 1:
                groups.append(group)
        
        return groups
    
    async def merge_skills(self, skill_names: list[str]) -> str:
        """将一组相似 Skill 合并成一个"""
        skills_content = []
        for name in skill_names:
            content = self._read_skill(name)
            skills_content.append(f"## Skill: {name}\n{content}")
        
        merge_prompt = f"""
        Merge these {len(skill_names)} similar skills into one comprehensive skill.
        Keep the best parts from each, remove redundancy.
        
        {chr(10).join(skills_content)}
        
        Output the merged skill in the standard Markdown format.
        """
        
        merged = await self.llm.generate(merge_prompt)
        return merged

6.3 内存使用优化

对于 7×24 运行的 Agent,内存泄漏是致命的。Hermes 使用了几种策略:

# hermes/agent/resource_manager.py (核心逻辑简化)

import gc
import resource
import psutil

class ResourceManager:
    """监控和优化运行时资源"""
    
    def __init__(self, max_memory_mb: int = 512):
        self.max_memory = max_memory_mb * 1024 * 1024
    
    def check_and_cleanup(self):
        """定期检查资源使用,必要时清理"""
        process = psutil.Process()
        memory_usage = process.memory_info().rss
        
        if memory_usage > self.max_memory * 0.8:
            logger.warning(
                f"Memory usage {memory_usage / 1024 / 1024:.0f}MB "
                f"exceeds 80% of limit"
            )
            self._aggressive_cleanup()
    
    def _aggressive_cleanup(self):
        """积极清理策略"""
        # 1. 清理对话缓存
        self._clear_old_conversation_cache(max_age_hours=24)
        
        # 2. 释放未使用的嵌入向量
        self._clear_embedding_cache()
        
        # 3. Python GC
        gc.collect()
        
        # 4. 关闭闲置的数据库连接
        self._close_idle_connections()
    
    def _clear_old_conversation_cache(self, max_age_hours: int):
        """清理过期的对话缓存"""
        cutoff = time.time() - max_age_hours * 3600
        keys_to_remove = [
            k for k, v in self._conversation_cache.items()
            if v["last_access"] < cutoff
        ]
        for k in keys_to_remove:
            del self._conversation_cache[k]
        logger.info(f"Cleared {len(keys_to_remove)} cached conversations")

七、与主流 Agent 框架的对比

维度Hermes AgentOpenClawLangChain AgentAutoGPT
核心定位自进化个人 Agent技能生态 Agent通用 Agent 框架自主任务 Agent
技能来源自动创建 + 迭代手动安装 Skill手动编码 Chain手动定义 Prompt
记忆架构四层分层配置式可插拔组件简单向量存储
自训练有(轨迹回流)
部署方式一行 curl一行 curlpip installgit clone
多平台内置 6 种网关插件式需自建
学习方式内置闭环外部技能市场人工编排Prompt 工程

Hermes 的差异点很明确:其他框架是"你给它能力",Hermes 是"它自己长能力"

但也要看到 Hermes 的局限:

  1. 单用户设计:不支持多租户,不适合做 SaaS
  2. SQLite 天花板:数据量大时暴力向量扫描会成为瓶颈
  3. Skill 质量依赖 LLM:Review 机制也是 LLM 做的,可能存在"自己审自己"的盲区
  4. 自训练成本:用 Claude Opus 做 Teacher 模型,成本不低

八、实战:30 分钟部署一个会学习的 Hermes

光说不练假把式。下面我带你从零部署 Hermes,并让它完成第一个自我学习的任务。

8.1 安装

# 一行安装(支持 Linux/macOS/WSL2)
curl -fsSL https://hermes.nousresearch.com/install.sh | bash

# 验证安装
hermes --version
# Output: hermes-agent 3.2.1

# 首次配置
hermes config set model anthropic/claude-sonnet-4
hermes config set api_key $ANTHROPIC_API_KEY

8.2 启动对话

hermes chat
🧠 Hermes Agent v3.2.1
Model: anthropic/claude-sonnet-4
Memory: ~/.hermes/memory.db (0 conversations)
Skills: ~/.hermes/skills/ (0 skills)

You: 帮我分析这个 Go 项目的代码结构,项目路径是 ~/projects/myapi

Hermes: 我来分析这个 Go 项目的结构。让我先看看项目目录...

[调用 terminal 工具执行 find ~/projects/myapi -type f -name "*.go"]

看到了,这是一个标准的 Go API 项目。让我看看关键文件...

[调用 file_read 工具读取 main.go, go.mod, router.go 等]

分析完成。这个项目的结构是:
- cmd/server/main.go:入口
- internal/handler/:HTTP handler 层
- internal/service/:业务逻辑层  
- internal/repository/:数据访问层
- pkg/middleware/:中间件

值得注意的是,你的 middleware 里没有限流组件,建议加上...

You: 好的,帮我加一个令牌桶限流中间件

Hermes: 好的,我来创建一个令牌桶限流中间件...

[创建文件 internal/middleware/ratelimit.go]
[修改 router.go 添加中间件注册]

中间件已添加。使用了 golang.org/x/time/rate 包实现令牌桶算法,
默认每秒100个请求,突发上限200...

[5+工具调用后,Hermes 自动触发 Skill 创建]

📝 Skill created: go-api-ratelimit-setup
   基于这次任务的经验,我记录了如何为 Go API 添加限流中间件的方法。

8.3 验证 Skill 创建

# 查看 Skill 列表
hermes skills list

# Output:
# NAME                       VERSION  USE_COUNT  SUCCESS_RATE
# go-api-ratelimit-setup     1        0          N/A

# 查看 Skill 内容
cat ~/.hermes/skills/go-api-ratelimit-setup.md

8.4 下次自动调用

下次你说"帮我给另一个项目加限流"时,Hermes 会自动检索到这个 Skill,直接复用经验,而不是从头推理。

8.5 部署为 7×24 服务

# 通过 Telegram 网关部署
hermes gateway add telegram --token $TELEGRAM_BOT_TOKEN

# 通过 Discord 网关部署
hermes gateway add discord --token $DISCORD_BOT_TOKEN

# 启动网关服务
hermes gateway start --daemon

# 现在你可以通过 Telegram/Discord 远程使用你的 Agent

九、二次开发:构建你自己的自进化 Agent

Hermes 的架构是可扩展的。如果你想基于它做二次开发,以下是几个关键入口:

9.1 添加自定义工具

# ~/.hermes/tools/custom_search.py

from hermes.tools import Tool, ToolParameter

class CustomSearchTool(Tool):
    """自定义搜索工具"""
    
    name = "custom_search"
    description = "Search internal knowledge base"
    parameters = [
        ToolParameter("query", "string", "Search query", required=True),
        ToolParameter("limit", "integer", "Max results", default=5),
    ]
    
    async def execute(self, query: str, limit: int = 5) -> str:
        # 你的搜索逻辑
        results = await your_search_api.search(query, limit=limit)
        return self._format_results(results)

# 注册工具
# 在 hermes config 中添加:
# tools_dir: ~/.hermes/tools/

9.2 自定义 Skill 触发策略

# ~/.hermes/plugins/conservative_skill_trigger.py

class ConservativeSkillTrigger:
    """更保守的 Skill 创建触发策略"""
    
    def should_create_skill(self, conversation: list[dict]) -> bool:
        """只有高质量对话才创建 Skill"""
        
        # 1. 工具调用次数 >= 8(比默认的5更严格)
        tool_calls = sum(1 for m in conversation if m.get("tool_calls"))
        if tool_calls < 8:
            return False
        
        # 2. 必须有用户明确肯定
        has_approval = any(
            "好的" in m.get("content", "") or "可以" in m.get("content", "")
            for m in conversation if m["role"] == "user"
        )
        if not has_approval:
            return False
        
        # 3. 没有产生过错误
        has_errors = any(
            "error" in m.get("content", "").lower()
            for m in conversation if m["role"] == "tool"
        )
        if has_errors:
            return False
        
        return True

十、总结与展望

Hermes Agent 的核心贡献不是又做了一个 Agent 框架,而是验证了一个设计理念:Agent 应该从经验中学习,而不是依赖人工编排

它的 Learning Loop 架构——前台执行 + 后台复盘、四层记忆、Skill 自动创建与迭代、自训练飞轮——每一层都有可以借鉴的工程实践。特别是几个设计决策值得记住:

  1. 选择性沉淀:不是所有经验都值得存,触发条件的设计比存储机制更重要
  2. 前台提议 + 后台验证:解决了 LLM 自我审查不可靠的问题
  3. SQLite 优先:个人 Agent 不需要分布式数据库,零依赖 > 功能丰富
  4. 贝叶斯平滑:评估 Skill 质量时,用先验平滑小样本偏差
  5. 轨迹回流训练:对话不只是服务,也是数据

当然,Hermes 也有明显的局限:单用户架构、SQLite 的性能天花板、Skill 质量依赖 LLM 自审、自训练的成本问题。这些都是它未来需要解决的。

但无论如何,它为 Agent 的下一步演进指明了一个方向:不是更多的工具和插件,而是让 Agent 自己学会使用工具。这个方向的探索才刚刚开始。


参考资源

  • Hermes Agent GitHub:https://github.com/NousResearch/hermes-agent
  • Hermes Agent 官方文档:https://hermes.nousresearch.com
  • agentskills.io 开放标准:https://agentskills.io

推荐文章

全栈工程师的技术栈
2024-11-19 10:13:20 +0800 CST
在 Rust 生产项目中存储数据
2024-11-19 02:35:11 +0800 CST
Nginx 状态监控与日志分析
2024-11-19 09:36:18 +0800 CST
JavaScript设计模式:单例模式
2024-11-18 10:57:41 +0800 CST
Vue3中的组件通信方式有哪些?
2024-11-17 04:17:57 +0800 CST
使用Vue 3实现无刷新数据加载
2024-11-18 17:48:20 +0800 CST
ElasticSearch 结构
2024-11-18 10:05:24 +0800 CST
JavaScript 实现访问本地文件夹
2024-11-18 23:12:47 +0800 CST
Vue中的异步更新是如何实现的?
2024-11-18 19:24:29 +0800 CST
MySQL数据库的36条军规
2024-11-18 16:46:25 +0800 CST
MyLib5,一个Python中非常有用的库
2024-11-18 12:50:13 +0800 CST
Roop是一款免费开源的AI换脸工具
2024-11-19 08:31:01 +0800 CST
PHP解决XSS攻击
2024-11-19 02:17:37 +0800 CST
Nginx负载均衡详解
2024-11-17 07:43:48 +0800 CST
php 连接mssql数据库
2024-11-17 05:01:41 +0800 CST
程序员茄子在线接单