Hermes Agent 深度拆解：当 AI 学会「自我进化」——从动态技能生成、三层记忆架构到强化学习闭环的工程全貌（2026）
引言：为什么 Hermes Agent 值得深读
2026年上半年，一个名为 Hermes Agent 的开源项目在 GitHub 上以惊人的速度攀升至 21.6 万 Stars。它由 Nous Research 开发，口号是 "self-improving AI agent"——自进化。
这不是营销词汇。当你翻阅其 3 万行 Python 代码（截至 v0.14.0），你会发现一个真正实现了 "从经验中学习" 的工程方案：
- 每次对话结束后自动评估自己的表现
- 决定哪些经验值得写入技能库
- 哪些事情应该记录进长期记忆
- 如何在使用过程中持续优化自身能力
传统 Agent（包括早期的 OpenClaw、LangChain 等）的上下文管理只服务于"当前会话"的稳定性——任务执行完毕，试错经验随之丢失，智能上限被死死锚定在基座模型与静态 Skill 上。
Hermes 的架构突破在于：它构建了一条从"任务执行"到"经验沉淀"再到"权重内化"的完整数据闭环，实现了真正的 Self-Evolving。
本文将深度拆解 Hermes Agent 的核心架构，涵盖动态技能生成、三层记忆系统、ACP 跨 Agent 通信协议、强化学习闭环、以及生产级部署实践。
一、核心架构概览：从静态执行到动态进化
1.1 传统 Agent 的根本局限
传统 AI Agent 架构通常是这样的：
用户请求 → 意图识别 → 工具调用 → 生成响应 → 结束
↓
静态 Skill 库
问题在于：
- Skill 是静态的：预定义好，不更新，遇到新问题只能硬凑
- 经验不沉淀：每次对话从零开始，不会"记住"上次怎么解决的
- 能力上限固定：智能水平 = 基座模型 + 静态工具，不会随使用增长
Hermes 的架构设计正是为了解决这三个痛点。
1.2 Hermes 的双轮驱动架构
Hermes 采用 "外驱 + 内驱" 双轮驱动的自进化架构：
┌─────────────────────────────────────────────────────────────┐
│ 用户交互层 │
│ Telegram / Discord / Slack / WhatsApp / CLI / Email │
└─────────────────────┬───────────────────────────────────────┘
│
┌─────────────────────▼───────────────────────────────────────┐
│ Gateway 消息网关 │
│ 多平台消息统一接入、协议转换、会话管理 │
└─────────────────────┬───────────────────────────────────────┘
│
┌─────────────────────▼───────────────────────────────────────┐
│ Agent 核心引擎 │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │
│ │ 对话主循环 │ │ 上下文引擎 │ │ 工具调度器 │ │
│ │conversation_ │ │context_engine│ │tool_router │ │
│ │loop.py │ │ │ │ │ │
│ └──────────────┘ └──────────────┘ └──────────────┘ │
└─────────────────────┬───────────────────────────────────────┘
│
┌───────────┴───────────┐
│ │
┌─────────▼─────────┐ ┌────────▼────────┐
│ 外驱演进引擎 │ │ 内驱演进引擎 │
│ Skill Curator │ │ GRPO RL Loop │
│ 记忆管理器 │ │ 权重优化器 │
│ 经验审查 Agent │ │ │
└───────────────────┘ └─────────────────┘
│ │
└───────────┬───────────┘
│
┌─────────────────────▼───────────────────────────────────────┐
│ 数据持久层 │
│ SQLite(FTS5) / Redis / PostgreSQL / 8种外部记忆后端 │
└─────────────────────────────────────────────────────────────┘
外驱演进：通过动态 Skill 创建、记忆沉淀、经验审查，让 Agent 在使用过程中积累能力。
内驱演进：基于 GRPO（Group Relative Policy Optimization）的强化学习闭环，通过权重微调让模型本身变得更适合特定任务场景。
这两条线并行推进，构成了 Hermes 的"自进化"内核。
二、外驱演进：动态技能生成与经验沉淀
2.1 从"静态调用"到"动态资产"
传统框架中，Skill 是预定义的静态模块：
# 传统 Skill 定义（静态）
class WeatherSkill(BaseSkill):
name = "get_weather"
description = "获取天气信息"
def execute(self, location: str) -> dict:
# 固定逻辑
return weather_api.get(location)
Hermes 将 Skill 升级为 "动态进化资产"：
# Hermes 的 Skill 动态生成流程（简化示意）
class SkillCurator:
"""后台审查 Agent，负责从交互轨迹中提取 Skill"""
async def review_trajectory(self, trajectory: Trajectory):
"""
trajectory = [
{"role": "user", "content": "帮我分析这个 PDF 的关键条款"},
{"role": "assistant", "tool_calls": [{"name": "read_pdf", ...}]},
{"role": "tool", "content": "...PDF 内容..."},
{"role": "assistant", "content": "基于分析，我提取了三个关键条款..."},
...
]
"""
# Step 1: 记忆审查 — 提取关键事实
facts = await self.extract_facts(trajectory)
# facts = ["用户偏好 PDF 先摘要再详细", "用户关注合同条款风险点"]
# Step 2: 技能审查 — 评估解决路径的通用性
skill_candidate = await self.extract_skill_pattern(trajectory)
# skill_candidate = {
# "name": "contract_clause_analysis",
# "pattern": "read_pdf → extract_clauses → risk_assessment",
# "trigger": "用户请求分析合同/协议/条款",
# "success_rate": 0.92
# }
# Step 3: 综合审查 — 生成优化策略
if skill_candidate["success_rate"] > 0.85:
await self.create_skill(skill_candidate)
return facts, skill_candidate
2.2 触发机制：自动化的学习循环
Hermes 内置 _iters_since_skill 计数器，当 Agent 连续执行多轮对话（默认 10 轮）且未更新技能时，会自动触发经验整理机制：
# agent/conversation_loop.py（核心对话主循环，约 3900 行）
class ConversationLoop:
def __init__(self, agent: "AIAgent"):
self.agent = agent
self._iters_since_skill = 0 # 技能更新计数器
async def run_turn(self, user_input: str) -> str:
# ... 执行对话逻辑 ...
self._iters_since_skill += 1
# 每 10 轮对话检查一次是否需要经验整理
if self._iters_since_skill >= 10:
await self._maybe_curate_experience()
self._iters_since_skill = 0
return response
async def _maybe_curate_experience(self):
"""
后台异步触发经验审查
主 Agent 立即响应用户，审查 Agent 在后台运行
"""
trajectory = self.get_recent_trajectory()
# Fork 一个轻量级审查 Agent
curator = SkillCurator(self.agent.llm_client)
# 异步执行，不阻塞主对话
asyncio.create_task(curator.review_trajectory(trajectory))
这种 "前台即时响应、后台异步进化" 的架构设计，既保证了用户体验，又实现了持续学习。
2.3 技能的结构化表示
Hermes 生成的 Skill 不是简单的"记录"，而是结构化的可执行模块：
# ~/.hermes/skills/contract_clause_analysis.yaml
name: contract_clause_analysis
version: 1.2
created_at: 2026-06-15T10:30:00Z
updated_at: 2026-07-02T14:22:00Z
trigger:
patterns:
- "分析.*合同"
- "审核.*条款"
- "这份协议.*风险"
intent_keywords: ["合同", "条款", "协议", "法律", "风险"]
workflow:
steps:
- name: read_document
tool: read_pdf
params:
extract_mode: structured
focus_areas: ["权利义务", "违约责任", "争议解决"]
- name: extract_clauses
tool: clause_extractor
params:
clause_types: ["termination", "liability", "payment", "confidentiality"]
- name: risk_assessment
tool: legal_risk_analyzer
params:
jurisdiction: "CN" # 可从用户配置读取
risk_levels: ["high", "medium", "low"]
output_format:
summary: "{{extract_clauses.summary}}"
risks: "{{risk_assessment.risks}}"
recommendations: "{{risk_assessment.recommendations}}"
performance:
success_rate: 0.92
avg_duration_ms: 3400
user_satisfaction: 4.7 # 1-5 分
这种结构化表示使得 Skill 可以被：
- 版本管理：跟踪 Skill 的演进历史
- 性能监控：实时统计成功率、耗时、满意度
- 持续优化：基于性能数据自动调整参数
三、三层记忆架构：让 AI 真正"记住"
3.1 为什么需要三层？
传统 Agent 的记忆通常只有一种：对话历史。问题显而易见：
- 短对话还行，长对话上下文爆炸
- 无法区分"当前任务上下文"和"长期知识"
- 用户偏好、习惯等信息散落在海量对话中，无法提取
Hermes 的记忆架构是三层堆叠：
┌─────────────────────────────────────────────────────────────┐
│ Layer 1: Built-in Memory (始终激活) │
│ MEMORY.md (Agent 个人笔记, 2200 字符上限) │
│ USER.md (用户画像, 1375 字符上限) │
│ 作用: 高频关键事实的零成本访问 │
└─────────────────────────────────────────────────────────────┘
↓
┌─────────────────────────────────────────────────────────────┐
│ Layer 2: External Memory Providers (可插拔) │
│ 8 种后端: Honcho / Holographic / Mem0 / Hindsight / │
│ OpenViking / RetainDB / ByteRover / Supermemory │
│ 作用: 语义化深度记忆, 向量检索 │
└─────────────────────────────────────────────────────────────┘
↓
┌─────────────────────────────────────────────────────────────┐
│ Layer 3: Session Search (无限容量) │
│ SQLite + FTS5 全文索引 │
│ 作用: 历史回溯, 精确查找 │
└─────────────────────────────────────────────────────────────┘
3.2 Layer 1：Built-in Memory
这是最轻量的一层，两个 Markdown 文件：
MEMORY.md（Agent 个人笔记）：
# Agent Memory
## 核心原则
- 用户偏好简洁回答，避免冗长解释
- 涉及金融数据时，始终先校验再输出
- 用户主要使用 Go 和 Python
## 学到的模式
- 用户习惯在早上 9-10 点提交代码审查任务
- 用户对性能优化特别关注
- 用户喜欢带代码示例的技术方案
## 避免事项
- 不要用 Markdown 表格，用户觉得在 Telegram 上阅读体验差
- 不要在对话中突然切换语言
USER.md（用户画像）：
# User Profile
## 基本信息
- 角色: 后端架构师
- 技术栈: Go, PostgreSQL, Kubernetes
- 时区: Asia/Shanghai
## 偏好
- 代码风格: 显式优于隐式
- 文档偏好: 先给结论再展开
- 沟通风格: 直接，不喜欢客套话
## 项目上下文
- 当前项目: 支付网关重构
- 痛点: 高并发下的分布式事务
这两个文件在每次会话启动时注入系统提示，零延迟访问，确保 Agent 始终"记得"关键信息。
3.3 Layer 2：External Memory Providers
当记忆超过几百字符，就需要外部存储。Hermes 支持 8 种可插拔后端：
# agent/memory_provider.py
class MemoryProvider(ABC):
"""记忆提供者抽象基类"""
@abstractmethod
async def store(self, key: str, value: str, metadata: dict = None):
"""存储记忆"""
pass
@abstractmethod
async def retrieve(self, query: str, top_k: int = 5) -> List[MemoryItem]:
"""语义检索"""
pass
@abstractmethod
async def forget(self, key: str):
"""删除记忆"""
pass
class Mem0Provider(MemoryProvider):
"""Mem0 云端记忆服务"""
def __init__(self, api_key: str, user_id: str):
self.client = mem0.Client(api_key)
self.user_id = user_id
async def store(self, key: str, value: str, metadata: dict = None):
self.client.add({
"content": value,
"user_id": self.user_id,
"metadata": metadata or {}
})
async def retrieve(self, query: str, top_k: int = 5) -> List[MemoryItem]:
results = self.client.search(query, user_id=self.user_id, limit=top_k)
return [MemoryItem(
content=r["memory"],
score=r["score"],
metadata=r.get("metadata", {})
) for r in results]
class HonchoProvider(MemoryProvider):
"""
Honcho: 对话式用户建模
特点: 通过对话理解用户意图，构建动态用户模型
"""
async def build_user_model(self) -> UserModel:
"""
基于历史对话构建用户模型
不是简单的记忆检索，而是"理解用户"
"""
# Honcho 的独特之处:
# 1. 分析用户的提问模式
# 2. 识别用户的思维偏好
# 3. 预测用户可能的需求
pass
选择建议：
| 后端 | 特点 | 适用场景 |
|---|---|---|
| Mem0 | 云端托管，开箱即用 | 个人用户，快速上手 |
| Honcho | 对话式建模，深度理解 | 需要精细用户画像 |
| RetainDB | 本地向量库，隐私优先 | 企业内网，数据不出域 |
| Supermemory | 超大规模，分布式 | 大量用户，高并发 |
3.4 Layer 3：Session Search
所有历史会话都进 SQLite，带 FTS5 全文索引：
# agent/session_search.py
import sqlite3
import json
class SessionSearch:
def __init__(self, db_path: str = "~/.hermes/sessions.db"):
self.conn = sqlite3.connect(db_path)
self._init_tables()
def _init_tables(self):
self.conn.executescript("""
CREATE TABLE IF NOT EXISTS sessions (
id INTEGER PRIMARY KEY,
user_id TEXT,
started_at TIMESTAMP,
ended_at TIMESTAMP,
summary TEXT
);
-- FTS5 全文索引
CREATE VIRTUAL TABLE IF NOT EXISTS messages_fts USING fts5(
session_id,
role,
content,
tokenize = 'porter unicode61'
);
""")
async def search(self, query: str, limit: int = 10) -> List[SearchResult]:
"""
全文搜索历史对话
支持中文（unicode61 tokenizer）
"""
cursor = self.conn.execute("""
SELECT
m.session_id,
m.role,
m.content,
s.summary,
bm25(messages_fts) as score
FROM messages_fts m
JOIN sessions s ON m.session_id = s.id
WHERE messages_fts MATCH ?
ORDER BY score
LIMIT ?
""", (query, limit))
return [SearchResult(
session_id=row[0],
role=row[1],
content=row[2],
summary=row[3],
score=row[4]
) for row in cursor.fetchall()]
FTS5 的优势：
- 毫秒级检索：百万级消息也能快速响应
- BM25 排序：结果相关性有保障
- Unicode 支持：中文搜索无障碍
3.5 记忆的上下文围栏
防止记忆污染是关键问题。Hermes 引入 "上下文围栏"（Context Fence） 机制：
class ContextFence:
"""
隔离不同来源的记忆，防止相互干扰
"""
def __init__(self):
self.fences = {
"user_profile": Fence(
max_tokens=500,
priority=10, # 最高优先级
allowed_tools=["user_lookup"]
),
"learned_skills": Fence(
max_tokens=1500,
priority=7,
allowed_tools=["skill_execute"]
),
"recent_sessions": Fence(
max_tokens=2000,
priority=5,
allowed_tools=["session_search"]
)
}
def build_context(self, query: str) -> str:
"""
根据当前查询，从各围栏中提取相关上下文
拼接时遵循优先级和 token 预算
"""
context_parts = []
total_tokens = 0
max_context_tokens = 4000 # 上限
for name, fence in sorted(
self.fences.items(),
key=lambda x: x[1].priority,
reverse=True
):
if total_tokens >= max_context_tokens:
break
relevant = fence.retrieve(query)
tokens = count_tokens(relevant)
if total_tokens + tokens <= fence.max_tokens:
context_parts.append(f"[{name}]\n{relevant}")
total_tokens += tokens
return "\n\n".join(context_parts)
这确保了：
- 用户画像始终可见
- 学到的技能按需加载
- 历史对话不淹没关键信息
四、ACP 协议：跨 Agent 通信的标准
4.1 为什么需要 Agent 间通信？
单 Agent 有能力上限。复杂场景需要多 Agent 协作：
- 专业分工：一个 Agent 擅长代码，另一个擅长数据分析
- 负载均衡：高峰期分流请求
- 权限隔离：不同 Agent 不同权限级别
但问题来了：不同 Agent 框架如何通信？
4.2 ACP：Agent Communication Protocol
Hermes 定义了 ACP（Agent Communication Protocol），一种轻量级的 Agent 间通信协议：
# ACP 消息格式
version: "1.0"
message_id: "uuid-xxx"
timestamp: "2026-07-18T10:30:00Z"
from:
agent_id: "hermes-primary"
capabilities: ["code_execution", "web_search", "file_ops"]
to:
agent_id: "hermes-analyst"
capabilities: ["data_analysis", "visualization"]
task:
id: "task-123"
type: "delegation" # delegation | query | notification
action: "analyze_dataset"
params:
dataset_path: "/data/sales_2026.csv"
analysis_type: "trend"
context:
session_id: "sess-456"
user_id: "user-789"
parent_task_id: "task-100"
priority: 7 # 1-10
timeout_ms: 30000
4.3 ACP 适配器实现
# agent/acp_adapter.py
class ACPAdapter:
"""ACP 协议适配器"""
def __init__(self, agent_id: str, registry_url: str):
self.agent_id = agent_id
self.registry = ACPRegistry(registry_url)
async def delegate(
self,
target_capability: str,
task: dict,
timeout_ms: int = 30000
) -> ACPResponse:
"""
委托任务给具有特定能力的 Agent
"""
# 1. 从注册中心查找合适的 Agent
candidates = await self.registry.find_by_capability(target_capability)
if not candidates:
raise NoAgentFoundError(f"No agent with capability: {target_capability}")
# 2. 选择最优候选（基于负载、延迟等）
target = self._select_best_candidate(candidates)
# 3. 构造 ACP 消息
message = ACPMessage(
version="1.0",
message_id=str(uuid.uuid4()),
timestamp=datetime.utcnow().isoformat(),
from_agent=AgentRef(
agent_id=self.agent_id,
capabilities=self.get_own_capabilities()
),
to_agent=AgentRef(
agent_id=target.agent_id,
capabilities=target.capabilities
),
task=Task(
id=str(uuid.uuid4()),
type="delegation",
action=task["action"],
params=task.get("params", {})
),
context=Context(
session_id=self.current_session_id,
user_id=self.current_user_id
),
priority=task.get("priority", 5),
timeout_ms=timeout_ms
)
# 4. 发送并等待响应
response = await self._send_and_wait(message, timeout_ms)
return ACPResponse(
success=response.success,
result=response.result,
error=response.error
)
async def handle_incoming(self, message: ACPMessage) -> ACPResponse:
"""
处理收到的 ACP 消息
"""
# 验证消息
if not self._validate_message(message):
return ACPResponse(success=False, error="Invalid message")
# 检查是否有对应能力
action = message.task.action
if action not in self._action_handlers:
return ACPResponse(success=False, error=f"Action not supported: {action}")
# 执行任务
handler = self._action_handlers[action]
result = await handler(message.task.params)
return ACPResponse(success=True, result=result)
4.4 ACP 注册中心
# agent/acp_registry.py
class ACPRegistry:
"""
Agent 注册中心
支持 etcd / Consul / 内存模式
"""
async def register(self, agent: AgentInfo):
"""
注册 Agent 及其能力
"""
key = f"/agents/{agent.agent_id}"
value = {
"agent_id": agent.agent_id,
"capabilities": agent.capabilities,
"endpoint": agent.endpoint,
"status": "online",
"load": 0, # 当前负载
"last_heartbeat": time.time()
}
await self.store.set(key, json.dumps(value))
async def find_by_capability(self, capability: str) -> List[AgentInfo]:
"""
按能力查找 Agent
"""
agents = []
for key, value in await self.store.list("/agents/"):
agent_data = json.loads(value)
if capability in agent_data["capabilities"]:
if agent_data["status"] == "online":
agents.append(AgentInfo(**agent_data))
return agents
async def heartbeat(self, agent_id: str):
"""
心跳更新
"""
key = f"/agents/{agent_id}"
value = await self.store.get(key)
if value:
agent_data = json.loads(value)
agent_data["last_heartbeat"] = time.time()
await self.store.set(key, json.dumps(agent_data))
五、内驱演进：强化学习闭环
5.1 为什么需要内驱演进？
外驱演进（Skill + 记忆）解决的是 "知识积累" 问题，但有一个上限：基座模型的能力。
如果模型本身不擅长某个任务，积累再多 Skill 也是在"硬凑"。
内驱演进的目标：通过强化学习微调模型权重，让它本身变得更适合特定任务场景。
5.2 GRPO：Group Relative Policy Optimization
Hermes 采用 GRPO（Group Relative Policy Optimization） 算法，这是 PPO 的高效变体：
# rl/grpo_trainer.py
import torch
from transformers import AutoModelForCausalLM, AutoTokenizer
class GRPOTrainer:
"""
Group Relative Policy Optimization
相比 PPO 的改进:
1. 不需要单独的 value model
2. 使用组内相对优势，减少方差
3. 更稳定，更适合 LLM
"""
def __init__(
self,
model: AutoModelForCausalLM,
ref_model: AutoModelForCausalLM, # 参考模型，冻结
tokenizer: AutoTokenizer,
beta: float = 0.04, # KL 散度系数
learning_rate: float = 1e-6
):
self.model = model
self.ref_model = ref_model
self.tokenizer = tokenizer
self.beta = beta
self.optimizer = torch.optim.AdamW(model.parameters(), lr=learning_rate)
def compute_advantages(self, scores: List[float]) -> torch.Tensor:
"""
GRPO 的核心: 组内相对优势
对一组样本的得分进行标准化
"""
scores_tensor = torch.tensor(scores)
mean = scores_tensor.mean()
std = scores_tensor.std() + 1e-8
advantages = (scores_tensor - mean) / std
return advantages
async def train_step(
self,
prompts: List[str],
responses: List[str],
scores: List[float]
):
"""
训练步骤
scores 由 reward model 给出（或人工标注）
"""
# 编码
prompt_ids = self.tokenizer(prompts, return_tensors="pt", padding=True)
response_ids = self.tokenizer(responses, return_tensors="pt", padding=True)
# 计算当前模型的 log probability
with torch.no_grad():
# 参考模型的 log prob（用于 KL 惩罚）
ref_log_probs = self._compute_log_probs(self.ref_model, prompt_ids, response_ids)
# 当前模型的 log prob
log_probs = self._compute_log_probs(self.model, prompt_ids, response_ids)
# 计算优势
advantages = self.compute_advantages(scores)
# GRPO loss
# L = -E[A * log(pi) - beta * KL(pi || pi_ref)]
ratio = torch.exp(log_probs - ref_log_probs)
kl_penalty = self.beta * (log_probs - ref_log_probs)
loss = -(advantages * log_probs - kl_penalty).mean()
# 反向传播
self.optimizer.zero_grad()
loss.backward()
self.optimizer.step()
return {"loss": loss.item(), "mean_score": sum(scores) / len(scores)}
def _compute_log_probs(self, model, prompt_ids, response_ids):
"""计算序列的 log probability"""
input_ids = torch.cat([prompt_ids.input_ids, response_ids.input_ids], dim=1)
attention_mask = torch.cat([prompt_ids.attention_mask, response_ids.attention_mask], dim=1)
outputs = model(input_ids=input_ids, attention_mask=attention_mask)
logits = outputs.logits
# 只取 response 部分的 logits
prompt_len = prompt_ids.input_ids.shape[1]
response_logits = logits[:, prompt_len-1:-1, :]
# 计算 log prob
log_probs = torch.log_softmax(response_logits, dim=-1)
# 收集实际 token 的 log prob
response_token_ids = response_ids.input_ids
token_log_probs = log_probs.gather(
2,
response_token_ids.unsqueeze(-1)
).squeeze(-1)
# 序列 log prob = sum of token log probs
sequence_log_probs = token_log_probs.sum(dim=1)
return sequence_log_probs
5.3 Reward Model：从用户反馈到训练信号
GRPO 需要 scores，这些分数从哪来？
Hermes 设计了多层 reward 来源：
# rl/reward_model.py
class RewardModel:
"""
多源奖励聚合
"""
def __init__(self):
self.sources = {
"explicit_feedback": ExplicitFeedbackSource(), # 用户直接评分
"implicit_signal": ImplicitSignalSource(), # 隐式信号（是否采纳、是否追问）
"task_success": TaskSuccessSource(), # 任务是否完成
"efficiency": EfficiencySource() # 效率（耗时、token 数）
}
async def compute_score(
self,
session: Session,
response: str
) -> float:
"""
综合计算奖励分数
"""
scores = {}
weights = {
"explicit_feedback": 0.4,
"implicit_signal": 0.3,
"task_success": 0.2,
"efficiency": 0.1
}
for name, source in self.sources.items():
scores[name] = await source.evaluate(session, response)
# 加权平均
total = sum(scores[name] * weights[name] for name in scores)
return total
class ImplicitSignalSource:
"""
隐式信号评估
用户不一定给显式反馈，但行为会透露偏好
"""
async def evaluate(self, session: Session, response: str) -> float:
score = 0.5 # 基准分
# 正向信号
if session.user_edited_response:
# 用户编辑了响应 → 原响应不够好
score -= 0.2
if session.user_ended_conversation:
# 用户结束对话 → 任务可能完成
score += 0.2
if session.user_said_thanks:
# 用户说谢谢 → 满意
score += 0.3
if session.user_reasked_same:
# 用户重复问同一问题 → 理解有误
score -= 0.3
return max(0, min(1, score))
5.4 在线学习流程
# rl/online_learner.py
class OnlineLearner:
"""
在线学习：边用边学
"""
def __init__(
self,
trainer: GRPOTrainer,
reward_model: RewardModel,
buffer_size: int = 1000,
min_samples: int = 50
):
self.trainer = trainer
self.reward_model = reward_model
self.buffer = ReplayBuffer(buffer_size)
self.min_samples = min_samples
async def add_experience(
self,
prompt: str,
response: str,
session: Session
):
"""
添加新经验到 buffer
"""
score = await self.reward_model.compute_score(session, response)
self.buffer.add(Experience(
prompt=prompt,
response=response,
score=score,
timestamp=time.time()
))
# 达到阈值时触发训练
if len(self.buffer) >= self.min_samples:
await self._train_batch()
async def _train_batch(self):
"""
批量训练
"""
batch = self.buffer.sample(self.min_samples)
prompts = [e.prompt for e in batch]
responses = [e.response for e in batch]
scores = [e.score for e in batch]
metrics = await self.trainer.train_step(prompts, responses, scores)
print(f"[GRPO] loss={metrics['loss']:.4f}, mean_score={metrics['mean_score']:.2f}")
六、工具链：40+ 内置工具与自定义扩展
6.1 内置工具分类
Hermes 内置 40+ 工具，覆盖 8 大类别：
| 类别 | 工具示例 | 用途 |
|---|---|---|
| 终端执行 | execute_command, run_script | 本地命令执行 |
| 文件操作 | read_file, write_file, search_files | 文件系统访问 |
| 网页搜索 | web_search, fetch_url | 信息检索 |
| 浏览器自动化 | browser_navigate, browser_click | Web 交互 |
| 图像生成 | generate_image, edit_image | AI 绘图 |
| 语音处理 | tts, stt | 语音合成/识别 |
| 数据分析 | analyze_csv, plot_data | 数据处理 |
| 系统信息 | get_weather, get_time | 环境查询 |
6.2 工具定义规范
# tools/base.py
from pydantic import BaseModel, Field
from typing import Optional, List, Dict, Any
from abc import ABC, abstractmethod
class ToolParameter(BaseModel):
"""工具参数定义"""
name: str
type: str # "string" | "number" | "boolean" | "array" | "object"
description: str
required: bool = True
default: Optional[Any] = None
enum: Optional[List[str]] = None
class ToolResult(BaseModel):
"""工具执行结果"""
success: bool
data: Optional[Any] = None
error: Optional[str] = None
metadata: Dict[str, Any] = {}
class BaseTool(ABC):
"""工具基类"""
name: str
description: str
parameters: List[ToolParameter]
@abstractmethod
async def execute(self, **kwargs) -> ToolResult:
"""执行工具"""
pass
def to_openai_tool(self) -> dict:
"""转换为 OpenAI tool 格式"""
return {
"type": "function",
"function": {
"name": self.name,
"description": self.description,
"parameters": {
"type": "object",
"properties": {
p.name: {
"type": p.type,
"description": p.description,
**({"enum": p.enum} if p.enum else {}),
**({"default": p.default} if p.default is not None else {})
}
for p in self.parameters
},
"required": [p.name for p in self.parameters if p.required]
}
}
}
# tools/web_search.py
class WebSearchTool(BaseTool):
"""网页搜索工具"""
name = "web_search"
description = "搜索网络信息，返回相关结果"
parameters = [
ToolParameter(
name="query",
type="string",
description="搜索关键词",
required=True
),
ToolParameter(
name="count",
type="number",
description="返回结果数量，默认 10",
required=False,
default=10
),
ToolParameter(
name="freshness",
type="string",
description="时间范围: day | week | month | year",
required=False,
default="week",
enum=["day", "week", "month", "year"]
)
]
async def execute(self, query: str, count: int = 10, freshness: str = "week") -> ToolResult:
# 实际搜索逻辑
results = await self._search_provider.search(query, count, freshness)
return ToolResult(
success=True,
data=results,
metadata={
"query": query,
"count": count,
"provider": self._search_provider.name
}
)
6.3 自定义工具开发
# my_tools/company_api.py
from hermes.tools import BaseTool, ToolParameter, ToolResult
class CompanyAPITool(BaseTool):
"""企业内部 API 工具"""
name = "company_api"
description = "查询企业内部数据：员工信息、项目状态、代码仓库等"
parameters = [
ToolParameter(
name="resource",
type="string",
description="资源类型: employee | project | repo",
required=True,
enum=["employee", "project", "repo"]
),
ToolParameter(
name="query",
type="string",
description="查询条件",
required=True
)
]
def __init__(self, api_base: str, api_key: str):
self.api_base = api_base
self.api_key = api_key
async def execute(self, resource: str, query: str) -> ToolResult:
import aiohttp
async with aiohttp.ClientSession() as session:
url = f"{self.api_base}/{resource}"
headers = {"Authorization": f"Bearer {self.api_key}"}
async with session.get(url, params={"q": query}, headers=headers) as resp:
if resp.status == 200:
data = await resp.json()
return ToolResult(success=True, data=data)
else:
return ToolResult(success=False, error=f"API error: {resp.status}")
# 注册工具
# config.yaml
tools:
custom:
- type: my_tools.company_api.CompanyAPITool
config:
api_base: "https://internal.company.com/api"
api_key: "${COMPANY_API_KEY}" # 从环境变量读取
七、多平台接入：Gateway 消息网关
7.1 统一接入架构
Hermes 支持同时接入多个消息平台，通过 Gateway 进程统一管理：
┌─────────────────────────────────────────────────────────────┐
│ Gateway 进程 │
│ ┌─────────┐ ┌─────────┐ ┌─────────┐ ┌─────────┐ │
│ │Telegram │ │Discord │ │ Slack │ │WhatsApp │ ... │
│ │Handler │ │Handler │ │Handler │ │Handler │ │
│ └────┬────┘ └────┬────┘ └────┬────┘ └────┬────┘ │
│ │ │ │ │ │
│ └───────────┴─────┬─────┴───────────┘ │
│ │ │
│ ┌──────▼──────┐ │
│ │ Message Bus │ │
│ │ (Redis Pub/Sub) │
│ └──────┬──────┘ │
└─────────────────────────┼───────────────────────────────────┘
│
┌─────────────────────────▼───────────────────────────────────┐
│ Agent 进程 │
│ 消费消息，生成响应 │
└─────────────────────────────────────────────────────────────┘
7.2 消息处理器实现
# gateway/handlers/telegram.py
from telegram import Update
from telegram.ext import Application, MessageHandler, filters
class TelegramHandler:
"""Telegram 消息处理器"""
def __init__(self, token: str, message_bus: MessageBus):
self.app = Application.builder().token(token).build()
self.message_bus = message_bus
# 注册消息处理器
self.app.add_handler(MessageHandler(filters.TEXT, self._handle_message))
async def _handle_message(self, update: Update, context):
"""收到消息时触发"""
message = update.message
# 构造统一消息格式
unified_msg = UnifiedMessage(
platform="telegram",
user_id=str(message.from_user.id),
chat_id=str(message.chat_id),
content=message.text,
metadata={
"username": message.from_user.username,
"message_id": message.message_id,
"timestamp": message.date.isoformat()
}
)
# 发布到消息总线
await self.message_bus.publish("incoming", unified_msg)
async def send_response(self, chat_id: str, text: str, **kwargs):
"""发送响应"""
await self.app.bot.send_message(
chat_id=int(chat_id),
text=text,
parse_mode=kwargs.get("parse_mode", "Markdown")
)
7.3 消息总线
# gateway/message_bus.py
import asyncio
import json
from typing import Callable
import redis.asyncio as redis
class MessageBus:
"""
基于 Redis Pub/Sub 的消息总线
解耦 Gateway 和 Agent
"""
def __init__(self, redis_url: str = "redis://localhost"):
self.redis = redis.from_url(redis_url)
self.handlers = {}
async def publish(self, channel: str, message: UnifiedMessage):
"""发布消息"""
await self.redis.publish(
channel,
json.dumps(message.to_dict())
)
async def subscribe(self, channel: str, handler: Callable):
"""订阅消息"""
pubsub = self.redis.pubsub()
await pubsub.subscribe(channel)
async for message in pubsub.listen():
if message["type"] == "message":
data = json.loads(message["data"])
unified_msg = UnifiedMessage.from_dict(data)
await handler(unified_msg)
八、部署实践：从本地到生产
8.1 本地快速启动
# 一键安装（Linux / macOS / WSL2）
curl -fsSL https://hermes-agent.nousresearch.com/install.sh | bash
source ~/.bashrc
# 配置
hermes config set model openrouter # 使用 OpenRouter 接入 300+ 模型
hermes config set default_model anthropic/claude-sonnet-4
# 配置 API Key
export OPENROUTER_API_KEY="sk-or-xxx"
# 启动 CLI 模式
hermes chat
# 启动 Telegram Bot
hermes start --platform telegram
8.2 Docker 部署
# docker-compose.yml
version: "3.9"
services:
hermes-agent:
image: nousresearch/hermes-agent:latest
restart: unless-stopped
environment:
- OPENROUTER_API_KEY=${OPENROUTER_API_KEY}
- TELEGRAM_BOT_TOKEN=${TELEGRAM_BOT_TOKEN}
volumes:
- ./config:/root/.hermes
- ./skills:/root/.hermes/skills
ports:
- "8080:8080" # 管理 API
depends_on:
- redis
redis:
image: redis:7-alpine
restart: unless-stopped
volumes:
- redis-data:/data
volumes:
redis-data:
8.3 生产级配置
# config.yaml（完整生产配置）
core:
name: "企业助手"
description: "智能化办公助手"
timezone: "Asia/Shanghai"
model:
provider: openrouter
default: anthropic/claude-sonnet-4
fallbacks:
- openai/gpt-4.1
- google/gemini-2.5-flash
temperature: 0.7
max_tokens: 4096
memory:
layer1:
enabled: true
memory_md_max_chars: 2200
user_md_max_chars: 1375
layer2:
provider: mem0
api_key: "${MEM0_API_KEY}"
layer3:
type: postgres
connection: "postgresql://user:pass@localhost/hermes"
skills:
auto_create: true
min_success_rate: 0.85
storage_path: "./skills"
rl:
enabled: true
algorithm: grpo
buffer_size: 1000
min_samples_for_training: 50
learning_rate: 1e-6
beta: 0.04
tools:
built_in:
- web_search
- file_ops
- execute_command
- browser_automation
custom:
- type: my_tools.company_api.CompanyAPITool
config:
api_base: "https://internal.company.com/api"
gateway:
platforms:
- type: telegram
token: "${TELEGRAM_BOT_TOKEN}"
- type: discord
token: "${DISCORD_BOT_TOKEN}"
- type: slack
app_token: "${SLACK_APP_TOKEN}"
bot_token: "${SLACK_BOT_TOKEN}"
monitoring:
metrics:
enabled: true
port: 9090
logging:
level: INFO
format: json
8.4 性能优化建议
# 性能调优清单
# 1. 上下文压缩
# 启用 LLM 摘要，避免 context 爆炸
context_compressor:
enabled: true
max_tokens: 3000
strategy: "summary" # summary | sliding_window | hybrid
# 2. 并行工具调用
# 支持一次响应中调用多个独立工具
tool_execution:
parallel: true
max_concurrent: 5
# 3. 响应缓存
# 相同问题直接返回缓存答案
cache:
enabled: true
ttl_seconds: 3600
semantic_match: true # 语义相似也命中
# 4. 流式输出
# 边生成边返回，提升用户体验
streaming: true
# 5. 负载均衡
# 多 Agent 实例分担请求
gateway:
load_balancing:
strategy: "least_connections"
health_check_interval: 30
九、与其他框架对比
9.1 能力对比
| 特性 | Hermes Agent | OpenClaw | LangChain | AutoGPT |
|---|---|---|---|---|
| 自进化学习 | ✅ Skill + RL | ✅ Skills | ❌ | ❌ |
| 记忆系统 | ✅ 三层架构 | ✅ 两层 | ⚠️ 需扩展 | ⚠️ 简单 |
| 跨 Agent 通信 | ✅ ACP 协议 | ✅ 会话管理 | ❌ | ❌ |
| 多平台接入 | ✅ 8+ 平台 | ✅ 10+ 平台 | ⚠️ 需自建 | ❌ |
| 工具生态 | ✅ 40+ 内置 | ✅ 丰富 | ✅ 丰富 | ⚠️ 有限 |
| RL 微调 | ✅ 内置 GRPO | ❌ | ❌ | ❌ |
| 开源协议 | MIT | MIT | MIT | MIT |
9.2 选择建议
| 场景 | 推荐框架 |
|---|---|
| 需要持续学习、越用越聪明 | Hermes Agent |
| 需要多 Agent 编排、复杂工作流 | OpenClaw |
| 需要快速原型、生态丰富 | LangChain |
| 需要简单自动化任务 | AutoGPT |
十、总结与展望
10.1 Hermes Agent 的核心价值
Hermes Agent 的突破不在于某个单点技术，而在于 "自进化闭环" 的完整工程实现：
- 外驱演进：从经验中创建 Skill，从交互中沉淀记忆
- 内驱演进：通过 RL 微调模型，让基座能力适配场景
- 三层记忆：解决"记住什么"和"如何检索"的问题
- ACP 协议：让 Agent 间协作有标准可循
这些模块协同工作，让 Agent 真正实现了 "越用越聪明"。
10.2 未来发展方向
从代码和设计看，Hermes Agent 未来可能演进的方向：
- 多模态学习：目前主要是文本，未来会扩展到图像、语音、视频
- 联邦学习：多个 Agent 共享学习成果，但数据不出域
- 主动探索：不只是等用户提问，主动发现需要学习的领域
- 人格定制：不只是能力进化，还能进化"性格"和"风格"
10.3 对开发者的启示
Hermes Agent 的代码给我们几个重要启示：
- 架构分层要清晰：Agent / Gateway / Memory 各司其职
- 性能与体验要平衡：后台学习不阻塞前台响应
- 扩展性要留足：插件化的工具、可插拔的记忆后端
- 可观测性要内置：日志、指标、追踪从一开始就设计好
附录：核心代码模块索引
| 模块 | 文件 | 行数 | 核心功能 |
|---|---|---|---|
| 对话主循环 | agent/conversation_loop.py | ~3900 | 消息处理、工具调用、技能触发 |
| Agent 初始化 | agent/agent_init.py | ~1200 | 配置加载、模块初始化 |
| 上下文引擎 | agent/context_engine.py | ~800 | 上下文构建、压缩、围栏 |
| 记忆管理 | agent/memory_manager.py | ~600 | 三层记忆协调 |
| 技能审查 | agent/skill_curator.py | ~1500 | 技能提取、验证、存储 |
| ACP 适配器 | agent/acp_adapter.py | ~900 | Agent 间通信 |
| GRPO 训练器 | rl/grpo_trainer.py | ~700 | 强化学习训练 |
参考资源：
- GitHub: https://github.com/NousResearch/hermes-agent
- 文档: https://hermes-agent.nousresearch.com
- Nous Research: https://nousresearch.com