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 APIResponsesApiTransport：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 | 内置 |
| WhatsAppAdapter / WhatsAppCloudAdapter | 内置 | |
| IRC | IRCAdapter | 内置 |
| 飞书 | FeishuAdapter | 插件 |
| 微信 | WeixinAdapter | 插件 |
| Microsoft Teams | MSGraphWebhookAdapter | 插件 |
| Signal | SignalAdapter | 插件 |
| 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
Agent 垂直技术社区，欢迎活跃、内容共建。
更多推荐
3
0- 0
已为社区贡献3条内容
所有评论(0)