编程 Hermes Agent 深度解析:当 AI Agent 学会「自我进化」——从四层记忆架构到闭环学习系统的技术内核

2026-05-17 09:20:03 +0800 CST views 5

Hermes Agent 深度解析:当 AI Agent 学会「自我进化」——从四层记忆架构到闭环学习系统的技术内核

一款能「记住你」「理解你」「与你共同成长」的开源智能体,如何在三个月内突破 15 万 Stars,超越 OpenClaw 登顶 OpenRouter 全球 Token 消耗榜首?

引言:AI Agent 的「失忆症」困境

如果你用过 ChatGPT、Claude 或任何 AI 对话工具,一定经历过这样的场景:

你:帮我写一个用户认证模块
AI:好的,这是代码...(生成代码)

(三天后)

你:继续完善上次那个认证模块
AI:抱歉,我没有之前的对话记录,能再描述一下吗?

这就是 AI Agent 的「失忆症」——每次对话都是独立的,无法跨会话记住上下文、用户偏好、项目背景。就像一条金鱼,每 7 秒重置一次记忆。

Hermes Agent 的诞生,正是为了解决这个问题。

2026 年 2 月,硅谷 AI 研究机构 Nous Research 开源了这款「自我进化型」AI Agent 框架。短短三个月,它在 GitHub 上突破 15 万 Stars,并在 2026 年 5 月超越 OpenClaw,登顶 OpenRouter 全球 Token 消耗量榜首——单日消耗 2710 亿 Token。

它的核心理念只有一句话:"The agent that grows with you"(与你共同成长的智能体)。

本文将从技术架构、核心原理、代码实现三个维度,深度解析 Hermes Agent 如何实现「自我进化」,以及它与传统 Agent 框架的本质差异。


一、技术背景:从「对话工具」到「执行系统」的演进

1.1 AI Agent 的三代演进

人工智能助手的发展经历了三个阶段:

代际时间代表产品核心能力局限性
第一代:问答型 AI2018-2022Siri、Alexa、Google Assistant语音识别、简单问答、设备控制无法理解复杂上下文,只能执行预定义指令
第二代:对话型 AI2022-2024ChatGPT、Claude、Gemini自然语言理解、多轮对话、知识问答只能「动口」不能「动手」,无法执行实际操作
第三代:执行型 AI Agent2024 至今OpenClaw、Devin、Hermes Agent理解意图、规划任务、调用工具、执行操作突破:从「回答问题」到「完成任务」的质变

Hermes Agent 属于第三代,但它在「执行型 Agent」的基础上,增加了一个关键能力:自我进化

1.2 为什么需要「自我进化」?

传统 Agent 的工作模式是线性的:

任务 A → 解决 → 结束 → 遗忘
任务 B → 解决 → 结束 → 遗忘
任务 N → 解决 → 结束 → 遗忘
(无法从经验中积累)

这导致几个严重问题:

  1. 重复劳动:相同的任务每次都要从头执行
  2. 无法学习:即使犯了错误,下次还会再犯
  3. 缺乏个性化:无法记住用户的偏好和习惯
  4. 效率瓶颈:复杂任务的执行时间无法优化

Hermes Agent 的设计目标是:

任务 A → 解决 → 提炼技能 → 存储
任务 B → 召回相关技能 → 更快解决 → 优化技能
任务 N → 技能库更丰富 → 解决更高效
(真正的经验积累)

根据 Nous Research 的测试数据,连续运行 3 个月后,65% 的新任务可以直接调用已有技能,平均执行时间减少 40%。


二、核心架构:四层记忆系统 + 闭环学习系统

Hermes Agent 的技术架构可以分为三个核心模块:

  1. 四层记忆系统:实现跨会话知识持久化
  2. 闭环学习系统:实现技能自动沉淀与优化
  3. 多平台消息网关:实现全渠道接入

2.1 四层记忆系统:模拟人类记忆逻辑

这是 Hermes Agent 最核心的技术创新。它模拟人类的记忆机制,设计了四层记忆体系:

┌─────────────────────────────────────────────────────────┐
│                    元记忆层 (Meta Memory)                │
│         管理其他三层记忆的刷新、检索和整合策略              │
├─────────────────────────────────────────────────────────┤
│                   长期记忆层 (Long-term Memory)          │
│         存储经过长期学习积累的知识、技能、用户偏好          │
├─────────────────────────────────────────────────────────┤
│                   工作记忆层 (Working Memory)            │
│         维护短期内的对话和交互信息,确保单次会话连贯性       │
├─────────────────────────────────────────────────────────┤
│                   即时记忆层 (Instant Memory)            │
│         处理当前输入信息,提供实时上下文感知               │
└─────────────────────────────────────────────────────────┘

2.1.1 即时记忆层(Instant Memory)

负责处理当前的输入信息,为 Agent 提供实时的上下文感知能力。

# 即时记忆的简化实现
class InstantMemory:
    def __init__(self):
        self.current_input = None
        self.context_window = []
        self.max_tokens = 4096
    
    def process_input(self, user_input: str):
        """处理用户输入,构建即时上下文"""
        self.current_input = user_input
        self.context_window.append({
            "role": "user",
            "content": user_input,
            "timestamp": datetime.now()
        })
        return self.build_context()
    
    def build_context(self) -> str:
        """构建发送给 LLM 的上下文"""
        # 截断到最大 token 数
        truncated = self._truncate_to_max_tokens(
            self.context_window, 
            self.max_tokens
        )
        return self._format_messages(truncated)

2.1.2 工作记忆层(Working Memory)

维护短期内的对话和交互信息,确保在单次会话中的连贯性。这与人类的「工作记忆」概念一致——容量有限,但可以快速访问。

class WorkingMemory:
    def __init__(self, session_id: str):
        self.session_id = session_id
        self.conversation_history = []
        self.active_tasks = []
        self.temporary_variables = {}
    
    def add_message(self, role: str, content: str):
        """添加消息到对话历史"""
        message = {
            "role": role,
            "content": content,
            "timestamp": datetime.now().isoformat(),
            "session_id": self.session_id
        }
        self.conversation_history.append(message)
    
    def set_variable(self, key: str, value: Any):
        """设置临时变量,用于任务执行过程中的状态传递"""
        self.temporary_variables[key] = value
    
    def get_relevant_context(self, query: str, top_k: int = 5) -> List[dict]:
        """获取与当前查询相关的上下文"""
        # 使用语义相似度检索
        embeddings = self._compute_embeddings(query)
        relevant = self._semantic_search(
            self.conversation_history, 
            embeddings, 
            top_k
        )
        return relevant

2.1.3 长期记忆层(Long-term Memory)

这是实现「真正智能」的关键。存储 Agent 经过长期学习和交互积累的知识,包括:

  • 事实知识:用户告诉 Agent 的客观事实
  • 程序性知识:如何完成某类任务的方法(即「技能」)
  • 偏好知识:用户的个人偏好和习惯

Hermes Agent 使用 SQLite + FTS5(全文检索) 实现长期记忆:

import sqlite3
from typing import List, Optional
import json

class LongTermMemory:
    def __init__(self, db_path: str = "~/.hermes/memory.db"):
        self.db_path = os.path.expanduser(db_path)
        self._init_database()
    
    def _init_database(self):
        """初始化数据库和 FTS5 索引"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        # 创建记忆表
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS memories (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                content TEXT NOT NULL,
                memory_type TEXT NOT NULL,  -- 'fact', 'skill', 'preference'
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                last_accessed TIMESTAMP,
                access_count INTEGER DEFAULT 0,
                importance_score REAL DEFAULT 0.5,
                metadata TEXT  -- JSON 格式的额外元数据
            )
        ''')
        
        # 创建 FTS5 全文检索虚拟表
        cursor.execute('''
            CREATE VIRTUAL TABLE IF NOT EXISTS memories_fts 
            USING fts5(
                content, 
                memory_type,
                content='memories', 
                content_rowid='id'
            )
        ''')
        
        conn.commit()
        conn.close()
    
    def store_memory(
        self, 
        content: str, 
        memory_type: str,
        importance: float = 0.5,
        metadata: dict = None
    ) -> int:
        """存储新的长期记忆"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        cursor.execute('''
            INSERT INTO memories (content, memory_type, importance_score, metadata)
            VALUES (?, ?, ?, ?)
        ''', (content, memory_type, importance, json.dumps(metadata or {})))
        
        memory_id = cursor.lastrowid
        conn.commit()
        conn.close()
        
        return memory_id
    
    def retrieve_memories(
        self, 
        query: str, 
        memory_type: Optional[str] = None,
        limit: int = 10
    ) -> List[dict]:
        """使用 FTS5 检索相关记忆"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        # 构建 FTS5 查询
        if memory_type:
            fts_query = f'''
                SELECT m.id, m.content, m.memory_type, m.importance_score, m.metadata
                FROM memories m
                JOIN memories_fts fts ON m.id = fts.rowid
                WHERE memories_fts MATCH ? AND m.memory_type = ?
                ORDER BY m.importance_score DESC, m.access_count DESC
                LIMIT ?
            '''
            cursor.execute(fts_query, (query, memory_type, limit))
        else:
            fts_query = f'''
                SELECT m.id, m.content, m.memory_type, m.importance_score, m.metadata
                FROM memories m
                JOIN memories_fts fts ON m.id = fts.rowid
                WHERE memories_fts MATCH ?
                ORDER BY m.importance_score DESC, m.access_count DESC
                LIMIT ?
            '''
            cursor.execute(fts_query, (query, limit))
        
        results = []
        for row in cursor.fetchall():
            results.append({
                "id": row[0],
                "content": row[1],
                "type": row[2],
                "importance": row[3],
                "metadata": json.loads(row[4]) if row[4] else {}
            })
        
        # 更新访问计数
        for result in results:
            cursor.execute('''
                UPDATE memories 
                SET access_count = access_count + 1, 
                    last_accessed = CURRENT_TIMESTAMP
                WHERE id = ?
            ''', (result["id"],))
        
        conn.commit()
        conn.close()
        
        return results

2.1.4 元记忆层(Meta Memory)

这是整个记忆系统的「大脑」,负责管理其他三层记忆的刷新、检索和整合策略。

class MetaMemory:
    def __init__(self):
        self.instant_memory = InstantMemory()
        self.working_memory = WorkingMemory(session_id=self._generate_session_id())
        self.long_term_memory = LongTermMemory()
        
        # 记忆整合策略配置
        self.consolidation_threshold = 0.7  # 重要性阈值
        self.forgetting_curve_factor = 0.1  # 遗忘曲线系数
    
    def consolidate_memories(self):
        """将工作记忆中的重要内容整合到长期记忆"""
        for message in self.working_memory.conversation_history:
            # 计算重要性分数
            importance = self._calculate_importance(message)
            
            if importance >= self.consolidation_threshold:
                # 判断记忆类型
                memory_type = self._classify_memory_type(message)
                
                # 存储到长期记忆
                self.long_term_memory.store_memory(
                    content=message["content"],
                    memory_type=memory_type,
                    importance=importance,
                    metadata={
                        "session_id": self.working_memory.session_id,
                        "original_timestamp": message["timestamp"]
                    }
                )
    
    def _calculate_importance(self, message: dict) -> float:
        """计算消息的重要性分数"""
        score = 0.5  # 基础分数
        
        # 因素 1:消息长度(更长的消息可能更重要)
        content_length = len(message["content"])
        if content_length > 500:
            score += 0.1
        
        # 因素 2:包含关键词
        important_keywords = ["重要", "记住", "偏好", "习惯", "规则", "约定"]
        for keyword in important_keywords:
            if keyword in message["content"]:
                score += 0.15
        
        # 因素 3:用户明确表示重要
        if message.get("marked_important"):
            score += 0.2
        
        return min(score, 1.0)
    
    def build_full_context(self, current_query: str) -> str:
        """构建完整的上下文,整合所有记忆层"""
        context_parts = []
        
        # 1. 即时上下文
        instant_ctx = self.instant_memory.build_context()
        context_parts.append(f"[即时上下文]\n{instant_ctx}")
        
        # 2. 工作记忆中的相关内容
        working_ctx = self.working_memory.get_relevant_context(current_query)
        if working_ctx:
            context_parts.append(f"[会话上下文]\n{self._format_working_context(working_ctx)}")
        
        # 3. 长期记忆中的相关内容
        long_term_ctx = self.long_term_memory.retrieve_memories(current_query, limit=5)
        if long_term_ctx:
            context_parts.append(f"[长期记忆]\n{self._format_long_term_context(long_term_ctx)}")
        
        return "\n\n".join(context_parts)

2.2 闭环学习系统:从「执行」到「进化」

四层记忆解决了「记住」的问题,闭环学习系统则解决「学习」的问题。

Hermes Agent 的学习闭环包含五个阶段:

┌──────────────────────────────────────────────────────────────┐
│                                                              │
│   ┌─────────┐    ┌─────────┐    ┌─────────┐                │
│   │  执行   │───▶│  提炼   │───▶│  沉淀   │                │
│   │ Execute │    │ Extract │    │Consolid │                │
│   └─────────┘    └─────────┘    └─────────┘                │
│        ▲                              │                      │
│        │                              ▼                      │
│   ┌─────────┐                   ┌─────────┐                │
│   │  自省   │◀──────────────────│  复用   │                │
│   │Reflect  │                   │ Reuse   │                │
│   └─────────┘                   └─────────┘                │
│                                                              │
└──────────────────────────────────────────────────────────────┘

2.2.1 执行阶段(Execute)

Agent 完成用户指派的任务,记录完整的执行轨迹。

from dataclasses import dataclass, field
from typing import List, Dict, Any
from datetime import datetime

@dataclass
class ExecutionTrace:
    """执行轨迹记录"""
    task_id: str
    task_description: str
    start_time: datetime
    end_time: datetime = None
    steps: List[Dict[str, Any]] = field(default_factory=list)
    tools_used: List[str] = field(default_factory=list)
    success: bool = False
    error_message: str = None
    
    def add_step(self, step_type: str, content: str, metadata: dict = None):
        self.steps.append({
            "type": step_type,
            "content": content,
            "timestamp": datetime.now().isoformat(),
            "metadata": metadata or {}
        })
    
    def record_tool_use(self, tool_name: str, inputs: dict, outputs: dict):
        self.tools_used.append(tool_name)
        self.add_step(
            step_type="tool_call",
            content=f"调用工具: {tool_name}",
            metadata={"inputs": inputs, "outputs": outputs}
        )

class ExecutionEngine:
    def __init__(self, agent):
        self.agent = agent
        self.current_trace = None
    
    def execute_task(self, task_description: str) -> ExecutionTrace:
        """执行任务并记录轨迹"""
        self.current_trace = ExecutionTrace(
            task_id=self._generate_task_id(),
            task_description=task_description,
            start_time=datetime.now()
        )
        
        try:
            # 解析任务
            plan = self.agent.plan(task_description)
            self.current_trace.add_step("planning", f"生成计划: {plan}")
            
            # 执行计划中的每个步骤
            for step in plan.steps:
                result = self._execute_step(step)
                self.current_trace.add_step(
                    "execution", 
                    f"执行步骤: {step.description}",
                    {"result": result}
                )
            
            self.current_trace.success = True
            
        except Exception as e:
            self.current_trace.success = False
            self.current_trace.error_message = str(e)
        
        finally:
            self.current_trace.end_time = datetime.now()
        
        return self.current_trace

2.2.2 提炼阶段(Extract)

分析执行轨迹,提取可复用的模式。

class PatternExtractor:
    def extract_patterns(self, trace: ExecutionTrace) -> List[dict]:
        """从执行轨迹中提取模式"""
        patterns = []
        
        # 1. 提取工具调用序列模式
        tool_sequence = self._extract_tool_sequence(trace)
        if self._is_reusable_pattern(tool_sequence):
            patterns.append({
                "type": "tool_sequence",
                "pattern": tool_sequence,
                "frequency": self._calculate_frequency(tool_sequence)
            })
        
        # 2. 提取决策点模式
        decision_points = self._extract_decision_points(trace)
        for dp in decision_points:
            patterns.append({
                "type": "decision_point",
                "pattern": dp,
                "context": dp["context"]
            })
        
        # 3. 提取成功策略
        if trace.success:
            successful_strategies = self._extract_strategies(trace)
            patterns.extend([
                {"type": "strategy", "pattern": s} 
                for s in successful_strategies
            ])
        
        return patterns
    
    def _is_reusable_pattern(self, tool_sequence: List[str]) -> bool:
        """判断工具序列是否可复用"""
        # 条件 1:序列长度适中(太短没价值,太长太具体)
        if not (3 <= len(tool_sequence) <= 10):
            return False
        
        # 条件 2:序列在历史中出现过多次
        frequency = self._calculate_frequency(tool_sequence)
        return frequency >= 2

2.2.3 沉淀阶段(Consolidate)

将提取的模式封装为可复用的技能(Skill)。

@dataclass
class Skill:
    """技能定义"""
    name: str
    description: str
    trigger_conditions: List[str]  # 触发条件
    steps: List[Dict[str, Any]]    # 执行步骤
    parameters: Dict[str, Any]     # 参数定义
    version: str = "1.0.0"
    created_at: datetime = field(default_factory=datetime.now)
    usage_count: int = 0
    success_rate: float = 0.0

class SkillConsolidator:
    def __init__(self, memory: LongTermMemory):
        self.memory = memory
        self.skills_dir = Path("~/.hermes/skills").expanduser()
    
    def consolidate_to_skill(
        self, 
        pattern: dict, 
        trace: ExecutionTrace
    ) -> Skill:
        """将模式沉淀为技能"""
        # 生成技能名称和描述
        skill_name = self._generate_skill_name(pattern, trace)
        description = self._generate_description(trace)
        
        # 提取触发条件
        trigger_conditions = self._extract_triggers(trace)
        
        # 规范化执行步骤
        steps = self._normalize_steps(trace.steps)
        
        # 提取参数
        parameters = self._extract_parameters(steps)
        
        skill = Skill(
            name=skill_name,
            description=description,
            trigger_conditions=trigger_conditions,
            steps=steps,
            parameters=parameters
        )
        
        # 保存技能
        self._save_skill(skill)
        
        # 存储到长期记忆
        self.memory.store_memory(
            content=f"技能: {skill_name}\n{description}",
            memory_type="skill",
            importance=0.8,
            metadata={"skill_name": skill_name, "version": skill.version}
        )
        
        return skill
    
    def _save_skill(self, skill: Skill):
        """保存技能到文件系统"""
        skill_path = self.skills_dir / f"{skill.name}.yaml"
        
        skill_data = {
            "name": skill.name,
            "description": skill.description,
            "trigger_conditions": skill.trigger_conditions,
            "steps": skill.steps,
            "parameters": skill.parameters,
            "version": skill.version,
            "created_at": skill.created_at.isoformat()
        }
        
        with open(skill_path, 'w', encoding='utf-8') as f:
            yaml.dump(skill_data, f, allow_unicode=True)

2.2.4 复用阶段(Reuse)

在遇到相似任务时,自动调用已有技能。

class SkillMatcher:
    def __init__(self, skills_dir: Path):
        self.skills = self._load_all_skills(skills_dir)
    
    def find_matching_skill(
        self, 
        task_description: str
    ) -> Optional[Skill]:
        """找到匹配当前任务的技能"""
        # 计算任务与每个技能触发条件的相似度
        best_match = None
        best_score = 0.0
        
        for skill in self.skills:
            for trigger in skill.trigger_conditions:
                score = self._compute_similarity(task_description, trigger)
                if score > best_score:
                    best_score = score
                    best_match = skill
        
        # 只有相似度超过阈值才返回
        if best_score >= 0.7:
            return best_match
        
        return None
    
    def _compute_similarity(self, text1: str, text2: str) -> float:
        """计算文本相似度(使用嵌入向量)"""
        emb1 = self._get_embedding(text1)
        emb2 = self._get_embedding(text2)
        
        # 余弦相似度
        similarity = np.dot(emb1, emb2) / (np.linalg.norm(emb1) * np.linalg.norm(emb2))
        return float(similarity)

2.2.5 自省阶段(Reflect)

分析技能执行结果,优化技能定义。

class SkillOptimizer:
    def optimize_skill(
        self, 
        skill: Skill, 
        execution_result: ExecutionTrace
    ) -> Skill:
        """根据执行结果优化技能"""
        # 更新使用统计
        skill.usage_count += 1
        
        # 更新成功率
        if execution_result.success:
            current_successes = skill.success_rate * (skill.usage_count - 1)
            skill.success_rate = (current_successes + 1) / skill.usage_count
        else:
            current_successes = skill.success_rate * (skill.usage_count - 1)
            skill.success_rate = current_successes / skill.usage_count
        
        # 如果成功率下降,分析失败原因并调整
        if not execution_result.success and skill.success_rate < 0.5:
            failure_analysis = self._analyze_failure(execution_result)
            skill = self._adjust_skill(skill, failure_analysis)
        
        # 如果成功率很高且使用次数多,考虑泛化
        if skill.success_rate > 0.9 and skill.usage_count > 10:
            skill = self._generalize_skill(skill)
        
        return skill
    
    def _analyze_failure(self, trace: ExecutionTrace) -> dict:
        """分析失败原因"""
        analysis = {
            "failed_step": None,
            "error_type": None,
            "suggested_fix": None
        }
        
        # 找到失败的步骤
        for step in reversed(trace.steps):
            if step.get("error"):
                analysis["failed_step"] = step
                analysis["error_type"] = self._classify_error(step["error"])
                analysis["suggested_fix"] = self._suggest_fix(step)
                break
        
        return analysis

2.3 多平台消息网关

Hermes Agent 支持通过多种消息平台访问:

平台状态特点
Telegram✅ 原生支持最稳定,推荐生产使用
Discord✅ 原生支持适合团队协作
Slack✅ 原生支持企业集成
WhatsApp✅ 原生支持个人使用
飞书✅ 原生支持国内企业
钉钉✅ 原生支持国内企业
微信✅ 原生支持实验性功能

消息网关的统一接口:

from abc import ABC, abstractmethod
from typing import AsyncIterator

class MessageGateway(ABC):
    """消息网关抽象基类"""
    
    @abstractmethod
    async def connect(self):
        """建立连接"""
        pass
    
    @abstractmethod
    async def listen(self) -> AsyncIterator[Message]:
        """监听消息流"""
        pass
    
    @abstractmethod
    async def send(self, message: Message):
        """发送消息"""
        pass

class TelegramGateway(MessageGateway):
    def __init__(self, bot_token: str):
        self.bot_token = bot_token
        self.client = None
    
    async def connect(self):
        from telegram import Bot
        self.client = Bot(token=self.bot_token)
    
    async def listen(self) -> AsyncIterator[Message]:
        from telegram import Update
        from telegram.ext import Application
        
        app = Application.builder().token(self.bot_token).build()
        
        async def message_handler(update: Update, context):
            yield Message(
                platform="telegram",
                user_id=str(update.effective_user.id),
                content=update.message.text,
                timestamp=update.message.date
            )
        
        app.add_handler(MessageHandler(Filters.text, message_handler))
        await app.run_polling()

三、技术实现:核心模块代码解析

3.1 项目结构

hermes-agent/
├── hermes/
│   ├── __init__.py
│   ├── agent.py              # Agent 主类
│   ├── memory/
│   │   ├── __init__.py
│   │   ├── instant.py        # 即时记忆
│   │   ├── working.py        # 工作记忆
│   │   ├── long_term.py      # 长期记忆
│   │   └── meta.py           # 元记忆
│   ├── learning/
│   │   ├── __init__.py
│   │   ├── executor.py       # 执行引擎
│   │   ├── extractor.py      # 模式提取器
│   │   ├── consolidator.py   # 技能沉淀器
│   │   ├── matcher.py        # 技能匹配器
│   │   └── optimizer.py      # 技能优化器
│   ├── skills/
│   │   ├── __init__.py
│   │   ├── base.py           # 技能基类
│   │   └── builtin/          # 内置技能
│   ├── gateways/
│   │   ├── __init__.py
│   │   ├── base.py           # 网关基类
│   │   ├── telegram.py
│   │   ├── discord.py
│   │   └── slack.py
│   └── tools/
│       ├── __init__.py
│       └── builtin/          # 内置工具
├── configs/
│   ├── default.yaml
│   └── memory.yaml
├── skills/                   # 用户技能目录
├── memory.db                 # 记忆数据库
└── MEMORY.md                 # 长期记忆文件

3.2 Agent 主类实现

from typing import Optional, List, Dict, Any
from pathlib import Path
import yaml

class HermesAgent:
    """Hermes Agent 主类"""
    
    def __init__(self, config_path: str = None):
        # 加载配置
        self.config = self._load_config(config_path)
        
        # 初始化记忆系统
        self.memory = MetaMemory()
        
        # 初始化学习系统
        self.executor = ExecutionEngine(self)
        self.extractor = PatternExtractor()
        self.consolidator = SkillConsolidator(self.memory.long_term_memory)
        self.matcher = SkillMatcher(Path("~/.hermes/skills").expanduser())
        self.optimizer = SkillOptimizer()
        
        # 初始化消息网关
        self.gateways: Dict[str, MessageGateway] = {}
        
        # 加载内置技能
        self.builtin_skills = self._load_builtin_skills()
    
    async def process_message(
        self, 
        message: str, 
        user_id: str,
        platform: str = "cli"
    ) -> str:
        """处理用户消息"""
        # 1. 构建完整上下文
        full_context = self.memory.build_full_context(message)
        
        # 2. 检查是否有匹配的技能
        matched_skill = self.matcher.find_matching_skill(message)
        
        if matched_skill:
            # 使用技能执行
            result = await self._execute_skill(matched_skill, message)
            
            # 记录执行轨迹
            trace = self.executor.current_trace
            trace.add_step("skill_execution", f"使用技能: {matched_skill.name}")
            
            # 优化技能
            optimized = self.optimizer.optimize_skill(matched_skill, trace)
            self._save_skill(optimized)
            
            return result
        
        # 3. 没有匹配技能,正常执行
        trace = self.executor.execute_task(message)
        
        # 4. 提取模式并沉淀技能
        patterns = self.extractor.extract_patterns(trace)
        for pattern in patterns:
            if pattern["type"] == "tool_sequence":
                skill = self.consolidator.consolidate_to_skill(pattern, trace)
                self.matcher.skills.append(skill)
        
        # 5. 整合记忆
        self.memory.consolidate_memories()
        
        # 6. 返回结果
        return trace.steps[-1]["content"] if trace.steps else "任务执行完成"
    
    async def _execute_skill(
        self, 
        skill: Skill, 
        context: str
    ) -> str:
        """执行技能"""
        results = []
        
        for step in skill.steps:
            if step["type"] == "tool_call":
                tool_name = step["tool"]
                params = self._resolve_parameters(step["parameters"], context)
                
                result = await self._call_tool(tool_name, params)
                results.append(result)
            
            elif step["type"] == "llm_call":
                prompt = self._render_prompt(step["prompt_template"], context)
                result = await self._call_llm(prompt)
                results.append(result)
        
        return results[-1] if results else ""
    
    async def _call_tool(self, tool_name: str, params: dict) -> Any:
        """调用工具"""
        # 查找工具
        tool = self._get_tool(tool_name)
        if not tool:
            raise ValueError(f"工具不存在: {tool_name}")
        
        # 执行工具
        return await tool.execute(**params)
    
    async def _call_llm(self, prompt: str) -> str:
        """调用大语言模型"""
        # 这里使用配置的 LLM 提供商
        provider = self.config["llm"]["provider"]
        
        if provider == "openai":
            return await self._call_openai(prompt)
        elif provider == "anthropic":
            return await self._call_anthropic(prompt)
        elif provider == "openrouter":
            return await self._call_openrouter(prompt)
        else:
            raise ValueError(f"不支持的 LLM 提供商: {provider}")

3.3 内置技能示例

Hermes Agent 内置了 75+ 个技能,以下是几个典型示例:

3.3.1 TDD 开发技能

# skills/test-driven-development.yaml
name: test-driven-development
description: 测试驱动开发工作流
trigger_conditions:
  - "写一个"
  - "实现"
  - "开发"
  - "创建.*功能"
parameters:
  feature_description:
    type: string
    description: 功能描述
  language:
    type: string
    default: python
steps:
  - type: llm_call
    name: write_test
    prompt_template: |
      为以下功能编写测试用例:
      {{feature_description}}
      
      使用 {{language}} 语言和 pytest 框架。
      遵循 AAA 模式(Arrange-Act-Assert)。
  
  - type: llm_call
    name: run_test
    prompt_template: |
      运行测试,预期失败:
      {{write_test_result}}
  
  - type: llm_call
    name: write_impl
    prompt_template: |
      编写最小实现使测试通过:
      测试代码:{{write_test_result}}
      
      只编写必要的代码,不要过度设计。
  
  - type: llm_call
    name: verify
    prompt_template: |
      再次运行测试验证实现:
      测试:{{write_test_result}}
      实现:{{write_impl_result}}

3.3.2 代码审查技能

# skills/code-review.yaml
name: code-review
description: 代码审查工作流
trigger_conditions:
  - "审查"
  - "review"
  - "检查代码"
  - "code review"
parameters:
  target_files:
    type: array
    description: 要审查的文件列表
steps:
  - type: tool_call
    tool: git_diff
    parameters:
      files: "{{target_files}}"
  
  - type: llm_call
    name: analyze
    prompt_template: |
      审查以下代码变更:
      {{git_diff_result}}
      
      从以下维度分析:
      1. 代码质量
      2. 潜在 Bug
      3. 性能问题
      4. 安全风险
      5. 可维护性
  
  - type: llm_call
    name: suggest
    prompt_template: |
      基于分析结果,提供改进建议:
      {{analyze_result}}
      
      格式:
      - 问题:xxx
        位置:xxx
        建议:xxx
        优先级:高/中/低

四、与 OpenClaw 的对比分析

作为 2026 年最火的两款开源 AI Agent 框架,Hermes Agent 和 OpenClaw 经常被拿来比较。

4.1 设计理念对比

维度Hermes AgentOpenClaw
核心理念单 Agent 深度自我进化多 Agent 协作
记忆系统四层记忆架构,跨会话持久化会话级记忆,重启后丢失
技能系统自动沉淀,持续优化手动配置,静态技能
适用场景长期个人助理,深度任务快速响应,多任务并行
学习曲线较陡,需要理解记忆和技能概念较平,开箱即用

4.2 技术架构对比

┌─────────────────────────────────────────────────────────────┐
│                      Hermes Agent                           │
│  ┌─────────┐    ┌─────────┐    ┌─────────┐                 │
│  │ 单Agent │───▶│ 四层记忆│───▶│ 技能沉淀│                 │
│  └─────────┘    └─────────┘    └─────────┘                 │
│       │              │              │                       │
│       └──────────────┴──────────────┘                       │
│                      ▼                                       │
│              【深度自我进化】                                 │
└─────────────────────────────────────────────────────────────┘

┌─────────────────────────────────────────────────────────────┐
│                       OpenClaw                              │
│  ┌─────────┐    ┌─────────┐    ┌─────────┐                 │
│  │Agent 1  │    │Agent 2  │    │Agent N  │                 │
│  └────┬────┘    └────┬────┘    └────┬────┘                 │
│       │              │              │                       │
│       └──────────────┴──────────────┘                       │
│                      ▼                                       │
│              【多Agent协作编排】                              │
└─────────────────────────────────────────────────────────────┘

4.3 性能对比

根据社区测试数据:

指标Hermes AgentOpenClaw
首次任务执行时间基准基准
重复任务执行时间减少 40%(使用技能)相同
长期运行后准确率提升 25%(学习优化)保持不变
内存占用较高(记忆系统)较低
启动时间较慢(加载记忆和技能)较快

4.4 选型建议

选择 Hermes Agent 如果你需要:

  • 长期运行的个人助理
  • 任务有重复性,需要积累经验
  • 对个性化有较高要求
  • 愿意投入时间调教 Agent

选择 OpenClaw 如果你需要:

  • 快速响应的团队协作
  • 任务多样化,每次都不同
  • 对启动速度有要求
  • 不需要长期记忆

五、实战:从零部署 Hermes Agent

5.1 环境准备

# 系统要求
# - macOS / Linux: 直接使用
# - Windows: 需要 WSL2

# 检查 Python 版本(需要 3.11+)
python3 --version

# 检查 Node.js(可选,用于某些工具)
node --version

5.2 一键安装

# 官方安装脚本
curl -fsSL https://get.hermes.agent.nousresearch.com | sh

# 或使用 pip
pip install hermes-agent

5.3 配置 LLM 提供商

# 使用 OpenRouter(推荐,支持 200+ 模型)
hermes config set llm.provider openrouter
hermes config set llm.openrouter.api_key YOUR_API_KEY

# 或使用 Anthropic
hermes config set llm.provider anthropic
hermes config set llm.anthropic.api_key YOUR_API_KEY

# 或使用 OpenAI
hermes config set llm.provider openai
hermes config set llm.openai.api_key YOUR_API_KEY

5.4 配置消息平台

# 配置 Telegram
hermes gateway add telegram --token YOUR_BOT_TOKEN

# 配置 Discord
hermes gateway add discord --token YOUR_BOT_TOKEN

# 配置 Slack
hermes gateway add slack --app-id YOUR_APP_ID --token YOUR_TOKEN

5.5 启动 Agent

# 终端模式
hermes run

# 后台模式
hermes run --daemon

# 指定消息平台
hermes run --gateway telegram

5.6 使用示例

# 在 Telegram 中与 Agent 对话

你: 帮我写一个 Python 爬虫,抓取 Hacker News 首页

Hermes: 好的,我来帮你写一个爬虫...
[生成代码]

你: 记住,我偏好使用 httpx 而不是 requests

Hermes: 已记住你的偏好。下次生成爬虫代码时我会使用 httpx。

(三天后)

你: 再帮我写一个爬虫,抓取 ProductHunt

Hermes: 好的,根据你之前的项目经验,我使用 httpx 编写...
[使用偏好生成代码]

(Agent 自动应用了之前记住的偏好)

六、高级特性:深度定制与扩展

6.1 自定义技能

创建自定义技能非常简单,只需在 ~/.hermes/skills/ 目录下创建 YAML 文件:

# ~/.hermes/skills/my-custom-skill.yaml
name: my-custom-skill
description: 我的自定义技能
trigger_conditions:
  - "帮我分析"
  - "数据分析"
parameters:
  data_source:
    type: string
    description: 数据源路径
steps:
  - type: tool_call
    tool: read_file
    parameters:
      path: "{{data_source}}"
  
  - type: llm_call
    name: analyze
    prompt_template: |
      分析以下数据:
      {{read_file_result}}
      
      提供以下分析:
      1. 数据概览
      2. 关键指标
      3. 异常检测
      4. 趋势预测
  
  - type: tool_call
    tool: save_file
    parameters:
      path: "{{data_source}}.analysis.md"
      content: "{{analyze_result}}"

6.2 自定义工具

# ~/.hermes/tools/my_tool.py
from hermes.tools import Tool, ToolResult

class MyCustomTool(Tool):
    """自定义工具示例"""
    
    name = "my_custom_tool"
    description = "我的自定义工具"
    parameters = {
        "input": {
            "type": "string",
            "description": "输入参数"
        }
    }
    
    async def execute(self, input: str) -> ToolResult:
        # 实现工具逻辑
        result = self._process(input)
        
        return ToolResult(
            success=True,
            output=result,
            metadata={"tool": self.name}
        )
    
    def _process(self, input: str) -> str:
        # 你的处理逻辑
        return f"处理结果: {input}"

6.3 记忆系统扩展

可以通过继承 LongTermMemory 类来扩展记忆存储:

from hermes.memory import LongTermMemory

class VectorMemory(LongTermMemory):
    """使用向量数据库的记忆存储"""
    
    def __init__(self, collection_name: str = "hermes_memory"):
        super().__init__()
        from chromadb import Client
        self.client = Client()
        self.collection = self.client.get_or_create_collection(collection_name)
    
    def store_memory(self, content: str, memory_type: str, **kwargs):
        # 生成嵌入向量
        embedding = self._get_embedding(content)
        
        # 存储到向量数据库
        self.collection.add(
            documents=[content],
            embeddings=[embedding],
            metadatas=[{"type": memory_type, **kwargs}],
            ids=[self._generate_id()]
        )
    
    def retrieve_memories(self, query: str, limit: int = 10):
        # 查询向量数据库
        query_embedding = self._get_embedding(query)
        
        results = self.collection.query(
            query_embeddings=[query_embedding],
            n_results=limit
        )
        
        return self._format_results(results)

七、性能优化与最佳实践

7.1 记忆系统优化

# 配置记忆刷新策略
hermes config set memory.consolidation_threshold 0.7
hermes config set memory.max_long_term_items 1000
hermes config set memory.embedding_model "text-embedding-3-small"

# 定期清理低价值记忆
hermes memory prune --min-importance 0.3 --older-than 30d

7.2 技能系统优化

# 技能去重
hermes skill dedupe

# 技能泛化
hermes skill generalize --skill-name "my-skill" --min-usage 10

# 技能导出/导入
hermes skill export --output skills-backup.tar.gz
hermes skill import --input skills-backup.tar.gz

7.3 性能监控

# 查看运行状态
hermes status

# 查看记忆统计
hermes memory stats

# 查看技能统计
hermes skill stats

# 查看性能指标
hermes metrics

八、总结与展望

8.1 核心价值总结

Hermes Agent 的核心创新在于:

  1. 四层记忆系统:解决了 AI Agent 的「失忆症」,实现真正的跨会话知识持久化
  2. 闭环学习系统:从「执行」到「进化」,让 Agent 越用越聪明
  3. 技能自动沉淀:无需手动配置,Agent 自动从经验中学习
  4. 多平台统一接入:一个 Agent,多端使用

8.2 适用场景

  • 个人 AI 助理:长期陪伴,越用越懂你
  • 开发辅助:记住项目背景、编码风格、技术栈偏好
  • 研究助手:积累领域知识,自动沉淀方法论
  • 自动化运维:学习运维经验,优化操作流程

8.3 未来展望

根据 Nous Research 的路线图,Hermes Agent 未来将支持:

  • 多 Agent 协作:在保持自我进化能力的同时,支持 Agent 间协作
  • Web UI:提供图形化界面,降低使用门槛
  • 技能市场:社区共享技能,加速能力积累
  • 本地模型优化:更好地支持本地部署的大模型

8.4 与其他框架的融合趋势

2026 年的 AI Agent 领域正在出现一个有趣的趋势:Hermes Agent 的「深度进化」与 OpenClaw 的「广度协作」正在融合

未来的 Agent 可能同时具备:

  • 深度自我进化能力(来自 Hermes Agent)
  • 多 Agent 协作能力(来自 OpenClaw)
  • 企业级安全与合规(来自 Microsoft Agent 365)

这将是 AI Agent 从「工具」真正进化为「数字员工」的关键一步。


附录:资源链接

  • GitHub 仓库:https://github.com/NousResearch/hermes-agent
  • 官方文档:https://hermes-agent.nousresearch.com
  • 技能市场:https://skills.hermes-agent.nousresearch.com
  • 社区论坛:https://community.nousresearch.com

一句话总结:Hermes Agent 不是又一个 AI 对话工具,而是一个会「记住你」「理解你」「与你共同成长」的数字伙伴。它的四层记忆系统和闭环学习机制,代表了 AI Agent 从「对话工具」向「智能伙伴」的根本性跨越。

推荐文章

CSS 中的 `scrollbar-width` 属性
2024-11-19 01:32:55 +0800 CST
PHP 微信红包算法
2024-11-17 22:45:34 +0800 CST
一个收银台的HTML
2025-01-17 16:15:32 +0800 CST
windows安装sphinx3.0.3(中文检索)
2024-11-17 05:23:31 +0800 CST
程序员茄子在线接单