Hermes Agent 源码解析
第 6 讲:记忆与技能系统——双态记忆、MemoryProvider ABC、Skills 渐进式披露与 Slash Command 路由
基于 Hermes Agent v0.16.0 源码 · 2026-06-19
一、记忆与技能:Agent 的"大脑皮层"
前五讲我们建立了全局架构地图、深入了对话循环、解析了工具系统的三层架构、拆解了插件系统、研究了 Provider 抽象层。这一讲聚焦 Hermes 最贴近"智能"的两个子系统——记忆系统(Memory)与技能系统(Skills)。
记忆让 Agent 跨会话"记住"东西:环境配置、项目约定、用户偏好。技能让 Agent 获得领域专长:代码审查、测试驱动开发、论文检索、图像生成……两者共同构成 Agent 的"长期记忆 + 程序性知识"。
📦 源码仓库
https://github.com/NousResearch/hermes-agent
本地源码:~/.hermes/hermes-agent/
本讲核心文件:
tools/memory_tool.py(811 行)— 内置记忆存储
agent/memory_manager.py(917 行)— 记忆管理器
agent/memory_provider.py(296 行)— MemoryProvider ABC
tools/skills_tool.py(1612 行)— 技能工具
agent/skill_utils.py(666 行)— 技能元数据处理
agent/skill_commands.py(527 行)— Slash Command 路由
agent/skill_bundles.py(410 行)— 技能组合
agent/skill_preprocessing.py(140 行)— 技能预处理
plugins/memory/(8+ 外部记忆后端)
二、记忆系统:双态存储与 Frozen Snapshot 模式
记忆系统的设计有一个核心矛盾:记忆要持久化(跨会话保留),但系统提示词要冻结(保 prefix cache)。Hermes 的解决方案是双态存储(Dual-State Storage)——MemoryStore同时维护两套状态:
MemoryStore 双态模型:
+---------------------+ +--------------------------+
| _system_prompt_ | | memory_entries / |
| snapshot | | user_entries (live) |
+---------------------+ +--------------------------+
| 冻结态 (Frozen) | | 活跃态 (Live) |
+---------------------+ +--------------------------+
| • 加载时一次性捕获 | | • 工具调用实时更新 |
| • 整个会话期间不变 | | • 立即持久化到磁盘 |
| • 注入系统提示词 | | • 工具响应反映最新状态 |
| • 保 prefix cache | | • 用户可 inspect + remove |
+---------------------+ +--------------------------+
文件: tools/memory_tool.py:113-122
两个文件:
~/.hermes/memories/MEMORY.md — Agent 的个人笔记 (2200 字符)
~/.hermes/memories/USER.md — 用户画像 (1375 字符)
条目分隔符: § (section sign)
字符限制 (非 token): 模型无关的硬约束1. 加载时双态分离
load_from_disk()是双态模型的起点。它从磁盘读取条目,然后分叉为两条路径:
📄 tools/memory_tool.py (第 132-170 行)
def load_from_disk(self):
"""Load entries, capture system prompt snapshot.
Frozen snapshot = system prompt (stable for prefix cache).
Live state = tool responses (always current).
"""
mem_dir = get_memory_dir()
mem_dir.mkdir(parents=True, exist_ok=True)
self.memory_entries = self._read_file(mem_dir / "MEMORY.md")
self.user_entries = self._read_file(mem_dir / "USER.md")
# Deduplicate (preserves order, keeps first occurrence)
self.memory_entries = list(dict.fromkeys(self.memory_entries))
self.user_entries = list(dict.fromkeys(self.user_entries))
# ── 关键分叉点 ──
# 快照态: 经过安全扫描,中毒条目被替换为 [BLOCKED: ...]
sanitized_memory = self._sanitize_entries_for_snapshot(
self.memory_entries, "MEMORY.md")
sanitized_user = self._sanitize_entries_for_snapshot(
self.user_entries, "USER.md")
# 冻结快照 — 整个会话不再改变
self._system_prompt_snapshot = {
"memory": self._render_block("memory", sanitized_memory),
"user": self._render_block("user", sanitized_user),
}
# 活跃态: 保留原始文本,用户可 inspect + remove 中毒条目
安全设计:快照态经过威胁模式扫描(tools/threat_patterns.py),任何匹配注入模式的条目在快照中被替换为 [BLOCKED: ...] 占位符。但活跃态保留原始文本——这样用户仍然可以看到中毒条目并通过 memory(action=remove) 删除它。如果连活跃态也替换,攻击就会被隐藏。
2. 写入路径:文件锁 + 漂移检测
每次写入(add/replace/remove)都经过三层防护:
MemoryStore.add() 写入路径:
1. 内容扫描 (threat_patterns, scope="strict")
│ 拒绝注入/泄露模式 → 返回 error
│
2. 文件锁 (_file_lock, fcntl.flock / msvcrt.locking)
│ 独立 .lock 文件,不阻塞原子替换
│
3. 漂移检测 (_detect_external_drift)
│ 磁盘文件内容 ≠ 工具可解析格式?
│ ├─ 是 → 备份到 .bak. ,拒绝写入
│ │ (防止 patch tool/shell append/姊妹会话
│ │ 写入的内容被静默丢弃 — issue #26045)
│ └─ 否 → 继续
│
4. 去重 + 字符限制检查
│ ├─ 精确重复 → "Entry already exists"
│ └─ 超限 → 提示 consolidate (replace/remove 后再 add)
│
5. 写入内存 + 持久化到磁盘 (atomic_replace)
│
6. 通知外部 Provider (on_memory_write hook)
关键: 系统提示词快照不更新 — prefix cache 保活!3. 字符限制的设计意图
为什么用字符而非 token?
字符计数是模型无关的。同一个 MEMORY.md 在 GPT-4、Claude、Qwen 下占用的 token 数不同,但字符数永远一样。2200 字符 ≈ 300-500 token(取决于模型分词器),足够容纳 5-10 条简洁的笔记。
三、MemoryProvider ABC:外部记忆后端的契约
内置记忆(MEMORY.md/USER.md)是"小脑"——轻量、快速、始终可用。外部记忆 Provider 是"大脑皮层"——语义搜索、向量索引、跨会话关联分析。两者通过 MemoryProvider ABC 统一接入。
1. ABC 接口设计:核心 + 可选钩子
📄 agent/memory_provider.py (第 42-153 行)
class MemoryProvider(ABC):
"""Abstract base class for memory providers."""
# ── 核心生命周期 (必须实现) ──
@property
@abstractmethod
def name(self) -> str:
"""短标识: 'builtin', 'honcho', 'hindsight'..."""
@abstractmethod
def is_available(self) -> bool:
"""配置好 + 有凭证 + 依赖就绪? 不做网络调用。"""
@abstractmethod
def initialize(self, session_id: str, **kwargs) -> None:
"""会话初始化: 创建资源、建连接、启动后台线程。
kwargs: hermes_home, platform, agent_context,
agent_identity, user_id, parent_session_id..."""
@abstractmethod
def get_tool_schemas(self) -> List[Dict[str, Any]]:
"""OpenAI 格式工具 schema 列表。"""
# ── 核心方法 (有默认实现,可覆盖) ──
def system_prompt_block(self) -> str:
"""系统提示词静态块。"""
return ""
def prefetch(self, query: str, *, session_id: str = "") -> str:
"""每轮调用前的背景召回。要求快速 — 用后台线程。"""
return ""
def queue_prefetch(self, query: str, *, session_id: str = "") -> None:
"""排队下一轮的背景召回。"""
pass
def sync_turn(self, user_content, assistant_content, ...) -> None:
"""每轮结束后持久化。非阻塞 — 队列后台处理。"""
pass
def handle_tool_call(self, tool_name, args, **kwargs) -> str:
"""处理工具调用。返回 JSON 字符串。"""
raise NotImplementedError(...)
def shutdown(self) -> None:
"""清理: 刷盘、关连接。"""
pass
# ── 可选钩子 (覆盖即启用) ──
def on_turn_start(turn, message, **kwargs) -> None: ...
def on_session_end(messages) -> None: ...
def on_session_switch(new_session_id, **kwargs) -> None: ...
def on_pre_compress(messages) -> str: ...
def on_memory_write(action, target, content, metadata) -> None: ...
def on_delegation(task, result, **kwargs) -> None: ...
def get_config_schema() -> List[Dict]: ...
def save_config(values, hermes_home) -> None: ...
设计亮点:可选钩子默认是 no-op。Provider 只需覆盖它需要的钩子——不需要实现一个 20 方法的空壳类。这是典型的默认适配器模式(Default Adapter Pattern),但通过 ABC 强制核心方法。
2. MemoryManager:单一外部 Provider 约束
MemoryManager是记忆系统的编排器。它有一个关键设计决策:只允许一个外部 Provider。
📄 agent/memory_manager.py (第 333-396 行)
def add_provider(self, provider: MemoryProvider) -> None:
is_builtin = provider.name == "builtin"
if not is_builtin:
if self._has_external:
existing = next(
(p.name for p in self._providers
if p.name != "builtin"), "unknown")
logger.warning(
"Rejected memory provider '%s' — external provider "
"'%s' is already registered. Only one external "
"memory provider is allowed.",
provider.name, existing,
)
return
self._has_external = True
self._providers.append(provider)
# ── 核心工具名保护 ──
# Provider 的工具不能覆盖内置工具 (clarify, delegate_task...)
# 否则 _tool_to_provider 路由表会被劫持 (#40466)
from toolsets import _HERMES_CORE_TOOLS
_core_tool_names = set(_HERMES_CORE_TOOLS)
for schema in provider.get_tool_schemas():
tool_name = schema.get("name", "")
if tool_name in _core_tool_names:
logger.warning(
"Memory provider '%s' tool '%s' shadows a reserved "
"core tool name; registration ignored.",
provider.name, tool_name,
)
continue
为什么只允许一个外部 Provider?两个原因:(1) 工具 schema 膨胀——每个 Provider 的工具都进入系统提示词,两个 Provider 意味着双倍 overhead;(2) 冲突避免——两个语义搜索后端同时写入可能产生不一致的记忆图。
3. 后台同步:永不阻塞对话
MemoryManager 的后台同步设计非常精巧——一个 Provider 故障绝不能阻塞对话:
MemoryManager 后台同步架构:
sync_all() / queue_prefetch_all()
│
└─ _submit_background(fn)
│
├─ _get_sync_executor() (懒创建, 单 worker)
│ ThreadPoolExecutor(max_workers=1)
│ thread_name_prefix="mem-sync"
│
├─ executor.submit(fn)
│ │
│ └─ fn() 遍历所有 provider
│ 每个 provider 独立 try/except
│ 一个失败 → 记录日志 → 继续下一个
│
└─ 失败回退: executor 不可用 → fn() 内联执行
(慢但正确, 不丢写)
真实案例: 一个 Hindsight daemon 配置错误,
sync_turn 阻塞 ~298s 后超时。如果内联执行,
CLI/TUI/Gateway 都会显示 "running" 数分钟,
导致用户重复发送消息 → 激进中断。
单 worker 保证: turn N 的写先于 turn N+1 完成,
Provider 实现不需要自己的排序保证。4. 流式上下文清洗器:StreamingContextScrubber
MemoryProvider 的 prefetch 结果被注入到 LLM 响应中,用 <memory-context> 标签包裹。但流式输出时,标签可能跨 chunk 断裂——普通正则无法处理。Hermes 实现了一个状态机清洗器:
📄 agent/memory_manager.py (第 130-293 行)
class StreamingContextScrubber:
"""Stateful scrubber for streaming text with split memory-context spans.
一次性正则无法存活 chunk 边界:
<memory-context> 在一个 delta 打开,
</memory-context> 在后面的 delta 关闭,
中间内容会泄漏到 UI。这个 scrubber 跨 delta
运行状态机, 扣留部分标签尾部, 丢弃 span 内所有内容。
"""
def feed(self, text: str) -> str:
"""返回清洗后的可见部分。
可能成为标签开头的尾部片段被扣留,
下次 feed() 或 flush() 时释放。"""
def flush(self) -> str:
"""流结束时释放扣留的缓冲区。
如果仍在未终止 span 内, 丢弃剩余内容
(更安全: 泄漏部分记忆上下文 > 截断回答)。"""
安全原则:泄漏记忆上下文(让用户看到 Agent 的内部回忆)比截断回答更糟——前者可能导致信息泄露,后者只是体验下降。所以 flush 时如果 span 未关闭,直接丢弃。
四、技能系统:渐进式披露与三层架构
技能系统的设计灵感来自 Anthropic 的 Claude Skills,但 Hermes 做了大量扩展。核心设计理念是渐进式披露(Progressive Disclosure)——LLM 先看到元数据,按需加载完整内容,再按需加载关联文件。每一层都控制 token 消耗。
1. 技能目录结构
技能目录结构 (agentskills.io 兼容):
~/.hermes/skills/
├── creative/
│ ├── comfyui/
│ │ ├── SKILL.md ← 主指令 (必需)
│ │ ├── references/ ← 参考文档
│ │ │ ├── api.md
│ │ │ └── examples.md
│ │ ├── templates/ ← 输出模板
│ │ ├── assets/ ← 补充文件
│ │ └── scripts/ ← 可执行脚本
│ └── excalidraw/
│ └── SKILL.md
├── productivity/
│ ├── google-workspace/
│ │ └── SKILL.md
│ └── ocr-and-documents/
│ └── SKILL.md
├── research/
│ └── arxiv/
│ └── SKILL.md
└── developer-tools/
└── ...
来源:
1. 内置技能 (skills/ 目录, 随 Hermes 发布)
2. Skills Hub 安装 (hermes skills install)
3. 用户手动添加
所有技能共存于 ~/.hermes/skills/, 不污染 git repo2. SKILL.md Frontmatter 规范
每个技能的 SKILL.md 文件以 YAML frontmatter 开头,声明元数据、平台要求、前置条件、环境变量等:
📄 SKILL.md 格式 (agentskills.io 兼容)
---
name: github-code-review # 必需, ≤64 字符
description: Perform code review... # 必需, ≤1024 字符
version: 1.0.0 # 可选
license: MIT # 可选
platforms: [macos, linux] # 可选 — 平台限制
prerequisites: # 可选 — 运行时要求
env_vars: [GITHUB_TOKEN]
commands: [git, curl]
required_environment_variables: # 新版声明方式
- name: GITHUB_TOKEN
prompt: "Enter your GitHub token"
help: "Get one at https://github.com/settings/tokens"
required_for: API authentication
- name: SLACK_WEBHOOK_URL
optional: true
setup: # 可选 — 交互式设置
help: "Configure GitHub access"
collect_secrets:
- env_var: GITHUB_TOKEN
prompt: "GitHub Personal Access Token"
provider_url: https://github.com/settings/tokens
metadata: # 可选 — 任意键值
hermes:
tags: [code-review, github]
related_skills: [test-driven-development]
config: # 技能级配置
default_reviewer: senior
---
# GitHub Code Review Skill
Full instructions here...
3. 渐进式披露:三层工具暴露
技能系统通过两个工具实现渐进式披露,每层控制 token 消耗:
渐进式披露三层:
Layer 1: skills_list()
└─ 返回: [{name, description, path, ...}]
每个技能 ~50-100 chars (name + description)
100 个技能 ≈ 5-10K chars — 可控
Layer 2: skill_view("skill-name")
└─ 返回: 完整 SKILL.md 内容
按需加载 — 只加载 LLM 请求的技能
包含 frontmatter 解析 + 平台过滤 + 环境检查
Layer 3: skill_view("skill-name", "references/file.md")
└─ 返回: 关联文件内容
最细粒度 — 只加载特定参考文件
路径安全: 拒绝 '..' 穿越, 限制在 skills dir
Token 效率: 100 技能 × 3 层 vs 全部注入 ≈ 10x 节省4. 技能预处理管线
技能内容在注入前经过预处理管线,支持模板变量替换和内联 shell 执行:
📄 agent/skill_preprocessing.py (第 124-140 行)
def preprocess_skill_content(
content: str,
skill_dir: Path | None,
session_id: str | None = None,
skills_cfg: dict | None = None,
) -> str:
"""Apply configured SKILL.md template and inline-shell preprocessing."""
cfg = skills_cfg or load_skills_config()
# 1. 模板变量替换
if cfg.get("template_vars", True):
content = substitute_template_vars(content, skill_dir, session_id)
# ${HERMES_SKILL_DIR} → /home/user/.hermes/skills/my-skill
# ${HERMES_SESSION_ID} → abc123-def456
# 2. 内联 shell 执行 (需显式启用)
if cfg.get("inline_shell", False):
timeout = int(cfg.get("inline_shell_timeout", 10) or 10)
content = expand_inline_shell(content, skill_dir, timeout)
# !`date +%Y-%m-%d` → 2026-06-19
# 每个命令独立 try/except, 超时 10s, 输出上限 4000 chars
安全设计:内联 shell 默认关闭(inline_shell: false),需要显式在 config.yaml 启用。每个命令有超时限制(默认 10s)和输出上限(4000 字符),防止单个命令失控。
五、Slash Command 路由与技能组合
用户通过 /skill-name 命令激活技能。路由系统需要处理技能名规范化、平台过滤、技能组合(bundles)、配置注入等多个环节。
1. Slash Command 路由管线
Slash Command 路由管线:
用户输入: "/github-code-review"
│
├─ 1. Slug 规范化
│ "GitHub Code Review" → "github-code-review"
│ 规则: 小写, 空格→-, 去特殊字符, 合并多重连字符
│
├─ 2. Bundle 优先查找 (skill_bundles.py)
│ ├─ 命中 bundle → 加载多个技能 + 组合指令
│ └─ 未命中 → 继续
│
├─ 3. 技能查找 (skills_tool.py)
│ ├─ 平台过滤 (platforms frontmatter)
│ ├─ 环境过滤 (kanban/docker/s6)
│ ├─ 路径安全检查 (拒绝 '..' 穿越)
│ └─ 注入模式检测 (_INJECTION_PATTERNS)
│
├─ 4. 预处理管线 (skill_preprocessing.py)
│ ├─ 模板变量替换 (${HERMES_SKILL_DIR}, ${HERMES_SESSION_ID})
│ └─ 内联 shell 扩展 (!`cmd`)
│
├─ 5. 配置注入 (skill_commands.py)
│ └─ metadata.hermes.config → [Skill config: ...] 块
│
└─ 6. 组装用户消息
"Here is a skill you have been given:
<skill content>
<optional config block>
<optional bundle instruction>"
缓存: _skill_commands 字典 + mtime 感知刷新
平台切换时自动失效缓存 (#14536)2. 技能组合(Bundles)
Bundle 是一个 YAML 文件,将多个技能打包为一个命令。比如 /backend-dev 可以同时加载代码审查、TDD、PR 工作流三个技能:
📄 agent/skill_bundles.py — Bundle 格式
# ~/.hermes/skill-bundles/backend-dev.yaml
name: backend-dev
description: Backend feature work — code review, testing, PR workflow.
skills:
- github-code-review
- test-driven-development
- github-pr-workflow
instruction: |
Optional extra guidance to inject above the skill bodies.
# 存储: ~/.hermes/skill-bundles/*.yaml
# 冲突: bundle 优先于同名 skill (用户显式命名 = 用户意图)
# 缓存: mtime 感知, 文件变更自动刷新
3. 技能安全:注入检测与路径保护
技能内容来自文件系统(可能包含不受信任的内容),因此有多层安全保护:
技能安全防护层:
1. 路径安全 (skills_tool.py:_skill_lookup_path_error)
└─ 拒绝: 绝对路径, '..' 穿越, Windows 盘符路径
└─ 信任根: SKILLS_DIR + 外部技能目录
2. 注入模式检测 (skills_tool.py:_INJECTION_PATTERNS)
└─ "ignore previous instructions"
└─ "you are now"
└─ "disregard your"
└─ "new instructions:"
└─ "<system>"
命中 → 技能标记为不安全, 拒绝加载
3. 环境变量隔离 (skills_tool.py:load_env)
└─ 从 HERMES_HOME/.env 加载, 非全局 env
└─ 远程后端 (docker/singularity/modal/ssh) 特殊处理
4. 平台过滤 (skill_utils.py:skill_matches_platform)
└─ frontmatter.platforms: [macos, linux]
└─ Termux 特殊处理 (android 平台兼容 linux 技能)六、记忆 × 技能:协同工作流
记忆和技能在运行时形成闭环。记忆告诉 Agent "你是谁、用户是谁、环境是什么",技能告诉 Agent "你能做什么、怎么做"。
记忆 × 技能协同工作流:
系统提示词组装 (system_prompt.py):
stable 层:
├─ 身份 + 工具指引
├─ Skills prompt (可用技能列表的指引)
└─ 环境/平台提示
volatile 层:
├─ Memory snapshot (Frozen!) ← MEMORY.md + USER.md
├─ 外部 Provider 块 (system_prompt_block)
└─ 时间戳/会话/模型/Provider 信息
每轮对话:
Phase 1 (Turn Prologue):
├─ Memory prefetch (外部 Provider 背景召回)
│ → <memory-context> 块注入
└─ 系统提示词恢复/重建
Phase 2 (LLM 调用):
└─ Agent 可能调用:
├─ memory(action=add/replace/remove/read)
│ → MemoryStore 双态更新
│ → on_memory_write hook → 外部 Provider 镜像
├─ skills_list() / skill_view()
│ → 渐进式技能加载
└─ /skill-name (slash command)
→ 完整技能注入用户消息
Phase 3 (Turn Epilogue):
├─ sync_all() → 后台持久化到所有 Provider
└─ queue_prefetch_all() → 下一轮预取外部 Provider 的 on_memory_write 镜像
当内置记忆工具写入一条新记录时,外部 Provider 通过 on_memory_write 钩子收到通知,可以将内置记忆同步到自己的后端。这实现了内置记忆 + 外部语义索引的无缝融合——用户只需要操作一个 memory() 工具,两个后端同时更新。
📄 agent/memory_provider.py (第 279-296 行)
def on_memory_write(
self,
action: str, # 'add', 'replace', 'remove'
target: str, # 'memory', 'user'
content: str, # 条目内容
metadata: Optional[Dict[str, Any]] = None,
) -> None:
"""Called when the built-in memory tool writes an entry.
metadata 包含结构化溯源信息:
write_origin, execution_context, session_id,
parent_session_id, platform, tool_name
用于将内置记忆写入镜像到你的后端。"""
七、外部记忆 Provider 生态
Hermes 支持 8+ 个外部记忆后端,通过插件机制接入:
plugins/memory/ 目录:
honcho/ — Honcho (Nous 官方记忆后端)
client.py REST 客户端
session.py 会话管理
cli.py 命令行工具
hindsight/ — Hindsight (本地 daemon 架构)
__init__.py 通过 Unix socket 与 daemon 通信
mem0/ — Mem0 (开源记忆框架)
__init__.py 向量搜索 + 图记忆
holographic/ — Holographic (全息记忆)
holographic.py 核心逻辑
store.py 存储层
retrieval.py 检索层
byterover/ — ByteRover (字节级记忆)
retaindb/ — RetainDB (图数据库记忆)
supermemory/ — SuperMemory (语义记忆)
openviking/ — OpenViking (开源向量检索)
选择: config.yaml → memory.provider: "honcho"
配置: hermes memory setup (交互式向导)每个 Provider 实现 MemoryProvider ABC,通过 plugins/memory/__init__.py 的发现机制自动注册。用户通过 memory.provider 配置项选择激活哪个后端。
八、设计总结
记忆系统关键设计
- 双态存储:冻结快照保 prefix cache,活跃态保实时性
- 字符限制:模型无关的硬约束,防止系统提示词膨胀
- 安全扫描:威胁模式检测 + 漂移检测 + 文件锁
- 单一外部 Provider:避免 schema 膨胀和冲突
- 后台同步:Provider 故障永不阻塞对话
- 流式清洗:状态机处理跨 chunk 的内存上下文标签
技能系统关键设计
- 渐进式披露:三层工具暴露,按需加载控制 token
- agentskills.io 兼容:YAML frontmatter 标准化
- 预处理管线:模板变量 + 内联 shell(默认关闭)
- Slash Command 路由:Bundle 优先,平台/环境过滤
- 安全防护:路径保护 + 注入检测 + 环境隔离
协同设计
- 记忆 × 技能闭环:记忆定义身份,技能定义能力
- on_memory_write 镜像:内置 + 外部后端无缝同步
- 系统提示词分层:stable/volatile 分离,保缓存
下一讲预告:Gateway 与平台适配器——Telegram/Discord/微信等 20+ 平台如何共享同一个 Agent 内核
夜雨聆风