Hermes Agent 深度实战:当 AI Agent 学会了"自我进化"——从 E-A-A-S 学习闭环到三层记忆、从 Skill 自动生成到多平台网关的生产级完全指南(2026)
一、引言:为什么"健忘"是 AI Agent 最大的痛点?
你有没有这样的体验——
每天用 AI 助手写代码,但它永远不记得你昨天重构了哪个模块;每次部署都要重新解释一遍 CI/CD 流程;让 AI 审查 PR,它每次都像第一次见到这份代码一样。这不是模型不够聪明,而是整个架构从根上就没有"记忆"和"进化"的能力。
2026 年的 AI Agent 赛道已经卷到天际,GitHub 上 Agent 相关项目总 Star 数突破 500 万,月新增项目超过 3000 个。AutoGPT、OpenClaw、Browser-use 等框架各有千秋,但它们共享一个根本性的局限:用完即忘,从零开始。
Nous Research 团队看准了这个痛点,推出了 Hermes Agent——一个从架构层面解决"进化"问题的新一代 Agent 框架。2026 年 2 月上线 GitHub,MIT 协议完全开源,一个月内突破 6 万 Stars,日均增长 2000+。它的 Slogan 直击要害:"The agent that grows with you"——与你共同成长的 AI 智能体。
本文将从架构设计、记忆系统、自学习闭环、技能引擎、多平台网关、自定义工具开发、生产部署等七个维度,全面拆解 Hermes Agent 的技术实现,并配以完整的代码实战。读完本文,你不仅能理解 Hermes 为什么能"越用越强",还能动手构建自己的自进化 Agent 系统。
二、架构全景:六层分层的精妙设计
2.1 为什么分层比单体更难?
大多数 Agent 框架采用单体架构——一个大的推理循环包揽一切:接收输入、规划任务、调用工具、返回结果。简单粗暴,但也意味着修改任何环节都可能牵一发动全身。
Hermes Agent 选择了一条更难走但更可持续的路:六层分层架构。每一层只做一件事,层与层之间通过明确定义的接口通信。
┌─────────────────────────────────────────────────┐
│ Layer 1: 入口层 │
│ CLI ┃ TUI ┃ Gateway(Telegram/Discord/Slack/飞书/企微)│
├─────────────────────────────────────────────────┤
│ Layer 2: AI Agent 核心层 │
│ AIAgent 同步推理循环 ┃ 任务规划器 ┃ 工具执行器 │
├─────────────────────────────────────────────────┤
│ Layer 3: 工具注册中心 │
│ 工具注册表 ┃ 47 个内置工具 ┃ 自定义工具 │
├─────────────────────────────────────────────────┤
│ Layer 4: 插件扩展系统 │
│ 插件加载器 ┃ 生命周期钩子 ┃ 事件总线 │
├─────────────────────────────────────────────────┤
│ Layer 5: 记忆与压缩管线 │
│ 短期工作记忆 ┃ 长期情景记忆 ┃ 程序化技能记忆 ┃ 压缩器 │
├─────────────────────────────────────────────────┤
│ Layer 6: 后台自学习器 │
│ Curator 服务 ┃ 经验提取器 ┃ GEPA 优化引擎 ┃ 技能生成器│
└─────────────────────────────────────────────────┘
这个分层不是随意划分的。每一层都对应着一个独立的变化维度:
- 入口层变化最频繁:今天加个 Telegram 适配,明天加个飞书支持
- 核心层最稳定:推理循环的逻辑不太会变
- 工具层横向扩展:内置 47 个工具,还能无限扩展
- 插件层第三方定制:不改核心代码也能改变行为
- 记忆层纵向深化:从短期到长期,从事实到流程
- 学习层自动进化:这是 Hermes 最核心的差异化能力
从代码量来看,Python 占比 93.6%,TypeScript 仅用于 TUI 界面渲染。核心文件 run_agent.py 约 12K 行,cli.py 约 11K 行——这不是"代码多就厉害"的炫耀,而是说明每一层的实现都有足够的深度。
2.2 同步推理循环:Agent 的心脏
核心层的 AIAgent 是整个系统的"心脏",它实现了一个同步推理循环:
# hermes/core/run_agent.py - 核心推理循环(简化版)
class AIAgent:
def __init__(self, config: AgentConfig):
self.planner = TaskPlanner(config.model)
self.executor = ToolExecutor(config.tools)
self.memory = MemorySystem(config.memory)
self.max_iterations = config.max_iterations # 默认 25
async def run(self, user_input: str) -> str:
"""同步推理循环:规划 → 执行 → 观察 → 继续"""
# 1. 注入记忆上下文
context = await self.memory.build_context(user_input)
# 2. 加载相关技能
relevant_skills = await self.memory.search_skills(user_input)
if relevant_skills:
context = self._inject_skills(context, relevant_skills)
iteration = 0
while iteration < self.max_iterations:
# 3. 规划下一步行动
action = await self.planner.plan(context)
if action.type == "FINISH":
# 4. 任务完成,触发学习
await self._trigger_learning(context, iteration)
return action.response
# 5. 执行工具调用
observation = await self.executor.execute(action)
# 6. 更新上下文
context = self._update_context(context, action, observation)
iteration += 1
return "达到最大迭代次数,任务未完成"
async def _trigger_learning(self, context, iterations):
"""任务完成后触发后台学习"""
if iterations >= 5: # 复杂任务才触发学习
trajectory = self._extract_trajectory(context)
await self.curator.submit(trajectory)
注意第 6 步的 _trigger_learning——这是 Hermes 与其他 Agent 的根本区别。当一个任务涉及 5 次以上的工具调用时(说明任务有足够复杂度),Agent 会把整个执行轨迹提交给后台的 Curator 服务进行学习。这个过程是异步的,不阻塞当前对话。
三、三层记忆架构:不是"记住",而是"理解"
3.1 为什么一层记忆不够?
大多数 Agent 的记忆就一层:上下文窗口。上下文满了就压缩,压缩不了就丢。这就像一个人只能记住最近 10 分钟的对话,前天的事一概不知。
Hermes 的三层记忆不是简单地把存储分层,而是让不同类型的记忆有不同的生命周期、访问模式和容量限制:
| 记忆类型 | 存储介质 | 生命周期 | 容量 | 访问方式 | 解决什么问题 |
|---|---|---|---|---|---|
| Layer 1: 内置记忆 | MEMORY.md + USER.md | 永久 | 2200 + 1375 字符 | 启动时直接注入 | "高频关键事实的零成本访问" |
| Layer 2: 外部记忆 | 8 个可插拔后端 | 永久 | 无限 | 语义检索 | "深度语义化记忆" |
| Layer 3: 会话搜索 | SQLite + FTS5 | 永久 | 无限 | 全文检索 + LLM 摘要 | "无限容量的历史回溯" |
3.2 Layer 1:内置记忆——零成本的"长期知识"
Hermes 用两个 Markdown 文件实现了最轻量的持久记忆:
# MEMORY.md(Agent 个人笔记,2200 字符上限)
## 项目信息
- 当前项目:电商后端重构,使用 Go 1.24 + Fiber v3
- 数据库:PostgreSQL 17,连接池使用 pgx/v5
- CI/CD:GitHub Actions,部署到 K8s 集群
- 代码规范:golangci-lint,gofmt,go vet
## 常用路径
- 项目根目录:/home/user/projects/ecommerce
- 配置文件:/home/user/projects/ecommerce/config/prod.yaml
- 日志目录:/var/log/ecommerce
## 技术决策
- 2026-06-15:选择 pgx 替代 database/sql,性能提升 40%
- 2026-06-17:引入 Temporal 处理订单状态机
# USER.md(用户画像,1375 字符上限)
## 偏好
- 语言:中文为主,技术术语用英文
- 代码风格:函数不超过 50 行,错误处理用 fmt.Errorf + %w
- 沟通风格:先给结论,再给细节
- 工作时间:9:00-18:00 CST,午休 12:00-13:00
## 习惯
- 每次提交前跑完整测试套件
- PR 描述必须包含动机、变更、测试三个段落
- 喜欢用 Makefile 管理常用命令
这两个文件在每次会话启动时自动加载到系统提示中,不需要任何检索操作——零延迟、零成本。2200 + 1375 字符的限制看似很小,但经过精心组织,足以覆盖"当前项目环境"和"用户核心偏好"这两类最高频信息。
3.3 Layer 2:外部记忆——语义化的"深度知识"
当信息量超出内置记忆的容量,或者需要语义化检索时,Hermes 启用外部记忆提供者。它支持 8 个可插拔后端:Honcho、Holographic、Mem0、Hindsight、OpenViking、RetainDB、ByteRover、Supermemory。
关键设计决策:同时只激活一个 Provider。这不是限制,而是为了保证记忆的一致性——如果同时写入多个后端,数据同步和冲突解决会让系统变得脆弱。
# hermes/memory/providers/base.py
class MemoryProvider(ABC):
"""外部记忆提供者基类"""
@abstractmethod
async def store(self, key: str, content: str, metadata: dict) -> str:
"""存储一条记忆,返回记忆 ID"""
...
@abstractmethod
async def search(self, query: str, top_k: int = 5) -> list[MemoryItem]:
"""语义搜索,返回最相关的 k 条记忆"""
...
@abstractmethod
async def delete(self, memory_id: str) -> bool:
"""删除一条记忆"""
...
@abstractmethod
async def snapshot(self) -> list[MemoryItem]:
"""导出所有记忆(用于冻结快照)"""
...
以 Mem0 为例(最常用的 Provider 之一):
# hermes/memory/providers/mem0_provider.py
class Mem0Provider(MemoryProvider):
def __init__(self, config: Mem0Config):
from mem0 import Memory
self.client = Memory.from_config(config_path=config.config_path)
self.user_id = config.user_id
async def store(self, key: str, content: str, metadata: dict) -> str:
result = self.client.add(
content,
user_id=self.user_id,
metadata=metadata
)
return result["id"]
async def search(self, query: str, top_k: int = 5) -> list[MemoryItem]:
results = self.client.search(
query,
user_id=self.user_id,
limit=top_k
)
return [
MemoryItem(
id=r["id"],
content=r["memory"],
score=r["score"],
metadata=r.get("metadata", {})
)
for r in results
]
3.4 Layer 3:会话搜索——无限容量的"历史回溯"
所有对话历史都存入 SQLite,配合 FTS5 全文索引:
-- 消息表:记录所有对话
CREATE TABLE messages (
id INTEGER PRIMARY KEY,
session_id TEXT NOT NULL,
role TEXT NOT NULL CHECK(role IN ('user', 'assistant', 'system')),
content TEXT NOT NULL,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
token_count INTEGER
);
-- FTS5 虚拟表:全文搜索
CREATE VIRTUAL TABLE messages_fts USING fts5(
content,
content='messages',
content_rowid='id',
tokenize='unicode61' -- 支持 CJK 分词
);
-- 触发器:自动同步
CREATE TRIGGER messages_ai AFTER INSERT ON messages BEGIN
INSERT INTO messages_fts(rowid, content) VALUES (new.id, new.content);
END;
当用户问"上周我们讨论的那个 K8s 部署问题怎么解决的?",Hermes 的工作流是:
- 用 FTS5 搜索关键词"K8s 部署"
- 找到相关历史会话
- 用 Gemini Flash 做快速摘要(因为完整历史可能很长)
- 将摘要注入当前上下文
# hermes/memory/session_search.py
class SessionSearch:
def __init__(self, db_path: str = "~/.hermes/data/sessions.db"):
self.db_path = Path(db_path).expanduser()
self.conn = sqlite3.connect(str(self.db_path))
self.summarizer = GeminiFlash() # 快速摘要模型
async def search(self, query: str, top_k: int = 5) -> list[str]:
"""搜索历史会话,返回摘要"""
# 1. FTS5 全文搜索
cursor = self.conn.execute("""
SELECT m.session_id, m.content, m.created_at
FROM messages_fts f
JOIN messages m ON m.id = f.rowid
WHERE messages_fts MATCH ?
ORDER BY rank
LIMIT ?
""", (query, top_k * 10)) # 多取一些,后面去重
# 2. 按 session 分组
sessions = defaultdict(list)
for row in cursor.fetchall():
sessions[row[0]].append((row[1], row[2]))
# 3. 每个会话取摘要
summaries = []
for session_id, messages in sessions.items():
full_text = "\n".join(m[0] for m in messages)
if len(full_text) > 2000:
summary = await self.summarizer.summarize(full_text, focus=query)
else:
summary = full_text
summaries.append(summary)
return summaries[:top_k]
3.5 三层记忆的协作机制
三层不是孤立工作的,它们形成一个"漏斗"结构:
用户输入 → Layer 1(零成本,始终命中)
↓ 未命中
Layer 2(语义检索,成本适中)
↓ 仍不够
Layer 3(全文回溯 + LLM 摘要,成本最高)
这种设计的好处是:大多数查询在 Layer 1 就能解决("我的项目用的什么数据库?"→ 直接从 MEMORY.md 读取),只有复杂的历史回溯才需要走到 Layer 3。性能和成本实现了分层优化。
3.6 冻结快照与双轨记忆
Hermes 引入了"冻结快照"(Frozen Snapshot)机制——在关键时间点对整个记忆状态做快照:
# hermes/memory/snapshot.py
class MemorySnapshot:
"""冻结快照:不可变的时间点记忆状态"""
async def create(self, label: str) -> str:
"""创建快照"""
snapshot_id = f"snap_{int(time.time())}_{label}"
# 1. 导出 Layer 1
memory_md = (Path("~/.hermes") / "MEMORY.md").read_text()
user_md = (Path("~/.hermes") / "USER.md").read_text()
# 2. 导出 Layer 2
provider_data = await self.provider.snapshot()
# 3. 导出 Layer 3(SQL dump)
session_dump = self._dump_sqlite()
# 4. 打包保存
snapshot = {
"id": snapshot_id,
"created_at": datetime.now().isoformat(),
"layer1": {"MEMORY.md": memory_md, "USER.md": user_md},
"layer2": provider_data,
"layer3": session_dump,
}
await self._save(snapshot_id, snapshot)
return snapshot_id
"双轨记忆"是指:活跃记忆可修改,快照记忆只读。当你发现 Agent 最近的行为异常,可以回滚到之前的快照——类似于数据库的 PITR(Point-In-Time Recovery)。
四、E-A-A-S 学习闭环:从"做过"到"会做"
4.1 什么是 E-A-A-S?
E-A-A-S 是 Hermes 自学习闭环的四个阶段:
- Experience(经验采集):记录复杂任务的完整执行轨迹
- Analysis(分析提取):从轨迹中识别成功模式和失败原因
- Adaptation(适应优化):用 GEPA + DSPy 优化技能模板
- Synthesis(技能合成):生成可复用的 Skill 文件
这四步形成了一个闭环:越用越强,越强越用。
4.2 Experience:什么才算"值得学习"?
不是所有任务都值得学习。"帮我写个 hello world"不需要沉淀成技能。Hermes 的触发条件很务实:
# hermes/curator/trigger.py
class LearningTrigger:
"""决定何时触发学习"""
def should_learn(self, trajectory: ExecutionTrajectory) -> bool:
# 条件 1:工具调用次数 >= 5(复杂度)
if len(trajectory.tool_calls) < 5:
return False
# 条件 2:任务成功完成或有明确的失败模式
if trajectory.status not in ("SUCCESS", "PARTIAL_SUCCESS", "ANALYZABLE_FAILURE"):
return False
# 条件 3:5 分钟冷却期(避免同一类型任务频繁触发)
if self._in_cooldown(trajectory.task_signature):
return False
return True
4.3 Analysis:怎么从轨迹中"提取"经验?
经验提取器的核心逻辑:对比成功路径和失败路径,找出差异。
# hermes/curator/extractor.py
class ExperienceExtractor:
async def extract(self, trajectory: ExecutionTrajectory) -> list[Experience]:
"""从执行轨迹中提取经验"""
experiences = []
# 1. 提取成功模式
if trajectory.status == "SUCCESS":
pattern = self._identify_pattern(trajectory)
if pattern:
experiences.append(Experience(
type="success_pattern",
description=f"当执行{pattern.task_type}时,"
f"按 {pattern.optimal_sequence} 顺序调用工具,"
f"平均耗时 {pattern.avg_time}秒",
evidence=trajectory,
confidence=pattern.confidence
))
# 2. 提取失败原因
for failure in trajectory.failures:
root_cause = await self._analyze_failure(failure, trajectory)
experiences.append(Experience(
type="failure_insight",
description=f"在{failure.step}步骤,"
f"因为{root_cause}导致失败,"
f"建议{root_cause.fix_suggestion}",
evidence=failure,
confidence=root_cause.confidence
))
# 3. 提取工具组合模式
tool_combos = self._find_tool_combinations(trajectory)
for combo in tool_combos:
if combo.frequency >= 3: # 出现 3 次以上的组合
experiences.append(Experience(
type="tool_combination",
description=f"在{combo.context}场景下,"
f"常用工具组合:{combo.tools}",
evidence=combo.examples,
confidence=combo.confidence
))
return experiences
4.4 Adaptation:GEPA 优化引擎
GEPA(Grammar-based Evolutionary Prompt Optimization)是 Hermes 最硬核的组件。它用遗传算法迭代优化技能模板的语法结构,同时用 DSPy 框架做语义层面的优化。
# hermes/curator/gepa_optimizer.py
class GEPAOptimizer:
"""基于语法的进化式提示词优化"""
async def optimize(self, raw_skill: RawSkill,
examples: list[ExecutionTrajectory]) -> OptimizedSkill:
# 1. 初始化种群:生成多个技能模板变体
population = self._initialize_population(raw_skill, population_size=20)
# 2. 进化迭代
for generation in range(self.max_generations):
# 2.1 评估适应度:用历史执行轨迹做交叉验证
fitness_scores = []
for candidate in population:
score = await self._evaluate_fitness(candidate, examples)
fitness_scores.append(score)
# 2.2 选择:保留 top 50%
survivors = self._select(population, fitness_scores, ratio=0.5)
# 2.3 交叉:组合两个父代的语法结构
offspring = []
for i in range(0, len(survivors), 2):
if i + 1 < len(survivors):
child = self._crossover(survivors[i], survivors[i+1])
offspring.append(child)
# 2.4 变异:随机修改语法结构
for child in offspring:
if random.random() < self.mutation_rate:
child = self._mutate(child)
# 2.5 替换种群
population = survivors + offspring
# 3. 返回最优个体
best = max(zip(population, fitness_scores), key=lambda x: x[1])
return OptimizedSkill(template=best[0], fitness=best[1])
async def _evaluate_fitness(self, candidate, examples) -> float:
"""用 DSPy 评估技能模板的质量"""
# 将候选模板转为 DSPy 签名
signature = dspy.Signature(
"task_description -> action_sequence",
instruction=candidate.instruction
)
# 在历史轨迹上做交叉验证
correct = 0
total = len(examples)
for example in examples:
pred = await self.dspy_executor(signature, example.task_description)
if self._matches_expected(pred.action_sequence, example.expected_actions):
correct += 1
return correct / total
根据官方测试数据,经过 3-5 次执行后,技能执行效率平均提升 40%,错误率降低 65%。这听起来像营销数字,但背后的逻辑是合理的:每次执行都在修正技能模板中的"模糊地带",让指令越来越精确。
4.5 Synthesis:技能文件长什么样?
最终生成的 Skill 文件遵循 agentskills.io 开放标准,由 YAML 元数据和 Markdown 指令组成:
# ~/.hermes/skills/pr_review_workflow/skill.yaml
name: pr_review_workflow
version: 1.3.0
description: "标准化 PR 审查工作流:从代码规范到架构评审"
trigger:
keywords: ["PR审查", "代码审查", "review PR", "审查代码"]
min_complexity: 3 # 需要至少 3 步操作
tools_required:
- read_file
- git_diff
- search_code
- web_search
evolution:
created_from: "session_20260615_a3f2"
times_used: 12
times_refined: 4
success_rate: 0.92
last_refined: "2026-06-18"
# ~/.hermes/skills/pr_review_workflow/instructions.md
# PR 审查标准工作流
## 触发条件
当用户要求审查 PR 或代码时激活此技能。
## 执行步骤
### 第一步:获取变更概览
使用 `git_diff` 工具获取 PR 的完整 diff:
- 重点关注:新增文件、修改行数超过 100 行的文件、删除的文件
- 记录:变更涉及哪些模块、大致影响范围
### 第二步:分层审查
按以下优先级依次审查:
1. **安全性**(最高优先级)
- 检查是否有硬编码的密钥、token
- SQL 拼接 vs 参数化查询
- 用户输入是否经过验证
2. **正确性**
- 边界条件处理
- 错误处理是否完善(特别是 fmt.Errorf + %w 链)
- 并发安全性
3. **可维护性**
- 函数长度是否超过 50 行(用户偏好)
- 命名是否清晰
- 是否有必要的注释
4. **性能**
- N+1 查询
- 不必要的全量加载
- 连接池使用是否合理
### 第三步:输出审查报告
格式:
审查摘要
[1-2 句话总结]
关键问题 🔴
- [必须修复的问题]
建议改进 🟡
- [建议但非必须的改进]
做得好的 ✅
- [值得肯定的代码]
## 踩坑记录(来自历史执行)
- 2026-06-15:审查时注意用户偏好 fmt.Errorf + %w,不要建议 log.Printf
- 2026-06-17:当 diff 超过 500 行时,建议分模块审查而非一次全部读完
注意最后那个"踩坑记录"——这就是自学习的直接体现。每次技能执行出现问题,Curator 会把修正追加到 instructions.md 中,让技能在使用中不断完善。
五、技能引擎:从"经验"到"能力"的桥梁
5.1 Memory 记事实,Skills 记流程,Tools 干动作
这是理解 Hermes 的关键概念模型:
┌─────────────────────────────────────────────┐
│ Memory(记事实) │
│ 项目用 Go 1.24,用户喜欢先看结论 │
│ → 静态知识,回答"是什么"的问题 │
├─────────────────────────────────────────────┤
│ Skills(记流程) │
│ PR 审查怎么一步步做,K8s 发版的标准步骤 │
│ → 动态流程,回答"怎么做"的问题 │
├─────────────────────────────────────────────┤
│ Tools(干动作) │
│ 读文件、跑命令、调浏览器、访问 API │
│ → 原子操作,回答"做什么"的问题 │
└─────────────────────────────────────────────┘
三者的关系:Memory 告诉 Agent "什么",Skills 告诉 Agent "怎么",Tools 让 Agent "去做"。
5.2 技能的加载与检索
技能存放在 ~/.hermes/skills/ 目录下,采用渐进式披露(Progressive Disclosure)来减少 token 消耗:
# hermes/skills/loader.py
class SkillLoader:
"""技能加载器:渐进式披露"""
async def load_for_task(self, task_description: str) -> list[Skill]:
"""根据任务描述加载相关技能"""
# 1. 第一阶段:只扫描 skill.yaml 的 trigger.keywords
candidates = self._keyword_match(task_description)
if not candidates:
return []
# 2. 第二阶段:对候选技能做语义相似度排序
ranked = await self._semantic_rank(candidates, task_description)
# 3. 第三阶段:只加载 top-k 的完整指令
skills = []
for skill_meta in ranked[:3]: # 最多加载 3 个技能
instructions = (Path(skill_meta.path) / "instructions.md").read_text()
skills.append(Skill(meta=skill_meta, instructions=instructions))
return skills
渐进式披露的核心思想:不要一次性把所有技能的完整指令都塞进上下文。先看关键词是否匹配,再做语义排序,最后只加载最相关的 2-3 个技能的完整指令。这在技能数量增多后特别重要——50 个技能的完整指令可能有 100K token,但关键词元数据只有几 KB。
5.3 技能的自动创建流程
当你第一次让 Hermes 帮你审查一个复杂的 PR,它不会调用任何技能——因为还没有相关技能。它会用通用能力完成任务。但任务完成后,后台的 Curator 会自动启动:
任务完成(5+ 工具调用)
→ Curator 接收执行轨迹
→ 经验提取器识别模式:"这个任务涉及 read_file → git_diff → search_code 的组合调用"
→ GEPA 优化引擎生成技能模板
→ 技能生成器写入文件到 ~/.hermes/skills/auto_generated_xxx/
→ 下次类似任务,Agent 会自动加载这个技能
这就是"越用越强"的闭环。第一次手动导航,第二次自动导航。
六、多平台消息网关:一个 Agent,六个平台
6.1 统一的消息抽象
Hermes 支持六个平台的消息接入:Telegram、Discord、Slack、飞书、企业微信、Web Dashboard。所有平台的差异都被网关层抽象掉了:
# hermes/gateway/base.py
class MessageGateway(ABC):
"""消息网关基类"""
@abstractmethod
async def start(self, agent: AIAgent):
"""启动网关,监听消息"""
...
@abstractmethod
async def normalize(self, raw_message: Any) -> NormalizedMessage:
"""将平台原生消息格式转为统一格式"""
...
@abstractmethod
async def send(self, response: AgentResponse, context: MessageContext):
"""将 Agent 回复转为平台原生格式并发送"""
...
# hermes/gateway/telegram.py
class TelegramGateway(MessageGateway):
def __init__(self, token: str):
self.bot = telegram.Bot(token=token)
async def normalize(self, update) -> NormalizedMessage:
return NormalizedMessage(
text=update.message.text,
user_id=str(update.message.from_user.id),
platform="telegram",
reply_to=update.message.message_id,
attachments=self._extract_attachments(update.message)
)
async def send(self, response: AgentResponse, context: MessageContext):
# Markdown 分块发送(Telegram 单条消息上限 4096 字符)
chunks = self._split_message(response.text, max_length=4000)
for chunk in chunks:
await self.bot.send_message(
chat_id=context.user_id,
text=chunk,
parse_mode="MarkdownV2",
reply_to_message_id=context.reply_to
)
6.2 安全五层防线
多平台接入意味着更大的攻击面。Hermes 实现了五层安全防线:
Layer 1: 平台身份验证(Telegram Bot Token / Discord Bot Token)
Layer 2: 用户白名单(只响应特定 user_id)
Layer 3: 指令过滤(敏感操作需二次确认)
Layer 4: 上下文围栏(跨平台消息不共享上下文)
Layer 5: 审计日志(所有操作记录到 SQLite)
"上下文围栏"是一个特别重要的设计——当同一个 Agent 同时接入 Telegram 和 Discord 时,Telegram 上的对话内容不应该泄露到 Discord。每个平台的上下文是完全隔离的。
6.3 网关配置实战
配置 Telegram 网关的完整步骤:
# 1. 在 Telegram @BotFather 创建 Bot,获取 Token
# 2. 编辑配置文件
cat > ~/.hermes/gateways/telegram.yaml << 'EOF'
platform: telegram
token: "your-bot-token-here"
allowed_users:
- "123456789" # 你的 Telegram user_id
settings:
max_message_length: 4096
parse_mode: "MarkdownV2"
rate_limit: 30 # 每分钟最多 30 条消息
EOF
# 3. 启动网关
hermes gateway start telegram
# 4. 验证
hermes gateway status
Discord 网关配置类似:
# ~/.hermes/gateways/discord.yaml
platform: discord
token: "your-discord-bot-token"
allowed_guilds:
- "987654321"
allowed_channels:
- "channel-id-1"
settings:
max_message_length: 2000
embed_large_outputs: true # 长输出用 Embed 展示
thread_for_long_tasks: true # 复杂任务自动创建线程
七、自定义工具开发:5 分钟扩展 Agent 能力
7.1 为什么自定义工具如此简单?
Hermes 的工具系统基于三个设计原则:
- 继承基类:只需继承
BaseTool,实现execute方法 - Pydantic 校验:参数定义即文档,自动生成 LLM 可读的描述
- 自动注册:放入
~/.hermes/skills/目录,重启即生效
7.2 实战:开发一个 K8s 资源查询工具
假设你日常需要频繁查询 K8s 集群状态,我们开发一个专用工具:
# ~/.hermes/skills/k8s_tools/k8s_resource_tool.py
from hermes.tools.base import BaseTool
from pydantic import Field
from typing import Optional
import subprocess
import json
class K8sResourceTool(BaseTool):
"""K8s 资源查询工具"""
# 工具基础信息
name: str = "k8s_query"
description: str = (
"查询 Kubernetes 集群资源状态。"
"支持查询 pod、deployment、service、ingress、configmap 等资源。"
"可以指定 namespace 和 label selector 过滤。"
"使用前请确保 kubectl 已配置正确的 kubeconfig。"
)
return_direct: bool = False
# 参数定义
resource_type: str = Field(
description="资源类型:pod/deployment/service/ingress/configmap/pvc"
)
namespace: str = Field(
default="default",
description="K8s 命名空间,默认 default"
)
label_selector: Optional[str] = Field(
default=None,
description="标签选择器,例如 'app=nginx,tier=frontend'"
)
output_format: str = Field(
default="wide",
description="输出格式:wide/json/yaml"
)
async def execute(self) -> str:
"""执行 K8s 资源查询"""
# 构造 kubectl 命令
cmd = [
"kubectl", "get", self.resource_type,
"-n", self.namespace,
f"-o {self.output_format}"
]
if self.label_selector:
cmd.extend(["-l", self.label_selector])
try:
result = subprocess.run(
cmd, capture_output=True, text=True, timeout=30
)
if result.returncode != 0:
return f"K8s 查询失败:{result.stderr.strip()}"
output = result.stdout.strip()
# 如果是 JSON 格式,做简要分析
if self.output_format == "json":
data = json.loads(output)
if isinstance(data, dict) and "items" in data:
count = len(data["items"])
return f"找到 {count} 个 {self.resource_type} 资源\n\n{output[:3000]}"
return output
except subprocess.TimeoutExpired:
return "K8s 查询超时(30秒),请检查集群连接"
except json.JSONDecodeError:
return output
except FileNotFoundError:
return "kubectl 未安装或不在 PATH 中"
# 注册工具
def register_tools(tool_manager):
tool_manager.register_tool(K8sResourceTool())
7.3 实战:开发一个代码评审辅助工具
# ~/.hermes/skills/review_tools/code_metrics_tool.py
from hermes.tools.base import BaseTool
from pydantic import Field
from typing import Optional
import os
import ast
class CodeMetricsTool(BaseTool):
"""代码度量工具:分析 Python 代码复杂度"""
name: str = "code_metrics"
description: str = (
"分析 Python 文件的代码度量指标,"
"包括函数圈复杂度、类行数、依赖关系等。"
"用于代码审查时快速评估代码质量。"
)
return_direct: bool = False
file_path: str = Field(
description="Python 文件的绝对路径"
)
complexity_threshold: int = Field(
default=10,
description="圈复杂度告警阈值,默认 10"
)
async def execute(self) -> str:
"""执行代码度量分析"""
if not os.path.exists(self.file_path):
return f"文件不存在:{self.file_path}"
try:
with open(self.file_path, 'r', encoding='utf-8') as f:
source = f.read()
tree = ast.parse(source)
metrics = {
"total_lines": len(source.splitlines()),
"functions": [],
"classes": [],
"imports": []
}
for node in ast.walk(tree):
if isinstance(node, ast.FunctionDef):
complexity = self._calc_complexity(node)
func_info = {
"name": node.name,
"line": node.lineno,
"complexity": complexity,
"line_count": node.end_lineno - node.lineno + 1 if hasattr(node, 'end_lineno') else "?"
}
metrics["functions"].append(func_info)
elif isinstance(node, ast.ClassDef):
class_info = {
"name": node.name,
"line": node.lineno,
"methods": [n.name for n in node.body if isinstance(n, ast.FunctionDef)]
}
metrics["classes"].append(class_info)
elif isinstance(node, (ast.Import, ast.ImportFrom)):
if isinstance(node, ast.ImportFrom):
metrics["imports"].append(node.module or "")
# 生成报告
report_lines = [
f"## 代码度量:{os.path.basename(self.file_path)}",
f"总行数:{metrics['total_lines']}",
f"函数数:{len(metrics['functions'])}",
f"类数:{len(metrics['classes'])}",
f"依赖模块数:{len(set(metrics['imports']))}",
""
]
# 高复杂度函数告警
high_complexity = [
f for f in metrics["functions"]
if f["complexity"] > self.complexity_threshold
]
if high_complexity:
report_lines.append("### 🔴 高复杂度函数")
for f in high_complexity:
report_lines.append(
f"- `{f['name']}`(L{f['line']}):"
f"圈复杂度 {f['complexity']},{f['line_count']} 行"
)
# 超长函数告警
long_functions = [
f for f in metrics["functions"]
if isinstance(f["line_count"], int) and f["line_count"] > 50
]
if long_functions:
report_lines.append("\n### 🟡 超长函数(>50行)")
for f in long_functions:
report_lines.append(
f"- `{f['name']}`(L{f['line']}):{f['line_count']} 行"
)
return "\n".join(report_lines)
except SyntaxError as e:
return f"Python 语法错误:{e}"
def _calc_complexity(self, node: ast.FunctionDef) -> int:
"""计算圈复杂度(简化版)"""
complexity = 1 # 基础复杂度
for child in ast.walk(node):
if isinstance(child, (ast.If, ast.While, ast.For, ast.ExceptHandler)):
complexity += 1
elif isinstance(child, ast.BoolOp):
complexity += len(child.values) - 1
return complexity
def register_tools(tool_manager):
tool_manager.register_tool(CodeMetricsTool())
7.4 工具开发的最佳实践
- Description 是关键:LLM 根据工具的
description决定是否调用。写清楚"什么时候用"、"能做什么"、"不能做什么" - 参数要有默认值:让 LLM 只需提供关键参数,不必要的信息用合理默认值
- 返回字符串:
execute方法必须返回字符串,LLM 才能理解和继续推理 - 错误处理要友好:不要抛异常,返回清晰的错误描述,让 LLM 能自行判断下一步
- 避免同步阻塞:
execute是 async 方法,如果必须用同步操作(如subprocess),用asyncio.to_thread包裹
八、v0.16.0 "Surface Release":从 CLI 走向桌面
8.1 原生桌面端
2026 年 6 月 5 日发布的 v0.16.0 是一个里程碑版本——代号 "The Surface Release",标志着 Hermes 从命令行工具进化为桌面应用。
主要更新:
- 原生桌面端:基于 Tauri 构建,比 Electron 更轻量(安装包 < 20MB vs Electron 的 150MB+)
- Web Dashboard:
http://127.0.0.1:9119,可视化监控 Agent 状态、技能列表、记忆内容 - 简体中文界面:全面支持中文
- 模糊模型选择器:不需要记住模型全名,输入"claude"就能匹配到可用的 Claude 模型
- Profile Builder:5 步配置 AI 智能体的人设和偏好
8.2 Dashboard 功能
# 启动 Dashboard(v0.16.0+)
hermes dashboard start
# 默认地址
# http://127.0.0.1:9119
Dashboard 提供以下核心功能:
- 会话管理:查看所有活跃会话,按平台分类
- 技能浏览器:列出所有已加载技能,查看触发条件和执行历史
- 记忆检查器:浏览 MEMORY.md、USER.md 和外部记忆内容
- 工具调试器:测试工具执行,查看输入输出
- RL 轨迹查看器:浏览 Curator 生成的学习轨迹
8.3 远程 Gateway
v0.16.0 新增了远程 Gateway 支持——你可以把 Hermes 部署在服务器上,通过 Dashboard 远程管理:
# ~/.hermes/gateway_remote.yaml
host: "0.0.0.0"
port: 9119
auth:
type: "token"
token: "your-secure-token-here"
tls:
enabled: false # 生产环境建议启用
cert: "/path/to/cert.pem"
key: "/path/to/key.pem"
九、RL 轨迹导出:数据飞轮的核心
9.1 为什么导出 RL 轨迹?
Hermes 的自学习闭环解决的是"单用户"场景——一个 Agent 从自己的经验中学习。但更强大的模式是多用户数据飞轮:
用户 A 的 Agent 学会了 "K8s 排障工作流"
→ 导出 RL 轨迹
→ 贡献到社区
→ 用户 B 的 Agent 加载这个轨迹
→ 用户 B 的 Agent 也能做 K8s 排障了
9.2 轨迹格式
# hermes/curator/trajectory_exporter.py
class TrajectoryExporter:
"""导出 RL 训练轨迹"""
async def export(self, session_id: str, output_format: str = "jsonl") -> str:
"""导出指定会话的 RL 轨迹"""
trajectory = await self._load_trajectory(session_id)
if output_format == "jsonl":
# 每一行是一个 (state, action, reward, next_state) 元组
lines = []
for step in trajectory.steps:
entry = {
"state": {
"task": step.task_description,
"context_summary": step.context_summary,
"available_tools": step.available_tools,
},
"action": {
"tool": step.tool_name,
"parameters": step.tool_params,
"reasoning": step.planning_reasoning,
},
"reward": step.reward, # 基于任务结果自动计算
"next_state": {
"observation": step.observation_summary,
"task_progress": step.progress,
}
}
lines.append(json.dumps(entry, ensure_ascii=False))
return "\n".join(lines)
导出的轨迹遵循标准的 RL 训练格式 (s, a, r, s'),可以直接用于微调语言模型的工具调用能力。
十、生产部署:从本地到云端
10.1 部署方式对比
| 部署方式 | 成本 | 延迟 | 隐私 | 适用场景 |
|---|---|---|---|---|
| 本地运行 | $0 | 最低 | 最高 | 个人开发 |
| Docker | $0 | 低 | 高 | 团队共享 |
| VPS | $5-20/月 | 中 | 中 | 7×24 在线 |
| Serverless | 按用量 | 高 | 低 | 低频使用 |
10.2 Docker 部署实战
# Dockerfile.hermes
FROM python:3.12-slim
# 安装系统依赖
RUN apt-get update && apt-get install -y \
sqlite3 \
git \
curl \
&& rm -rf /var/lib/apt/lists/*
# 安装 Hermes
RUN pip install --no-cache-dir hermes-agent
# 创建数据目录
RUN mkdir -p /data/hermes/{skills,data,gateways}
# 复制配置
COPY hermes.yaml /data/hermes/config.yaml
COPY MEMORY.md /data/hermes/MEMORY.md
COPY USER.md /data/hermes/USER.md
# 暴露端口(Dashboard + Gateway)
EXPOSE 9119 8080
WORKDIR /data/hermes
ENTRYPOINT ["hermes", "serve", "--config", "/data/hermes/config.yaml"]
# docker-compose.yml
version: "3.8"
services:
hermes:
build:
context: .
dockerfile: Dockerfile.hermes
ports:
- "9119:9119" # Dashboard
- "8080:8080" # Gateway
volumes:
- hermes_data:/data/hermes/data # SQLite 数据持久化
- hermes_skills:/data/hermes/skills # 技能文件持久化
environment:
- ANTHROPIC_API_KEY=${ANTHROPIC_API_KEY}
- OPENAI_API_KEY=${OPENAI_API_KEY}
restart: unless-stopped
volumes:
hermes_data:
hermes_skills:
10.3 性能优化建议
- 模型选择:日常对话用 Gemini Flash(便宜、快),复杂推理用 Claude Opus(贵、好)
- 技能数量控制:加载超过 10 个技能时,关键词匹配阶段可能成为瓶颈。定期清理不再使用的技能
- FTS5 索引优化:当 SQLite 数据库超过 1GB 时,考虑定期 VACUUM 和重建索引
- 记忆压缩:长期运行后,Layer 3 的历史会话会膨胀。配置自动压缩策略:
# ~/.hermes/config.yaml
memory:
session_retention_days: 90 # 保留 90 天的会话历史
auto_compress: true # 自动压缩旧会话
compress_after_days: 30 # 30 天后压缩
max_fts_results: 50 # FTS5 最多返回 50 条
十一、与竞品对比:Hermes 的优势与不足
11.1 横向对比
| 维度 | Hermes Agent | OpenClaw | AutoGPT | Browser-use |
|---|---|---|---|---|
| 自学习闭环 | ✅ 内置 E-A-A-S | ❌ | ❌ | ❌ |
| 三层记忆 | ✅ | ✅ 两层 | ❌ 单层 | ❌ 单层 |
| 技能自动生成 | ✅ Curator | ❌ 手动 Skill | ❌ | ❌ |
| 多平台网关 | ✅ 6 个 | ✅ 多平台 | ❌ Web only | ❌ |
| 桌面端 | ✅ Tauri | ✅ | ❌ | ❌ |
| 模型无关 | ✅ | ✅ | ❌ GPT only | ✅ |
| RL 轨迹导出 | ✅ | ❌ | ❌ | ❌ |
| MCP 协议支持 | ✅ | ✅ | ❌ | ✅ |
| 开源协议 | MIT | MIT | MIT | MIT |
| Stars(2026.06) | 61K+ | 302K | 182K | 45K |
11.2 Hermes 的不足
- 小模型驱动效果差:自学习闭环依赖 LLM 做经验提取和技能优化,7B 以下模型效果明显下降
- 记忆检索有时不准:FTS5 全文搜索在语义理解上不如向量检索,"K8s 部署"可能搜不到"Kubernetes 部署"的记录
- Gateway 连接超时:Telegram/Discord 在国内网络环境下偶尔超时,需要代理
- 学习延迟:Curator 是异步的,从任务完成到技能可用有 5-10 分钟延迟
- 技能冲突:多个自动生成的技能可能功能重叠,目前缺少自动合并机制
十二、最佳实践与踩坑记录
12.1 7 条最佳实践
- MEMORY.md 要精不要多:2200 字符的限制是特性,不是 bug。只放高频使用的信息
- 手动创建种子技能:Hermes 自动生成的技能可能不够好。手动创建几个高质量的种子技能,让 Curator 在此基础上优化
- 设置冷却期:避免同一类型任务频繁触发学习(生成大量低质量技能)
- 定期审查技能库:删除低 success_rate 的技能,合并功能重叠的技能
- 模型分层使用:对话用 Flash,规划用 Sonnet,技能优化用 Opus
- 敏感操作加确认:在 skill.yaml 中设置
requires_confirmation: true - 导出 RL 轨迹:即使不做 RL 训练,轨迹也是排查 Agent 行为的宝贵资料
12.2 5 个踩坑记录
坑 1:小模型驱动效果差
现象:用 Qwen-7B 做 Agent 后端,技能生成质量很低,经常生成无意义的模板。
解决:Agent 推理可以用小模型,但 Curator 的经验提取和 GEPA 优化必须用 70B+ 模型。在配置中指定:
# ~/.hermes/config.yaml
models:
agent: "qwen-7b" # 日常推理用小模型
curator: "claude-sonnet" # 学习闭环用大模型
summarizer: "gemini-flash" # 摘要用快速模型
坑 2:记忆检索不准
现象:问"上次数据库迁移怎么做的",搜不到之前讨论"PostgreSQL schema migration"的记录。
解决:在 MEMORY.md 中手动添加同义词映射:
## 术语映射
- 数据库迁移 = schema migration = PostgreSQL 迁移
- K8s = Kubernetes = 容器编排
- CI/CD = 持续集成 = GitHub Actions
坑 3:Gateway 连接 Telegram 超时
解决:配置 HTTP 代理:
# ~/.hermes/gateways/telegram.yaml
proxy:
type: "socks5"
host: "127.0.0.1"
port: 1080
坑 4:自动生成技能过多
现象:Curator 一周内生成了 30+ 个技能,很多是重复的。
解决:设置技能合并策略和数量上限:
# ~/.hermes/config.yaml
curator:
max_skills: 20 # 最多保留 20 个技能
merge_threshold: 0.85 # 相似度 > 0.85 自动合并
min_success_rate: 0.6 # success_rate < 0.6 自动清理
坑 5:execute 方法不是 async
现象:自定义工具的 execute 方法写成同步函数,运行时报错。
解决:所有 execute 方法必须是 async def。如果内部有同步阻塞操作:
# 正确写法:用 asyncio.to_thread 包裹同步操作
async def execute(self) -> str:
result = await asyncio.to_thread(self._sync_operation)
return result
def _sync_operation(self) -> str:
# 同步操作放在这里
import subprocess
return subprocess.run(["kubectl", "get", "pods"], capture_output=True, text=True).stdout
十三、总结与展望
13.1 核心结论
Hermes Agent 的核心价值不在于"又一个 Agent 框架",而在于它从架构层面解决了 Agent 的进化问题:
- 三层记忆解决了"记住你"——不是简单地保存对话历史,而是分层管理不同类型的知识
- E-A-A-S 闭环解决了"越用越强"——从经验中自动提取、优化、生成技能
- 多平台网关解决了"随时可用"——一个 Agent 覆盖六个平台
- RL 轨迹导出解决了"群体进化"——单个 Agent 的经验可以惠及整个生态
13.2 适用场景推荐
- ✅ 个人 AI 助手:每天和它交互,它越来越懂你——这是最理想的场景
- ✅ 团队知识沉淀:把团队的排障流程、发版 SOP 沉淀为技能
- ✅ AI Agent 研究:RL 轨迹导出是独特的数据源
- ⚠️ 一次性任务:如果只是偶尔用一下,自学习的价值体现不出来
- ❌ 高安全性要求:三层记忆 + 多平台接入增加了攻击面
13.3 展望
Hermes Agent 目前的发展方向有三个值得关注:
- 技能市场:agentskills.io 正在建设技能共享市场,类似 npm 之于 Node.js
- 多 Agent 协作:Curator 目前只服务单个 Agent,未来可能支持跨 Agent 的经验共享
- 本地模型优化:团队在研究如何让 7B 模型也能有效运行 Curator
Hermes Agent 证明了一件事:AI Agent 的竞争不在于"谁的模型更强",而在于"谁能从使用中进化"。这个方向,大概率是对的。
本文所有代码基于 Hermes Agent v0.16.0,MIT 协议。项目地址:github.com/nous-research/hermes-agent