Hermes Agent 源码解析 Hermes Agent 源码解析 Hermes Agent 源码解析

❝

仓库: NousResearch/hermes-agent 许可证: MIT 分析日期: 2026-04-15 代码规模: ~21.2 万行 Python 源码，~1.85 万行测试（11,397 个测试函数） 最后提交: 2026-04-14

Hermes Agent 是一个自带"自我进化闭环"的多平台 AI Agent 框架——它通过后台独立 Agent 实例定期回顾对话历史，自动将经验沉淀为 Skill（操作技能）和 Memory（用户画像），并在跨会话中持续改进这些知识。

核心代码在 run_agent.py:run_conversation() (L7745-10639)。

用户消息 → 构建系统提示 → 主循环(LLM调用 → 工具执行 → 结果注入) → 持久化 → 后台审查

主循环关键代码段：

while api_call_count < max_iterations and iteration_budget.remaining > 0:
    1. 检查中断（用户打断）
    2. 准备 API 消息（注入记忆、插件上下文、推理内容）
    3. 应用 Anthropic Prompt 缓存
    4. 调用 LLM API（带 3 次重试、降级链、流式）
    5. 如果 response.tool_calls → 并行/串行执行工具
    6. 如果纯文本回复 → 循环结束

关键参数：

• ▸max_iterations: 默认 90（run_agent.py:651）
• ▸IterationBudget: 线程安全计数器，run_agent.py:L170-211
• ▸重试次数：3 次，抖动退避 5s 基准、120s 上限（retry_utils.py）

注册表模式，三层 import 链：

tools/registry.py（无依赖，最先导入）
    ↑
tools/*.py（每个文件 import 后自动 register()）
    ↑
model_tools.py（导入所有工具文件，触发注册）
    ↑
run_agent.py / cli.py（通过 handle_function_call() 分发）

每个工具注册示例（tools/registry.py）：

registry.register(
    name="web_search",
    toolset="web",
    schema={...},
    handler=lambda args, **kw: web_search(...),
    check_fn=lambda: bool(os.getenv("TAVILY_API_KEY")),
    requires_env=["TAVILY_API_KEY"],
)

并行工具执行策略（run_agent.py:L267-310）：

这是 Hermes 最独特的设计。自我进化不依赖任何外部训练，完全通过运行时对话回顾实现。

在 run_agent.py 中，两个独立的计数器跟踪何时触发回顾：

Memory 回顾触发（run_agent.py:L8871-8878, L10572-10578）：

• ▸_turns_since_memory 计数器：每轮用户对话 +1
• ▸当 _turns_since_memory >= _memory_nudge_interval（默认 10）时触发
• ▸触发后重置为 0

Skill 回顾触发（run_agent.py:L8121-8125, L10588-10592）：

• ▸_iters_since_skill 计数器：每次工具调用迭代 +1
• ▸当 _iters_since_skill >= _skill_nudge_interval（默认 10）时触发
• ▸使用 skill_manage 工具时重置为 0

关键点：这两个计数器不在 run_conversation() 开头重置（L7833-7835 注释明确说明），确保累积逻辑跨对话正确。

触发后，Hermes 会fork 一个完整的 AIAgent 实例在后台线程运行：

_spawn_background_review()（run_agent.py:L2169-2268）：

1. 选择审查提示词（memory / skill / combined）
2. 在 daemon 线程中创建新的 AIAgent：
   - 使用相同 model、provider、platform
   - max_iterations = 8（限制审查成本）
   - quiet_mode = True（静默运行）
   - stdout/stderr 重定向到 /dev/null
   - 共享 _memory_store（直接写入同一内存）
3. 将当前对话快照作为 conversation_history 传入
4. 发送审查提示词，让后台 Agent 自主决定是否保存
5. 审查完成后扫描 _session_messages，提取成功操作摘要
6. 向用户显示 "Memory updated" 或 "Skill created"

Memory 审查提示词（run_agent.py:L2134-2143）：

Review the conversation and consider saving to memory if appropriate.
Focus on:
1. Has the user revealed things about themselves — persona, desires,
   preferences, or personal details?
2. Has the user expressed expectations about how you should behave,
   work style, or ways they want you to operate?
If nothing is worth saving, just say 'Nothing to save.' and stop.

Skill 审查提示词（run_agent.py:L2145-2153）：

Review the conversation and consider saving or updating a skill.
Focus on: was a non-trivial approach used that required trial and error,
or changing course due to experiential findings?
If a relevant skill already exists, update it with what you learned.
Otherwise, create a new skill if the approach is reusable.

skill_manager_tool.py 提供 6 种操作：

Skill 目录结构：

~/.hermes/skills/
├── my-skill/
│   ├── SKILL.md          # 核心指令（YAML frontmatter + Markdown body）
│   ├── references/       # 参考资料
│   ├── templates/        # 模板文件
│   ├── scripts/          # 辅助脚本
│   └── assets/           # 资源文件

安全扫描（tools/skills_guard.py）：Agent 创建的 Skill 和社区安装的 Skill 经过同样的安全扫描，检测危险命令、prompt injection、敏感文件访问等。

memory_tool.py 采用双存储设计：

_system_prompt_snapshot:  启动时冻结的快照，注入系统提示，会话内永不改变
memory_entries:           运行时可变状态，工具调用立即更新并写入磁盘

这个设计确保 Prompt 缓存前缀稳定——会话中间写入 memory 不会改变系统提示，但写入立即持久化，下次会话启动时加载新快照。

文件：~/.hermes/memories/MEMORY.md（Agent 观察）和 USER.md（用户画像），用 \n§\n 作为条目分隔符。

安全扫描（memory_tool.py:L65-102）：memory 内容同样经过 prompt injection 和密钥 exfiltration 模式检测。

agent/memory_manager.py 允许最多一个外部插件与内置 memory 并行运行：

MemoryProvider 抽象接口（agent/memory_provider.py）：

initialize()         → 连接、预热
system_prompt_block() → 静态系统提示文本
prefetch(query)      → 每轮前后台召回
sync_turn(user, asst) → 每轮后异步写入
get_tool_schemas()   → 工具 schema
handle_tool_call()   → 工具分发
shutdown()           → 清理退出

Hermes 的系统提示词构建是一个分层组装策略，核心在 run_agent.py:_build_system_prompt() (L3121-3286)。

系统提示词按以下层级组装（顺序不可变，以保证缓存前缀稳定）：

为什么重要：Anthropic prefix caching 可降低 ~75% 输入 token 成本。

四层保护：

1. ▸

   系统提示只构建一次（_cached_system_prompt，run_agent.py:L7899-7938）

   • ▸首次调用 _build_system_prompt()
   • ▸之后复用缓存值
   • ▸只有上下文压缩后（重建了对话历史）才会重建
2. ▸

   继续会话时从 SQLite 恢复（run_agent.py:L7902-7908）

   • ▸Gateway 为每条消息新建 AIAgent 实例
   • ▸从 SessionDB 加载上次保存的系统提示词
   • ▸不重建——重建会拾取 memory 变化，产生不同提示词，破坏缓存前缀
3. ▸

   插件注入走 user message（run_agent.py:L8000-8031）

   • ▸pre_llm_call hook 的返回值注入到当前轮的 user message
   • ▸不碰 system prompt——保持前缀缓存不变
4. ▸

   消息规范化（run_agent.py:L8221-8243）

   • ▸所有消息 content 做 .strip()
   • ▸tool_call 的 JSON arguments 做 json.dumps(sort_keys=True, separators=(",", ":"))
   • ▸确保 bit-perfect 前缀匹配，最大化本地 inference server（llama.cpp、vLLM）的 KV 缓存复用

prefill_messages（run_agent.py:L781）——可在每次 API 调用时自动注入 few-shot 示例：

• ▸不在 messages 列表中持久化
• ▸不在 session DB / trajectory 中保存
• ▸每次 API 调用自动重新应用（包括继续会话）
• ▸通过 load_prefill_messages() 从 JSON 文件加载

Hermes 根据不同模型的弱点注入不同指南：

Hermes 的上下文管理通过可插拔 Context Engine 架构实现（agent/context_engine.py），默认使用压缩引擎（ContextCompressor）。

agent/context_engine.py 定义了一个抽象基类，所有上下文管理引擎必须实现：

class ContextEngine(ABC):
    @property
    @abstractmethod
    def name(self) -> str          # "compressor", "lcm" 等

    @abstractmethod
    def update_from_response(self, usage):   # 每次 API 响应后更新 token 计数

    @abstractmethod
    def should_compress(self, prompt_tokens):  # 是否触发压缩

    @abstractmethod
    def compress(self, messages, current_tokens)  # 执行压缩

生命周期：

1. ▸实例化 → 注册（plugin register() 或默认）
2. ▸on_session_start() → 会话开始
3. ▸update_from_response() → 每次 API 响应后更新 token 使用量
4. ▸should_compress() → 每轮后检查是否触发
5. ▸compress() → 执行压缩
6. ▸on_session_end() → 会话真正结束时清理

agent/context_compressor.py 的压缩策略：

1. 裁剪旧工具输出（便宜，不需要 LLM）
   - 将旧 tool result 替换为占位符 "[Old tool output cleared...]"
   
2. 保护头部消息
   - system prompt + 前 3 条消息（protect_first_n = 3）
   
3. 保护尾部消息（token 预算，不是条数）
   - 保留最近 ~20K tokens 的对话（protect_last_n 按 token 计算）
   
4. 用辅助模型总结中间部分
   - 摘要前缀："[CONTEXT COMPACTION — REFERENCE ONLY]"
   - 明确告知模型：这是背景参考，不是活跃指令
   - 摘要预算：min(压缩内容的 20%, 12,000 tokens)
   
5. 迭代更新（后续压缩时）
   - 保留上次摘要，与新中间部分合并再总结

关键常量（context_compressor.py:L46-57）：

触发条件：

• ▸配置阈值：默认 50% 上下文使用率（config.yaml: compression.threshold）
• ▸目标压缩比：20%（压缩后保留原始大小的 20%）
• ▸Preflight 检查：加载历史对话时就可能触发（run_agent.py:L7949-7997）

Hermes 不是在单一阈值触发压缩，而是渐进式管理：

关键设计决策（run_agent.py:L796+ 注释）：

• ▸不在中间对话中向 LLM 注入压力警告——这会导致模型"放弃"复杂任务（#7915 bug）
• ▸只通知用户——纯粹的信息显示，不影响 LLM 的行为

Context Engine 是可替换的——通过 context.engine 配置项选择：

第三方引擎可以放在 plugins/context_engine/<name>/ 目录，通过 plugin 系统注册。

Harness 是 Agent 的"操作层"——负责模型调用、错误恢复、凭证管理、路由决策等。

agent/error_classifier.py 提供了一个结构化的 API 错误分类体系：

分类流程（error_classifier.py:L86+）：

1. ▸提取 HTTP 状态码
2. ▸匹配错误消息模式（billing、rate_limit、context_overflow 等）
3. ▸区分使用量限制（可能是临时限速或永久欠费）
4. ▸输出 ClassifiedError，包含 retryable、should_compress、should_rotate_credential、should_fallback 等恢复提示

agent/credential_pool.py + hermes_cli/auth.py 实现了多凭证负载均衡：

支持策略：

凭证池管理：

• ▸支持 OAuth 和 API Key 两种认证
• ▸429 限速 / 402 欠费均冷却 1 小时（可被 provider 的 reset_at 覆盖）
• ▸OAuth token 自动刷新（带 30 秒 skew）
• ▸支持 Copilot、Kimi、Z.ai 等多种 OAuth 流程

配置在 config.yaml 中的 fallback_model 或 fallback_providers：

主 Provider 失败 → 分类错误 → 决定策略：
  - retryable → 重试（带抖动退避）
  - should_compress → 压缩上下文后重试
  - should_rotate_credential → 轮换凭证后重试
  - should_fallback → 切换到下一个 provider

降级激活过程（run_agent.py:L5580-5657）：

1. ▸从 fallback_chain 取下一个 provider
2. ▸用 resolve_provider_client() 路由客户端
3. ▸自动判断 api_mode（chat_completions / codex_responses / anthropic_messages）
4. ▸重建 OpenAI/Anthropic 客户端
5. ▸重置 retry_count 和 compression_attempts
6. ▸如果 fallback_chain 也耗尽，返回错误

空响应立即降级（run_agent.py:L8456-8465）：

• ▸空/malformed 响应被视为限速症状
• ▸立即切换 fallback，不重试

agent/smart_model_routing.py —— 简单对话自动使用廉价模型：

判断逻辑：

• ▸消息长度 ≤ 160 字符 且 ≤ 28 单词
• ▸不包含多行、代码块、URL
• ▸不包含复杂关键词（debug、implement、refactor、architecture 等 35+ 个）

如果命中，路由到 cheap_model 配置（通常是便宜的模型如 GPT-4o-mini）。

Hermes 支持三种 API 模式，自动检测并切换：

自动检测逻辑（run_agent.py:L679-724）：

• ▸provider == "anthropic" 或 URL 包含 "api.anthropic.com" → anthropic_messages
• ▸provider == "openai-codex" 或 URL 包含 "chatgpt.com/backend-api/codex" → codex_responses
• ▸GPT-5.x 模型（如 "o1"、"o3"）→ 自动升级到 codex_responses
• ▸URL 以 "/anthropic" 结尾 → anthropic_messages（第三方 Anthropic 兼容端点）

Hermes Agent 不仅是一个交互式 Agent 框架，还提供了一套完整的 强化学习训练基础设施，通过与 Atropos（NousResearch 的 RL 训练框架）集成，实现从 Agent 交互轨迹到模型训练的闭环。

核心设计在 environments/hermes_base_env.py。Hermes 的 RL 环境支持两种操作模式，对应训练的不同阶段：

模式检测（hermes_base_env.py:328-344）：

def _use_managed_server(self) -> bool:
    server = self.server.servers[0]
    from atroposlib.envs.server_handling.openai_server import OpenAIServer
    return not isinstance(server, OpenAIServer)  # VLLM/SGLang → True

Phase 2 为什么需要 ManagedServer：RL 训练（如 GRPO/PPO）需要精确的 token IDs、logprobs 和 loss mask。ManagedServer 通过 vLLM 的 /generate 端点获取这些数据，而非简单的 /v1/chat/completions。

environments/agent_loop.py 实现了一个独立的、可复用的工具调用循环，专门用于 RL 训练场景。它与 run_agent.py 的主循环结构相同，但做了精简：

用户任务 → 系统提示 → 循环(LLM 调用 → 工具执行 → 结果注入) → AgentResult

核心差异（相比 run_agent.py 主循环）：

HermesAgentLoop.run() 关键代码段（agent_loop.py:175-523）：

async def run(self, messages):
    for turn in range(self.max_turns):
        # 1. API 调用（标准 OpenAI 格式，传入 tools= schemas）
        response = await self.server.chat_completion(**chat_kwargs)

        # 2. 提取推理内容（reasoning_content / reasoning / reasoning_details）
        reasoning = _extract_reasoning_from_message(assistant_msg)

        # 3. 回退解析：若无结构化 tool_calls 但内容含 

完整代码流程（agent_loop.py:204-512）：

1. ▸每轮调用 server.chat_completion()，传入 tool schemas
2. ▸检查 response.choices[0].message.tool_calls
3. ▸如果无结构化 tool_calls 但内容含 `` 标签，使用回退解析器（environments/tool_call_parsers/get_parser('hermes')）提取工具调用
4. ▸工具调用分发：
   • ▸todo / memory / session_search 本地处理或禁用
   • ▸其他工具通过 handle_function_call() 派发，在 ThreadPoolExecutor（128 workers）中运行
5. ▸工具结果经过 maybe_persist_tool_result() 预算裁剪后注入消息
6. ▸如果模型不再调用工具，返回 AgentResult(finished_naturally=True)
7. ▸如果达到 max_turns，返回 AgentResult(finished_naturally=False)

AgentResult 数据结构（agent_loop.py:63-79）：

@dataclass
class AgentResult:
    messages: List[Dict]         # 完整对话历史（OpenAI 格式）
    managed_state: Optional[Dict] # Phase 2 的 SequenceNodes（含 tokens/logprobs/masks）
    turns_used: int              # 实际 LLM 调用次数
    finished_naturally: bool      # 模型是否自然停止（vs 达到 max_turns）
    reasoning_per_turn: List      # 每轮的推理内容提取
    tool_errors: List[ToolError]  # 工具执行错误记录

线程池设计（agent_loop.py:33）：

_tool_executor = concurrent.futures.ThreadPoolExecutor(max_workers=128)

为什么要 128 个 worker？注释明确指出：大规模并行评估场景（如 89 个 TB2 任务同时做工具调用）需要足够的线程避免饥饿。Modal/Docker/Daytona 后端在内部使用 asyncio.run()，必须在独立线程中运行以避免与 Atropos 的事件循环死锁。

Phase 2 使用 VLLM ManagedServer 时，/generate 端点返回纯文本而非结构化的 tool_calls。Hermes 实现了 12 个独立的客户端解析器，每个对应一个模型家族的输出格式：

注册表机制（tool_call_parsers/__init__.py:62-100）：

PARSER_REGISTRY: Dict[str, Type[ToolCallParser]] = {}

@register_parser("hermes")
class HermesToolCallParser(ToolCallParser): ...

def get_parser(name: str) -> ToolCallParser:
    return PARSER_REGISTRY[name]()

Parser 接口（tool_call_parsers/__init__.py:35-58）：

class ToolCallParser(ABC):
    @abstractmethod
    def parse(self, text: str) -> ParseResult:
        # 返回 (content, tool_calls)
        # content = 工具调用标记被剥离后的文本
        # tool_calls = ChatCompletionMessageToolCall 对象列表

每个 Parser 都是独立实现——它们是对 vLLM 服务端 extract_tool_calls() 逻辑的客户端重实现，只依赖标准库（re, json, uuid）和 openai.types。这意味着 Hermes 的 RL 训练不需要 vLLM 作为依赖，只需要模型输出符合预期的文本格式。

environments/tool_context.py 是 RL 训练中最精巧的设计之一。它给 reward/verification 函数提供对 Hermes Agent 所有工具的无限制访问，作用域限定在单个 rollout 的 task_id。

核心设计理念：验证器作者决定使用哪些工具，没有硬编码限制。

class ToolContext:
    def __init__(self, task_id: str):
        self.task_id = task_id

    def terminal(self, command, timeout=180) -> Dict  # 在 rollout 的终端会话中执行命令
    def read_file(self, path) -> Dict                  # 读取文件
    def write_file(self, path, content) -> Dict        # 写入文本文件
    def upload_file(self, local, remote) -> Dict       # 二进制安全上传
    def download_file(self, remote, local) -> Dict     # 二进制安全下载
    def search(self, query, path) -> Dict              # 文件系统搜索
    def web_search(self, query) -> Dict                # 网络搜索
    def browser_navigate(self, url) -> Dict            # 浏览器导航
    def browser_snapshot(self) -> Dict                 # 浏览器快照
    def call_tool(self, name, args) -> str             # 通用工具调用（逃逸舱口）
    def cleanup(self)                                  # 释放所有资源

为什么 task_id 很重要：同一个 task_id 意味着 terminal/browser 会话与模型在 rollout 中使用的是同一个。所有文件系统状态、进程、浏览器标签页都保持不变。验证器可以直接在模型的沙箱中运行测试。

资源清理（tool_context.py:440-474）：cleanup() 按顺序清理：

1. ▸process_registry.kill_all(task_id) — 杀掉后台进程
2. ▸cleanup_vm(task_id) — 释放 VM（Modal/Docker/Daytona）
3. ▸cleanup_browser(task_id) — 关闭浏览器会话（抑制 debug 输出避免日志污染）

hermes_base_env.py:collect_trajectory()（L489-644）是 RL 训练的核心编排函数：

1. 生成唯一 task_id（用于终端/浏览器会话隔离）
2. 解析工具集（每组一次，在 collect_trajectories 中）
3. 构建初始消息（system prompt + user prompt）
4. 判断操作模式：
   - Phase 2: async with server.managed_server() → HermesAgentLoop(managed) → result
   - Phase 1: HermesAgentLoop(server) → result
5. 检查产出：如果只有 system+user 消息（API 首轮失败），跳过 reward 计算
6. 创建 ToolContext(task_id) → compute_reward(item, result, ctx)
7. finally: ctx.cleanup()
8. 构建 ScoredDataItem：
   - Phase 2: 从 managed_state['nodes'] 提取 tokens/masks/logprobs
   - Phase 1: 用 tokenizer 编码对话文本得到近似 tokens
9. 附加 messages（用于 wandb 展示）
10. 返回 (scored_item, [])

关键代码段（hermes_base_env.py:604-644）：

nodes = (result.managed_state or {}).get("nodes", [])
if nodes:
    # Phase 2: 真实 token 数据
    node = nodes[-1]  # 最终序列节点 = 完整轨迹
    scored_item = {
        "tokens": node.tokens,
        "masks": node.masked_tokens,
        "scores": reward,
    }
    if hasattr(node, "logprobs") and node.logprobs:
        scored_item["advantages"] = None  # 由 trainer 计算
        scored_item["ref_logprobs"] = None
else:
    # Phase 1: 占位 tokens（不适合训练，但允许 SFT 数据生成工作）
    tokens = self.tokenizer.encode(full_text, add_special_tokens=True)
    scored_item = {
        "tokens": tokens,
        "masks": [-100] + tokens[1:],  # 第一个 token 作为 prompt 被 mask
        "scores": reward,
    }

environments/hermes_swe_env/hermes_swe_env.py 是一个具体的 RL 环境实现，用于软件工程任务：

配置（hermes_swe_env.py:83-111）：

• ▸工具集：terminal + file + web（编码任务必需）
• ▸终端后端：modal（云隔离，每个 rollout 一个沙箱）
• ▸数据集：bigcode/humanevalpack（可替换）
• ▸最大轮次：30，最大 token：4096
• ▸系统提示："You are a skilled software engineer..."

Reward 计算（hermes_swe_env.py:160-193）：

async def compute_reward(self, item, result, ctx: ToolContext) -> float:
    test_code = item.get("test", item.get("test_code", ""))
    if test_code:
        # 在模型的 Modal 沙箱中运行测试
        test_result = ctx.terminal(f'cd /workspace && python3 -c "{test_code}"', timeout=60)
        if test_result["exit_code"] == 0:
            return 1.0  # 测试通过
    # 部分得分：检查是否创建了 Python 文件
    file_check = ctx.terminal("find /workspace -name '*.py' -newer /tmp/.start_marker")
    if file_check["exit_code"] == 0 and file_check.get("output", "").strip():
        return 0.1  # 至少创建了文件
    return 0.0

这个设计的关键点：ctx.terminal() 使用的是与模型 rollout 期间相同的 Modal 沙箱，所以模型创建的所有文件、安装的依赖、启动的进程都对验证器可见。

RL 训练不需要每次 rollout 都用同样的工具集。hermes_base_env.py:289-322 支持两种工具集解析策略：

1. ▸显式列表：enabled_toolsets=['terminal', 'file'] — 固定工具集
2. ▸分布采样：distribution='development' — 从 toolset_distributions.py 中预定义的分布中随机采样

每次 collect_trajectories() 组调用时解析一次工具集，组内所有并行 rollout 共享同一工具集。这确保了组内对比的公平性。

_SafeWriter（run_agent.py:L113-168）—— 防止 systemd / Docker 容器中 pipe 断开导致的 crash。

model_tools.py:L44-97 —— 三种场景分别管理 event loop：

有 running loop（gateway/Atropos） → 开新线程跑 asyncio.run()
worker 线程 → 线程本地持久化 loop
CLI 主线程 → 全局持久化 loop

解决 httpx/AsyncOpenAI 客户端 GC 时绑定到已关闭 loop 的经典问题。

prompt_builder.py:L36-72 —— 在注入 AGENTS.md、SOUL.md 前检测：

• ▸prompt injection 模式（10+ 种）
• ▸不可见 Unicode 字符
• ▸隐藏 div、curl exfil 等攻击
• ▸memory 内容同样经过此检测

IterationBudget 耗尽后设置 _budget_grace_call = True 给模型最后一次机会。

enforce_turn_budget()（tools/tool_result_storage.py）—— 工具输出超过阈值时自动裁剪，防止单个工具结果撑爆上下文窗口。

run_agent.py（~8,400+ 行）和 cli.py（~10,000+ 行）过大。

改进方向：

• ▸API 调用重试逻辑提取为独立模块
• ▸命令处理用策略模式分发到独立 handler

• ▸测试时需小心 import 顺序
• ▸无法懒加载
• ▸调试需跟踪 import 链

分析局限性：

• ▸未深入阅读每个工具的具体实现（40+ 工具）
• ▸RL 训练环境已深度解读核心文件，但未逐一分析每个 Tool Call Parser 的具体正则逻辑
• ▸ACP 适配器仅查看结构，未深入协议细节
• ▸未实际运行代码验证行为