Hermes Agent 源码解析：全栈 AI Agent 框架的极致设计

深入剖析 NousResearch/hermes-agent——一个集成了 CLI、TUI、消息网关、定时任务、MCP 工具协议、插件系统于一体的全栈 AI Agent 框架

项目简介

Hermes Agent 是由 NousResearch 开发的全栈 AI Agent 框架，定位为”多平台、多模型、可扩展的 AI Agent 工具”。

核心能力：

🤖 多模型支持：OpenAI、Anthropic、Gemini、xAI、DeepSeek、通义千问、Ollama、LM Studio 等 30+ 提供商
🌐 多平台接入：CLI、TUI、Telegram、Discord、Slack、WhatsApp、微信、飞书、IRC 等 20+ 消息平台
🔧 丰富工具系统：终端执行、文件操作、浏览器控制、代码生成、MCP 工具、图片/视频/语音生成
🧩 插件化架构：记忆提供者、上下文引擎、平台适配器、技能包均可通过插件扩展
🏢 企业级特性：凭据池轮换、出站防火墙（iron-proxy）、Bitwarden/1Password 集成、审批工作流、会话检查点

技术栈：Python（后端核心）+ TypeScript/React（桌面端/Web 仪表盘）

整体架构

┌───────────────────────────────────────────────────────────────┐
│                      用户界面层 (UI Layer)                      │
│  ┌──────────┐  ┌──────────┐  ┌──────────┐  ┌──────────────┐  │
│  │ CLI REPL │  │   TUI    │  │ Desktop  │  │ Web Dashboard│  │
│  │ (cli.py) │  │(ui-tui/) │  │(Electron)│  │  (Next.js)   │  │
│  └─────┬────┘  └─────┬────┘  └─────┬────┘  └──────┬───────┘  │
│        │             │             │               │           │
├────────┴─────────────┴─────────────┴───────────────┴───────────┤
│                   网关层 (Gateway Layer)                         │
│  ┌────────────────────────────────────────────────────────┐    │
│  │              GatewayRunner (gateway/run.py)             │    │
│  │  ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐  │    │
│  │  │ Telegram │ │ Discord  │ │  Slack   │ │ WhatsApp │  │    │
│  │  │ Adapter  │ │ Adapter  │ │ Adapter  │ │ Adapter  │  │    │
│  │  └──────────┘ └──────────┘ └──────────┘ └──────────┘  │    │
│  └────────────────────────────────────────────────────────┘    │
├───────────────────────────────────────────────────────────────┤
│                   Agent 引擎层 (Agent Engine)                   │
│  ┌────────────────────────────────────────────────────────┐    │
│  │              AIAgent (run_agent.py)                     │    │
│  │  ┌──────────────────────────────────────────────┐      │    │
│  │  │     run_conversation() — 对话主循环            │      │    │
│  │  │  • 系统提示构建  • API 调用  • 工具分发        │      │    │
│  │  │  • 流式输出     • 错误重试  • 上下文压缩       │      │    │
│  │  └──────────────────────────────────────────────┘      │    │
│  └────────────────────────────────────────────────────────┘    │
├───────────────────────────────────────────────────────────────┤
│                   工具与协议层 (Tools & Protocols)               │
│  ┌──────────┐  ┌──────────┐  ┌──────────┐  ┌──────────┐      │
│  │  Tool    │  │   MCP    │  │  Skills  │  │  ACP     │      │
│  │ Registry │  │  Client  │  │   Hub    │  │ Adapter  │      │
│  └──────────┘  └──────────┘  └──────────┘  └──────────┘      │
├───────────────────────────────────────────────────────────────┤
│                   提供商层 (Provider Layer)                      │
│  ┌──────────┐  ┌──────────┐  ┌──────────┐  ┌──────────┐      │
│  │ OpenAI   │  │Anthropic │  │ Gemini   │  │  xAI /   │      │
│  │Transport │  │Transport │  │Transport │  │ DeepSeek │      │
│  └──────────┘  └──────────┘  └──────────┘  └──────────┘      │
├───────────────────────────────────────────────────────────────┤
│                   基础设施层 (Infrastructure)                    │
│  ┌──────────┐  ┌──────────┐  ┌──────────┐  ┌──────────┐      │
│  │ Config   │  │Credential│  │ Session  │  │  Memory  │      │
│  │ (YAML)   │  │  Pool    │  │   DB     │  │ Provider │      │
│  └──────────┘  └──────────┘  └──────────┘  └──────────┘      │
└───────────────────────────────────────────────────────────────┘

启动流程

主入口 `hermes_cli/main.py::main()`

main()
 ├── _set_process_title()               # 修改进程名（ps 显示为 hermes）
 ├── configure_windows_stdio()          # Windows UTF-8 stdio
 ├── _cleanup_quarantined_exes()        # 清理旧版本隔离文件
 ├── _sweep_stale_bytecode_if_checkout_changed()
 ├── _recover_from_interrupted_install()
 │
 ├── _try_termux_fast_tui_launch()      # Termux TUI 快速启动路径
 ├── _try_termux_fast_cli_launch()      # Termux CLI 快速启动路径
 │
 └── build_top_level_parser()           # 构建 argparse 解析器
     ├── chat (默认) → cmd_chat()
     ├── gateway → cmd_gateway()
     ├── setup → cmd_setup()
     ├── model → cmd_model()
     ├── moa → cmd_moa()              # Mixture of Agents
     ├── fallback → cmd_fallback()
     ├── secrets → _dispatch_secrets()
     ├── egress → _dispatch_egress()
     ├── migrate → cmd_migrate()
     ├── memory → cmd_memory()
     ├── tools → cmd_tools()
     ├── skills → cmd_skills()
     ├── mcp → cmd_mcp()
     ├── acp → cmd_acp()
     ├── plugins → cmd_plugins()
     └── ... (更多子命令)

聊天启动路径 `cmd_chat()`

cmd_chat(args)
 ├── _resolve_use_tui(args)             # 决定使用 TUI 还是 CLI
 │   └── 优先级: --cli > --tui > 非 TTY > HERMES_TUI > config > 默认 CLI
 ├── _apply_safe_mode(args)             # 安全模式处理
 ├── 会话恢复 (--continue / --resume)
 ├── _has_any_provider_configured()     # 首次运行检查
 │   └── 若无配置 → 提示运行 hermes setup
 ├── _sync_bundled_skills_for_startup() # 同步内置技能
 │
 └── if use_tui:
     └── _launch_tui()                  # 启动 TUI（Node.js/React Ink）
     else:
     └── from cli import main as cli_main
         cli_main(**kwargs)             # 启动经典 CLI REPL

Agent 引擎

AIAgent 类

AIAgent 是整个框架的核心类，封装了 LLM 交互的完整生命周期：

class AIAgent:
    def __init__(self, ...):
        """
        关键参数:
        - base_url, api_key, provider, model: 模型连接信息
        - max_iterations: 工具调用最大迭代次数（默认 90）
        - enabled_toolsets / disabled_toolsets: 工具集开关
        - ephemeral_system_prompt: 自定义系统提示
        - 各种 callback: tool_progress, stream_delta, thinking, clarify…
        - session_id, platform, user_id: 会话与平台标识
        - credential_pool: 凭据池（多 key 轮换）
        - iteration_budget: 迭代预算（跨 turn 共享）
        - fallback_model: 降级模型配置
        """
        from agent.agent_init import init_agent
        init_agent(self, ...)   # 真正的初始化逻辑在 agent_init.py

初始化流程（agent/agent_init.py::init_agent()）：

解析提供商：provider → models.dev 映射 → 确定 base_url、api_key
选择传输层：根据 provider/api_mode 创建对应的 ProviderTransport
构建系统提示：prompt_builder.py 组装（规则文件、记忆、技能…）
注册工具：从 tools/registry.py 获取可用工具，按 toolsets 过滤
初始化记忆：加载 MemoryProvider（内置或插件如 Mem0、Honcho）
初始化上下文引擎：创建 ContextCompressor 或插件引擎
加载凭据池：CredentialPool 管理多个 API key 的轮换与刷新
会话持久化：创建/恢复 SessionDB 记录

对话循环（Conversation Loop）

对话循环是 Agent 的”心脏”，定义在 agent/conversation_loop.py::run_conversation()：

run_conversation(agent, user_message, ...)
 │
 ├── 1. MoA 解码（decode_moa_turn）
 │     └── 如果用户消息触发"混合代理"模式，配置多模型协作
 │
 ├── 2. 构建轮次上下文 (build_turn_context)
 │     ├── 标准 I/O 守卫
 │     ├── 用户消息清洗（surrogate 字符修复）
 │     ├── 系统提示恢复/构建
 │     ├── 预压缩检查（preflight compression）
 │     ├── 插件 pre_llm_call 钩子
 │     └── 外部记忆预取（ext_prefetch）
 │
 ├── 3. 主循环 (while api_call_count < max_iterations)
 │     │
 │     ├── 检查中断 (_interrupt_requested)
 │     ├── 消耗迭代预算 (iteration_budget.consume)
 │     ├── 触发 step_callback（网关钩子）
 │     │
 │     ├── 排空待处理的转向请求 (drain_pending_redirect)
 │     ├── 排空待处理的引导指令 (drain_pending_steer)
 │     │
 │     ├── 工具调用参数清洗 (sanitize_tool_call_arguments)
 │     ├── 消息序列修复 (repair_message_sequence)
 │     │
 │     ├── 构建 API 消息 (api_messages)
 │     │   ├── 去除内部字段 (api_content, display_kind…)
 │     │   └── 提供商消息转换 (convert_messages)
 │     │
 │     ├── 执行 API 调用 (_perform_api_call)
 │     │   ├── 选择传输层 (_get_transport)
 │     │   ├── 构建请求参数 (build_api_kwargs)
 │     │   ├── 发送请求（流式/非流式）
 │     │   ├── 处理响应（tool_calls, content, reasoning）
 │     │   └── 错误处理与重试
 │     │
 │     ├── 处理工具调用
 │     │   ├── 解析 tool_calls
 │     │   ├── 执行工具 (execute_tool)
 │     │   ├── 记录工具结果到消息列表
 │     │   └── 更新 turn summary
 │     │
 │     ├── 处理最终响应
 │     │   ├── 内容验证（verification gate）
 │     │   ├── 压缩检查（是否需要上下文压缩）
 │     │   └── 降级检查（fallback_model）
 │     │
 │     └── 循环条件判断
 │         ├── 有 tool_calls → 继续循环
 │         ├── 无 tool_calls → 退出（模型给出最终回答）
 │         ├── 达到 max_iterations → 退出
 │         └── 被中断 → 退出
 │
 ├── 4. 轮次收尾
 │     ├── 会话持久化 (_persist_session)
 │     ├── 记忆同步 (memory.sync_turn)
 │     ├── 上下文引擎 on_turn_complete
 │     └── 使用量统计
 │
 └── 返回结果字典
     {
       "final_response": str,
       "messages": list,
       "api_calls": int,
       "completed": bool,
       "tokens_used": int,
       ...
     }

关键机制

迭代预算（IterationBudget）：控制 Agent 在单轮/多轮中的最大工具调用次数，防止无限循环
错误分类器（error_classifier.py）：对 API 错误分类（429 限流、401 认证失败、500 服务器错误…），决定重试/降级/报错策略
上下文压缩：当消息长度接近模型上下文窗口时，自动触发压缩（摘要旧消息）
降级链（fallback_model）：主模型失败时，按配置链尝试备选模型
MoA（Mixture of Agents）：支持多模型协作——多个”顾问”模型各自分析，再由主模型综合

工具系统

工具注册表

class ToolRegistry:
    """全局单例——所有工具在此注册"""

    def register(self, name, function, description, parameters, toolset=None):
        """注册一个工具"""

    def deregister(self, name):
        """注销工具"""

    def get_tool(self, name) -> ToolDefinition:
        """按名获取工具"""

    def get_tools_for_api(self, tool_names=None) -> list:
        """获取 API 格式的工具定义（OpenAI function calling 格式）"""

    def execute(self, name, arguments) -> str:
        """执行工具并返回结果"""

# 模块级单例
registry = ToolRegistry()

内置工具集

工具集名	包含工具	说明
`terminal`	`terminal`, `close_terminal`	终端命令执行
`filesystem`	`read_file`, `write_file`, `edit_file`, `list_directory`…	文件操作
`browser`	`browser_navigate`, `browser_click`, `browser_screenshot`…	浏览器自动化
`code_execution`	`execute_code`	代码沙箱执行
`delegate`	`delegate_task`	子 Agent 委派
`mcp`	(动态发现)	MCP 协议工具
`image_gen`	`generate_image`	图片生成
`tts`	`text_to_speech`	文本转语音
`vision`	`vision_analyze`	图像分析
`send_message`	`send_message`	跨平台消息发送
`cron`	`cron_manage`	定时任务管理
`computer_use`	`computer_use_*`	CUA 计算机使用
`checkpoint`	`checkpoint_*`	会话检查点

工具审批机制

工具执行前经过审批门控：

工具调用请求
 │
 ├── 安全模式检查 (--yolo 跳过)
 ├── 工具白名单检查
 ├── 写操作审批 (write_approval)
 │   ├── 内存写操作 → memory.write_approval
 │   └── 技能写操作 → skills.write_approval
 ├── 危险命令审批 (terminal 工具)
 │   ├── 需要 sudo → 密码回调
 │   └── 危险操作 → 用户确认
 └── 通过 → 执行工具

MCP 协议集成

MCPServerTask

每个 MCP 服务器由一个 MCPServerTask 实例管理，运行在独立的 asyncio Task 中：

class MCPServerTask:
    """管理单个 MCP 服务器连接"""

    # 支持的传输方式
    # - stdio: 通过子进程的标准输入/输出通信
    # - HTTP/StreamableHTTP: 通过 HTTP 端点通信

    async def _run_stdio(self):
        """stdio 传输的生命周期"""
        # 1. 启动子进程
        # 2. 建立 JSON-RPC 流
        # 3. session.initialize() 握手
        # 4. 工具发现 (tools/list)
        # 5. 持续服务（通知处理、工具调用）
        # 6. 空闲超时回收 / 最大生命周期回收

    async def _run_http(self):
        """HTTP 传输的生命周期"""
        # 1. 建立 HTTP 连接
        # 2. OAuth 认证（如配置）
        # 3. session.initialize()
        # 4. 工具发现与服务

关键特性：

动态工具发现：监听 notifications/tools/list_changed
连接保活：ping / list_tools 探测
自动重连：认证失败后刷新凭据并重连
空闲回收：idle_timeout_seconds / max_lifetime_seconds
快速降级：连续失败后”park”服务器
RPC 串行：_rpc_lock 防止并发 JSON-RPC 冲突

工具注册流程

MCP 服务器连接成功
 │
 ├── _register_server_tools()
 │   ├── 遍历服务器通告的工具
 │   ├── 名称规范化（mcp_prefixed_tool_name）
 │   ├── 冲突检测（同名工具来自不同服务器）
 │   └── 注册到全局 ToolRegistry
 │
 ├── 动态更新: tools/list_changed 通知
 │   ├── _schedule_tools_refresh()
 │   └── _refresh_tools() — 增量更新注册表
 │
 └── 调用时:
     ├── agent 发出 tool_call
     ├── registry.execute() 路由到 MCP 处理器
     ├── MCPServerTask.call_tool() 发送 JSON-RPC
     └── 返回结果给 agent

Provider / Transport 层

传输层架构

class ProviderTransport(ABC):
    """提供商传输层抽象基类"""

    @abstractmethod
    def convert_messages(self, messages) -> list:
        """将内部消息格式转换为提供商格式"""

    @abstractmethod
    def convert_tools(self, tools) -> list:
        """将工具定义转换为提供商格式"""

    @abstractmethod
    def build_api_kwargs(self, ...) -> dict:
        """构建 API 请求参数"""

    @abstractmethod
    def normalize_response(self, response) -> dict:
        """将提供商响应标准化为内部格式"""

具体实现：

ChatCompletionsTransport：OpenAI 兼容格式（最通用）
AnthropicTransport：Anthropic Messages API
ResponsesApiTransport：OpenAI Responses API（Codex）
BedrockTransport：AWS Bedrock

支持的提供商

提供商	传输层	说明
OpenAI	ChatCompletions	GPT 系列
Anthropic	AnthropicTransport	Claude 系列
Google Gemini	ChatCompletions	Gemini 系列
xAI	ChatCompletions	Grok 系列
DeepSeek	ChatCompletions	DeepSeek 系列
通义千问 (Alibaba)	ChatCompletions	Qwen 系列
Ollama	ChatCompletions	本地模型
LM Studio	ChatCompletions	本地模型
OpenRouter	ChatCompletions	多模型路由
AWS Bedrock	BedrockTransport	企业级 AWS
Azure OpenAI	ChatCompletions	Azure 部署
Copilot	ChatCompletions	GitHub Copilot
Fireworks	ChatCompletions	快速推理
Groq	ChatCompletions	快速推理
Mistral	ChatCompletions	Mistral 系列
Together AI	ChatCompletions	开源模型
… 等 30+

Gateway 消息网关

GatewayRunner

class GatewayRunner:
    """管理所有平台适配器的生命周期"""

    def start_gateway(self):
        """启动网关"""
        # 1. 加载网关配置 (load_gateway_config)
        # 2. 发现并注册平台适配器 (PlatformRegistry)
        # 3. 对每个已配置的平台:
        #    a. 创建适配器实例 (create_adapter)
        #    b. 调用 adapter.connect()
        #    c. 启动消息处理循环
        # 4. 启动定时任务调度器 (cron.scheduler)
        # 5. 启动健康监控
        # 6. 等待关闭信号

    # Agent 缓存管理:
    # - _AGENT_CACHE_MAX_SIZE = 128
    # - _AGENT_CACHE_IDLE_TTL_SECS = 3600 (1小时)
    # - LRU 淘汰 + 空闲超时

网关消息处理流

平台收到用户消息
 │
 ├── 平台适配器 (e.g. TelegramAdapter)
 │   ├── 验证用户权限 (_is_user_authorized)
 │   ├── 解析消息内容（文本、图片、文件、语音）
 │   ├── 查找/创建会话 (session_key)
 │   └── 转发到消息处理管线
 │
 ├── GatewayRunner 消息管线
 │   ├── 获取/创建 AIAgent 实例 (agent cache)
 │   ├── 构建平台上下文 (platform_hint, user_id…)
 │   ├── 调用 run_conversation()
 │   │   └── 流式回调 → 分段发送给用户
 │   └── 持久化会话
 │
 └── 响应返回
     ├── 文本分段发送 (max_message_length 控制)
     ├── 文件/图片发送
     └── 工具结果格式化

平台适配器

BasePlatformAdapter

所有平台适配器的基类，定义了统一的接口：

class BasePlatformAdapter:
    """平台适配器抽象基类"""

    name: str               # 平台标识
    config: PlatformConfig  # 平台配置

    async def connect(self, *, is_reconnect=False) -> bool:
        """连接到平台"""

    async def disconnect(self):
        """断开连接"""

    async def send_message(self, chat_id, text, ...):
        """发送消息"""

    async def _handle_message(self, ...):
        """处理收到的消息"""

    # 通用能力:
    # - 自动重连（指数退避）
    # - 消息分段（超过平台长度限制时自动拆分）
    # - 用户授权检查
    # - 媒体处理（图片、文件、语音）
    # - 线程/话题支持
    # - 打字状态指示器
    # - 会话管理

支持的平台

平台	适配器	来源
Telegram	TelegramAdapter	内置
Discord	DiscordAdapter	内置/插件
Slack	SlackAdapter	内置
WhatsApp	WhatsAppAdapter / WhatsAppCloudAdapter	内置
IRC	IRCAdapter	内置
飞书	FeishuAdapter	插件
微信	WeixinAdapter	插件
Microsoft Teams	MSGraphWebhookAdapter	插件
Signal	SignalAdapter	插件
QQ	QQAdapter	插件
BlueBubbles	BlueBubblesAdapter	插件
Webhook	WebhookAdapter	内置
REST API	APIServerAdapter	内置

记忆系统

MemoryProvider 接口

class MemoryProvider(ABC):
    """记忆提供者抽象基类"""

    @property
    @abstractmethod
    def name(self) -> str: ...

    @abstractmethod
    def is_available(self) -> bool: ...

    @abstractmethod
    def initialize(self, session_id: str, **kwargs) -> None: ...

    @abstractmethod
    def get_tool_schemas(self) -> List[Dict]: ...

    # 核心生命周期:
    def system_prompt_block(self) -> str: ...      # 系统提示注入
    def prefetch(self, query: str) -> str: ...     # 轮次前预取
    def sync_turn(self, user, assistant, ...) -> None: ...  # 轮次后同步

    # 可选钩子:
    def on_turn_start(self, ...): ...
    def on_session_end(self, messages): ...
    def on_session_switch(self, new_session_id, ...): ...
    def on_pre_compress(self, messages) -> str: ...
    def on_delegation(self, task, result, ...): ...
    def on_memory_write(self, action, target, content, ...): ...

内置与插件实现

提供者	位置	说明
builtin	内置	MEMORY.md + USER.md 文件存储
Mem0	plugins/memory/mem0/	向量记忆
Honcho	plugins/memory/honcho/	用户建模
Hindsight	plugins/memory/hindsight/	回顾式记忆
Holographic	plugins/memory/holographic/	全息记忆
Supermemory	plugins/memory/supermemory/	多容器记忆
ByteRover	plugins/memory/byterover/	字节级记忆
OpenViking	plugins/memory/openviking/	维京式记忆
RetainDB	plugins/memory/retaindb/	数据库记忆
… (17 种)

上下文引擎与压缩

ContextCompressor

内置的上下文压缩器，当对话接近上下文窗口限制时自动压缩：

class ContextCompressor(ContextEngine):
    """当对话接近上下文窗口限制时自动压缩"""

    # 压缩策略:
    # 1. 预飞行压缩 (preflight): API 调用前检查压力
    # 2. 溢出压缩 (overflow): 413 响应后压缩
    # 3. 工具后压缩 (post-tool): 工具执行后检查
    # 4. 空闲压缩 (idle): 空闲时主动压缩

    # 压缩方式:
    # - 微压缩 (micro-compaction): 保留关键信息
    # - 全压缩: 用 LLM 摘要旧消息
    # - 就地压缩: 原地替换不新建消息列表

认证与凭据管理

凭据池

class CredentialPool:
    """管理多个 API key 的轮换"""

    # 特性:
    # - 多 key 轮换: 一个 provider 可配置多个 key
    # - 自动刷新: OAuth token 过期时自动刷新
    # - 401 恢复: 收到 401 后尝试刷新当前 token
    # - 刷新计数: 防止单 key 无限刷新循环
    # - 来源: 环境变量、配置文件、OAuth、外部密钥管理器

密钥源

class SecretSource(ABC):
    @abstractmethod
    def get_secret(self, key: str) -> Optional[str]: ...

# 实现:
# - EnvSecretSource: 从环境变量读取
# - BitwardenSecretSource: Bitwarden Secrets Manager
# - OnePasswordSecretSource: 1Password (op:// 引用)

定时任务（Cron）

CronJob 数据模型

interface CronJob {
  name: string           # 任务名称
  prompt: string         # Agent 提示词
  schedule: string       # cron 表达式
  deliver: string        # 投递目标 ("local" | 平台名)
  skills: string[]       # 加载的技能
  provider?: string      # 指定提供商
  model?: string         # 指定模型
  script?: string        # 纯脚本模式
  no_agent?: boolean     # 跳过 Agent（仅执行脚本）
  context_from?: string[] # 从其他任务继承上下文
  enabled_toolsets?: string[]  # 启用的工具集
  workdir?: string       # 工作目录
}

执行流程

调度器触发
 │
 ├── 加载 CronJob 配置
 ├── 创建 AIAgent 实例
 │   ├── 应用 provider/model 覆盖
 │   ├── 加载指定 skills
 │   └── 设置 enabled_toolsets
 │
 ├── 执行 (二选一)
 │   ├── no_agent=true → 直接运行 script，返回 stdout
 │   └── no_agent=false → agent.run_conversation(prompt)
 │
 └── 投递结果
     ├── deliver="local" → 写入本地日志
     └── deliver="telegram" → 通过平台适配器发送

安全与审批

安全层级

┌─────────────────────────────────┐
│         出站防火墙 (Egress)       │  iron-proxy: TLS 拦截，凭据注入
├─────────────────────────────────┤
│         工具审批 (Approval)       │  危险命令需用户确认
├─────────────────────────────────┤
│         写操作审批 (Write)        │  内存/技能写操作需审批
├─────────────────────────────────┤
│         安全模式 (Safe Mode)      │  --safe: 忽略用户配置和规则
├─────────────────────────────────┤
│         YOLO 模式                │  --yolo: 跳过所有审批
├─────────────────────────────────┤
│         凭据保护                  │  密钥不写入日志，.env 权限 0600
├─────────────────────────────────┤
│         配置管理 (Managed)        │  管理员锁定特定配置键
└─────────────────────────────────┘

配置系统

配置文件结构

~/.hermes/
├── config.yaml          # 主配置文件
├── .env                 # 环境变量（API keys）
├── memories/
│   ├── MEMORY.md        # Agent 记忆
│   └── USER.md          # 用户画像
├── sessions.db          # 会话数据库（SQLite）
├── logs/
│   ├── agent.log        # Agent 日志
│   └── errors.log       # 错误日志
├── skills/              # 已安装技能
├── plugins/             # 已安装插件
└── state/               # 状态文件

配置键结构

# 模型配置
model: "gpt-4"
provider: "openai"
base_url: ""

# Agent 行为
agent:
  max_turns: 90
  reasoning: {}

# 显示
display:
  interface: "cli"  # "cli" | "tui"
  tool_preview_length: 0
  friendly_tool_labels: true
  turn_summary: true

# 安全
security:
  redact_secrets: true

# 降级
fallback_model:
  - provider: "anthropic"
    model: "claude-sonnet-4-20250514"

# 压缩
compression:
  max_attempts: 3

# 记忆
memory:
  provider: ""  # 内置 | mem0 | honcho | hindsight…

# MCP 服务器
mcp:
  servers:
    my_server:
      command: "npx"
      args: ["-y", "@my/mcp-server"]

# 工具集
toolsets:
  enabled: []
  disabled: []

# 监控
monitoring:
  gateway_health_export:
    enabled: true
  export:
    otlp:
      enabled: false
      endpoint: ""

设计模式总结

模式	应用
策略模式	ProviderTransport 多实现、MemoryProvider 多实现
工厂模式	PlatformRegistry.create_adapter()、SkillSource 适配器
观察者模式	callback 系统（tool_progress, stream_delta, step_callback…）
单例模式	ToolRegistry (registry)、PlatformRegistry
模板方法	BasePlatformAdapter 定义框架，子类实现细节
延迟加载	PlatformRegistry 延迟导入、AIAgent 延迟实例化
注册表模式	工具注册、平台注册、技能源注册、密钥源注册
适配器模式	Transport 层将不同 API 格式统一为内部格式
管道模式	消息处理管线（预处理 → Agent → 后处理 → 输出）
命令模式	斜杠命令系统（SlashCommandCompleter）

关键工程实践

延迟导入 (Lazy Import)：大量使用函数内 import，避免启动时导入所有模块
原子写入：配置文件和数据库使用 atomic_yaml_write 防止写入中断导致损坏
线程安全：_CONFIG_LOCK (RLock)、_rpc_lock (asyncio.Lock)
优雅降级：大量 try/except + 日志，确保非关键功能失败不影响主流程
游标优化：repair_message_sequence_with_cursor、sanitize_tool_call_arguments 使用游标避免重复处理
Termux 快速启动：专门的快速路径避免完整解析器的开销
凭据安全：密钥不入日志、.env 权限 0600、密钥在 UI 中脱敏显示
配置默认值剥离：save_config 不写入与默认值相同的配置，保持配置文件简洁

总结

Hermes Agent 是一个工程规模庞大但架构清晰的全栈 AI Agent 框架，几个亮点：

真正的多平台：20+ 消息平台适配器，一套 Agent 逻辑接入所有平台
插件化到极致：记忆、上下文引擎、平台、技能都可插拔
企业级安全：凭据池、出站防火墙、Bitwarden/1Password 集成、审批工作流
MoA 多模型协作：多个”顾问”模型各自分析，主模型综合
上下文压缩：多种策略自动压缩，突破上下文窗口限制
定时任务：Agent 可以作为 cron 任务定期执行

如果你想构建一个生产级的、可接入多平台的 AI Agent 系统，Hermes Agent 的源码绝对值得深入研究。

项目地址：https://github.com/NousResearch/hermes-agent
文档：项目内 docs/ 目录

AI > 源码解析

#AI Agent #Python #开源 #Hermes Agent #MCP

Hermes Agent 源码解析：全栈 AI Agent 框架的极致设计

https://tingfeng347.github.io/2026/08/02/Hermes Agent 源码解析：全栈 AI Agent 框架的极致设计/

作者

Tingfeng

发布于

2026年8月2日

许可协议

Pi Agent Harness 源码解析：TypeScript 实现的模块化 Coding Agent 上一篇

OpenAI Codex CLI 源码解析：Rust 驱动的本地编程代理下一篇

Hermes Agent 源码解析：全栈 AI Agent 框架的极致设计

Hermes Agent 源码解析：全栈 AI Agent 框架的极致设计

项目简介

整体架构

启动流程

主入口 hermes_cli/main.py::main()

聊天启动路径 cmd_chat()

Agent 引擎

AIAgent 类

对话循环（Conversation Loop）

关键机制

工具系统

工具注册表

内置工具集

工具审批机制

MCP 协议集成

MCPServerTask

工具注册流程

Provider / Transport 层

传输层架构

支持的提供商

Gateway 消息网关

GatewayRunner

网关消息处理流

平台适配器

BasePlatformAdapter

支持的平台

记忆系统

MemoryProvider 接口

内置与插件实现

上下文引擎与压缩

ContextCompressor

认证与凭据管理

凭据池

密钥源

定时任务（Cron）

CronJob 数据模型

执行流程

安全与审批

安全层级

配置系统

配置文件结构

配置键结构

设计模式总结

关键工程实践

总结

主入口 `hermes_cli/main.py::main()`

聊天启动路径 `cmd_chat()`