Forem: Henry Lin

Hermes Agent 项目思维导图

Henry Lin — Sun, 19 Apr 2026 13:38:57 +0000

Hermes Agent 项目思维导图

这份文档用于导入 XMind 或其他支持 Markdown 层级导入的思维导图工具。主体内容尽量使用稳定的分级结构，减少长段解释，方便直接转换为脑图。

Hermes Agent
- 项目定位
- 自主型 AI Agent 运行时
- 不只是聊天界面
- 不只是 IDE copilot
- 目标
  - 调用工具完成任务
  - 跨会话保留长期能力
  - 跨平台运行
  - 支持自动化与扩展
- 典型场景
  - 本地开发助手
  - 远程消息平台代理
  - 定时任务与监控
  - 企业内部系统接入
  - 研究与批处理
- 用户入口
- CLI
  - hermes
  - 交互式主入口
  - 最完整的调试与观察界面
- Gateway
  - hermes gateway
  - Telegram
  - Discord
  - Slack
  - WhatsApp
  - Signal
  - 其他平台适配器
- ACP
  - 编辑器集成
  - VS Code
  - Zed
  - JetBrains
- API Server / Python Library / Batch Runner
  - 程序化调用
  - 轨迹批处理
  - 研究与训练相关入口
- 核心能力
- Tools
  - 文件操作
  - 终端执行
  - Web 搜索与抽取
  - Browser 自动化
  - Vision / TTS / Image Gen
  - Memory
  - Messaging
  - Delegation
  - Cron
- Toolsets
  - 核心 toolset
  - 组合 toolset
  - 平台 toolset
  - 动态 toolset
- Skills
  - 按需加载的流程知识
  - 本地 skills
  - Skills Hub
  - 动态 slash commands
- Memory
  - 跨会话事实信息
  - MEMORY.md / USER.md
  - 外部 memory provider
- Context Files
  - AGENTS.md
  - SOUL.md
  - .cursorrules
  - .cursor/rules/*.mdc
- MCP
  - 外部工具接入
  - stdio server
  - HTTP server
  - 动态工具注册
- Cron
  - 定时运行新 session
  - delivery 到平台
- Delegation
  - 子代理
  - 并行任务
  - 上下文隔离
- 核心运行系统
- AIAgent
  - run_agent.py
  - chat()
  - run_conversation()
  - 同步 agent loop
- Prompt Builder
  - agent/prompt_builder.py
  - 组装系统提示词
  - 注入 personality / memory / context / skills
- Prompt Caching
  - agent/prompt_caching.py
  - 保持 prefix cache
- Context Compression
  - agent/context_compressor.py
  - 长会话压缩
- Provider / Runtime Resolution
  - hermes_cli/runtime_provider.py
  - hermes_cli/auth.py
  - provider 选择
  - credentials 解析
- Tool Dispatch
  - model_tools.py
  - schema 收集
  - handle function call
  - tool availability gating
- Session Storage
  - hermes_state.py
  - gateway/session.py
  - SQLite + FTS5
- Response Delivery
  - CLI display
  - gateway delivery
  - cron delivery
- 核心模块
- 顶层核心文件
  - run_agent.py
  - cli.py
  - model_tools.py
  - toolsets.py
  - hermes_state.py
  - batch_runner.py
- agent/
  - prompt builder
  - context compressor
  - prompt caching
  - auxiliary client
  - display
  - skill commands
  - trajectory
- hermes_cli/
  - main.py
  - config.py
  - commands.py
  - setup.py
  - tools_config.py
  - skills_config.py
  - model_switch.py
  - auth.py
- tools/
  - registry.py
  - terminal_tool.py
  - file_tools.py
  - web_tools.py
  - browser_tool.py
  - code_execution_tool.py
  - delegate_tool.py
  - mcp_tool.py
  - environments/
- gateway/
  - run.py
  - session.py
  - delivery.py
  - pairing.py
  - platforms/
- cron/
  - jobs.py
  - scheduler.py
- 其他关键目录
  - acp_adapter/
  - plugins/
  - skills/
  - optional-skills/
  - website/
  - tests/
- 配置与状态体系
- ~/.hermes/config.yaml
- ~/.hermes/.env
- ~/.hermes/SOUL.md
- ~/.hermes/skills/
- ~/.hermes/sessions/
- HERMES_HOME
- Profiles
  - hermes profile list
  - hermes profile create
  - hermes profile use
  - hermes profile show
- 关键规则
- Prompt cache 不应随意打破
- 不要中途重建系统提示
- Toolset 不应在会话中随意变化
- profile-safe 路径规则
  - get_hermes_home()
  - display_hermes_home()
- Tool / Skill / MCP 的选择边界
  - 流程优先 skill
  - 外部系统优先 MCP
  - 原生底层能力再做 tool
- 测试不要写到真实 ~/.hermes
- 关键文件
- 入口
  - run_agent.py
  - cli.py
  - gateway/run.py
- 工具系统
  - model_tools.py
  - toolsets.py
  - tools/registry.py
- 配置系统
  - hermes_cli/config.py
  - hermes_cli/main.py
  - hermes_constants.py
- 会话与状态
  - hermes_state.py
  - gateway/session.py
- 扩展系统
  - tools/mcp_tool.py
  - agent/skill_commands.py
  - hermes_cli/commands.py
- 关键流程
- CLI 请求流程
  - 用户输入
  - CLI
  - AIAgent
  - Prompt Builder
  - Provider / Runtime Resolution
  - Model API Call
  - Tool Dispatch
  - Built-in Tools / MCP Tools
  - Session Storage
  - CLI display
- Gateway 请求流程
  - 平台消息
  - Gateway
  - 会话解析与授权
  - AIAgent
  - Tool Dispatch
  - Session Storage
  - 平台回传
- Cron 请求流程
  - Scheduler Tick
  - 新 session
  - AIAgent
  - Tool Dispatch
  - Session Storage
  - Delivery target
- MCP 工具接入流程
  - mcp_servers
  - server connect
  - tool discovery
  - dynamic toolset
  - model 可调用
- Tool 调用流程
  - schema collection
  - model tool call
  - handle_function_call()
  - registry dispatch
  - tool result
  - loop continuation
- 常用命令与观察点
- CLI
  - hermes
  - hermes setup
  - hermes model
  - hermes tools
  - hermes gateway
  - hermes cron
- Session Commands
  - /model
  - /usage
  - /compress
  - /tools
  - /toolsets
  - /skills
  - /reload-mcp
  - /background
- Debug
  - hermes doctor
  - hermes status
  - /verbose
  - python -m pytest tests/ -q

Hermes Agent 核心运行系统调用流程--源码分析

Henry Lin — Tue, 14 Apr 2026 12:44:22 +0000

Hermes Agent 核心运行系统调用流程

本文基于当前仓库实现梳理 Hermes Agent 从入口层到结果交付的真实调用链，重点覆盖：

核心运行系统 AIAgent
Prompt Builder / Prompt Caching / Context Compression
Provider / Runtime Resolution
Tool Dispatch
Session Storage
Response Delivery（CLI / gateway / cron）

不展开单个 tool 的内部业务逻辑，重点描述“谁调用谁、数据如何流动、哪些状态被缓存或持久化”。

1. 总体分层

入口层
  CLI: cli.py
  Gateway: gateway/run.py
  Cron: cron/scheduler.py
      ↓
运行时解析
  hermes_cli/runtime_provider.py
  hermes_cli/auth.py
      ↓
核心 Agent 初始化
  run_agent.py -> AIAgent.__init__()
      ↓
系统提示词装配
  AIAgent._build_system_prompt()
  agent/prompt_builder.py
      ↓
同步 agent loop
  AIAgent.run_conversation()
      ↓
模型响应二分
  无 tool_calls -> 最终响应
  有 tool_calls -> tool dispatch -> tool result 回填 -> 下一轮
      ↓
上下文管理
  agent/prompt_caching.py
  agent/context_compressor.py
      ↓
持久化
  hermes_state.py / gateway/session.py
      ↓
结果交付
  CLI Rich/TUI
  Gateway adapter.send(...)
  Cron output file + auto delivery

2. 三个入口如何进入 AIAgent

2.1 CLI 路径

入口主要在 cli.py：

HermesCLI.chat() 先调用 _ensure_runtime_credentials()。
_ensure_runtime_credentials() 通过 resolve_runtime_provider() 解析 provider、api_mode、base_url、api_key。
_init_agent() 在首次使用或路由变化时创建 AIAgent(...)。
HermesCLI.chat() 调用 self.agent.run_conversation(...)。
返回结果后，CLI 负责把 reasoning / response / TTS / streaming 内容显示到终端。

关键点：

CLI 会在 provider、model、route、credential 变化时重建 agent，而不是复用旧 client。
CLI 恢复会话时，会先从 SessionDB 取历史消息，再交给 AIAgent 继续运行。

相关位置：

cli.py:2169 _ensure_runtime_credentials()
cli.py:2269 _init_agent()
cli.py:6309 HermesCLI.chat()

2.2 Gateway 路径

入口主要在 gateway/run.py：

收到平台消息后，SessionStore.get_or_create_session(source) 建立/获取会话。
build_session_context(...) 和 build_session_context_prompt(...) 生成平台上下文。
session_store.load_transcript(session_id) 读取历史对话。
_run_agent(...) 内部解析 runtime、构造或复用 AIAgent。
调用 agent.run_conversation(message, conversation_history=agent_history, ...)。
将新消息写回 transcript / SQLite，并把最终结果返回给 adapter。

关键点：

Gateway 不把 session context 写进持久化的 system prompt，而是作为 ephemeral_system_prompt 每轮临时注入，避免破坏 prompt cache 前缀。
Gateway 会按 session_key + agent config signature 缓存 AIAgent，尽量复用同一个 agent 实例，保持 system prompt 和 tool schema 稳定。

相关位置：

gateway/run.py:2308 获取/创建 session
gateway/run.py:2342 构造 context prompt
gateway/run.py:2435 加载 transcript
gateway/run.py:6647 创建/复用 AIAgent
gateway/run.py:6856 调用 run_conversation()

2.3 Cron 路径

入口主要在 cron/scheduler.py 的 run_job(job)：

构造 cron 专用 prompt（包含 skill 展开、cron 执行提示）。
resolve_runtime_provider() 解析运行时 provider。
创建 AIAgent(..., platform="cron", disabled_toolsets=["cronjob", "messaging", "clarify"], skip_memory=True)。
在线程池里执行 agent.run_conversation(prompt)。
生成 markdown output 文档并保存到 ~/.hermes/cron/output/...。
如配置了 deliver，调用 _deliver_result() 发送到目标平台。

关键点：

cron 是 headless 执行，不允许 clarify。
cron 同样接入 SessionDB，因此历史可被 session_search 检索。
cron 有“无活动超时”机制，不是简单 wall-clock timeout。

相关位置：

cron/scheduler.py:512 run_job()
cron/scheduler.py:623 runtime 解析
cron/scheduler.py:643 创建 AIAgent
cron/scheduler.py:678 调用 run_conversation()
cron/scheduler.py:199 _deliver_result()

3. Provider / Runtime Resolution 调用链

运行时解析由 hermes_cli/runtime_provider.py 统一封装，CLI / gateway / cron 都走这里。

3.1 入口函数

统一入口：

resolve_runtime_provider(...) in hermes_cli/runtime_provider.py:566

它返回一个 runtime dict，典型字段包括：

provider
api_mode
base_url
api_key
command / args（外部进程型 provider）
credential_pool
requested_provider

3.2 解析步骤

resolve_runtime_provider() 的主流程：

resolve_requested_provider() 先决定“用户想要哪个 provider”。
_resolve_named_custom_runtime() 先尝试命名 custom provider。
resolve_provider() in hermes_cli/auth.py 解析最终 provider 名称。
_resolve_explicit_runtime() 处理显式 api_key/base_url 覆盖。
如有 credential pool，优先从 pool 取 runtime credential。
按 provider 类型进入对应分支：
- nous -> resolve_nous_runtime_credentials()
- openai-codex -> resolve_codex_runtime_credentials()
- copilot-acp -> resolve_external_process_provider_credentials()
- anthropic -> 走 Anthropic token 解析
- API-key provider -> resolve_api_key_provider_credentials()
- 其他 -> _resolve_openrouter_runtime()

3.3 `auth.py` 的职责

hermes_cli/auth.py 主要负责：

resolve_provider()：决定 provider 名称
resolve_codex_runtime_credentials()：读取/刷新 Codex token
resolve_nous_runtime_credentials()：刷新 Nous portal access token，并确保短期 inference key 可用
resolve_api_key_provider_credentials()：读取 API-key 类 provider 的密钥和 base_url
resolve_external_process_provider_credentials()：解析 copilot-acp 这类本地子进程 provider

关键位置：

hermes_cli/auth.py:790 resolve_provider()
hermes_cli/auth.py:1174 resolve_codex_runtime_credentials()
hermes_cli/auth.py:1683 resolve_nous_runtime_credentials()
hermes_cli/auth.py:2084 resolve_api_key_provider_credentials()
hermes_cli/auth.py:2122 resolve_external_process_provider_credentials()

3.4 运行时解析的输出如何进入 AIAgent

三个入口都会把 runtime dict 展开为 AIAgent(...) 参数：

api_key
base_url
provider
api_mode
command / args
credential_pool

随后 AIAgent.__init__() 决定使用哪类 client：

anthropic_messages -> Anthropic native client
codex_responses -> Responses API / raw codex path
其他 -> OpenAI-compatible client

4. AIAgent 初始化阶段

核心类位于 run_agent.py：

AIAgent 定义：run_agent.py:470
__init__()：run_agent.py:487
run_conversation()：run_agent.py:6801
chat()：run_agent.py:9170

4.1 初始化时完成的事情

AIAgent.__init__() 不是只做 client 构造，它会一次性初始化整套运行时状态：

保存 model / provider / api_mode / iteration budget。
基于 provider/base_url 判断实际 API 模式。
创建底层 LLM client。
调用 get_tool_definitions(...) 加载当前会话可用工具。
初始化 SessionDB 会话行（如果传入了 session_db）。
初始化 todo store、memory store、memory provider plugin。
初始化 context compressor。
计算 prompt caching 开关。
建立 _cached_system_prompt 缓存槽。
建立 session log 文件路径和 _session_messages 缓冲区。

4.2 初始化时和本题流程直接相关的状态

self.tools
- 来自 model_tools.get_tool_definitions()
self.valid_tool_names
- 当前会话真实可调用的工具名集合
self._cached_system_prompt
- 会话级稳定 system prompt 快照
self.context_compressor
- 长会话压缩器
self._session_db
- SQLite 会话存储
self._memory_store / self._memory_manager
- memory 注入和外部 memory provider
self._use_prompt_caching
- 是否启用 Anthropic/OpenRouter prompt cache 标记

5. Prompt Builder：系统提示词如何被组装

AIAgent._build_system_prompt() 是总装函数，位于：

run_agent.py:2609

真正的提示词拼装细节在：

agent/prompt_builder.py

5.1 组装顺序

_build_system_prompt() 的拼接顺序非常明确：

Agent identity
- 优先 SOUL.md
- 否则 DEFAULT_AGENT_IDENTITY
tool-aware guidance
- 仅当对应 tool 存在时才注入 memory/session_search/skills guidance
Nous subscription prompt
tool-use enforcement
- 按模型名和 config 决定是否注入
调用方传入的 system_message
内建 memory / user profile
外部 memory provider prompt block
skills system prompt
project context files
冻结时间戳 / session id / model / provider
platform hint

5.2 `prompt_builder.py` 提供的核心能力

5.2.1 上下文文件发现与注入

build_context_files_prompt() 负责扫描并注入项目上下文文件：

.hermes.md / HERMES.md
AGENTS.md
CLAUDE.md
.cursorrules / .cursor/rules/*.mdc
SOUL.md

关键特性：

有 prompt injection 扫描：_scan_context_content()
有截断：_truncate_content()
有优先级：只取一种 project context 类型

关键位置：

agent/prompt_builder.py:55 _scan_context_content()
agent/prompt_builder.py:807 load_soul_md()
agent/prompt_builder.py:920 build_context_files_prompt()

5.2.2 Skills 注入

build_skills_system_prompt() 会构建“可用技能索引”，供模型在系统提示词里看到。

它有两层缓存：

进程内 LRU cache
磁盘 snapshot：.skills_prompt_snapshot.json

注意：

这是“skills 索引构建缓存”，不是模型 provider 的 prefix cache。
它的目标是减少每次扫描 skills/ 目录的成本。

关键位置：

agent/prompt_builder.py:505 build_skills_system_prompt()

5.3 会话级 system prompt 稳定策略

Hermes 为了维持 prefix cache 命中，不会每轮都重建 system prompt。

策略是：

首轮调用时构建并缓存到 self._cached_system_prompt
持续会话时，如果有 SessionDB.system_prompt，优先复用库里的快照
只有在 context compression 后才调用 _invalidate_system_prompt() 并重建

相关位置：

run_agent.py:6953 首轮构建/复用 system prompt
run_agent.py:2929 _invalidate_system_prompt()

6. Prompt Caching：prefix cache 如何保持稳定

这一层涉及三种“缓存”，需要区分：

6.1 会话级 system prompt 缓存

由 AIAgent._cached_system_prompt 管理。

作用：

保证同一 session 内 system prompt 文本稳定
避免 memory/context/skills 每轮重建后发生微小漂移

6.2 Skills prompt 构建缓存

由 agent/prompt_builder.py 内部管理。

作用：

加速 skills 索引生成
不直接参与 provider prefix cache

6.3 Provider 级 prompt cache

真正的 provider prefix cache 逻辑在：

agent/prompt_caching.py

入口函数：

apply_anthropic_cache_control()：agent/prompt_caching.py:41

策略：

最多放 4 个 cache_control 断点
system prompt 1 个
最近 3 条非 system message 3 个

在 run_conversation() 里，API 消息构造完成后，如果当前 session 启用了 prompt caching，则调用：

run_agent.py:7239-7244

这就是当前仓库里“保持 prefix cache”的实际执行点。

7. `run_conversation()` 的同步 agent loop

AIAgent.run_conversation() 是核心同步循环，整体是：

准备消息
  ↓
拼 API payload
  ↓
调用模型
  ↓
有 tool_calls ?
  ├─ 否 -> 最终响应 -> 持久化 -> 返回
  └─ 是 -> 执行 tools -> tool result 写回 messages -> 下一轮

7.1 进入循环前的准备

主要步骤：

恢复 primary runtime（如果上轮触发了 fallback）。
清洗用户输入中的 surrogate 字符。
拷贝 conversation_history 到本轮 messages。
去掉旧 budget warning。
从历史中恢复 todo store。
追加本轮 user message。
复用或构建 _cached_system_prompt。
进行 preflight compression。
从 memory provider / plugin hook 获取额外上下文。

相关位置：

run_agent.py:6833-6938
run_agent.py:6953-6991
run_agent.py:6994-7050
run_agent.py:7064-7085

7.2 单轮 API 调用前，如何拼 payload

每次循环会把内部 messages 转成 api_messages：

对当前 user message 注入 external memory / plugin context
assistant message 里的 reasoning 映射成 provider 能理解的字段
移除内部字段：reasoning、finish_reason、_thinking_prefill
对 strict provider 清理 Codex 专用 tool fields
前置 system prompt
注入 prefill_messages
应用 Anthropic/OpenRouter prompt cache 标记
做 tool_call/tool_result 完整性修复
调用 _build_api_kwargs() 产出 provider-specific kwargs

关键位置：

run_agent.py:7169-7250
run_agent.py:5256 _build_api_kwargs()

7.3 `_build_api_kwargs()` 如何分发到不同 API 形态

_build_api_kwargs() 按 self.api_mode 分三路：

anthropic_messages
- 走 Anthropic adapter
codex_responses
- system prompt 变成 instructions
- history 变成 Responses input
- tools 转成 Responses schema
- 带 parallel_tool_calls=True
chat_completions
- 标准 OpenAI-compatible messages + tools
- 对 GPT-5 / Codex 会把第一条 system role 改写成 developer

关键位置：

run_agent.py:5256-5330
run_agent.py:5367-5379

7.4 模型返回后，先做什么

模型返回后会进入一系列规范化/防御逻辑：

处理 incomplete / empty / reasoning-only 情况
修正 hallucinated tool name
校验 tool args JSON
去重 tool calls
限制同轮 delegate_task 数量

然后才决定进入：

tool 分支
final response 分支

相关位置：

run_agent.py:8590-8775
run_agent.py:8845-8977

7.5 `chat()` 只是外层薄封装

AIAgent.chat() 只是：

调 run_conversation()
取出 result["final_response"]

位置：

run_agent.py:9170

8. Tool Dispatch：schema 收集、availability gating、function call 分发

8.1 tool schema 收集

model_tools.py 在 import 时就会执行 _discover_tools()：

import tools/*.py
每个 tool 模块通过 registry.register(...) 自注册
之后构建 TOOL_TO_TOOLSET_MAP 和 TOOLSET_REQUIREMENTS

位置：

model_tools.py:132 _discover_tools()
model_tools.py:170 调用 _discover_tools()

8.2 `get_tool_definitions()` 如何做 availability gating

get_tool_definitions(...) 负责给模型暴露“当前会话真实可见的 tools”。

步骤：

根据 enabled_toolsets / disabled_toolsets 解析 tool 名集合
调 registry.get_definitions(...)
用每个 tool 的 check_fn 做可用性过滤
对 execute_code 动态重建 schema，只列出当前 sandbox 中真实可用的 tools
对 browser_navigate 清理失效的 cross-tool 描述
把最终可用 tool 名写入 _last_resolved_tool_names

位置：

model_tools.py:234 get_tool_definitions()

8.3 模型发出 tool_calls 后的调用链

在 run_conversation() 中，如果 assistant_message.tool_calls 非空：

先校验 tool name 和 JSON 参数
生成 assistant message 并附加到 messages
调 _execute_tool_calls(...)

位置：

run_agent.py:8590-8775
run_agent.py:5948 _execute_tool_calls()

8.4 并行与串行执行策略

_execute_tool_calls() 会根据 _should_parallelize_tool_batch() 决定走：

_execute_tool_calls_sequential()
_execute_tool_calls_concurrent()

并行条件大致是：

多个 tool call
不包含 clarify
只包含并行安全工具
或 path-scoped tool 且目标路径不冲突

位置：

run_agent.py:262 _should_parallelize_tool_batch()
run_agent.py:5948 _execute_tool_calls()
run_agent.py:6045 _execute_tool_calls_concurrent()
run_agent.py:6261 _execute_tool_calls_sequential()

8.5 agent-level tool 与 registry-level tool 的分界

AIAgent._invoke_tool() 先拦截 agent 级工具：

todo
session_search
memory
clarify
delegate_task
memory provider tools

其他工具才走：

handle_function_call(...) in model_tools.py

位置：

run_agent.py:5971 _invoke_tool()
model_tools.py:459 handle_function_call()

8.6 `handle_function_call()` 做什么

handle_function_call() 是 registry 的同步分发入口：

coerce_tool_args() 按 JSON Schema 纠正参数类型
对 read/search loop 做计数重置
执行 plugin pre_tool_call hook
registry.dispatch(...)
执行 plugin post_tool_call hook
返回 JSON string

注意：

todo / memory / session_search / delegate_task 在这里会返回“必须由 agent loop 处理”的 stub，因为它们依赖 agent-level state。

位置：

model_tools.py:372 coerce_tool_args()
model_tools.py:459 handle_function_call()

9. Context Compression：长会话压缩如何接入主流程

核心类：

agent/context_compressor.py:53 ContextCompressor

9.1 初始化

AIAgent.__init__() 会创建 self.context_compressor，传入：

model
provider
base_url
api_key
context_length override
threshold / protect_last_n / summary_model 等 config

位置：

run_agent.py:1186

9.2 压缩触发点

当前实现里至少有三类触发：

Preflight compression
- 本轮 API 还没发出，粗估请求已经超过阈值
正常循环中的 post-tool compression
- 本轮 tool 执行后，根据真实 token 使用量判断
上下文错误恢复
- provider 返回 payload too large / context length exceeded 等错误时，进入压缩重试路径

本题主链里最核心的是前两种。

位置：

run_agent.py:6994 preflight compression
run_agent.py:8827 post-tool compression

9.3 `_compress_context()` 的动作

_compress_context() 不是只改 messages，它还会处理 session 边界。

主流程：

flush_memories() 先给模型一次机会把值得保留的信息写入 memory
调用 ContextCompressor.compress(messages, current_tokens=...)
追加 todo snapshot
_invalidate_system_prompt()
重建 system prompt
若启用 SessionDB：
- end_session(old_session_id, "compression")
- 创建新 session_id
- create_session(parent_session_id=old_session_id)
- 继承并自动编号 title
- update_system_prompt(new_session_id, new_system_prompt)
更新压缩后的 token 估算
重置 file read dedup cache

位置：

run_agent.py:5853 _compress_context()

9.4 `ContextCompressor.compress()` 的内部算法

真实算法不是“简单保留头尾 + 做摘要”，而是：

先裁剪老旧 tool result
保护头部消息
按 token budget 保护尾部消息
对中间消息做结构化摘要
修复 tool_call / tool_result 配对完整性

摘要模板包含：

Goal
Constraints & Preferences
Progress
Key Decisions
Relevant Files
Next Steps
Critical Context

位置：

agent/context_compressor.py:155 _prune_old_tool_results()
agent/context_compressor.py:253 _generate_summary()
agent/context_compressor.py:565 compress()

10. Session Storage：SQLite + FTS5 与 gateway session index 的分工

这里有两套相关但不重复的状态层。

10.1 `SessionDB`：结构化、可检索的长期存储

实现位于：

hermes_state.py

核心设计：

SQLite
WAL mode
sessions 表 + messages 表
messages_fts FTS5 虚表
支持 parent_session_id 链接 compression continuation

关键位置：

hermes_state.py:41 sessions 表
hermes_state.py:71 messages 表
hermes_state.py:93 FTS5 定义
hermes_state.py:115 SessionDB

主要接口：

create_session()
end_session()
update_system_prompt()
append_message()
get_messages_as_conversation()
search_messages()

10.2 `AIAgent` 如何写入 `SessionDB`

每次 turn 结束时调用：

_persist_session()

它会做两件事：

_save_session_log() 写 JSON session snapshot
_flush_messages_to_session_db() 把未刷新的消息增量写入 SQLite

_flush_messages_to_session_db() 用 _last_flushed_db_idx 避免重复写入。

位置：

run_agent.py:1869 _persist_session()
run_agent.py:1882 _flush_messages_to_session_db()

10.3 `SessionStore`：gateway 的会话路由与 transcript 兼容层

实现位于：

gateway/session.py

它负责：

根据 SessionSource 生成 session_key
管理 session reset policy
维护 SessionEntry
管理 transcript（SQLite + JSONL）
生成 gateway 用的 session context prompt

关键位置：

gateway/session.py:202 build_session_context_prompt()
gateway/session.py:692 get_or_create_session()
gateway/session.py:942 append_to_transcript()
gateway/session.py:970 rewrite_transcript()
gateway/session.py:1002 load_transcript()

10.4 `SessionDB` 与 `SessionStore` 的关系

可以把两者理解成：

SessionDB
- 面向 agent / search / 压缩 continuation
- 是结构化 SQLite 正式存储
SessionStore
- 面向 gateway 平台 session 生命周期
- 维护 session_key 与当前 session_id 的映射
- 同时保留 JSONL transcript 兼容层

也就是说：

AIAgent 直接写 SessionDB
gateway 通过 SessionStore 管理“当前聊天应该落在哪个 session_id 上”

11. Response Delivery：CLI、gateway、cron 三条出口

11.1 CLI display

CLI 的交付不是简单 print(final_response)，而是一套 callback + streaming UI：

_init_agent() 给 AIAgent 注入：
- tool_progress_callback
- tool_start_callback
- tool_complete_callback
- stream_delta_callback
- thinking_callback
- reasoning_callback
HermesCLI.chat() 在线程中执行 agent.run_conversation(...)
运行过程中：
- streaming token 通过 _stream_delta(...) 渲染
- reasoning 可单独显示
- tool progress 可实时显示
结束后：
- 若没走 streaming，则用 Rich Panel(...) 包裹最终响应
- 可选 TTS 播放

相关位置：

cli.py:2340-2378 注入 callbacks
cli.py:6481 调用 run_conversation()
cli.py:6621-6644 reasoning display
cli.py:6645-6678 最终 response panel

11.2 Gateway delivery

Gateway 的交付链分两段：

第一段：`_run_agent()` 负责生成 agent 结果

它会：

构造/复用 AIAgent
绑定 tool_progress_callback / step_callback / stream_delta_callback / status_callback
调 run_conversation()
处理 transcript 持久化
更新 SessionStore.last_prompt_tokens
返回最终 response

关键位置：

gateway/run.py:6647-6683
gateway/run.py:6856
gateway/run.py:2996-3078

第二段：平台 adapter 负责真正发送

Gateway 外层收到 _run_agent() 的返回值后，会：

在普通模式下调用 adapter.send(...)
在 streaming 模式下由 stream consumer 边生成边发送
如响应里包含 MEDIA: 或本地文件路径，调用 _deliver_media_from_response(...)
如需要自动语音回复，走 _send_voice_reply(...)

关键位置：

gateway/run.py:7273 queued message 场景下的 adapter.send(...)
gateway/run.py:3080-3098 already streamed 场景下的后处理
gateway/run.py:4287 _send_voice_reply()
gateway/run.py:4345 _deliver_media_from_response()

11.3 Cron delivery

cron 的结果交付分三步：

run_job() 返回 (success, output_doc, final_response, error)
save_job_output() 先把 markdown 文档落盘
如果需要对外发送，_deliver_result(job, content, adapters, loop) 再做平台投递

_deliver_result() 细节：

先 _resolve_delivery_target(job)
优先使用 gateway live adapter（支持 E2EE 等）
失败时 fallback 到 standalone _send_to_platform(...)
先抽取 MEDIA: 标签，附件作为原生文件发送

位置：

cron/scheduler.py:199 _deliver_result()
cron/scheduler.py:856-873 tick 内触发保存与投递

12. 一条完整调用链示例

下面用“gateway 收到一条普通文本消息”举例：

sequenceDiagram
    participant User
    participant Adapter
    participant Gateway as gateway/run.py
    participant Store as gateway/session.py
    participant Runtime as runtime_provider.py
    participant Agent as run_agent.py:AIAgent
    participant Prompt as agent/prompt_builder.py
    participant Tools as model_tools.py
    participant DB as hermes_state.py

    User->>Adapter: 发送消息
    Adapter->>Gateway: MessageEvent
    Gateway->>Store: get_or_create_session(source)
    Gateway->>Store: load_transcript(session_id)
    Gateway->>Runtime: resolve_runtime_provider()
    Gateway->>Agent: AIAgent(...)
    Agent->>Tools: get_tool_definitions(...)
    Agent->>DB: create_session(...) / update_system_prompt(...)
    Gateway->>Agent: run_conversation(message, history)
    Agent->>Prompt: _build_system_prompt()
    Prompt-->>Agent: system prompt
    Agent->>Agent: 同步 loop，调用模型
    alt 模型返回 tool_calls
        Agent->>Tools: handle_function_call(...) / agent-level tool
        Tools-->>Agent: tool result JSON
        Agent->>DB: append_message(...)
        Agent->>Agent: 下一轮 API 调用
    else 模型返回最终文本
        Agent->>DB: flush messages / token counts
    end
    Agent-->>Gateway: {final_response, messages, tokens...}
    Gateway->>Store: append_to_transcript(...) / update_session(...)
    Gateway->>Adapter: send(response)
    Adapter-->>User: 最终消息

13. 本系统里最重要的几个“稳定性约束”

13.1 system prompt 必须尽量稳定

这直接关系到：

Anthropic/OpenRouter prefix cache
多轮会话成本
gateway 复用 cached agent 的收益

因此：

gateway 的动态上下文放到 ephemeral_system_prompt
plugin context 放到 user message
只有 compression 后才主动重建 system prompt

13.2 tool_call / tool_result 配对必须完整

Hermes 在两个地方都做了防御：

API 调用前 _sanitize_api_messages()
compression 后 ContextCompressor._sanitize_tool_pairs()

否则 Anthropic / OpenAI / strict provider 都可能拒绝请求。

13.3 session persistence 不是“最后一次统一保存”，而是多层增量保存

当前实现同时维护：

JSON session snapshot
SQLite messages/session rows
gateway transcript JSONL 兼容层

目标是：即使中途中断、压缩、provider 异常，也尽量不丢会话。

14. 一句话总结

Hermes 的核心运行系统可以概括为：

入口层先通过 runtime_provider + auth 解析可执行 runtime，再由 AIAgent 初始化稳定的 tool surface、system prompt 和 context compressor；随后 run_conversation() 以同步 loop 反复执行“模型调用 -> tool dispatch -> tool result 回填”，并在整个过程中用 SessionDB/SessionStore 持久化状态，最后由 CLI、gateway 或 cron 各自的 display/delivery 层把结果送到用户。

Hermes Agent 06. 技能、记忆与上下文文件

Henry Lin — Mon, 13 Apr 2026 01:30:46 +0000

06. 技能、记忆与上下文文件

如果说工具决定 Hermes “能不能做”，那么技能、记忆和上下文文件决定它“会不会持续做对”。

本章目标

理解 skill、memory、context file 的边界
学会查找、安装和使用 skill
学会用 AGENTS.md 和 SOUL.md 提升稳定性
理解 prompt caching 与上下文注入的约束

适用读者

想长期高效使用 Hermes 的所有用户
想把重复流程沉淀下来的进阶用户
开发者

前置条件

已理解 CLI 和工具系统

核心概念

Skill 是“怎么做”

Skill 是按需加载的流程知识。它更像一份可执行的说明书，而不是一条记忆条目。

适合做成 skill 的内容：

固定工作流
多步骤操作规范
领域知识流程
常见排错套路

Memory 是“是什么”

Memory 适合保存事实信息，例如：

你的偏好
环境事实
常用项目位置
团队约定

经验法则很简单：

可复用流程放 skill
长期事实放 memory

Context File 是“默认注入的背景”

Hermes 会自动读取某些上下文文件，例如：

AGENTS.md
SOUL.md
.cursor/*

这些文件会直接影响每次会话的系统上下文，所以质量和长度都很重要。

操作步骤

查看和使用 skill

最常用的方式：

/skills
/skills search docker
/skills browse

技能安装后通常会直接变成 slash command，例如：

/plan 设计一个 REST API
/ascii-art Make a banner

CLI 外也可以：

hermes skills list
hermes skills install official/research/arxiv

自己创建 skill

用户自定义 skill 放在：

~/.hermes/skills/

典型结构：

~/.hermes/skills/my-category/my-skill/
└── SKILL.md

如果 skill 还需要参考文件、模板或脚本，也可以加 references/、templates/、scripts/。

使用 memory

Memory 用于跨会话保存信息。你可以直接要求 Hermes 记住某个事实，例如：

记住：我们的 CI 用 GitHub Actions，主工作流文件是 deploy.yml。

写好 `AGENTS.md`

对于代码项目，AGENTS.md 非常重要。它适合写：

架构约定
测试方式
代码规范
禁止事项
特定目录行为

它不是 README 的替代，而是“给 agent 的项目说明书”。

区分 `SOUL.md` 和 `AGENTS.md`

SOUL.md：定义 Hermes 的长期人格和表达倾向
AGENTS.md：定义当前项目或工作区的约束

一个偏“说话风格”，一个偏“做事规则”。

常见场景

场景 1：重复做同一种工作流

如果某个任务总要 5 步以上，而且你会重复做，优先考虑做成 skill，而不是每次重写 prompt。

场景 2：希望 Hermes 一直遵守项目约定

优先写 AGENTS.md，而不是每次聊天都重新说一遍。

场景 3：希望 Hermes 长期记住偏好

例如“回答尽量简洁”“默认用 pytest”“某仓库路径固定在某位置”，这些更适合进 memory。

常见问题与排错

为什么安装了 skill，当前会话却没看到

有些 skill 需要新会话才能稳定进入 prompt。必要时重开会话或 /reset。

为什么不建议把 `AGENTS.md` 写得特别长

因为它会被反复注入上下文，太长会拖慢会话、增加 token 消耗，并降低重点信息密度。

为什么 memory 改了但当前会话感觉没变

memory 是跨会话生效的长期上下文。很多情况下写入后要到下一个会话，系统提示里才会以稳定方式体现出来。

什么是 prompt cache 不该被打破

Hermes 很依赖稳定的系统提示前缀来命中缓存。频繁改系统提示、技能载入方式、上下文文件或模型，会降低缓存命中率，提高成本和延迟。

本章总结

工具解决“做事”，skill、memory 和 context file 解决“长期稳定地做对”。高质量的长期使用，离不开这三者的正确分工。

下一步建议

继续看 MCP 与扩展能力，学习如何把 Hermes 接到外部系统。

Hermes Agent 05. 工具与工具集

Henry Lin — Mon, 13 Apr 2026 01:29:46 +0000

05. 工具与工具集

这一章解释 Hermes 为什么能“做事”，以及你该如何控制它能做哪些事。

本章目标

理解 tool 与 toolset 的关系
了解 Hermes 内置工具的大类能力
学会按会话或平台启停工具
学会区分 terminal、file、web、browser、execute_code、delegation 的适用场景

适用读者

所有认真使用 Hermes 的用户
需要控制安全边界和执行能力的部署者
开发者

前置条件

已理解 CLI 与配置系统

核心概念

Tool 是能力单位，Toolset 是开关单位

Hermes 内置工具注册在中央 registry 中，每个工具属于一个 toolset。你平时更常操作的是 toolset，而不是单个 tool。

常见 toolset 包括：

file
terminal
web
browser
skills
memory
code_execution
delegation
cronjob

工具是否可用，取决于三个层面

该工具或 toolset 是否启用
所需环境变量或依赖是否存在
当前平台是否允许该能力

这意味着“文档里有这个工具”不等于“你当前会话里一定能用”。

内置工具的常见边界

file：适合读写搜索文本文件
terminal：适合执行 shell 命令和后台进程
web：适合搜索网页和抽取正文
browser：适合真正的页面交互、表单、按钮、复杂界面
execute_code：适合 3 步以上的机械化串联操作
delegate_task：适合上下文隔离、并行研究、独立子任务

操作步骤

查看当前 toolset

CLI 内：

/toolsets
/tools list

CLI 外：

hermes tools

按会话启用指定 toolset

hermes chat --toolsets web,file,terminal
hermes chat --toolsets debugging
hermes chat --toolsets all

其中 debugging 这类 composite toolset 会展开成多个核心 toolset。

在当前会话中禁用某个工具

/tools disable browser
/tools enable browser

这对“想限制行为”或“怀疑某个工具在误导模型”时非常有用。

学会按任务选对能力层级

优先顺序通常是：

单次网页信息获取：web
复杂网页交互：browser
代码仓库文件理解：file
实际跑命令、构建、测试：terminal
批量顺序处理：execute_code
并行或新上下文分析：delegate_task

理解凭据型工具

有些工具并不是默认可用，例如：

web_search 需要 EXA_API_KEY、PARALLEL_API_KEY、FIRECRAWL_API_KEY 或 TAVILY_API_KEY
image_generate 需要 FAL_KEY
homeassistant 需要 HASS_TOKEN

如果环境变量没配好，工具会被排除或不可用。

常见场景

场景 1：代码分析

优先用 file 和 terminal。如果只是查文件结构或读取文本，没必要打开 browser。

场景 2：网页调研

优先用 web_search 和 web_extract。只有在需要登录、点按钮、处理复杂页面时再切换到 browser。

场景 3：十几个重复步骤的数据处理

优先用 execute_code，因为它能把工具调用和中间逻辑放进脚本里，避免主会话被中间结果淹没。

场景 4：三条研究线并行推进

优先用 delegate_task，让不同子代理分别处理不同问题，再回收摘要。

常见问题与排错

为什么我明明启用了 toolset，工具还是没出现

最常见原因是该工具的 check_fn 失败了，比如缺 API key、缺依赖、缺平台前提。

`browser` 和 `web` 为什么要分开

因为它们成本和能力完全不同。web 更快、更便宜、更适合信息获取；browser 更重，但能真实操作页面。

什么时候不该乱开全部工具

当你需要安全边界、成本控制或更可预测的行为时。工具越多，模型的选择空间越大，行为也越难控。

本章总结

Hermes 的执行力来自工具，但稳定性来自对工具的约束。真正高效的用法不是“什么都开”，而是为任务提供最小且足够的能力集合。

下一步建议

继续看技能、记忆与上下文文件，理解 Hermes 如何长期保持方法和偏好。

Hermes Agent 04. 配置体系与模型管理

Henry Lin — Mon, 13 Apr 2026 01:29:19 +0000

04. 配置体系与模型管理

Hermes 的强大很大程度上来自“入口统一，但后端可替换”。要把这一点用好，就必须掌握配置体系。

本章目标

理解 config.yaml 与 .env 的角色分工
学会选择 provider 和模型
理解 profile 的隔离价值
学会管理常见配置项

适用读者

已经完成首次运行的用户
想在不同场景下切换模型或部署多实例的用户
需要排查配置问题的开发者

前置条件

已完成安装并能进入 CLI

核心概念

`config.yaml` 与 `.env` 的分工

你可以简单记成：

config.yaml：结构化配置，控制行为和默认值
.env：密钥、令牌、URL、环境变量

常见位置：

~/.hermes/config.yaml
~/.hermes/.env

模型选择不是单一命令，而是一整套解析流程

Hermes 会综合这些信息来决定最终的模型与 provider：

当前会话显式指定
hermes model 保存的默认值
config.yaml
.env 中的 provider 相关变量

这也是为什么“明明我填了 API key，但还是走了别的 provider”这类问题会出现。

Profile 是隔离实例，不是主题标签

Profile 的作用是把配置、会话、技能、记忆、gateway 状态等全部隔离开。它适合：

工作 / 私人分离
不同 bot 实例分离
测试环境 / 生产环境分离

操作步骤

查看和设置当前模型

最直接的方式：

hermes model

会话中也可以：

/model
/model claude-sonnet-4
/model openrouter:anthropic/claude-sonnet-4
/model custom:qwen-2.5

管理环境变量

最基础的是 provider 凭据，例如：

OPENROUTER_API_KEY=...
OPENAI_API_KEY=...
OPENAI_BASE_URL=...
ANTHROPIC_API_KEY=...
GLM_API_KEY=...

除模型相关之外，还有 web、browser、Gateway、MCP、memory provider 等能力依赖的变量。

管理 config.yaml

常用方式：

hermes config
hermes config get skills.config
hermes config set some.key value

你不需要一开始手改所有配置。很多场景下，用 hermes setup、hermes model、hermes tools、hermes skills 这类命令比直接编辑 YAML 更稳。

使用 profile

先看当前有哪些 profile：

hermes profile list

创建和切换：

hermes profile create work --clone
hermes profile use work
hermes profile show work

如果想回到默认实例：

hermes profile use default

为不同场景建立不同默认值

一个很实用的模式是：

default：个人本地使用
work：工作环境，配置公司 MCP、企业平台和相关技能
bot：部署在服务器上的长期运行实例

这样能减少“在一个配置里堆所有用途”的混乱。

常见场景

场景 1：接多个 provider，按任务切换

你可以默认使用一个通用 provider，然后在特定会话用 /model 临时切换。Hermes 适合这种“一个工作流，多种模型”的使用方式。

场景 2：连接自建 OpenAI-compatible endpoint

常见做法是配置：

OPENAI_API_KEY=...
OPENAI_BASE_URL=http://your-endpoint/v1

然后在会话中选择：

/model custom:your-model-name

场景 3：把工作和私人使用完全隔离

最推荐用 profile，而不是频繁替换同一个 config.yaml。

常见问题与排错

为什么切换模型后表现变化很大

除了模型本身变化，还可能是：

provider 不同
base URL 不同
上下文缓存被打断
会话中的工具使用策略发生变化

为什么我改了配置却没有马上感知

有些改动影响新会话，有些改动影响当前会话，有些需要重新加载相关系统。比如 MCP 配置常常需要 /reload-mcp。

什么时候要小心 prompt cache

如果你频繁改系统提示、模型、上下文文件或技能加载方式，缓存命中会变差，成本和延迟都可能上升。不要把中途乱改配置当成常规工作流。

本章总结

Hermes 的配置核心可以归纳为四件事：provider、model、tooling、profile。理解它们之后，你就能把 Hermes 从“能用”推进到“适合自己的使用方式”。

下一步建议

继续看工具与工具集，理解 Hermes 的实际执行能力边界。

Hermes Agent 03. 首次运行与 CLI 基础

Henry Lin — Mon, 13 Apr 2026 01:28:47 +0000

03. 首次运行与 CLI 基础

Hermes 的 CLI 是最重要的主入口。很多高级能力最终都要回到 CLI 里调试和观察，所以这一章建议认真读。

本章目标

学会启动 Hermes CLI 并完成一次基本对话
掌握常用 slash commands
学会恢复会话、切换模型、观察工具行为
理解 CLI 与消息平台交互面的差异

适用读者

刚完成安装的新用户
希望把 Hermes 当作主工作台的重度用户

前置条件

已完成安装
至少有一个可用模型

核心概念

CLI 是最完整的交互面

CLI 默认拥有最完整的观察能力和调试体验，包括：

slash command 自动补全
工具执行进度显示
会话恢复
中断与重定向
本地图片粘贴
背景会话

有两类命令需要区分

shell 命令：例如 hermes status
会话内 slash commands：例如 /model、/usage、/compress

你可以把前者理解为“启动或管理 Hermes”，把后者理解为“正在和 Hermes 对话时控制当前会话”。

操作步骤

启动 CLI

最简单的方式就是：

hermes

进入后先尝试几条基础命令：

/help
/model
/usage
/toolsets

发出第一条有效请求

不要一开始就发“你好”，更推荐给 Hermes 一个能触发真实能力的任务，例如：

阅读当前目录结构并总结这个仓库的核心组成。

这样你能更快看到它是否会搜索文件、是否会调用工具、输出风格是否符合预期。

学会多行输入与中断

CLI 支持多行输入，适合粘贴代码、traceback 和长提示。任务跑偏时可以按 Ctrl+C 中断，再直接发新消息重定向。

掌握最常用的 slash commands

下面这些是高频中的高频：

/help：查看帮助
/model [model]：查看或切换模型
/new / /reset：新会话
/title [name]：给当前会话命名
/resume [name]：恢复之前命名过的会话
/usage：看 token、成本、上下文使用情况
/compress：手动压缩上下文
/verbose：切换工具输出显示级别
/background <prompt>：在后台启动独立任务
/tools [list|enable|disable]：查看或修改当前会话工具
/skills：搜索、查看、安装技能

学会会话恢复

CLI 外部可以用：

hermes -c
hermes -r "会话标题"

CLI 内部可以用：

/resume 会话标题

如果你经常做连续任务，会话命名是非常值得养成的习惯。

观察工具行为

执行复杂任务时，建议切换 /verbose，观察 Hermes 是否：

真正调用了你期望的工具
在错误的地方浪费步骤
因为权限、配置或工具缺失而被卡住

这比盲目重试更有效。

常见场景

场景 1：恢复昨天没做完的任务

hermes -r "auth-refactor"

或者进入 CLI 后：

/resume auth-refactor

场景 2：切换到更合适的模型

/model
/model openrouter:anthropic/claude-sonnet-4
/model zai:glm-5

复杂推理任务和简单格式化任务，往往不该强行用同一个模型。

场景 3：让长任务后台运行

/background 读取当前仓库，整理一份贡献指南并给出改进建议。

这样当前会话仍然可用，后台任务完成后会返回结果。

常见问题与排错

为什么我在 CLI 里看不到工具细节

先执行 /verbose，Hermes 会在不同显示级别之间切换。

为什么一个任务越聊越慢

通常是上下文变长了。先看 /usage，必要时执行 /compress。

为什么同样的命令在消息平台里没有

因为有些 slash commands 是 CLI-only，例如 /tools、/toolsets、/browser、/config、/skills 等。CLI 和 Gateway 共享一部分命令，但并不完全一致。

本章总结

CLI 是你理解 Hermes 行为、观察工具调用、调整模型与会话状态的最佳入口。只要你把 CLI 用顺，后面使用 Gateway、MCP、Cron 和开发扩展时都会轻松很多。

下一步建议

继续看配置体系与模型管理，理解 Hermes 的配置到底怎么组织。

Hermes Agent 02. 安装与初始化

Henry Lin — Sun, 12 Apr 2026 01:22:42 +0000

02. 安装与初始化

这一章解决最实际的问题：怎样把 Hermes Agent 安装好、初始化好，并确认它真的可用。

本章目标

完成 Hermes Agent 安装
理解 ~/.hermes 目录和关键配置文件
完成第一次 hermes setup
知道开发仓库与普通安装的差别

适用读者

新用户
准备在本机或服务器上部署 Hermes 的用户
准备修改仓库源码的开发者

前置条件

Linux、macOS 或 WSL2 环境
网络可访问安装依赖
至少准备一个模型 provider 的 API key 或 OAuth 登录方式

核心概念

Hermes 的默认家目录

默认情况下，Hermes 的配置和状态文件都在 ~/.hermes/ 下。最常见的文件是：

~/.hermes/config.yaml
~/.hermes/.env
~/.hermes/SOUL.md
~/.hermes/skills/
~/.hermes/sessions/

如果你使用 profile 或设置了 HERMES_HOME，这些路径会被隔离到对应目录。

安装和开发是两条路径

普通使用者：用官方安装脚本最快
仓库开发者：在仓库里建虚拟环境并安装 .[all,dev]

操作步骤

方式一：官方安装脚本

这是最简单的路径：

curl -fsSL https://raw.githubusercontent.com/NousResearch/hermes-agent/main/scripts/install.sh | bash

安装完成后，重新加载 shell：

source ~/.bashrc
# 或者
source ~/.zshrc

然后直接启动：

hermes

方式二：开发者从仓库安装

如果你要修改这个仓库本身，建议使用下面的流程：

git clone https://github.com/NousResearch/hermes-agent.git
cd hermes-agent
uv venv venv --python 3.11
source venv/bin/activate
uv pip install -e ".[all,dev]"

如果你就在本仓库里工作，请记住 AGENTS.md 给出的开发约束：运行 Python 相关命令前应先执行：

source venv/bin/activate

第一次执行 setup

安装后建议立刻运行：

hermes setup

你也可以分模块执行：

hermes setup model
hermes setup terminal
hermes setup gateway
hermes setup tools
hermes setup agent

hermes setup 主要帮助你完成这些事情：

选择 provider 和默认模型
配置 terminal backend
配置 gateway 平台
配置工具
初始化 agent 行为设置

准备 `.env`

Hermes 会从 ~/.hermes/.env 读取大量环境变量。最基础的是模型 provider：

OPENROUTER_API_KEY=your_key_here

其他能力如 web、browser、TTS、Gateway、MCP 往往也需要额外变量。后面在配置章节会集中展开。

验证安装是否成功

推荐依次跑下面几个命令：

hermes version
hermes doctor
hermes status
hermes

如果你能进入 CLI，并正常发出一条消息，说明基础安装已经通了。

常见场景

场景 1：在自己电脑上做本地助手

最常见组合是：

官方安装脚本
默认 ~/.hermes
只先配置一个 provider
暂时不启用 Gateway

场景 2：在服务器上长期运行

更常见的组合是：

安装后先配好 hermes gateway
设置 allowlist
配置 home channel
再逐步加 cron 和 MCP

场景 3：准备参与仓库开发

你应该优先用仓库安装方式，并且在运行测试、脚本和 Python 命令前先激活 venv。

常见问题与排错

Windows 能直接用吗

官方不支持原生 Windows。推荐使用 WSL2。

`hermes` 命令找不到

通常是安装后没有重新加载 shell，或者 PATH 没更新。先尝试重新打开终端，或者 source ~/.bashrc / source ~/.zshrc。

`hermes doctor` 报 provider 未配置

这通常不是安装失败，而是还没有可用的模型凭据。先配置 ~/.hermes/.env，或者运行 hermes model / hermes auth 完成认证。

我什么时候应该开始用 profile

如果你已经知道自己会同时维护“工作助手”“私人助手”“机器人实例”这类多个隔离环境，那么在一开始就用 profile 会比后面迁移更省事。

本章总结

安装 Hermes 并不难，真正重要的是把配置目录、初始化流程和 provider 凭据这三件事弄清楚。只要这一步稳住，后面 CLI、Gateway、工具和自动化就容易很多。

下一步建议

继续看首次运行与 CLI 基础，把第一次真实使用跑通。

Hermes Agent 01. 概览与架构

Henry Lin — Sun, 12 Apr 2026 01:21:59 +0000

01. 概览与架构

这一章帮助你先建立整体认知：Hermes Agent 到底是什么、适合做什么、和普通聊天助手有什么区别，以及仓库里最重要的几个系统是如何配合工作的。

本章目标

理解 Hermes Agent 的产品定位
了解 CLI、Gateway、Tools、Skills、Memory、MCP、Cron 的关系
判断 Hermes 是否适合你的使用场景
建立后续阅读所需的整体地图

适用读者

第一次接触 Hermes Agent 的用户
准备长期使用 Hermes 的进阶用户
希望参与仓库开发的贡献者

前置条件

不需要任何安装经验
建议对命令行、API key、配置文件有基本概念

核心概念

Hermes Agent 不是“只能聊天”的助手

Hermes 的核心价值在于“会调用工具、会保留长期上下文、能跨平台持续运行”。它不是单纯把一个模型包进聊天窗口，而是把模型、工具、记忆、技能、会话存储、多平台入口和自动化机制组合成一个可持续工作的 agent。

Hermes 的几块核心系统

CLI：终端里的交互式主入口，适合开发、调试、重度使用
Gateway：把同一个 agent 能力接到 Telegram、Discord、Slack、WhatsApp、Signal 等平台
Tools：模型真正执行动作的能力边界，例如终端执行、文件读写、网页搜索、浏览器控制
Toolsets：对工具做分组和开关控制，决定某个平台或某次会话能用哪些工具
Skills：按需加载的流程知识，适合保存“怎么做”
Memory：跨会话保留的事实性信息，适合保存“是什么”
Context Files：如 AGENTS.md、SOUL.md 等长期注入的上下文文件
MCP：把外部系统通过统一协议暴露成 Hermes 可调用的工具
Cron：让 Hermes 按计划自动运行任务，而不是等你每次手动发消息

一个典型的数据流

当你在 CLI 或消息平台中发出请求时，大致会发生这些步骤：

入口层接收消息
AIAgent 构造系统提示词和上下文
模型决定直接回答，或者调用一个或多个工具
工具结果回到 agent 循环中
生成最终回复并保存到会话存储

这意味着 Hermes 的“智能”不只是模型本身，还来自整个执行回路。

操作步骤

第一步：记住 Hermes 的两个主入口

想在终端里工作，用 hermes
想在聊天平台里工作，用 hermes gateway ...

CLI 更适合：

写代码
看工具输出
调试配置
处理复杂任务

Gateway 更适合：

长期在线
远程使用
定时推送
团队或多设备访问

第二步：理解“能力来自工具”

Hermes 不会凭空浏览网页、修改文件或运行命令。它必须通过工具完成这些操作。你应当从一开始就建立这个判断标准：

只是回答知识问题：可能不需要工具
需要访问项目、网页、终端、外部系统：通常一定会用工具
需要长期复用的流程：通常还会涉及 skill

第三步：理解“长期表现来自技能、记忆和上下文文件”

同一个模型在 Hermes 里表现更稳定，往往不是因为模型突然变强，而是因为：

你给了它合适的 AGENTS.md
你把重复流程做成了 skill
你让它把长期事实写进了 memory
你没有频繁打断 prompt cache

常见场景

场景 1：本地开发助手

你在项目根目录启动 hermes，让它读 AGENTS.md、搜索代码、跑测试、改文件。这个场景最依赖 CLI、terminal、file、toolset 和上下文文件。

场景 2：消息平台上的远程代理

你把 Hermes 部署在一台服务器上，用 Telegram 或 Discord 与它对话，让它代你查信息、执行脚本、推送定时报告。这个场景最依赖 Gateway、allowlist、home channel 和 cron。

场景 3：连接企业内部系统

你希望 Hermes 能访问公司文档、数据库、内部 API 或 GitHub 工作流，但不想给 Hermes core 写一堆新工具。这时通常应该优先考虑 MCP。

常见问题与排错

为什么说 Hermes 不等于 IDE copilot

IDE copilot 更像“对当前编辑环境的增强”；Hermes 更像“拥有多种入口和工具的独立 agent 运行时”。它可以不在 IDE 里，不在你的本机上，甚至不在你当前打开的项目里。

为什么有时候 Hermes 看起来“不会做事”

最常见的原因不是模型不够强，而是：

对应工具没启用
需要的 API key 没配置
toolset 不包含该能力
上下文不够清楚

这些问题在后续章节都会展开。

本章总结

Hermes Agent 的核心不是某一个模型，而是一整套围绕 agent 循环组织起来的运行系统。后面所有章节，本质上都在回答三个问题：Hermes 能做什么、怎么让它做对、以及怎么让它长期稳定地做。

下一步建议

继续看安装与初始化，把 Hermes 真正跑起来。

Hermes Agent 完整使用教程

Henry Lin — Fri, 10 Apr 2026 06:48:14 +0000

Hermes Agent 完整使用教程

1. Hermes 是什么

Hermes Agent 不是单纯的聊天界面，也不是只能在 IDE 里工作的代码补全器。它更像一个统一的 agent 运行时：

在 CLI 里，它可以读写文件、执行命令、管理会话、切换模型、调用工具
在 Gateway 里，它可以通过 Telegram、Discord、Slack、WhatsApp、Signal 等平台长期在线
通过 skill、memory 和 context file，它能把流程知识、事实信息和项目约束长期保留下来
通过 MCP 和 plugins，它能对接外部系统
通过 cron 和背景任务，它能定时、异步、持续地工作

你可以把 Hermes 的核心结构概括成下面几层：

入口层：CLI、Gateway、ACP、batch runner
agent 层：AIAgent 和 prompt / provider / tool 调度
执行层：tools、toolsets、terminal backend、browser backend、MCP
持久层：sessions、memory、skills、profiles

这意味着 Hermes 的效果不仅取决于模型，还取决于工具可用性、上下文质量和你的使用方式。

2. 安装与初始化

普通安装

最快的方式是官方安装脚本：

curl -fsSL https://raw.githubusercontent.com/NousResearch/hermes-agent/main/scripts/install.sh | bash
source ~/.bashrc
# 或 source ~/.zshrc
hermes

官方支持 Linux、macOS 和 WSL2。原生 Windows 不在推荐路径里。

仓库开发安装

如果你打算直接修改仓库：

git clone https://github.com/NousResearch/hermes-agent.git
cd hermes-agent
uv venv venv --python 3.11
source venv/bin/activate
uv pip install -e ".[all,dev]"

对于这个仓库本身，AGENTS.md 明确要求：运行 Python 相关命令前要先 source venv/bin/activate。

初始化配置

首次安装后建议运行：

hermes setup

它会帮你处理：

模型与 provider
terminal backend
gateway 平台
tools
agent 行为

默认配置目录是 ~/.hermes/，你最需要知道的文件通常是：

~/.hermes/config.yaml
~/.hermes/.env
~/.hermes/SOUL.md

.env 里通常先配一个可用 provider，例如：

OPENROUTER_API_KEY=...

智谱 glm 配置： 
base-url 写这个：
https://open.bigmodel.cn/api/coding/paas/v4

安装验证建议用：

hermes version
hermes doctor
hermes status
hermes

3. 第一次进入 CLI

CLI 是最完整的主入口。推荐你第一次就用它，而不是先上消息平台。

启动：

hermes

进入后先跑这几条：

/help
/model
/usage
/toolsets

然后给一个带动作的任务，而不是“你好”，例如：

阅读当前目录结构并总结这个仓库的核心组成。

这样你能马上看到 Hermes 是否会用 file 或 terminal 工具。

你很快会常用到的命令

/new / /reset：新会话
/title：命名会话
/resume：恢复命名会话
/model：切换模型
/usage：看 token、成本、上下文
/compress：压缩上下文
/verbose：观察工具输出
/background <prompt>：后台启动独立任务
/skills：搜索和安装 skill
/tools list：看当前工具

CLI 外也别忘了：

hermes -c
hermes -r "session-title"

复杂任务建议尽早 /title，否则后续恢复会话会很痛苦。

4. 配置、模型与 profile

Hermes 的配置分两类：

config.yaml：结构化行为配置
.env：API key、base URL、平台 token、终端环境变量等

最常见的 provider 相关变量有：

OPENROUTER_API_KEY
OPENAI_API_KEY
OPENAI_BASE_URL
ANTHROPIC_API_KEY
GLM_API_KEY

选择模型最直接的方式是：

hermes model

会话中可以：

/model
/model openrouter:anthropic/claude-sonnet-4
/model zai:glm-5
/model custom:qwen-2.5

profile 为什么重要

如果你有多个用途，强烈建议用 profile 隔离，而不是把所有配置堆在一个默认实例里。

常见命令：

hermes profile list
hermes profile create work --clone
hermes profile use work
hermes profile show work
hermes profile use default

profile 隔离的是整套 Hermes home，包括配置、技能、会话、memory、gateway 状态等。

关于 prompt cache

Hermes 很依赖稳定的系统提示前缀来命中缓存。频繁改模型、系统提示、上下文文件、技能加载方式，会提高成本并降低会话稳定性。不要把“中途乱改配置”当作常态。

5. 工具、toolset 与执行边界

Hermes 的执行力来自工具。每个工具属于一个 toolset，你平时更常控制的是 toolset。

常见 toolset：

file
terminal
web
browser
skills
memory
code_execution
delegation
cronjob

查看当前会话可用能力：

/toolsets
/tools list

从命令行按会话选择：

hermes chat --toolsets web,file,terminal
hermes chat --toolsets debugging
hermes chat --toolsets all

怎样选对工具层级

读写搜索文件：file
真正执行命令：terminal
快速搜索网页或抽取正文：web
点击按钮、填表单、真实页面交互：browser
多步机械流程：execute_code
并行子任务或新上下文分析：delegate_task

高频误区是“什么都开、什么都用”，这会增加风险、成本和不可预测性。更好的方式是为任务提供最小且足够的能力集合。

6. skill、memory 与上下文文件

Hermes 的长期稳定性主要来自三类东西：

skill：如何做
memory：是什么
context file：默认背景和规则

使用 skill

查看和安装：

/skills
/skills search docker
/skills browse

或：

hermes skills list
hermes skills install official/research/arxiv

skill 安装后通常会直接变成 slash command，例如：

/plan 设计一个 REST API
/ascii-art Make a banner

使用 memory

适合保存长期事实，例如：

记住：我们的 CI 用 GitHub Actions，部署工作流是 deploy.yml。

使用 `AGENTS.md` 和 `SOUL.md`

AGENTS.md：项目规则、架构约束、测试要求
SOUL.md：Hermes 的长期风格和行为倾向

如果你总在重复同一批规则，就应该把它们写进 AGENTS.md，而不是继续每次重说。

一个重要限制

不要把 AGENTS.md 写成百科全书。它会进入系统上下文，太长会增加成本并稀释重点。

7. MCP 接入外部系统

MCP 适合把外部系统接成 Hermes 的工具。例如 GitHub、文件系统、内部 API、远程知识库。

当前仓库和代码使用的配置键是：

mcp_servers:
  ...

不是 mcp.servers。

最小 stdio 示例

mcp_servers:
  filesystem:
    command: "npx"
    args: ["-y", "@modelcontextprotocol/server-filesystem", "/home/user/projects"]

最小 HTTP 示例

mcp_servers:
  company_api:
    url: "https://mcp.internal.example.com"
    headers:
      Authorization: "Bearer ***"

改完配置后，记得：

/reload-mcp

先过滤，再暴露

非常推荐一开始就做白名单或黑名单控制：

mcp_servers:
  github:
    command: "npx"
    args: ["-y", "@modelcontextprotocol/server-github"]
    env:
      GITHUB_PERSONAL_ACCESS_TOKEN: "***"
    tools:
      include: [list_issues, create_issue, search_code]
      prompts: false
      resources: false

判断标准可以记为：

内置工具已经够用：优先内置工具
只是流程复杂：优先 skill
需要接外部系统：优先 MCP

8. Gateway 与多平台使用

Hermes 可以通过 gateway 暴露到 Telegram、Discord、Slack、WhatsApp、Signal 等平台。

核心命令：

hermes gateway setup
hermes gateway run
hermes gateway start
hermes gateway stop
hermes gateway status

常见平台变量包括：

TELEGRAM_BOT_TOKEN
DISCORD_BOT_TOKEN
SLACK_BOT_TOKEN
SLACK_APP_TOKEN

授权控制必须认真做

不要轻易开放所有用户。优先配置：

TELEGRAM_ALLOWED_USERS
DISCORD_ALLOWED_USERS
GATEWAY_ALLOWED_USERS

home channel 很重要

进入对应聊天后执行：

/sethome

cron 和后台结果才知道发去哪里。

消息平台常见命令包括：

/new
/status
/model
/personality
/retry
/undo
/sethome
/compress
/title
/resume
/usage
/reload-mcp
/background

而 /tools、/toolsets、/browser 之类通常更偏 CLI。

9. cron、后台任务与自动化

Hermes 的 cron 是“在新 session 中运行一个 agent 任务”，而不是单纯执行系统计划任务。所以最关键的规则是：

你的 prompt 必须完全自包含。

创建 job

/cron add "0 9 * * 1" 生成每周 AI 摘要并发送到 home channel。

或：

hermes cron create "0 9 * * 1" "生成每周 AI 摘要并发送到 home channel。" --name "weekly-ai"

常见管理：

/cron list
/cron run <job_id>
/cron pause <job_id>
/cron resume <job_id>

背景任务

如果只是想把当前任务异步化，而不是按时间重复执行：

/background 总结这个仓库的工具系统并写一份建议。

自动化设计的一个好模式

脚本负责抓数据、做机械步骤、保存状态
agent 负责解释、筛选、总结

很多监控类任务还应该显式约定：

如果没有变化，回复 [SILENT]。

这样不会把通知刷屏。

10. 开发扩展

如果你要修改这个仓库，先记住关键结构：

run_agent.py：AIAgent 核心循环
cli.py：CLI
gateway/run.py：Gateway
model_tools.py：工具发现和 dispatch
tools/registry.py：工具注册
toolsets.py：工具分组

新增 tool 的最小路径

一般要动 3 个地方：

tools/your_tool.py
toolsets.py
model_tools.py

其中 handler 应返回 JSON 字符串，错误也应包装成 JSON。

新增 slash command

一般要动：

hermes_cli/commands.py
cli.py
gateway/run.py（如果 Gateway 也支持）

profile-safe 规则

不要硬编码 ~/.hermes。代码里应使用：

get_hermes_home()

用户可见提示中应使用：

display_hermes_home()

本仓库开发测试

source venv/bin/activate
python -m pytest tests/ -q
python -m pytest tests/test_model_tools.py -q
python -m pytest tests/gateway/ -q
python -m pytest tests/tools/ -q

11. 调试、测试与排障

遇到问题时，先按层次判断：

安装层：hermes 能否启动
provider 层：认证、API key、base URL
tool 层：toolset 是否启用，依赖是否满足
MCP 层：server 是否加载
Gateway 层：token、allowlist、home channel
上下文层：skills、memory、会话状态、prompt cache

通用命令：

hermes doctor
hermes status
hermes version
hermes profile show <name>

CLI 内常用辅助：

/model
/usage
/toolsets
/tools list
/verbose

MCP 问题优先检查：

mcp_servers 键名是否正确
本地依赖命令是否存在
是否执行了 /reload-mcp

消息平台问题优先检查：

token 是否正确
allowlist 是否允许当前用户
是否需要 mention
home channel 是否配置

不要一上来就删整个 ~/.hermes。大多数问题都可以局部定位。

12. 最佳实践

最后给你一套更稳的长期使用方式：

1. 第一条消息尽量把上下文给够

不要只说“帮我修一下”。尽量直接给：

文件路径
错误信息
期望结果
约束条件

2. 复杂任务先命名会话

/title auth-refactor

以后恢复就简单很多。

3. 重要规则写进 `AGENTS.md`

不要依赖聊天记录记住所有项目约定。

4. 长会话定期看 `/usage`

必要时 /compress，不要等上下文快爆了才处理。

5. 选对执行方式

单一动作：直接工具
多步机械流程：execute_code
并行分析：delegate_task
可复用流程：skill

6. 用 profile 管隔离

工作、个人、bot、实验环境，最好分 profile，而不是混在一个实例里。

7. 对不可信代码使用隔离环境

优先 Docker、Daytona 等，而不是默认本地 terminal。

8. 最小暴露原则

toolset、MCP、Gateway 授权都遵循同一个原则：只开必须的，不要因为“以后可能用到”而默认暴露。

Hermes Agent 项目功能与通用使用场景分析

Henry Lin — Wed, 08 Apr 2026 13:06:09 +0000

Hermes Agent 项目功能与通用使用场景分析

1. 项目定位

Hermes Agent 是一个通用型 AI Agent 平台，不是单纯的聊天界面，也不是单一的代码助手。它把大模型推理、工具调用、终端执行、文件操作、网页检索、浏览器自动化、长期记忆、定时任务、多平台消息接入和外部系统扩展整合到一个统一框架里。

从仓库结构和主干代码来看，它的目标不是“回答问题”本身，而是让模型具备持续执行任务的能力，并且能在不同入口、不同环境和不同工具集之间稳定运行。

可以把它理解为三层系统：

Agent 执行层：负责会话循环、模型调用、工具调度、上下文压缩。
平台能力层：负责 CLI、消息网关、状态存储、调度、配置、权限边界。
扩展生态层：负责 skills、MCP、插件、自定义工具和多环境运行。

所以，Hermes Agent 更接近一个“可运行的 Agent Operating Layer”，而不是一个普通的 LLM 应用。

2. 项目主干结构

从代码组织看，几个核心模块职责很清楚：

run_agent.py
核心 AIAgent 实现，负责主对话循环、工具调用、消息历史和最终响应。
model_tools.py
工具发现与分发层。仓库里的工具模块在导入时注册到统一 registry，Agent 再按 toolset 组装可用能力。
tools/registry.py
工具注册中心，负责 schema、handler、可用性检测和 dispatch。
toolsets.py
定义不同场景下的工具组合，比如 CLI、gateway、API server、ACP 等。
cli.py 和 hermes_cli/
提供交互式终端 UI、slash command、模型切换、配置管理、setup、doctor、profile 等。
gateway/
提供 Telegram、Discord、Slack、WhatsApp、Signal、Email、Webhook 等消息平台入口。
hermes_state.py
使用 SQLite + WAL + FTS5 做会话持久化和全文检索。
cron/
提供定时任务能力，支持定时运行 agent 并把结果投递到配置的平台。
tools/mcp_tool.py
提供 MCP 接入，把外部系统能力注册成 Hermes 原生工具。
acp_adapter/
提供编辑器/ACP 集成能力。
batch_runner.py、environments/
提供批处理、轨迹生成和研究/训练相关能力。

从规模上看，这已经是一个完整平台：

tools/ 下有大量独立工具模块。
gateway/platforms/ 下有多种消息平台适配器。
tests/ 覆盖面很大，说明项目不是演示性质，而是实际维护中的系统。

3. 核心功能拆解

3.1 Agent 会话循环

Hermes 的核心是 AIAgent 的 tool-calling loop。基本模式是：

模型接收消息和可用工具定义。
模型决定是否调用工具。
Hermes 执行工具，并把结果写回消息历史。
循环直到模型产出最终答案，或者达到迭代限制。

这个设计的意义在于，Hermes 不是只让模型“建议怎么做”，而是让模型“真的去做”：

读取文件
搜索代码
执行命令
查询网页
调用浏览器
写入结果
再根据结果继续行动

这使它适合处理多步任务，而不是只做一轮问答。

3.2 工具注册与工具集

Hermes 的工具系统是项目的核心之一。

其特点是：

工具模块独立存在，一个文件通常负责一个工具域。
每个工具在导入时自注册到 registry。
Agent 不直接依赖具体工具，而是通过 toolset 组合决定当前会话能用什么。

这带来几个好处：

容易增删工具。
可以按平台控制权限边界。
不同入口可以有不同能力集合。
外部工具可以以统一方式接入。

从已有代码看，工具能力至少覆盖以下几类：

Web 搜索与网页抽取
终端执行与进程管理
文件读写、补丁和搜索
浏览器自动化
图像分析与生成
会话搜索
记忆与 TODO
代码执行沙箱
子代理委派
定时任务
跨平台消息发送
Home Assistant

这意味着 Hermes 不是“只有 shell 的 agent”，而是一个多能力编排器。

3.3 终端与多执行环境

tools/terminal_tool.py 表明 Hermes 支持多种命令执行后端，而不局限于本机 shell。

常见后端包括：

local
docker
ssh
modal
daytona
singularity

这使 Hermes 能适配不同部署模式：

本地开发机
远程服务器
容器化执行环境
云端弹性运行环境

对通用项目来说，这一点非常重要，因为很多 Agent 系统的问题不是“能不能推理”，而是“能不能在正确环境里做事”。Hermes 在这一层是有明确工程设计的。

3.4 CLI 交互系统

Hermes 的 CLI 不是一个简单的 input() 包装，而是一个完整的终端交互系统。

它具备：

多行输入
slash command
自动补全
tool progress 展示
会话切换和恢复
模型切换
配置修改
交互式 setup
doctor 诊断
皮肤和显示层定制

这意味着 Hermes 可以作为“长期使用的本地 Agent 工作台”，而不是一次性的命令入口。

3.5 消息网关

gateway/ 模块是 Hermes 与许多本地 agent 项目区别很大的地方。

它支持把同一个 agent 接到多种消息平台上，包括：

Telegram
Discord
Slack
WhatsApp
Signal
Email
Webhook
Home Assistant 等

这样带来的通用价值是：

Agent 不必绑定在终端里。
用户可以从不同平台继续同一个工作流。
自动任务结果可以直接投递到聊天平台。
Agent 可以成为团队协作接口，而不是个人玩具。

如果把 Hermes 放到服务器上，它更像一个“随处可用的个人/团队执行代理”。

3.6 会话持久化与检索

hermes_state.py 使用 SQLite + WAL + FTS5 管理 session 和 message。

这不是轻量日志，而是明确的长期状态存储设计，提供：

会话持久化
历史消息恢复
全文检索
标题化与检索式恢复
跨会话搜索

这类能力对长期使用的 Agent 非常关键，因为很多真正有价值的任务不是“当场回答”，而是：

记住上次做到哪里
找回过去解决过的问题
在多个 session 中复用历史上下文

Hermes 在这方面明显不是无状态聊天框架。

3.7 记忆与技能系统

Hermes 还有两个比较有辨识度的方向：

memory
skills

memory 的作用偏向长期事实沉淀，比如偏好、约束、常用模式。

skills 的作用偏向流程化知识沉淀，即：

某类任务怎么做
哪些步骤固定可复用
某个工作流如何标准化

从仓库设计看，skills 不是装饰功能，而是平台的一部分：

CLI 和 gateway 里有对应命令
prompt builder 会把相关指令纳入系统 prompt 逻辑
skill 文件体系是长期工作流复用的重要机制

这使 Hermes 更接近“能积累经验的 agent”，而不是每轮都从零开始。

3.8 上下文压缩与长任务支持

agent/context_compressor.py 显示 Hermes 对长对话场景有明确支持：

接近上下文上限时自动压缩
旧工具输出裁剪
结构化摘要
保留头尾关键内容

这很重要，因为真实任务里，很多 Agent 失败不是因为模型不会，而是因为：

对话太长
工具输出太多
历史信息失控

Hermes 在工程上显式处理了这个问题，因此更适合长流程任务。

3.9 子代理与程序化工具调用

Hermes 不只支持“一个 agent 干到底”，还支持：

delegate_task 子代理委派
execute_code 程序化工具调用

子代理的意义是把复杂问题拆开，让多个上下文隔离的 agent 并行处理。

程序化工具调用的意义是让模型先写一个 Python 脚本，再由脚本批量调用工具，通过 RPC 完成多步操作，减少模型回合数。

这种设计很适合：

需要并行调查多个问题
需要对多个对象执行重复步骤
需要压缩 token 成本
需要将步骤型操作转成脚本化流程

这说明 Hermes 在“任务执行效率”上做过比较深的思考。

3.10 定时任务与后台自动化

cron/ 模块使 Hermes 具备调度能力，而不是只能交互式运行。

典型能力包括：

定时运行 prompt
触发数据收集脚本
保存输出
推送结果到聊天平台
和 gateway 联动执行

这让 Hermes 具备“自动运行 agent job”的能力，适合做周期性报告、巡检、整理和提醒类任务。

3.11 MCP 与外部系统扩展

tools/mcp_tool.py 是 Hermes 变成“平台型系统”的关键。

它允许：

通过 stdio 或 HTTP 连接 MCP server
自动发现工具
把外部能力注入 Hermes 工具体系

这意味着 Hermes 不需要自己内置所有能力。它更像一个统一壳层，外部系统可以通过 MCP 接入，例如：

公司内部 API
数据平台
Issue 系统
文档系统
工单系统
DevOps 平台

这让 Hermes 的上限不再是仓库内置功能，而是可连接生态。

3.12 批处理与研究型能力

batch_runner.py 和相关环境目录说明 Hermes 还支持更偏研究/训练的能力：

批量运行 agent
保存 trajectory
统计工具调用情况
支持研究数据集驱动的运行方式

这部分对一般用户未必是核心入口，但对做 agent 评测、数据生成、研究实验的人来说很有价值。

4. 项目的通用能力画像

如果把上面这些功能浓缩成一句话，Hermes 的通用能力画像是：

“一个可长期运行、可跨平台接入、可调用多种工具、可持久化记忆与会话、可调度自动任务、可连接外部系统的通用 Agent 平台。”

更具体一点，它擅长以下五类事情：

把自然语言任务转成多步执行流程。
把多种工具和外部系统统一成一个可调用层。
把 agent 从终端扩展到聊天平台和后台任务。
把历史会话、技能和记忆沉淀成长期资产。
把一次性问答升级成持续运行的工作流。

5. 典型使用场景

以下场景不依赖任何特定行业，基本都适用于通用团队或个人用户。

5.1 个人长期助手

这是最直接的场景。

Hermes 可以作为个人长期助手，承担：

日常查询
文件整理
命令执行
提醒和定时任务
历史会话回顾
在 Telegram/Slack 上继续同一会话

和普通聊天助手相比，它的优势在于：

能执行动作
能记忆历史
能跨端继续
能定时自动运行

5.2 开发与调试助手

这是 Hermes 很强的适用面之一。

它适合：

阅读和搜索代码
修改文件
执行测试
分析日志
运行 shell 命令
编写脚本
使用子代理并行调查问题

尤其在大型代码库中，它比“只会生成代码的助手”更强，因为它能在本地环境真实执行和验证。

5.3 研究与资料整理助手

Hermes 很适合承担资料型任务：

搜索网页
抽取网页内容
汇总多来源信息
生成摘要
保存结果
周期性监控某类主题变化

如果配合 cron 和消息推送，可以形成自动研究简报工作流。

5.4 运维与巡检助手

通过 terminal、process、cron 和 messaging，Hermes 很适合做轻量运维助手：

定时检查服务状态
分析日志
发现异常后推送消息
执行标准化恢复动作
汇总巡检结果

它不一定适合作为高风险生产控制核心，但很适合作为巡检、观察、汇报和半自动处置层。

5.5 团队协作 Agent

由于 Hermes 有 gateway 和跨平台消息能力，它可以作为团队共享入口：

成员通过 Slack/Telegram 直接发起任务
Agent 拉取上下文并执行
结果回传到团队频道
定时任务自动发日报、周报、告警和摘要

这使它适合做团队内部的“智能机器人基础设施”。

5.6 文档与知识管理助手

Hermes 也适合做知识型工作：

搜索历史会话
归纳结论
生成知识文档
维护标准操作流程
把重复问题沉淀成 skills

相比只接数据库检索的知识机器人，它更强的地方在于它还能行动，能在整理知识时顺便验证、修正和生成产物。

5.7 自动化报告系统

这是 Hermes 很自然的落地形式。

例如：

每天整理新闻摘要
每周汇总项目状态
每小时巡检服务健康度
每晚生成工作进展报告

其关键能力来自：

cron
web / terminal / file 工具
持久化会话
messaging 投递

5.8 多系统统一入口

如果一个组织已经有多个内部系统，Hermes 很适合作为统一自然语言入口。

通过 MCP 或自定义工具，可以把这些系统统一接入：

文档平台
项目管理平台
CI/CD
工单系统
内部 API
知识库

这样用户就不必记住每个系统的入口和操作语法，而是通过 Hermes 统一调度。

6. 项目的工程价值

Hermes 的工程价值主要体现在以下方面。

6.1 它不是玩具式 agent

很多 agent 项目能 demo，但很难长期用。Hermes 明显考虑了长期运行问题：

状态持久化
工具分层
多入口
调度
配置体系
测试体系
权限和环境边界

6.2 它强调统一入口

CLI、gateway、ACP、API server、cron、MCP 这些模块放在一起，说明 Hermes 的设计目标不是单一产品，而是统一 agent runtime。

6.3 它强调可扩展性

registry、toolset、MCP、skills、plugins 这些机制说明 Hermes 的重点之一是“扩展”，不是“把一切写死在主循环里”。

6.4 它强调长期记忆和复用

session search、memory、skills、context compression 都在解决一个问题：

如何让 agent 不只在当前轮聪明，而是在长期使用中变得更有连续性。

7. 适合解决的问题类型

Hermes 特别适合下面这些任务：

多步骤、需要工具执行的任务
需要长期上下文的任务
需要跨平台访问的任务
需要周期性自动运行的任务
需要连接多个系统的任务
需要把经验沉淀成标准流程的任务

简单说，它适合“工作流型问题”，不只是“知识问答型问题”。

8. 不适合解决的问题类型

虽然 Hermes 很强，但也有明确边界。

8.1 不适合极低延迟场景

只要核心路径依赖大模型推理和多轮工具调用，就不适合：

毫秒级交互
超高频实时控制
严格时延敏感链路

8.2 不适合强确定性核心逻辑

Hermes 适合作为智能编排层，但不适合直接替代：

严格规则引擎
事务核心
高风险审批最终裁决器

8.3 不适合无边界的高权限自动执行

它可以执行命令、写文件、接外部系统，因此如果没有工具边界、审批机制和环境隔离，风险会很高。

项目本身其实已经考虑到这点，所以才有：

toolset
approval
environment backend
profile
配置隔离

9. 最适合的落地方式

如果从通用使用角度评价，Hermes 最适合以下三种落地方式。

9.1 本地个人工作台

适合开发者、研究者、重度终端用户。

特点：

直接用 CLI
连本地代码和 shell
有历史会话和技能沉淀

9.2 服务器上的长期运行 Agent

适合希望让 agent 常驻运行的用户。

特点：

部署到 VPS 或云主机
用 gateway 接 Telegram/Slack 等
用 cron 定时执行任务
用消息平台接收结果

9.3 组织内部的统一智能接口

适合中大型团队。

特点：

通过 MCP 接入内部系统
通过 gateway 提供统一入口
通过 skill 和 memory 沉淀组织经验

10. 最终评价

Hermes Agent 的核心价值不在于“模型本身多强”，而在于它把一个通用 agent 真正做成了可运行、可扩展、可持续使用的平台。

如果只看通用功能，它最突出的点有四个：

工具系统完整，能做真实任务执行。
多入口统一，既能本地用，也能在消息平台和后台运行。
长期状态能力强，支持会话、记忆、技能和上下文压缩。
扩展性好，能通过 MCP 和自定义工具连接外部世界。

因此，Hermes 最适合被理解为：

“面向长期任务执行和多系统编排的通用 Agent 平台。”

它不是最轻的方案，但如果目标是做一个真正能长期用、能接系统、能跑自动化、能在不同入口持续工作的 agent，Hermes 的设计是比较完整的。

第1章什么是 QuantConnect/Lean

Henry Lin — Sun, 22 Mar 2026 08:17:50 +0000

第1章什么是 Lean？

本章介绍 Lean 量化交易引擎的基本概念、核心功能以及 Lean 与 Lean CLI 的区别。

1.1 项目简介

Lean 是 QuantConnect 开源的事件驱动算法交易平台，专为量化交易开发者设计。它提供了完整的策略研究、历史回测、参数优化和实盘交易能力。

核心功能

功能模块	说明
策略研究	支持在 Jupyter Notebook 中进行数据探索和策略验证
历史回测	基于历史数据模拟交易策略表现
参数优化	自动搜索最优策略参数
实盘交易	支持多家券商的实盘接入
多资产类别	股票、期权、期货、外汇、加密货币等

支持的编程语言

C# - 原生语言，性能最佳
Python - 通过 Python.NET 集成，生态丰富

1.2 Lean vs Lean CLI

Lean Engine（本项目）

Lean Engine 是核心交易引擎，包含：

Lean/
├── Engine/              # 核心引擎
├── Algorithm/           # 算法基类
├── Algorithm.CSharp/    # C# 示例策略
├── Algorithm.Python/    # Python 示例策略
├── Algorithm.Framework/ # 框架化模型
├── Common/              # 通用组件
├── Brokerages/          # 券商接口
└── Launcher/            # 程序入口

Lean CLI（推荐用于策略开发）

Lean CLI 是命令行工具，简化了 Lean 的使用：

# 安装 CLI
pip install lean

# 创建项目
lean project-create

# 运行回测
lean backtest

# 启动研究环境
lean research

何时使用哪个？

场景	推荐工具
开发交易策略	Lean CLI
贡献 Lean 源码	Lean Engine
学习内部机制	Lean Engine
快速验证想法	Lean CLI

1.3 核心特性

事件驱动架构

Lean 采用事件驱动架构，算法响应各种事件：

from AlgorithmImports import *

class MyAlgorithm(QCAlgorithm):

    def Initialize(self):
        # 初始化设置
        self.AddEquity("SPY", Resolution.Minute)

    def OnData(self, data):
        # 每个数据点触发
        pass

    def OnOrderEvent(self, orderEvent):
        # 订单事件触发
        pass

    def OnEndOfDay(self, symbol):
        # 每日收盘时触发
        pass

多资产类别支持

资产类型	Add 方法	市场示例
股票	`AddEquity()`	USA, China, India
外汇	`AddForex()`	EURUSD, USDJPY
期货	`AddFuture()`	ES, NQ, CL
期权	`AddOption()`	SPY, AAPL
加密货币	`AddCrypto()`	BTCUSD, ETHUSD
CFD	`AddCfd()`	各种差价合约
指数	`AddIndex()`	SPX, VIX

模块化设计

Lean 的每个组件都可以自定义：

┌─────────────────────────────────────────────┐
│              QCAlgorithm                    │
│  ┌──────────────────────────────────────┐  │
│  │  Alpha Model        (交易信号生成)    │  │
│  │  Portfolio Model    (组合构建)        │  │
│  │  Execution Model    (订单执行)        │  │
│  │  Risk Model         (风险控制)        │  │
│  │  Universe Model     (证券选择)        │  │
│  └──────────────────────────────────────┘  │
│                                             │
│  ┌──────────────────────────────────────┐  │
│  │  Brokerage          (券商接口)        │  │
│  │  Data Feed          (数据源)          │  │
│  │  Fill Model         (成交模型)        │  │
│  │  Fee Model          (手续费模型)      │  │
│  │  Slippage Model     (滑点模型)        │  │
│  └──────────────────────────────────────┘  │
└─────────────────────────────────────────────┘

1.4 架构概览

                        ┌─────────────┐
                        │   用户输入   │
                        └──────┬──────┘
                               │
                        ┌──────▼──────┐
                        │  Launcher   │
                        └──────┬──────┘
                               │
        ┌──────────────────────┼──────────────────────┐
        │                      │                      │
┌───────▼────────┐    ┌────────▼────────┐    ┌───────▼────────┐
│  Algorithm     │    │   Data Feed     │    │   Transaction  │
│  (策略逻辑)    │    │  (数据提供器)    │    │   (交易处理)    │
└───────┬────────┘    └────────┬────────┘    └───────┬────────┘
        │                      │                      │
        └──────────────────────┼──────────────────────┘
                               │
                        ┌──────▼──────┐
                        │   Results    │
                        │  (结果输出)   │
                        └─────────────┘

1.5 本地与云端混合开发

Lean 支持本地开发和云端部署的混合模式：

┌─────────────┐         ┌─────────────┐
│  本地开发    │  <--->  │  云端平台    │
│  (Lean CLI) │         │(QuantConnect)│
└─────────────┘         └─────────────┘
     • 完整控制              • 数据获取
     • 调试方便              • 云计算资源
     • 数据隐私              • 策略分享

1.6 示例：第一个 Lean 算法

下面是一个最简单的 Lean 算法示例：

from AlgorithmImports import *

class HelloWorldAlgorithm(QCAlgorithm):
    """
    最简单的 Lean 算法示例
    功能：买入并持有 SPY
    """

    def Initialize(self):
        # 1. 设置回测时间范围
        self.SetStartDate(2023, 1, 1)    # 开始日期
        self.SetEndDate(2023, 12, 31)    # 结束日期
        self.SetCash(100000)             # 初始资金

        # 2. 添加证券
        self.AddEquity("SPY", Resolution.Daily)

        # 3. 设置benchmark
        self.SetBenchmark("SPY")

    def OnData(self, data):
        # 4. 如果还没有持仓，买入SPY
        if not self.Portfolio.Invested:
            self.SetHoldings("SPY", 1)   # 使用100%资金买入
            self.Debug("Purchased SPY")

# RD-Agent Tutorial - Chapter 3: Advanced Usage Techniques

Henry Lin — Sun, 22 Mar 2026 02:29:59 +0000

RD-Agent Tutorial - Chapter 3: Advanced Usage Techniques

3.1 Quantitative Finance Agent Deep Dive

Quantitative Finance Scenario Architecture

RD-Agent's quantitative finance agent RD-Agent(Q) is the industry's first data-driven multi-agent quantitative strategy development framework. It achieves full-stack quantitative research and development automation through coordinated factor-model joint optimization.

Core Architecture Design

graph TB
    A[Market Data] --> B[Data Preprocessing]
    B --> C[Factor Mining Agent]
    B --> D[Model Evolution Agent]
    C --> E[Factor Evaluation]
    D --> F[Model Evaluation]
    E --> G[Joint Optimization Agent]
    F --> G
    G --> H[Strategy Backtesting]
    H --> I[Performance Analysis]
    I --> J{Requirements Met?}
    J -->|No| K[Knowledge Feedback]
    K --> C
    K --> D
    J -->|Yes| L[Strategy Deployment]

    style C fill:#e1f5fe
    style D fill:#f3e5f5
    style G fill:#e8f5e8
    style H fill:#fff3e0

Qlib Framework Integration

RD-Agent(Q) deeply integrates with Microsoft's Qlib quantitative investment library, providing:

Data Management:

🗃️ Standardized financial data format
📈 Real-time and historical data interfaces
🔄 Automatic data update mechanism
🧹 Data cleaning and preprocessing

Backtesting Engine:

📊 High-performance vectorized backtesting
💰 Trading cost and slippage simulation
📋 Multiple portfolio construction strategies
📈 Risk metric calculation

Model Library:

🤖 Machine learning model integration
📊 Traditional quantitative factors
🧠 Deep learning models
🔄 Model ensemble strategies

Factor Mining and Optimization

`rdagent fin_factor` Explained

The factor mining agent focuses on discovering and optimizing effective quantitative factors.

Startup Command:

# Basic factor mining
rdagent fin_factor

# Specify configuration file
rdagent fin_factor --config factor_config.yaml

# Specify iteration count
export FACTOR_MAX_LOOP=15
rdagent fin_factor

Workflow:

Hypothesis Generation Phase

   # Generate factor hypotheses based on market theory and historical experience
   factor_hypotheses = [
       "technical_indicator_momentum_factor",
       "financial_quality_factor",
       "market_sentiment_factor",
       "macroeconomic_factor"
   ]

Factor Implementation Phase

   # Automatically generate factor calculation code
   class MomentumFactor:
       def calculate(self, data):
           # Auto-generated momentum factor calculation logic
           return (data['close'] / data['close'].shift(20) - 1)

Factor Evaluation Phase

   # Multi-dimensional factor evaluation
   evaluation_metrics = {
       'IC': 0.045,          # Information Coefficient
       'IC_IR': 1.2,         # Information Ratio
       'rank_IC': 0.038,     # Rank IC
       'turnover': 0.8,      # Turnover rate
       'max_drawdown': 0.15  # Maximum drawdown
   }

Factor Generation Strategies

Technical Analysis Factors:

technical_factors = {
    "momentum": [
        "price_momentum_5d", "price_momentum_20d",
        "volume_momentum", "volatility_momentum"
    ],
    "mean_reversion": [
        "rsi_divergence", "bollinger_position",
        "price_deviation"
    ],
    "trend": [
        "ma_trend", "macd_signal", "trend_strength"
    ]
}

Fundamental Factors:

fundamental_factors = {
    "profitability": [
        "roe_trend", "roa_improvement",
        "gross_margin_stability"
    ],
    "growth": [
        "revenue_growth_consistency", "eps_growth_quality",
        "cash_flow_growth"
    ],
    "valuation": [
        "pe_relative", "pb_sector_adjusted",
        "ev_ebitda_normalized"
    ]
}

Alternative Factors:

alternative_factors = {
    "sentiment": [
        "news_sentiment_score", "social_media_buzz",
        "analyst_revision_momentum"
    ],
    "network": [
        "supply_chain_strength", "industry_correlation",
        "peer_performance_influence"
    ]
}

Factor Effectiveness Evaluation

Evaluation Framework:

class FactorEvaluator:
    def __init__(self, benchmark_data, start_date, end_date):
        self.data = benchmark_data
        self.start_date = start_date
        self.end_date = end_date

    def evaluate_factor(self, factor_values):
        """Comprehensive factor effectiveness evaluation"""
        metrics = {}

        # 1. Information Coefficient analysis
        metrics['IC'] = self.calculate_IC(factor_values)
        metrics['IC_std'] = self.calculate_IC_stability(factor_values)
        metrics['IC_IR'] = metrics['IC'] / metrics['IC_std']

        # 2. Monotonicity test
        metrics['monotonicity'] = self.test_monotonicity(factor_values)

        # 3. Turnover rate analysis
        metrics['turnover'] = self.calculate_turnover(factor_values)

        # 4. Backtesting performance
        backtest_result = self.backtest_factor(factor_values)
        metrics.update(backtest_result)

        return metrics

Model Evolution and Optimization

`rdagent fin_model` Explained

The model evolution agent focuses on automated development and optimization of prediction models.

Supported Model Types:

Model Category	Specific Models	Applicable Scenarios	Features
Linear Models	LinearRegression, Ridge, Lasso	Simple fast prediction	High interpretability
Tree Models	XGBoost, LightGBM, CatBoost	Tabular data modeling	Feature importance
Neural Networks	MLP, TabNet, DeepFM	Complex relationship modeling	Non-linear fitting
Time Series Models	LSTM, GRU, Transformer	Sequence data prediction	Time dependency
Ensemble Models	Stacking, Voting, Blending	Improve prediction accuracy	Model fusion

Model Architecture Search

Automated Architecture Search:

class ModelArchitectureSearch:
    def __init__(self):
        self.search_space = {
            "n_layers": [2, 3, 4, 5],
            "hidden_dims": [64, 128, 256, 512],
            "dropout_rate": [0.1, 0.2, 0.3, 0.4],
            "activation": ["relu", "tanh", "gelu"],
            "optimizer": ["adam", "adamw", "sgd"],
            "learning_rate": [0.001, 0.01, 0.1]
        }

    def search_best_architecture(self, train_data, valid_data):
        """Search for optimal model architecture"""
        best_config = None
        best_score = float('-inf')

        for config in self.generate_configurations():
            model = self.build_model(config)
            score = self.evaluate_model(model, train_data, valid_data)

            if score > best_score:
                best_score = score
                best_config = config

        return best_config, best_score

Hyperparameter Optimization

Multiple Optimization Algorithms:

from optuna import create_study
from sklearn.model_selection import cross_val_score

class HyperparameterOptimizer:
    def __init__(self, model_class, param_space):
        self.model_class = model_class
        self.param_space = param_space

    def optimize_with_optuna(self, X_train, y_train, n_trials=100):
        """Hyperparameter optimization using Optuna"""
        def objective(trial):
            params = {}
            for param_name, param_range in self.param_space.items():
                if isinstance(param_range, list):
                    params[param_name] = trial.suggest_categorical(
                        param_name, param_range
                    )
                elif isinstance(param_range, tuple):
                    if isinstance(param_range[0], int):
                        params[param_name] = trial.suggest_int(
                            param_name, param_range[0], param_range[1]
                        )
                    else:
                        params[param_name] = trial.suggest_float(
                            param_name, param_range[0], param_range[1]
                        )

            model = self.model_class(**params)
            scores = cross_val_score(model, X_train, y_train, cv=5)
            return scores.mean()

        study = create_study(direction='maximize')
        study.optimize(objective, n_trials=n_trials)

        return study.best_params, study.best_value

Factor-Model Joint Optimization

`rdagent fin_quant` Explained

The joint optimization agent is the core innovation of RD-Agent(Q), implementing collaborative evolution of factors and models.

Joint Optimization Strategies:

Alternating Optimization Strategy

   class AlternatingOptimization:
       def __init__(self, factor_agent, model_agent):
           self.factor_agent = factor_agent
           self.model_agent = model_agent

       def optimize(self, max_iterations=10):
           for i in range(max_iterations):
               # Fix model, optimize factors
               new_factors = self.factor_agent.evolve_factors(
                   current_model=self.current_model
               )

               # Fix factors, optimize model
               new_model = self.model_agent.evolve_model(
                   current_factors=new_factors
               )

               # Evaluate joint performance
               performance = self.evaluate_joint_performance(
                   new_factors, new_model
               )

               if performance > self.best_performance:
                   self.update_best_solution(new_factors, new_model)

Multi-Objective Optimization Strategy

   class MultiObjectiveOptimization:
       def __init__(self):
           self.objectives = [
               "maximize_sharpe_ratio",
               "minimize_max_drawdown",
               "maximize_information_ratio",
               "minimize_turnover"
           ]

       def pareto_optimization(self, population):
           """Pareto front optimization"""
           pareto_front = []

           for individual in population:
               is_dominated = False
               for other in population:
                   if self.dominates(other, individual):
                       is_dominated = True
                       break

               if not is_dominated:
                   pareto_front.append(individual)

           return pareto_front

Multi-Objective Optimization Algorithms

NSGA-II Implementation:

class NSGA2Optimizer:
    def __init__(self, population_size=100, generations=50):
        self.population_size = population_size
        self.generations = generations

    def optimize(self, factor_space, model_space):
        # Initialize population
        population = self.initialize_population()

        for generation in range(self.generations):
            # Evaluate fitness
            fitness_values = self.evaluate_population(population)

            # Non-dominated sorting
            fronts = self.fast_non_dominated_sort(population, fitness_values)

            # Select next generation
            next_population = self.select_next_generation(
                fronts, population, fitness_values
            )

            # Crossover and mutation
            offspring = self.crossover_and_mutation(next_population)
            population = offspring

        return self.get_pareto_front(population)

Collaborative Evolution Mechanism

Knowledge Sharing Mechanism:

class KnowledgeSharing:
    def __init__(self):
        self.factor_knowledge_base = {}
        self.model_knowledge_base = {}
        self.interaction_patterns = {}

    def share_factor_insights(self, factor_id, insights):
        """Share factor insights"""
        self.factor_knowledge_base[factor_id] = {
            'effectiveness': insights['ic_score'],
            'stability': insights['ic_std'],
            'best_models': insights['compatible_models'],
            'market_regimes': insights['effective_periods']
        }

    def share_model_insights(self, model_id, insights):
        """Share model insights"""
        self.model_knowledge_base[model_id] = {
            'architecture': insights['model_config'],
            'performance': insights['validation_score'],
            'best_factors': insights['important_features'],
            'hyperparameters': insights['optimal_params']
        }

    def get_recommendations(self, current_factors, current_model):
        """Get optimization recommendations"""
        recommendations = {
            'factor_suggestions': [],
            'model_suggestions': [],
            'joint_strategies': []
        }

        # Generate recommendations based on historical knowledge
        for factor in current_factors:
            if factor in self.factor_knowledge_base:
                knowledge = self.factor_knowledge_base[factor]
                recommendations['model_suggestions'].extend(
                    knowledge['best_models']
                )

        return recommendations

Financial Report Analysis

`rdagent fin_factor_report` Usage

Automatically extracting quantitative factors from financial reports is a unique capability of RD-Agent.

Supported Report Types:

📋 Annual Reports (10-K, Annual Report)
📊 Quarterly Reports (10-Q, Quarterly Report)
📰 News Announcements (8-K, Press Release)
📈 Analyst Reports (Research Report)
🏛️ Regulatory Filings (SEC Filings)

Usage Example:

# Basic report analysis
rdagent fin_factor_report --report-folder ./financial_reports

# Specify report type
rdagent fin_factor_report \
    --report-folder ./reports \
    --report-type annual \
    --language zh

# Batch processing
rdagent fin_factor_report \
    --report-folder ./reports \
    --batch-size 50 \
    --parallel-workers 4

Report Parsing and Feature Extraction

Text Preprocessing Pipeline:

class ReportPreprocessor:
    def __init__(self):
        self.nlp = spacy.load("en_core_web_sm")
        self.financial_terms = self.load_financial_dictionary()

    def preprocess_report(self, report_text):
        """Preprocess financial report"""
        # 1. Document cleaning
        cleaned_text = self.clean_document(report_text)

        # 2. Section segmentation
        sections = self.extract_sections(cleaned_text)

        # 3. Entity recognition
        entities = self.extract_financial_entities(sections)

        # 4. Sentiment analysis
        sentiment_scores = self.analyze_sentiment(sections)

        return {
            'sections': sections,
            'entities': entities,
            'sentiment': sentiment_scores,
            'metadata': self.extract_metadata(report_text)
        }

Factor Extraction Strategy:

class FinancialFactorExtractor:
    def __init__(self):
        self.factor_extractors = {
            'growth_factors': GrowthFactorExtractor(),
            'quality_factors': QualityFactorExtractor(),
            'sentiment_factors': SentimentFactorExtractor(),
            'risk_factors': RiskFactorExtractor()
        }

    def extract_factors(self, processed_report):
        """Extract factors from processed report"""
        extracted_factors = {}

        for factor_type, extractor in self.factor_extractors.items():
            factors = extractor.extract(processed_report)
            extracted_factors[factor_type] = factors

        return extracted_factors

class GrowthFactorExtractor:
    def extract(self, report_data):
        """Extract growth factors"""
        growth_factors = {}

        # Revenue growth quality
        revenue_mentions = self.find_revenue_discussions(report_data)
        growth_factors['revenue_growth_quality'] = self.analyze_growth_quality(
            revenue_mentions
        )

        # Market expansion strategy
        expansion_mentions = self.find_expansion_discussions(report_data)
        growth_factors['expansion_strategy_score'] = self.score_expansion_strategy(
            expansion_mentions
        )

        # R&D investment commitment
        rd_mentions = self.find_rd_discussions(report_data)
        growth_factors['rd_commitment_score'] = self.analyze_rd_commitment(
            rd_mentions
        )

        return growth_factors

Sentiment Analysis and Signal Mining

Multi-dimensional Sentiment Analysis:

class FinancialSentimentAnalyzer:
    def __init__(self):
        self.sentiment_model = pipeline(
            "sentiment-analysis",
            model="ProsusAI/finbert"
        )
        self.uncertainty_detector = UncertaintyDetector()
        self.forward_looking_detector = ForwardLookingDetector()

    def analyze_comprehensive_sentiment(self, text_sections):
        """Comprehensive sentiment analysis"""
        sentiment_scores = {}

        for section_name, text in text_sections.items():
            # Basic sentiment analysis
            basic_sentiment = self.sentiment_model(text)

            # Uncertainty detection
            uncertainty_score = self.uncertainty_detector.detect(text)

            # Forward-looking statement detection
            forward_looking = self.forward_looking_detector.detect(text)

            sentiment_scores[section_name] = {
                'polarity': basic_sentiment[0]['score'],
                'label': basic_sentiment[0]['label'],
                'uncertainty': uncertainty_score,
                'forward_looking': forward_looking,
                'confidence': self.calculate_confidence(
                    basic_sentiment, uncertainty_score
                )
            }

        return sentiment_scores

class SignalGenerator:
    def __init__(self):
        self.signal_weights = {
            'management_tone': 0.3,
            'financial_health': 0.4,
            'market_position': 0.2,
            'future_outlook': 0.1
        }

    def generate_trading_signals(self, factor_scores, sentiment_scores):
        """Generate trading signals"""
        composite_score = 0.0

        # Weighted composite score
        for factor_type, weight in self.signal_weights.items():
            if factor_type in factor_scores:
                composite_score += factor_scores[factor_type] * weight

        # Sentiment adjustment
        sentiment_adjustment = self.calculate_sentiment_adjustment(
            sentiment_scores
        )

        final_score = composite_score * sentiment_adjustment

        # Generate signal
        if final_score > 0.6:
            signal = "STRONG_BUY"
        elif final_score > 0.3:
            signal = "BUY"
        elif final_score > -0.3:
            signal = "HOLD"
        elif final_score > -0.6:
            signal = "SELL"
        else:
            signal = "STRONG_SELL"

        return {
            'signal': signal,
            'confidence': abs(final_score),
            'score': final_score,
            'components': {
                'factor_score': composite_score,
                'sentiment_adjustment': sentiment_adjustment
            }
        }

3.2 CoSTEER Evolution Framework Deep Dive

Evolution Framework Core Principles

CoSTEER (Collaborative evolving STrategy for automatic data-cEntric dEvelopment fRamework) is RD-Agent's core evolution engine, implementing autonomous learning and continuous improvement for agents.

Collaborative Evolution Strategy Design

The CoSTEER framework is based on the following core principles:

Multi-Agent Collaboration - Different specialized agents work together
Knowledge Accumulation - Learn and improve from historical experience
Adaptive Evolution - Dynamically adjust strategies based on feedback
RAG Enhancement - Retrieval-augmented generation improves decision quality

class CoSTEER(Developer[Experiment]):
    """CoSTEER evolution framework core class"""

    def __init__(
        self,
        settings: CoSTEERSettings,
        eva: RAGEvaluator,
        es: EvolvingStrategy,
        evolving_version: int = 2,
        with_knowledge: bool = True,
        knowledge_self_gen: bool = True,
        max_loop: int | None = None,
    ):
        self.settings = settings
        self.max_loop = settings.max_loop if max_loop is None else max_loop
        self.knowledge_base_path = Path(settings.knowledge_base_path)
        self.with_knowledge = with_knowledge
        self.knowledge_self_gen = knowledge_self_gen
        self.evolving_strategy = es
        self.evaluator = eva

        # Initialize RAG system
        self.rag_strategy = self._init_rag_strategy(evolving_version)

Knowledge Management Mechanism

Multi-level Knowledge Representation:

class KnowledgeManagementSystem:
    def __init__(self):
        self.knowledge_layers = {
            'factual_knowledge': FactualKnowledgeBase(),
            'procedural_knowledge': ProceduralKnowledgeBase(),
            'experiential_knowledge': ExperientialKnowledgeBase(),
            'meta_knowledge': MetaKnowledgeBase()
        }

    def store_knowledge(self, knowledge_item):
        """Layered knowledge storage"""
        knowledge_type = self.classify_knowledge(knowledge_item)
        self.knowledge_layers[knowledge_type].store(knowledge_item)

    def retrieve_relevant_knowledge(self, query_context):
        """Retrieve relevant knowledge"""
        relevant_knowledge = {}

        for layer_name, knowledge_base in self.knowledge_layers.items():
            relevant_items = knowledge_base.query(
                query_context,
                similarity_threshold=0.7
            )
            relevant_knowledge[layer_name] = relevant_items

        return self.synthesize_knowledge(relevant_knowledge)

Knowledge Representation Format:

@dataclass
class KnowledgeItem:
    """Knowledge item data structure"""
    id: str
    content: str
    knowledge_type: KnowledgeType
    domain: str
    confidence: float
    source: str
    timestamp: datetime
    usage_count: int = 0
    success_rate: float = 0.0

    def to_embedding(self) -> np.ndarray:
        """Convert to vector representation"""
        return self.embedding_model.encode(self.content)

    def update_effectiveness(self, success: bool):
        """Update knowledge effectiveness"""
        self.usage_count += 1
        if success:
            self.success_rate = (
                (self.success_rate * (self.usage_count - 1) + 1.0)
                / self.usage_count
            )
        else:
            self.success_rate = (
                self.success_rate * (self.usage_count - 1)
                / self.usage_count
            )

Adaptive Optimization Algorithm

Multi-Strategy Evolution Algorithm:

class AdaptiveEvolutionStrategy:
    def __init__(self):
        self.evolution_strategies = {
            'genetic_algorithm': GeneticAlgorithmStrategy(),
            'differential_evolution': DifferentialEvolutionStrategy(),
            'particle_swarm': ParticleSwarmStrategy(),
            'simulated_annealing': SimulatedAnnealingStrategy()
        }
        self.strategy_performance = {}
        self.adaptive_weights = {}

    def evolve_population(self, population, generation):
        """Adaptive population evolution"""
        # Select evolution strategy
        selected_strategies = self.select_strategies(generation)

        new_population = []
        for strategy_name, weight in selected_strategies.items():
            strategy = self.evolution_strategies[strategy_name]

            # Allocate population by weight
            sub_population_size = int(len(population) * weight)
            sub_population = population[:sub_population_size]

            # Evolve sub-population
            evolved_sub_pop = strategy.evolve(sub_population)
            new_population.extend(evolved_sub_pop)

        # Update strategy performance
        self.update_strategy_performance(new_population, generation)

        return new_population

    def select_strategies(self, generation):
        """Select evolution strategy based on historical performance"""
        if generation < 5:
            # Uniform distribution in early stages
            return {name: 1.0/len(self.evolution_strategies)
                   for name in self.evolution_strategies.keys()}

        # Adjust weights based on performance
        total_performance = sum(self.strategy_performance.values())
        return {
            name: performance / total_performance
            for name, performance in self.strategy_performance.items()
        }

Custom Evolution Strategies

Strategy Configuration and Adjustment

Configuration File Example:

# costeer_config.yaml
costeer_settings:
  max_loop: 10
  knowledge_base_path: "./knowledge_base"
  new_knowledge_base_path: "./new_knowledge"
  evolving_version: 2

evolution_strategy:
  population_size: 50
  mutation_rate: 0.1
  crossover_rate: 0.8
  selection_method: "tournament"
  tournament_size: 5

rag_settings:
  embedding_model: "text-embedding-3-small"
  similarity_threshold: 0.75
  max_retrieved_docs: 10
  rerank_top_k: 5

evaluation_settings:
  metrics: ["accuracy", "f1_score", "auc_roc"]
  cross_validation_folds: 5
  test_ratio: 0.2

knowledge_management:
  auto_cleanup: true
  cleanup_threshold: 0.3
  knowledge_update_frequency: 100
  max_knowledge_items: 10000

Custom Evolution Strategy Implementation:

class CustomEvolutionStrategy(EvolvingStrategy):
    """Custom evolution strategy"""

    def __init__(self, config: dict):
        super().__init__()
        self.config = config
        self.population_size = config.get('population_size', 50)
        self.mutation_rate = config.get('mutation_rate', 0.1)
        self.crossover_rate = config.get('crossover_rate', 0.8)

    def initialize_population(self, size: int = None):
        """Initialize population"""
        size = size or self.population_size
        population = []

        for _ in range(size):
            individual = self.create_random_individual()
            population.append(individual)

        return population

    def evolve_generation(self, population, fitness_scores):
        """Evolve one generation"""
        new_population = []

        # Elite selection - preserve best individuals
        elite_count = int(len(population) * 0.1)
        elite_indices = np.argsort(fitness_scores)[-elite_count:]
        for idx in elite_indices:
            new_population.append(population[idx])

        # Generate new individuals through crossover and mutation
        while len(new_population) < len(population):
            # Select parents
            parent1 = self.tournament_selection(population, fitness_scores)
            parent2 = self.tournament_selection(population, fitness_scores)

            # Crossover
            if random.random() < self.crossover_rate:
                child1, child2 = self.crossover(parent1, parent2)
            else:
                child1, child2 = parent1, parent2

            # Mutation
            if random.random() < self.mutation_rate:
                child1 = self.mutate(child1)
            if random.random() < self.mutation_rate:
                child2 = self.mutate(child2)

            new_population.extend([child1, child2])

        return new_population[:len(population)]

    def crossover(self, parent1, parent2):
        """Crossover operation"""
        # Implement specific crossover logic
        # Parameter crossover example here
        child1_params = {}
        child2_params = {}

        for key in parent1.parameters.keys():
            if random.random() < 0.5:
                child1_params[key] = parent1.parameters[key]
                child2_params[key] = parent2.parameters[key]
            else:
                child1_params[key] = parent2.parameters[key]
                child2_params[key] = parent1.parameters[key]

        child1 = parent1.copy()
        child2 = parent2.copy()
        child1.parameters = child1_params
        child2.parameters = child2_params

        return child1, child2

    def mutate(self, individual):
        """Mutation operation"""
        mutated = individual.copy()

        for key, value in mutated.parameters.items():
            if random.random() < self.mutation_rate:
                if isinstance(value, (int, float)):
                    # Gaussian mutation
                    noise = np.random.normal(0, 0.1 * abs(value))
                    mutated.parameters[key] = value + noise
                elif isinstance(value, str):
                    # String mutation (e.g., model names)
                    mutated.parameters[key] = self.mutate_string(value)

        return mutated

Evaluator Customization

Multi-Metric Evaluator:

class MultiMetricEvaluator(RAGEvaluator):
    """Multi-metric evaluator"""

    def __init__(self, metrics_config: dict):
        super().__init__()
        self.metrics_config = metrics_config
        self.metric_weights = metrics_config.get('weights', {})
        self.metric_calculators = self._init_metric_calculators()

    def evaluate(self, experiment: Experiment) -> Dict[str, float]:
        """Multi-metric evaluation"""
        metrics = {}

        for metric_name, calculator in self.metric_calculators.items():
            try:
                score = calculator.calculate(experiment)
                metrics[metric_name] = score
            except Exception as e:
                logger.warning(f"Metric {metric_name} calculation failed: {e}")
                metrics[metric_name] = 0.0

        # Calculate weighted composite score
        weighted_score = self._calculate_weighted_score(metrics)
        metrics['weighted_score'] = weighted_score

        return metrics

    def _calculate_weighted_score(self, metrics: Dict[str, float]) -> float:
        """Calculate weighted composite score"""
        total_score = 0.0
        total_weight = 0.0

        for metric_name, score in metrics.items():
            if metric_name in self.metric_weights:
                weight = self.metric_weights[metric_name]
                total_score += score * weight
                total_weight += weight

        return total_score / total_weight if total_weight > 0 else 0.0

class CustomMetricCalculator:
    """Custom metric calculator"""

    def __init__(self, metric_name: str, calculation_func: callable):
        self.metric_name = metric_name
        self.calculation_func = calculation_func

    def calculate(self, experiment: Experiment) -> float:
        """Calculate metric value"""
        return self.calculation_func(experiment)

# Usage example
def sharpe_ratio_calculator(experiment):
    """Calculate Sharpe ratio"""
    returns = experiment.get_returns()
    if len(returns) == 0 or returns.std() == 0:
        return 0.0
    return returns.mean() / returns.std() * np.sqrt(252)

def max_drawdown_calculator(experiment):
    """Calculate maximum drawdown"""
    cumulative_returns = (1 + experiment.get_returns()).cumprod()
    running_max = cumulative_returns.expanding().max()
    drawdown = (cumulative_returns - running_max) / running_max
    return -drawdown.min()  # Return positive value

# Create custom evaluator
metrics_config = {
    'weights': {
        'sharpe_ratio': 0.4,
        'max_drawdown': 0.3,
        'information_ratio': 0.3
    }
}

evaluator = MultiMetricEvaluator(metrics_config)
evaluator.add_metric_calculator('sharpe_ratio', sharpe_ratio_calculator)
evaluator.add_metric_calculator('max_drawdown', max_drawdown_calculator)

Knowledge Base Management

Knowledge Base Optimization Strategy:

class KnowledgeBaseManager:
    """Knowledge base manager"""

    def __init__(self, config: dict):
        self.config = config
        self.knowledge_store = {}
        self.embedding_cache = {}
        self.access_frequency = defaultdict(int)
        self.last_access_time = {}

    def add_knowledge(self, knowledge_item: KnowledgeItem):
        """Add knowledge item"""
        # Check knowledge base size limit
        if len(self.knowledge_store) >= self.config.get('max_items', 10000):
            self._cleanup_knowledge()

        self.knowledge_store[knowledge_item.id] = knowledge_item
        self.embedding_cache[knowledge_item.id] = knowledge_item.to_embedding()
        self.last_access_time[knowledge_item.id] = datetime.now()

    def query_knowledge(
        self,
        query: str,
        top_k: int = 10,
        similarity_threshold: float = 0.7
    ) -> List[KnowledgeItem]:
        """Query relevant knowledge"""
        query_embedding = self._get_query_embedding(query)
        similarities = {}

        for item_id, item_embedding in self.embedding_cache.items():
            similarity = cosine_similarity(
                query_embedding.reshape(1, -1),
                item_embedding.reshape(1, -1)
            )[0][0]

            if similarity >= similarity_threshold:
                similarities[item_id] = similarity
                self.access_frequency[item_id] += 1
                self.last_access_time[item_id] = datetime.now()

        # Sort by similarity and return top-k
        sorted_items = sorted(
            similarities.items(),
            key=lambda x: x[1],
            reverse=True
        )[:top_k]

        return [self.knowledge_store[item_id] for item_id, _ in sorted_items]

    def _cleanup_knowledge(self):
        """Clean up knowledge base"""
        cleanup_count = int(len(self.knowledge_store) * 0.1)  # Clean 10%

        # Composite score based on access frequency and time
        scores = {}
        current_time = datetime.now()

        for item_id, knowledge_item in self.knowledge_store.items():
            frequency_score = self.access_frequency[item_id]
            time_score = (
                current_time - self.last_access_time[item_id]
            ).days
            effectiveness_score = knowledge_item.success_rate

            # Composite score (high frequency, recent, effective = high score)
            composite_score = (
                frequency_score * 0.4 +
                effectiveness_score * 0.4 -
                time_score * 0.2
            )
            scores[item_id] = composite_score

        # Remove lowest scored items
        items_to_remove = sorted(
            scores.items(),
            key=lambda x: x[1]
        )[:cleanup_count]

        for item_id, _ in items_to_remove:
            del self.knowledge_store[item_id]
            del self.embedding_cache[item_id]
            del self.access_frequency[item_id]
            del self.last_access_time[item_id]

        logger.info(f"Cleaned up {cleanup_count} knowledge items")

Multi-Agent Collaboration Mechanism

Research Agent Responsibilities

class ResearchAgent:
    """Research agent - responsible for hypothesis generation and strategy planning"""

    def __init__(self, knowledge_base: KnowledgeBase):
        self.knowledge_base = knowledge_base
        self.hypothesis_generator = HypothesisGenerator()
        self.strategy_planner = StrategyPlanner()

    def generate_hypotheses(self, problem_context: dict) -> List[Hypothesis]:
        """Generate research hypotheses"""
        # Retrieve relevant theories from knowledge base
        relevant_knowledge = self.knowledge_base.query(
            problem_context['description']
        )

        # Generate hypotheses based on domain knowledge
        hypotheses = []
        for knowledge_item in relevant_knowledge:
            hypothesis = self.hypothesis_generator.generate_from_knowledge(
                knowledge_item, problem_context
            )
            hypotheses.append(hypothesis)

        # Generate novel hypotheses
        novel_hypotheses = self.hypothesis_generator.generate_novel(
            problem_context, existing_hypotheses=hypotheses
        )
        hypotheses.extend(novel_hypotheses)

        return self.rank_hypotheses(hypotheses)

    def plan_research_strategy(self, hypothesis: Hypothesis) -> ResearchPlan:
        """Plan research strategy"""
        return self.strategy_planner.create_plan(
            hypothesis=hypothesis,
            available_resources=self.get_available_resources(),
            time_constraints=self.get_time_constraints()
        )

Development Agent Functions

class DevelopmentAgent:
    """Development agent - responsible for code implementation and testing validation"""

    def __init__(self, code_generator: CodeGenerator, validator: CodeValidator):
        self.code_generator = code_generator
        self.validator = validator
        self.implementation_cache = {}

    def implement_hypothesis(self, hypothesis: Hypothesis) -> Implementation:
        """Implement research hypothesis"""
        # Generate implementation code
        implementation_code = self.code_generator.generate_code(
            specification=hypothesis.specification,
            constraints=hypothesis.constraints
        )

        # Validate code correctness
        validation_result = self.validator.validate(implementation_code)

        if not validation_result.is_valid:
            # Fix code issues
            fixed_code = self.code_generator.fix_issues(
                implementation_code,
                validation_result.issues
            )
            implementation_code = fixed_code

        # Create implementation object
        implementation = Implementation(
            code=implementation_code,
            hypothesis=hypothesis,
            validation_result=validation_result
        )

        # Cache implementation
        self.implementation_cache[hypothesis.id] = implementation

        return implementation

    def optimize_implementation(
        self,
        implementation: Implementation,
        feedback: Feedback
    ) -> Implementation:
        """Optimize implementation based on feedback"""
        optimization_suggestions = self.analyze_feedback(feedback)

        optimized_code = self.code_generator.optimize_code(
            implementation.code,
            optimization_suggestions
        )

        return Implementation(
            code=optimized_code,
            hypothesis=implementation.hypothesis,
            parent_implementation=implementation
        )

Inter-Agent Communication Protocol

class AgentCommunicationProtocol:
    """Inter-agent communication protocol"""

    def __init__(self):
        self.message_queue = {}
        self.subscriptions = defaultdict(list)
        self.message_handlers = {}

    def register_agent(self, agent_id: str, agent: Agent):
        """Register agent"""
        self.message_queue[agent_id] = []
        agent.set_communication_protocol(self)

    def subscribe(self, agent_id: str, message_type: str):
        """Subscribe to message type"""
        self.subscriptions[message_type].append(agent_id)

    def publish(self, sender_id: str, message_type: str, message: dict):
        """Publish message"""
        # Add message metadata
        full_message = {
            'sender': sender_id,
            'type': message_type,
            'timestamp': datetime.now(),
            'content': message
        }

        # Send to all subscribers
        for subscriber_id in self.subscriptions[message_type]:
            if subscriber_id != sender_id:  # Don't send to self
                self.message_queue[subscriber_id].append(full_message)

    def get_messages(self, agent_id: str) -> List[dict]:
        """Get agent's messages"""
        messages = self.message_queue.get(agent_id, [])
        self.message_queue[agent_id] = []  # Clear queue
        return messages

class CollaborativeWorkflow:
    """Collaborative workflow"""

    def __init__(self, agents: List[Agent], protocol: AgentCommunicationProtocol):
        self.agents = {agent.id: agent for agent in agents}
        self.protocol = protocol
        self.workflow_state = WorkflowState.INITIALIZED

    def execute_collaborative_task(self, task: CollaborativeTask):
        """Execute collaborative task"""
        # Decompose task
        subtasks = self.decompose_task(task)

        # Assign subtasks to suitable agents
        task_assignments = self.assign_tasks(subtasks)

        # Execute subtasks
        results = {}
        for agent_id, assigned_tasks in task_assignments.items():
            agent = self.agents[agent_id]
            agent_results = agent.execute_tasks(assigned_tasks)
            results[agent_id] = agent_results

        # Coordinate and integrate results
        integrated_result = self.integrate_results(results)

        return integrated_result

    def coordinate_agents(self):
        """Coordinate agent activities"""
        # Process inter-agent messages
        for agent_id, agent in self.agents.items():
            messages = self.protocol.get_messages(agent_id)
            if messages:
                agent.process_messages(messages)

        # Check if coordination is needed
        coordination_needed = self.check_coordination_needs()
        if coordination_needed:
            self.perform_coordination()

This chapter demonstrates RD-Agent's advanced features and deep customization capabilities, including specialized functions of the quantitative finance agent, core mechanisms of the CoSTEER evolution framework, and implementation methods for multi-agent collaboration. The next chapter will demonstrate practical applications of these advanced features through real-world case studies.

Forem: Henry Lin

Hermes Agent 项目思维导图

Hermes Agent 项目思维导图

Hermes Agent 核心运行系统调用流程--源码分析

Hermes Agent 核心运行系统调用流程

1. 总体分层

2. 三个入口如何进入 AIAgent

2.1 CLI 路径

2.2 Gateway 路径

2.3 Cron 路径

3. Provider / Runtime Resolution 调用链

3.1 入口函数

3.2 解析步骤

3.3 auth.py 的职责

3.4 运行时解析的输出如何进入 AIAgent

4. AIAgent 初始化阶段

4.1 初始化时完成的事情

4.2 初始化时和本题流程直接相关的状态

5. Prompt Builder：系统提示词如何被组装

5.1 组装顺序

5.2 prompt_builder.py 提供的核心能力

5.2.1 上下文文件发现与注入

5.2.2 Skills 注入

5.3 会话级 system prompt 稳定策略

6. Prompt Caching：prefix cache 如何保持稳定

6.1 会话级 system prompt 缓存

6.2 Skills prompt 构建缓存

6.3 Provider 级 prompt cache

7. run_conversation() 的同步 agent loop

7.1 进入循环前的准备

7.2 单轮 API 调用前，如何拼 payload

7.3 _build_api_kwargs() 如何分发到不同 API 形态

7.4 模型返回后，先做什么

7.5 chat() 只是外层薄封装

8. Tool Dispatch：schema 收集、availability gating、function call 分发

8.1 tool schema 收集

8.2 get_tool_definitions() 如何做 availability gating

8.3 模型发出 tool_calls 后的调用链

8.4 并行与串行执行策略

8.5 agent-level tool 与 registry-level tool 的分界

8.6 handle_function_call() 做什么

9. Context Compression：长会话压缩如何接入主流程

9.1 初始化

9.2 压缩触发点

9.3 _compress_context() 的动作

9.4 ContextCompressor.compress() 的内部算法

10. Session Storage：SQLite + FTS5 与 gateway session index 的分工

10.1 SessionDB：结构化、可检索的长期存储

10.2 AIAgent 如何写入 SessionDB

10.3 SessionStore：gateway 的会话路由与 transcript 兼容层

10.4 SessionDB 与 SessionStore 的关系

11. Response Delivery：CLI、gateway、cron 三条出口

11.1 CLI display

11.2 Gateway delivery

第一段：_run_agent() 负责生成 agent 结果

第二段：平台 adapter 负责真正发送

11.3 Cron delivery

12. 一条完整调用链示例

13. 本系统里最重要的几个“稳定性约束”

13.1 system prompt 必须尽量稳定

13.2 tool_call / tool_result 配对必须完整

13.3 session persistence 不是“最后一次统一保存”，而是多层增量保存

14. 一句话总结

Hermes Agent 06. 技能、记忆与上下文文件

06. 技能、记忆与上下文文件

本章目标

适用读者

前置条件

核心概念

Skill 是“怎么做”

Memory 是“是什么”

Context File 是“默认注入的背景”

操作步骤

查看和使用 skill

自己创建 skill

使用 memory

写好 AGENTS.md

区分 SOUL.md 和 AGENTS.md

常见场景

场景 1：重复做同一种工作流

3.3 `auth.py` 的职责

5.2 `prompt_builder.py` 提供的核心能力

7. `run_conversation()` 的同步 agent loop

7.3 `_build_api_kwargs()` 如何分发到不同 API 形态

7.5 `chat()` 只是外层薄封装

8.2 `get_tool_definitions()` 如何做 availability gating

8.6 `handle_function_call()` 做什么

9.3 `_compress_context()` 的动作

9.4 `ContextCompressor.compress()` 的内部算法

10.1 `SessionDB`：结构化、可检索的长期存储

10.2 `AIAgent` 如何写入 `SessionDB`

10.3 `SessionStore`：gateway 的会话路由与 transcript 兼容层

10.4 `SessionDB` 与 `SessionStore` 的关系

第一段：`_run_agent()` 负责生成 agent 结果

写好 `AGENTS.md`

区分 `SOUL.md` 和 `AGENTS.md`

为什么不建议把 `AGENTS.md` 写得特别长

`browser` 和 `web` 为什么要分开

`config.yaml` 与 `.env` 的分工