🧠 Hermes Agent 源码分析

第一章 入门指南

1.1 什么是 Hermes Agent？

Hermes Agent 是由 Nous Research 构建的自改进 AI Agent 框架，具备内置学习循环能力。它不仅仅是一个聊天机器人，而是一个完整的 AI 开发环境，支持多轮对话、工具调用、技能学习、记忆管理等高级功能。

关键特性

版本信息

• 当前版本：v0.8.0
• Python 要求：≥ 3.11
• 许可证：MIT
• Stars：50.9k
• Forks：6.6k
• Commits：3,733
• Branches：730

1.2 快速安装

方式一：pip 安装（推荐）

# 基本安装
pip install hermes-ai

# 完整安装（包含所有依赖）
pip install "hermes-ai[all]"

# 检查安装
hermes --version

方式二：从源码安装

# 克隆仓库
git clone https://github.com/NousResearch/hermes-agent.git
cd hermes-agent

# 安装依赖
pip install -e .

# 或使用 poetry
poetry install

方式三：使用安装脚本

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

依赖环境

1.3 基本用法

交互式模式

# 启动交互式 REPL
hermes

# 指定模型
hermes --model anthropic/claude-sonnet-4

# 使用特定配置
hermes --config ~/.hermes/prod-config.yaml

单次查询模式

# 执行单次查询后退出
hermes -q "解释量子计算的基本原理"

# 复杂查询
hermes --query "分析 /var/log/syslog 中的错误模式"

子代理模式

# 作为子代理运行（用于 delegation）
hermes-agent --subagent --model openai/gpt-4o

# 批量处理
hermes --batch prompts.txt --output results/

Gateway 模式

# 启动 Telegram Bot
hermes gateway --platform telegram

# 启动 Discord Bot
hermes gateway --platform discord

# 启动所有配置的平台
hermes gateway

1.4 环境配置

创建配置文件

# 创建默认配置
hermes setup

# 交互式配置向导
hermes setup --interactive

# 查看当前配置
hermes config show

配置示例

# ~/.hermes/config.yaml

model:
  default: "anthropic/claude-sonnet-4-5"
  provider: "openrouter"
  base_url: "https://openrouter.ai/api/v1"

terminal:
  env_type: "local"  # local/docker/singularity/modal/daytona
  timeout: 120

browser:
  provider: "browserbase"  # browserbase/browseruse/camofox
  inactivity_timeout: 300

agent:
  max_turns: 90
  personality: "kawaii"

display:
  skin: "default"
  compact: false
  streaming: true

环境变量

# ~/.hermes/.env

# API Keys
OPENAI_API_KEY=sk-...
ANTHROPIC_API_KEY=sk-ant-...
OPENROUTER_API_KEY=sk-or-...

# Provider Tokens
NOTION_TOKEN=secret_...
TELEGRAM_BOT_TOKEN=123:abc
DISCORD_BOT_TOKEN=...

# Feature Flags
HERMES_HUMAN_DELAY_MODE=true
HERMES_RECORD_SESSIONS=false

第二章 系统架构

2.1 整体架构

Hermes Agent 采用分层模块化架构，核心由以下几个层次构成：

架构层次

2.2 目录结构

hermes-agent/
├── agent/                      # Agent 核心逻辑
│   ├── __init__.py
│   ├── prompt_builder.py       # 系统提示构建
│   ├── context_compressor.py    # 上下文压缩
│   ├── memory_manager.py        # 记忆管理
│   ├── credential_pool.py       # 凭证池
│   ├── display.py               # TUI 显示
│   ├── error_classifier.py      # 错误分类
│   ├── insights.py              # 使用分析
│   ├── skill_commands.py        # 技能命令
│   ├── skill_utils.py          # 技能工具
│   └── trajectory.py            # 轨迹保存
│
├── tools/                       # 工具系统 (50+ 工具)
│   ├── __init__.py
│   ├── registry.py              # 中心注册表
│   ├── model_tools.py           # 工具编排
│   ├── toolsets.py              # 工具集定义
│   │
│   ├── terminal_tool.py         # 终端执行
│   ├── browser_tool.py          # 浏览器自动化
│   ├── file_tools.py            # 文件操作
│   ├── web_tools.py             # 网页搜索
│   ├── delegate_tool.py         # 子代理委托
│   ├── code_execution_tool.py   # 代码执行
│   ├── skills_tool.py           # 技能管理
│   ├── skill_manager_tool.py    # 技能 CRUD
│   ├── memory_tool.py           # 记忆管理
│   ├── cronjob_tools.py         # 定时任务
│   ├── todo_tool.py             # TODO 列表
│   ├── mcp_tool.py              # MCP 客户端
│   ├── rl_training_tool.py      # RL 训练
│   └── environments/            # 执行后端
│       ├── __init__.py
│       ├── local.py
│       ├── docker.py
│       ├── ssh.py
│       ├── modal.py
│       ├── singularity.py
│       └── daytona.py
│
├── skills/                      # 技能系统 (28 分类)
│   ├── software-development/
│   ├── mlops/
│   ├── github/
│   ├── media/
│   └── ... (共 28 个分类)
│
├── gateway/                     # 消息网关
│   ├── run.py                   # 网关主循环
│   ├── session.py               # 会话管理
│   ├── config.py                # 配置
│   ├── delivery.py              # 消息投递
│   ├── pairing.py               # 配对管理
│   ├── channel_directory.py     # 频道目录
│   └── platforms/              # 平台适配器
│       ├── base.py              # 基类
│       ├── telegram.py
│       ├── discord.py
│       ├── slack.py
│       ├── whatsapp.py
│       ├── feishu.py
│       ├── weixin.py
│       ├── wecom.py
│       ├── matrix.py
│       └── ... (15+ 平台)
│
├── hermes_cli/                  # CLI 系统
│   ├── main.py                  # 主入口
│   ├── config.py                # 配置管理
│   ├── setup.py                 # 设置向导
│   ├── commands.py              # Slash 命令
│   ├── auth.py                  # 认证
│   ├── models.py                # 模型管理
│   ├── tools_config.py          # 工具配置
│   ├── skin_engine.py           # 皮肤系统
│   ├── profiles.py              # 多配置
│   ├── banner.py                # 横幅
│   ├── mcp_config.py            # MCP 配置
│   └── ...
│
├── environments/                # RL 研究环境
│   ├── hermes_base_env.py       # 基础环境 (51KB)
│   ├── agent_loop.py            # Agent 循环
│   ├── tool_context.py          # 工具上下文
│   └── benchmarks/              # 基准测试
│
├── plugins/                      # 插件系统
│   └── memory/                  # 记忆插件
│       ├── honcho/
│       ├── mem0/
│       ├── holographic/
│       └── ... (8+ providers)
│
├── tests/                       # 测试套件 (~3000 tests)
├── docs/                        # 文档
├── scripts/                     # 安装脚本
│
├── run_agent.py                 # Agent 主入口 (502KB, 9911行)
├── cli.py                       # CLI 主入口 (412KB, 9268行)
├── gateway/run.py               # Gateway 主入口 (365KB)
├── batch_runner.py              # 批量处理
├── trajectory_compressor.py     # 轨迹压缩
├── rl_cli.py                    # RL CLI
├── mcp_serve.py                 # MCP 服务器
├── mini_swe_runner.py           # SWE-bench
└── pyproject.toml               # 项目配置

2.3 核心文件详解

2.4 数据流图

┌─────────────────────────────────────────────────────────────────────────┐
│                           Hermes Agent 数据流                               │
└─────────────────────────────────────────────────────────────────────────┘

用户消息
    │
    ▼
┌─────────────────────────────────────────────────────────────────────────┐
│                            CLI / Gateway                                  │
│  ┌──────────────┐    ┌──────────────┐    ┌──────────────────────────┐  │
│  │  hermes CLI  │    │  Telegram    │    │  Discord / Slack / ...   │  │
│  │  (交互式)    │    │  Adapter     │    │  Adapter                 │  │
│  └──────────────┘    └──────────────┘    └──────────────────────────┘  │
└─────────────────────────────────────────────────────────────────────────┘
    │
    ▼
┌─────────────────────────────────────────────────────────────────────────┐
│                         AIAgent.run_conversation()                       │
│                                                                         │
│  messages = [{"role": "user", "content": "..."}]                        │
│      │                                                                 │
│      ▼                                                                 │
│  ┌─────────────────────────────────────────────────────────────────┐    │
│  │                    while 迭代循环                                │    │
│  │  ┌───────────────┐    ┌───────────────┐    ┌───────────────┐   │    │
│  │  │ API 调用      │───▶│ 处理响应      │───▶│ 工具调用      │   │    │
│  │  │ (LLM API)    │    │ (完成/继续)   │    │ (执行工具)    │   │    │
│  │  └───────────────┘    └───────────────┘    └───────┬───────┘   │    │
│  │                                                       │           │    │
│  │         ┌─────────────────────────────────────────────┘           │    │
│  │         ▼                                                         │    │
│  │  ┌───────────────┐    ┌───────────────┐    ┌───────────────┐      │    │
│  │  │ 记忆注入      │    │ 上下文压缩    │    │ 错误恢复      │      │    │
│  │  │ (prefetch)   │    │ (自动触发)    │    │ (重试/回退)   │      │    │
│  │  └───────────────┘    └───────────────┘    └───────────────┘      │    │
│  └─────────────────────────────────────────────────────────────────┘    │
│                                                                         │
└─────────────────────────────────────────────────────────────────────────┘
    │
    ▼
┌─────────────────────────────────────────────────────────────────────────┐
│                           工具系统                                       │
│                                                                         │
│  ┌─────────────────┐    ┌─────────────────┐    ┌─────────────────┐     │
│  │ Terminal Tool   │    │ Browser Tool    │    │ File Tools      │     │
│  │ ┌─────────────┐ │    │ ┌─────────────┐ │    │ ┌─────────────┐ │     │
│  │ │ Local       │ │    │ │ Browserbase │ │    │ │ read_file   │ │     │
│  │ │ Docker      │ │    │ │ BrowserUse  │ │    │ │ write_file  │ │     │
│  │ │ SSH         │ │    │ │ Camofox     │ │    │ │ patch       │ │     │
│  │ └─────────────┘ │    │ └─────────────┘ │    │ └─────────────┘ │     │
│  └─────────────────┘    └─────────────────┘    └─────────────────┘     │
│                                                                         │
│  ┌─────────────────┐    ┌─────────────────┐    ┌─────────────────┐     │
│  │ Web Tools       │    │ Delegate Tool   │    │ Skills Tool     │     │
│  │ ┌─────────────┐ │    │ ┌─────────────┐ │    │ ┌─────────────┐ │     │
│  │ │ Parallel    │ │    │ │ 子代理池    │ │    │ │ 技能读取    │ │     │
│  │ │ Firecrawl   │ │    │ │ 独立上下文  │ │    │ │ 技能创建    │ │     │
│  │ │ Tavily      │ │    │ └─────────────┘ │    │ └─────────────┘ │     │
│  │ └─────────────┘ │    │                  │    │                  │     │
│  └─────────────────┘    └─────────────────┘    └─────────────────┘     │
└─────────────────────────────────────────────────────────────────────────┘
    │
    ▼
┌─────────────────────────────────────────────────────────────────────────┐
│                           持久化层                                       │
│                                                                         │
│  ┌─────────────────┐    ┌─────────────────┐    ┌─────────────────┐     │
│  │ SessionDB       │    │ Skills Store    │    │ Memory          │     │
│  │ (SQLite + FTS5) │    │ (~/.hermes/    │    │ ┌─────────────┐ │     │
│  │                 │    │  skills/)       │    │ │ MEMORY.md   │ │     │
│  │ ┌─────────────┐ │    │                 │    │ │ USER.md     │ │     │
│  │ │ sessions    │ │    │ ┌─────────────┐ │    │ └─────────────┘ │     │
│  │ │ messages    │ │    │ │ SKILL.md   │ │    │                  │     │
│  │ │ messages_fts│ │    │ │ references/│ │    │ External:       │     │
│  │ └─────────────┘ │    │ └─────────────┘ │    │ honcho/mem0/... │     │
│  └─────────────────┘    └─────────────────┘    └─────────────────┘     │
└─────────────────────────────────────────────────────────────────────────┘

第三章 核心机制详解

3.1 AIAgent 类

AIAgent 是 Hermes Agent 的核心类，位于 run_agent.py 第 439 行开始，共约 9,200 行代码。它封装了 Agent 的所有逻辑，包括对话循环、工具调用、错误处理等。

类层次结构

run_agent.py
│
└── AIAgent (行 439-9681)
    │
    ├── __init__(行 463-1280)         # 初始化
    │   ├── 配置加载
    │   ├── 客户端创建
    │   ├── 工具初始化
    │   └── 会话管理
    │
    ├── reset_session_state(行 1281-1325)
    │   # 重置会话状态
    │
    ├── switch_model(行 1326-1448)
    │   # 运行时切换模型
    │
    ├── run_conversation(行 7124-9680)  # ⭐ 核心方法
    │   │
    │   └── while 循环
    │       ├── _prepare_api_messages()
    │       ├── _streaming_api_call()
    │       ├── _execute_tool_calls()
    │       └── 返回 / 继续
    │
    └── chat(行 9681-9700)            # 简单接口

核心属性

构造函数详解

class AIAgent:
    def __init__(
        self,
        model: str = "anthropic/claude-opus-4.6",
        max_iterations: int = 90,
        enabled_toolsets: list = None,
        disabled_toolsets: list = None,
        quiet_mode: bool = False,
        save_trajectories: bool = False,
        platform: str = None,           # "cli", "telegram", etc.
        session_id: str = None,
        skip_context_files: bool = False,
        skip_memory: bool = False,
        # ... plus provider, api_mode, callbacks, routing params
    ): ...

构造函数完成以下初始化：

1. 配置合并：CLI 配置 + 环境变量 + 代码参数
2. 客户端创建：根据 api_mode 创建对应的 API 客户端
3. 工具发现：调用 _discover_tools() 扫描并注册所有工具
4. 会话初始化：创建或恢复会话状态
5. 压缩器初始化：创建上下文压缩器

3.2 Agent 循环

Agent 循环是 Hermes 的核心，位于 run_conversation() 方法（行 7124-9680，共约 2,500 行）。

主循环结构

def run_conversation(self, user_message: str, system_message: str = None,
                     conversation_history: list = None, task_id: str = None) -> dict:
    """
    主对话循环
    返回: {"final_response": "...", "messages": [...], "timing": {...}}
    """
    
    # ========== 初始化阶段 ==========
    messages = list(conversation_history) if conversation_history else []
    user_msg = {"role": "user", "content": user_message}
    messages.append(user_msg)
    
    # 重置状态
    self._reset_conversation_state()
    
    # ========== 预压缩检查 ==========
    if self.compression_enabled and _compressor.should_compress(initial_tokens):
        messages, system_message = self._compress_context(messages, system_message)
    
    # ========== 主循环 ==========
    while api_call_count < self.max_iterations and self.iteration_budget.remaining > 0:
        
        # 1. interrupt 检查
        if self._interrupt_requested:
            break
        
        # 2. 消耗迭代预算
        if not self.iteration_budget.consume():
            break
        
        # 3. 构建 API 消息
        api_messages = self._prepare_api_messages(messages, system_message, ...)
        
        # 4. 调用 API（支持 streaming）
        response = self._interruptible_streaming_api_call(api_messages, ...)
        
        # 5. 处理响应
        if response.tool_calls:
            # 执行工具
            results = self._execute_tool_calls(response.tool_calls, ...)
            
            # 追加结果到消息
            messages.append({"role": "assistant", "tool_calls": response.tool_calls})
            messages.append({"role": "tool", "content": results})
            
            # 继续循环
            continue
        
        else:
            # 正常完成
            final_response = response.content
            break
    
    # ========== 清理和返回 ==========
    self._flush_messages_to_session_db(messages)
    return {"final_response": final_response, "messages": messages, ...}

IterationBudget 迭代预算

class IterationBudget:
    """线程安全的迭代预算计数器"""
    
    def __init__(self, max_total: int):
        self._max = max_total
        self._used = 0
        self._lock = threading.Lock()
    
    def consume(self) -> bool:
        """尝试消耗 1 次迭代，返回是否允许"""
        with self._lock:
            if self._used >= self._max:
                return False
            self._used += 1
            return True
    
    def refund(self) -> None:
        """退还 1 次迭代（如 execute_code 专用）"""
        with self._lock:
            self._used = max(0, self._used - 1)
    
    @property
    def remaining(self) -> int:
        return self._max - self._used

API 调用预处理

def _prepare_api_messages(self, messages, system_message, ...):
    """构建发送给 API 的消息列表"""
    
    api_messages = []
    
    # 1. 添加系统消息
    if system_message:
        api_messages.append({"role": "system", "content": system_message})
    
    # 2. 注入外部 memory provider 预取结果
    if self.memory_manager:
        memory_context = self.memory_manager.prefetch(messages)
        if memory_context:
            # 注入到当前用户消息
            user_msg = messages[-1].copy()
            user_msg["content"] = memory_context + "\n\n" + user_msg["content"]
            messages[-1] = user_msg
    
    # 3. 处理 reasoning 字段
    for msg in messages:
        api_msg = msg.copy()
        if "reasoning" in msg:
            # 复制到 reasoning_content（兼容 Moonshot 等）
            api_msg["reasoning_content"] = msg["reasoning"]
            del api_msg["reasoning"]  # 仅用于 trajectory 存储
        api_messages.append(api_msg)
    
    # 4. Anthropic Prompt Caching
    if self._use_prompt_caching:
        api_messages = self._apply_prompt_caching(api_messages)
    
    return api_messages

流式 API 调用

def _interruptible_streaming_api_call(self, api_kwargs, *, on_first_delta):
    """
    可中断的流式 API 调用
    后台线程执行 HTTP 请求，主线程每 0.3s 检查中断标志
    """
    
    result_holder = []
    exception_holder = []
    
    def background_thread():
        try:
            # 执行实际的 API 调用
            response = self.client.chat.completions.create(
                stream=True,
                **api_kwargs
            )
            
            for chunk in response:
                # 处理 delta
                delta = chunk.choices[0].delta
                if delta.content:
                    result_holder.append(delta.content)
                    if on_first_delta:
                        on_first_delta()  # 触发回调
                
                # 检查中断
                if self._interrupt_requested:
                    break
                    
        except Exception as e:
            exception_holder.append(e)
    
    # 启动后台线程
    thread = threading.Thread(target=background_thread)
    thread.start()
    
    # 主线程等待，每 0.3s 检查中断
    while thread.is_alive():
        thread.join(timeout=0.3)
        if self._interrupt_requested:
            # 设置标志让后台线程优雅退出
            self._streaming_interrupt = True
    
    # 处理结果或异常
    if exception_holder:
        raise exception_holder[0]
    
    return "".join(result_holder)

3.3 消息系统

消息格式

Hermes 使用 OpenAI 兼容的消息格式：

tool_calls 结构

{
    "tool_calls": [
        {
            "id": "call_abc123",           # 唯一 ID
            "type": "function",
            "function": {
                "name": "read_file",       # 工具名
                "arguments": "{\"path\": \"/etc/passwd\"}"  # JSON 字符串
            }
        }
    ]
}

reasoning 字段

模型思考内容存储在 reasoning 字段中，发送 API 时转换到 reasoning_content（兼容部分厂商）：

# 存储时
assistant_msg["reasoning"] = "我认为应该先读取文件..."

# 发送 API 时
api_msg["reasoning_content"] = "我认为应该先读取文件..."
del api_msg["reasoning"]  # 仅用于本地 trajectory 存储

会话历史管理

# 追加新消息
messages.append({"role": "user", "content": user_message})

# 追加助手回复（带工具调用）
messages.append({
    "role": "assistant",
    "tool_calls": [...],
    "content": None
})

# 追加工具结果
messages.append({
    "role": "tool",
    "tool_call_id": "call_abc123",
    "content": "file content here..."
})

3.4 错误处理与恢复

错误分类

class FailoverReason(Enum):
    rate_limit           # 429 - 限流
    billing              # 账单问题
    context_overflow     # 上下文超限 (413/400)
    payload_too_large    # 请求体过大
    thinking_signature   # Anthropic thinking 签名无效
    auth / credential    # 认证错误 (401/403)
    server_error         # 服务器内部错误 (500)
    network_error        # 网络错误
    timeout              # 请求超时

重试机制

# 重试装饰器
@tenacity.retry(
    stop=tenacity.stop_after_attempt(3),
    wait=tenacity.wait_exponential(multiplier=2, min=2, max=60),
    retry=tenacity.retry_if_exception_type(RateLimitError)
)
def _make_api_call_with_retry(**kwargs):
    return client.chat.completions.create(**kwargs)

错误恢复策略

Credential Pool

class CredentialPool:
    """API 密钥池，支持轮换"""
    
    def __init__(self, credentials: list[str]):
        self._creds = credentials
        self._index = 0
        self._lock = threading.Lock()
    
    def get(self) -> str:
        with self._lock:
            cred = self._creds[self._index]
            self._index = (self._index + 1) % len(self._creds)
            return cred
    
    def reset(self):
        """重置到第一个凭证"""
        with self._lock:
            self._index = 0

上下文压缩触发

if self.compression_enabled:
    real_tokens = count_tokens(messages) + count_tokens(completion)
    threshold = self.context_length * 0.5  # 50% 阈值
    
    if real_tokens > threshold:
        messages, active_system_prompt = self._compress_context(
            messages, system_message
        )
        # 压缩后重试当前请求

3.5 回调系统

Hermes 提供了丰富的回调钩子，允许在关键节点插入自定义逻辑：

使用示例

agent = AIAgent(
    model="anthropic/claude-sonnet-4",
    
    # 步骤回调
    step_callback=lambda count, tools: print(f"Step {count}: {tools}"),
    
    # 工具回调
    tool_progress_callback=lambda name, args: print(f"Executing {name}..."),
    tool_complete_callback=lambda name, result: print(f"Done {name}"),
    
    # 流式回调
    stream_delta_callback=lambda delta: print(delta, end="", flush=True),
    
    # 澄清回调
    clarify_callback=lambda q, choices: input(f"{q}\n{choices}\n> "),
)

思考动画回调

# thinking_callback 用于显示模型的"思考"过程
def show_thinking(text):
    # 清除当前行
    print(f"\r🤔 {text[:50]}...", end="", flush=True)

agent = AIAgent(
    model="anthropic/claude-sonnet-4",
    thinking_callback=show_thinking
)

第四章 工具系统

Hermes Agent 的工具系统是其核心能力之一，支持 50+ 工具，涵盖文件操作、终端执行、浏览器自动化、网络搜索等。

4.1 工具注册机制

核心组件

ToolEntry 数据结构

@dataclass
class ToolEntry:
    name: str                    # 工具名，如 "read_file"
    toolset: str                 # 所属工具集，如 "file"
    schema: dict                 # OpenAI 格式的 JSON Schema
    handler: callable            # 处理函数
    check_fn: callable = None   # 可用性检查函数
    requires_env: list = field(default_factory=list)  # 所需环境变量
    is_async: bool = False       # 是否异步
    emoji: str = "🔧"           # emoji 图标
    description: str = ""        # 描述
    max_result_size_chars: float = float('inf')  # 最大结果大小

注册流程

Hermes 使用模块级导入自动注册模式，每个工具文件在底部调用 registry.register()：

# tools/registry.py
registry = ToolRegistry()  # 单例

# tools/file_tools.py (模块底部)
def _handle_read_file(path: str, offset: int = 1, limit: int = 500, **kwargs):
    """读取文件内容"""
    ...

def _check_file_reqs() -> bool:
    """检查工具是否可用"""
    return os.path.exists(path) or True  # 延迟到执行时检查

registry.register(
    name="read_file",
    toolset="file",
    schema={
        "name": "read_file",
        "description": "Read content from a file...",
        "parameters": {
            "type": "object",
            "properties": {
                "path": {"type": "string", "description": "Path to the file"},
                "offset": {"type": "integer", "default": 1},
                "limit": {"type": "integer", "default": 500}
            }
        }
    },
    handler=_handle_read_file,
    check_fn=_check_file_reqs,
    emoji="📖"
)

工具发现

# model_tools.py
def _discover_tools():
    """在模块导入时触发所有工具注册"""
    
    # 按顺序导入所有工具模块
    # 每个模块底部会调用 registry.register()
    
    from tools import terminal_tool
    from tools import browser_tool
    from tools import file_tools
    from tools import web_tools
    # ... 共 50+ 个模块
    
    # 构建工具名到工具集的映射
    TOOL_TO_TOOLSET_MAP = registry.get_tool_to_toolset_map()
    
    return registry

# 初始化时自动发现
registry = _discover_tools()

4.2 工具执行流程

LLM 返回 function_call
         │
         ▼
┌─────────────────────────────────────────────────────────────────┐
│  handle_function_call(tool_name, arguments, task_id)           │
│  (model_tools.py)                                               │
└─────────────────────────────────────────────────────────────────┘
         │
         ▼
┌─────────────────────────────────────────────────────────────────┐
│  coerce_tool_args()  — 类型修正                                  │
│  "42" → 42, "true" → True, "[1,2,3]" → [1,2,3]                  │
└─────────────────────────────────────────────────────────────────┘
         │
         ▼
┌─────────────────────────────────────────────────────────────────┐
│  pre_tool_call hook  — 插件钩子                                   │
└─────────────────────────────────────────────────────────────────┘
         │
         ▼
┌─────────────────────────────────────────────────────────────────┐
│  registry.dispatch(name, args)                                   │
│         │                                                       │
│         ├─→ 同步工具 → 直接调用 handler(args, **kwargs)           │
│         │                                                       │
│         └─→ 异步工具 → _run_async(handler(...))                  │
│                      (桥接到事件循环)                             │
└─────────────────────────────────────────────────────────────────┘
         │
         ▼
┌─────────────────────────────────────────────────────────────────┐
│  异常捕获 → tool_error() → {"error": "..."}                       │
│  正常 → tool_result() → {"success": true, "data": ...}          │
└─────────────────────────────────────────────────────────────────┘
         │
         ▼
┌─────────────────────────────────────────────────────────────────┐
│  post_tool_call hook  — 插件钩子                                 │
└─────────────────────────────────────────────────────────────────┘
         │
         ▼
返回 JSON 字符串给 LLM

handle_function_call 实现

def handle_function_call(tool_name: str, arguments: dict, task_id: str = None) -> str:
    """
    核心工具调度函数
    """
    
    # 1. 查找工具
    entry = registry.get(tool_name)
    if not entry:
        return tool_error(f"Unknown tool: {tool_name}")
    
    # 2. 检查可用性
    if entry.check_fn and not entry.check_fn():
        return tool_error(f"Tool {tool_name} is not available")
    
    # 3. 类型强制转换
    try:
        args = coerce_tool_args(arguments, entry.schema)
    except Exception as e:
        return tool_error(f"Invalid arguments: {e}")
    
    # 4. 执行
    try:
        result = entry.handler(args, task_id=task_id)
        
        # 异步工具
        if entry.is_async:
            result = _run_async(lambda: entry.handler(args, task_id=task_id))
        
        return tool_result(result)
    
    except Exception as e:
        return tool_error(str(e))

4.3 并行执行

Hermes 支持工具的并行执行，通过 _should_parallelize_tool_batch() 决策：

并行决策规则

# 永远不并行的工具
_NEVER_PARALLEL_TOOLS = frozenset({"clarify"})

# 可并行执行的只读工具
_PARALLEL_SAFE_TOOLS = frozenset({
    "read_file",
    "search_files", 
    "web_search",
    "session_search",
    "skills_list",
    "skill_view",
    "terminal_tool",  # 仅 background 模式
})

# 路径范围工具（需检查路径冲突）
_PATH_SCOPED_TOOLS = {"read_file", "write_file", "patch"}

# 最大并行线程数
_MAX_TOOL_WORKERS = 8

def _should_parallelize_tool_batch(tool_calls) -> bool:
    """决定是否并行执行一批工具调用"""
    
    # 1. 单个工具 → 不并行
    if len(tool_calls) == 1:
        return False
    
    # 2. 包含 clarify → 串行（交互式）
    if any(tc.function.name in _NEVER_PARALLEL_TOOLS for tc in tool_calls):
        return False
    
    # 3. 路径工具 → 检查路径是否重叠
    path_tools = [tc for tc in tool_calls if tc.function.name in _PATH_SCOPED_TOOLS]
    if path_tools and _paths_conflict(path_tools):
        return False
    
    # 4. 其他工具 → 必须在 _PARALLEL_SAFE_TOOLS 中
    all_safe = all(
        tc.function.name in _PARALLEL_SAFE_TOOLS 
        for tc in tool_calls
    )
    if not all_safe:
        return False
    
    # 5. 限制线程数
    return len(tool_calls) <= _MAX_TOOL_WORKERS

并发执行实现

def _execute_tool_calls_concurrent(self, tool_calls, messages, ...):
    """并发执行多个工具调用"""
    
    # 创建线程池
    with ThreadPoolExecutor(max_workers=len(tool_calls)) as executor:
        
        # 提交所有任务
        future_to_tc = {
            executor.submit(self._execute_single_tool, tc, ...): tc
            for tc in tool_calls
        }
        
        # 收集结果（按提交顺序）
        results = []
        for future in as_completed(future_to_tc):
            tc = future_to_tc[future]
            try:
                result = future.result()
                results.append((tc.id, result))
            except Exception as e:
                results.append((tc.id, tool_error(str(e))))
    
    # 按原始顺序排序
    results.sort(key=lambda x: tool_call_order[x[0]])
    return results

4.4 Terminal 终端工具

Terminal 工具（tools/terminal_tool.py，1791 行）是 Hermes 最复杂的工具之一，支持多种执行后端。

执行后端

核心功能

@registry.register(
    name="terminal_tool",
    toolset="terminal",
    schema={...},
    handler=_handle_terminal
)
def _handle_terminal(
    command: str,
    task_id: str = None,
    background: bool = False,     # 后台执行
    workdir: str = None,           # 工作目录
    timeout: int = 180,            # 超时秒数
    env_type: str = None,          # 执行后端
    **kwargs
) -> str:
    """终端工具主入口"""
    
    # 1. 选择执行后端
    env = _get_environment(env_type or os.getenv("TERMINAL_ENV", "local"))
    
    # 2. 检查危险命令
    if _is_dangerous_command(command):
        approved = _request_approval(command)
        if not approved:
            return json.dumps({"error": "Command not approved"})
    
    # 3. 执行命令
    if background:
        return env.execute_background(command, task_id, workdir, timeout)
    else:
        return env.execute(command, task_id, workdir, timeout)

Sudo 处理

def _transform_sudo_command(command: str, sudo_password: str = None) -> str:
    """
    将 sudo 命令转换为无交互模式
    sudo → sudo -S -p '' (从 stdin 读取密码)
    """
    
    if "sudo " not in command or "-S" in command:
        return command
    
    # 转换为使用 stdin 读取密码
    command = command.replace("sudo ", "sudo -S -p '' ", 1)
    
    # 如果没有提供密码，提示用户
    if not sudo_password:
        sudo_password = _get_cached_sudo_password() or _prompt_sudo_password()
    
    return command, sudo_password  # 返回修改后的命令和密码

危险命令审批

# 危险模式正则
_DESTRUCTIVE_PATTERNS = [
    r'\brsync\b.*--delete',
    r'\brm\s+-rf\s+/',
    r'\bdd\b.*of=',
    r'\bmkfs\b',
    r'\bparted\b',
    r'\bchef-client\b',
    r'\bpuppet\b.*apply',
    r'\bdocker\s+rm\s+-f',
    r'\bkubectl\s+delete',
]

# 路径白名单（允许的路径字符）
_WORKDIR_SAFE_RE = re.compile(r'^[A-Za-z0-9/_-]+$')

def _check_dangerous_command(command: str) -> tuple[bool, str]:
    """检查命令是否危险"""
    
    for pattern in _DESTRUCTIVE_PATTERNS:
        if re.search(pattern, command):
            return True, f"Dangerous pattern detected: {pattern}"
    
    # 检查路径遍历
    if "../" in command or "%" in command:
        return True, "Path traversal detected"
    
    return False, ""

4.5 Browser 浏览器工具

Browser 工具（tools/browser_tool.py，2218 行）支持多种浏览器自动化后端。

后端架构

browser_tool.py (主逻辑)
    │
    ├── CloudBrowserProvider ABC
    │   ├── BrowserbaseProvider     # Browserbase 云服务
    │   ├── BrowserUseProvider     # Nous 订阅
    │   └── FirecrawlProvider      # Firecrawl 服务
    │
    ├── CamofoxProvider           # 本地反检测浏览器
    │
    └── LocalProvider             # Agent-Browser CLI

核心工具函数

元素引用系统

# 快照返回可访问性树，每个元素有 ref ID
snapshot:
  - link "Sign in" [ref=e5]
  - button "Submit" [ref=e12]
  - input "Email" [ref=e20]

# 点击时使用 ref
browser_click(ref="@e5")   # 点击 Sign in 链接
browser_click(ref="@e12")  # 点击 Submit 按钮

# 输入时
browser_type(ref="@e20", text="user@example.com")

安全机制

# SSRF 防护：阻止私有地址
_PRIVATE_IP_PATTERNS = [
    r'^127\.',           # localhost
    r'^10\.',            # 10.0.0.0/8
    r'^172\.(1[6-9]|2[0-9]|3[0-1])\.',  # 172.16.0.0/12
    r'^192\.168\.',      # 192.168.0.0/16
    r'^localhost$',
    r'\.local$',
]

def _is_safe_url(url: str) -> bool:
    """检查 URL 是否安全（防止 SSRF）"""
    
    try:
        parsed = urlparse(url)
        host = parsed.hostname or ""
        
        # 检查主机名
        for pattern in _PRIVATE_IP_PATTERNS:
            if re.match(pattern, host):
                return False
        
        # 检查端口
        if parsed.port in [22, 23, 3389, 5900]:
            return False
        
        return True
    except:
        return False

4.6 File 文件工具

工具列表

read_file 实现

@registry.register(name="read_file", toolset="file", ...)
def _handle_read_file(args, task_id=None) -> str:
    path = args["path"]
    offset = args.get("offset", 1)   # 1-indexed
    limit = args.get("limit", 500)
    
    try:
        with open(path, 'r', encoding='utf-8') as f:
            lines = f.readlines()
        
        # 边界处理
        total_lines = len(lines)
        offset = max(1, min(offset, total_lines))
        end = min(offset + limit - 1, total_lines)
        
        # 读取指定范围
        content = ''.join(lines[offset-1:end])
        
        return json.dumps({
            "content": content,
            "total_lines": total_lines,
            "read_lines": end - offset + 1,
            "path": path
        })
    except FileNotFoundError:
        return tool_error(f"File not found: {path}")
    except PermissionError:
        return tool_error(f"Permission denied: {path}")
    except Exception as e:
        return tool_error(str(e))

write_file 实现

@registry.register(name="write_file", toolset="file", ...)
def _handle_write_file(args, task_id=None) -> str:
    path = args["path"]
    content = args["content"]
    
    # 原子写入（临时文件 + os.replace）
    tmp_path = path + f".{os.getpid()}.tmp"
    
    try:
        with open(tmp_path, 'w', encoding='utf-8') as f:
            f.write(content)
        
        os.replace(tmp_path, path)  # 原子操作
        
        return json.dumps({
            "success": True,
            "path": path,
            "bytes": len(content.encode('utf-8'))
        })
    except Exception as e:
        if os.path.exists(tmp_path):
            os.remove(tmp_path)
        return tool_error(str(e))

patch 实现（模糊匹配）

@registry.register(name="patch", toolset="file", ...)
def _handle_patch(args, task_id=None) -> str:
    path = args["path"]
    old_string = args["old_string"]
    new_string = args["new_string"]
    replace_all = args.get("replace_all", False)
    
    try:
        with open(path, 'r', encoding='utf-8') as f:
            content = f.read()
        
        # 模糊匹配：允许空白符/缩进差异
        if replace_all:
            new_content = content.replace(old_string, new_string)
        else:
            # fuzzy_find_and_replace 实现模糊匹配
            new_content = fuzzy_replace(content, old_string, new_string)
        
        if new_content == content:
            return tool_error("Pattern not found in file")
        
        # 写回
        with open(path, 'w', encoding='utf-8') as f:
            f.write(new_content)
        
        return json.dumps({"success": True, "path": path})
    except Exception as e:
        return tool_error(str(e))

第五章 技能系统

Hermes 的技能系统是其自改进能力的核心，允许 Agent 从经验中创建、存储和复用知识。

5.1 技能存储结构

~/.hermes/skills/                          # 技能根目录
│
├── software-development/
│   └── systematic-debugging/
│       ├── SKILL.md                    # 主指令文件
│       ├── references/
│       │   └── api.md                 # 参考文档
│       ├── templates/
│       │   └── config.yaml            # 模板
│       └── scripts/
│           └── validate.py           # 脚本
│
├── mlops/
│   └── axolotl/
│       └── SKILL.md
│
└── ... (共 28 个分类)

SKILL.md 格式

---
name: systematic-debugging       # 必填，技能标识（slug 格式）
description: |                    # 必填，≤1024 字符
  A comprehensive debugging methodology for identifying,
  isolating, and resolving software bugs systematically.
version: 1.0.0                  # 可选，语义版本
platforms: [macos, linux]       # 可选，限定操作系统
metadata:                       # 可选，元数据
  hermes:
    tags: [debugging, problem-solving]
    related_skills: [testing, logging]
    requires_toolsets: [terminal, file]
    fallback_for_toolsets: [web]
---

# Systematic Debugging

## Overview
This skill provides a structured approach to debugging...

## Steps

1. **Reproduce the Issue**
   - Create a minimal test case...

2. **Isolate the Problem**
   - Use binary search on code...

3. **Identify Root Cause**
   ...

5.2 渐进式披露

技能系统采用渐进式披露架构，避免一次性加载所有内容浪费 token：

Tier 0 - 分类列表

def skills_categories() -> str:
    """返回技能分类（Tier 0）"""
    
    categories = [
        ("software-development", "8 skills", "Coding, testing, debugging..."),
        ("github", "8 skills", "Repository management, PRs..."),
        ("mlops", "10 skills", "Model deployment, monitoring..."),
        ("media", "6 skills", "Video, audio, image processing..."),
        # ... 共 28 个分类
    ]
    
    return json.dumps({"categories": categories})

Tier 1 - 技能列表

def skills_list(category: str = None) -> str:
    """返回技能列表（Tier 1）"""
    
    skills = []
    for skill_dir in scan_skills_directories():
        if category and skill_dir.parent.name != category:
            continue
        
        skill = parse_skill_md(skill_dir / "SKILL.md")
        skills.append({
            "name": skill.name,
            "description": skill.description,
            "category": skill_dir.parent.name
        })
    
    return json.dumps({"skills": skills})

Tier 2 - 完整技能

def skill_view(name: str, file_path: str = None) -> str:
    """
    返回完整技能内容（Tier 2）
    如果 file_path 指定，返回特定支持文件（Tier 3）
    """
    
    skill_md = load_skill(name)
    
    if file_path:
        # Tier 3：支持文件
        return load_skill_file(skill_md, file_path)
    
    # Tier 2：完整 SKILL.md
    return {
        "content": skill_md.content,
        "frontmatter": skill_md.frontmatter,
        "linked_files": skill_md.get_linked_files(),
        "environment_variables": extract_env_vars(skill_md)
    }

5.3 技能执行机制

Slash Command 流程

用户输入: /systematic-debugging
         │
         ▼
┌─────────────────────────────────────────────────────────────────┐
│  scan_skill_commands()  — 扫描所有 SKILL.md 建立映射           │
│  /systematic-debugging → skill: "software-development/..."    │
└─────────────────────────────────────────────────────────────────┘
         │
         ▼
┌─────────────────────────────────────────────────────────────────┐
│  build_skill_invocation_message()                               │
│                                                                 │
│  activation_msg = """                                            │
│  [SYSTEM: The user has invoked the "systematic-debugging" skill]│
│                                                                  │
│  ## Skill: Systematic Debugging                                 │
│                                                                  │
│  (加载 SKILL.md 完整内容)                                        │
│                                                                  │
│  ## Required Environment Variables                              │
│  (列出所需环境变量)                                               │
│                                                                  │
│  ## Setup Notes                                                 │
│  (列出设置说明)                                                  │
│  """                                                            │
└─────────────────────────────────────────────────────────────────┘
         │
         ▼
作为 user message 注入对话上下文

技能激活消息格式

[SYSTEM: The user has invoked the "systematic-debugging" skill. 
This skill provides a comprehensive debugging methodology. 
Load its full content and follow its instructions exactly.]

---

## Skill: Systematic Debugging

(version 1.0.0 | platforms: macos, linux)

### Description
A comprehensive debugging methodology for identifying, isolating, 
and resolving software bugs systematically.

### Required Environment Variables
- None required

### Setup Notes
1. Ensure you have access to terminal tools
2. This skill works best with file reading capabilities

---

# Systematic Debugging

## Overview
This skill provides a structured approach to debugging...

[... 完整技能内容 ...]

5.4 技能创建与改进

触发时机

Agent 会在以下情况被提示创建或更新技能：

• 复杂任务成功完成（5+ 工具调用）
• 克服错误后找到正确方法
• 用户纠正了 Agent 的方法
• 用户明确要求"记住这个流程"

skill_manage 操作

# 创建技能
skill_manage(
    action='create',
    name='my-custom-skill',
    content=,
    category='domain'
)

# 更新技能（首选）
skill_manage(
    action='patch',
    name='my-custom-skill',
    old_string='原文本',      # 不需要精确匹配
    new_string='新文本'
)

# 整体重写（用于重大修改）
skill_manage(
    action='edit',
    name='my-custom-skill',
    content=
)

# 删除
skill_manage(action='delete', name='my-custom-skill')

模糊匹配 Patch

# fuzzy_find_and_replace 支持：
# 1. 空白符差异
# 2. 缩进差异  
# 3. 注释差异

# 例如：
old_string = "for i in range(10):"
new_string = "for i in range(100):"

# 可以匹配：
#   for i in range(10):
#   for  i  in  range(10):    # 多余空格
#     for i in range(10):     # 缩进差异

验证流程

def _validate_skill(name: str, content: str) -> None:
    """技能验证"""
    
    # 1. 名称验证
    if not re.match(r'^[a-z0-9-]+$', name):
        raise ValueError("Name must be lowercase alphanumeric + hyphen")
    
    if len(name) > 64:
        raise ValueError("Name must be ≤64 characters")
    
    # 2. Frontmatter 验证
    frontmatter = parse_yaml_frontmatter(content)
    if 'name' not in frontmatter:
        raise ValueError("Missing required 'name' field")
    if 'description' not in frontmatter:
        raise ValueError("Missing required 'description' field")
    
    # 3. 安全扫描
    if skills_guard.scan_skill(content).blocked:
        raise ValueError("Skill content blocked by security scan")
    
    # 4. 原子写入
    _atomic_write_text(skill_path, content)

技能回顾线程

# run_agent.py - 每次对话结束后触发
def _spawn_background_review(self, messages, task_id):
    """后台线程评估是否需要创建/更新技能"""
    
    thread = threading.Thread(target=self._review_for_skills, args=(...))
    thread.daemon = True
    thread.start()

def _review_for_skills(self, messages, ...):
    """
    评估最近对话，决定是否：
    1. 创建新技能
    2. 更新现有技能
    """
    
    # 使用 _SKILL_REVIEW_PROMPT 分析对话
    review_prompt = f"""
    Review this conversation for值得创建为技能的模式:
    
    {messages}
    
    Consider:
    - 复杂的多次工具调用
    - 克服的错误和解决方法
    - 用户明确要求记住的流程
    """
    
    response = self.llm.complete(review_prompt)
    
    if response.should_create:
        skill_manage(action='create', name=response.name, ...)
    elif response.should_update:
        skill_manage(action='patch', name=response.existing_skill, ...)

第六章 CLI 与配置系统

6.1 CLI 架构

Hermes CLI（cli.py，9268 行）是用户交互的核心界面。

入口点

# 安装后可用命令
$ hermes --help

# 或直接运行
$ python cli.py --help

# 通过模块运行
$ python -m hermes_cli.main

命令模式

HermesCLI 类

class HermesCLI:
    """主 CLI 类"""
    
    def __init__(self):
        self.agent = None
        self.console = ChatConsole()  # Rich 适配
        self.session_store = SessionStore()
    
    def run_interactive(self):
        """交互式 REPL"""
        while True:
            try:
                user_input = self.console.input()
                
                # 处理特殊命令
                if user_input.startswith('/'):
                    self._handle_slash_command(user_input)
                    continue
                
                # 发送到 Agent
                response = self.agent.chat(user_input)
                self.console.print(response)
                
            except KeyboardInterrupt:
                self._handle_interrupt()
            except EOFError:
                break
    
    def _handle_slash_command(self, cmd: str):
        """处理 / 命令"""
        parts = cmd.split()
        command = parts[0][1:]  # 去掉 /
        
        if command == 'help':
            self._show_help()
        elif command == 'model':
            self._switch_model(parts[1])
        elif command == 'exit':
            sys.exit(0)

TUI 实现

# 使用 prompt_toolkit 构建 TUI
from prompt_toolkit import PromptSession
from prompt_toolkit.key_binding import KeyBindings

# 固定输入区域 + 可滚动输出
session = PromptSession(
    message='🤖 ',
    key_bindings=kb,
    multiline=False,
    mouse_support=True
)

# 快捷键
kb = KeyBindings()

@kb.add('c-c', eager=True)
def interrupt(event):
    agent.interrupt()

@kb.add('c-d', eager=True)
def exit(event):
    raise EOFError()

@kb.add('c-z')
def suspend(event):
    # 挂起到后台
    signal.SIGTSTP

6.2 配置系统

配置加载优先级

# 优先级（高到低）
1. 命令行参数 (--config, --model, etc.)
2. 环境变量 (HERMES_MODEL, HERMES_CONFIG, etc.)
3. ~/.hermes/config.yaml (用户配置)
4. ./cli-config.yaml (项目配置，备用)
5. 内置默认值

配置结构

# ~/.hermes/config.yaml

# 模型配置
model:
  default: "anthropic/claude-sonnet-4"
  provider: "openrouter"        # openrouter/openai/anthropic/nous
  base_url: ""                 # 自定义 API URL

# 终端配置
terminal:
  env_type: "local"           # local/docker/ssh/modal/singularity/daytona
  cwd: "."                     # 默认工作目录
  timeout: 120                # 超时秒数
  lifetime_seconds: 300        # 最大生命周期

# 浏览器配置
browser:
  provider: "browserbase"     # browserbase/browseruse/camofox
  inactivity_timeout: 300
  record_sessions: false

# Agent 配置
agent:
  max_turns: 90               # 最大迭代
  personality: "kawaii"       #人格: helpful/concise/kawaii/catgirl/pirate...

# 显示配置
display:
  skin: "default"             # 皮肤: default/ares/mono/slate
  compact: false
  show_reasoning: false
  streaming: true

# 代码执行
code_execution:
  timeout: 300
  max_tool_calls: 50

# 压缩配置
compression:
  enabled: true
  threshold: 0.5            # 50% 上下文阈值

load_cli_config 实现

def load_cli_config() -> dict:
    """加载 CLI 配置"""
    
    config = {}
    
    # 1. 加载默认配置
    from hermes_cli.config import DEFAULT_CONFIG
    config = DEFAULT_CONFIG.copy()
    
    # 2. 加载用户配置
    user_config_path = get_config_path()  # ~/.hermes/config.yaml
    if user_config_path.exists():
        with open(user_config_path) as f:
            user = yaml.safe_load(f)
            deep_update(config, user)
    
    # 3. 加载项目配置（备用）
    project_config = Path("cli-config.yaml")
    if project_config.exists():
        with open(project_config) as f:
            project = yaml.safe_load(f)
            deep_update(config, project)
    
    # 4. 应用环境变量覆盖
    env_overrides = {
        "model.default": os.getenv("HERMES_MODEL"),
        "model.provider": os.getenv("HERMES_PROVIDER"),
        "terminal.timeout": os.getenv("HERMES_TERMINAL_TIMEOUT"),
    }
    for path, value in env_overrides.items():
        if value:
            set_nested(config, path, value)
    
    return config

6.3 皮肤系统

皮肤系统（hermes_cli/skin_engine.py）允许自定义 CLI 外观。

内置皮肤

皮肤配置示例

# ~/.hermes/skins/cyberpunk.yaml

name: cyberpunk
description: Neon-soaked terminal theme

colors:
  banner_border: "#FF00FF"
  banner_title: "#00FFFF"
  banner_accent: "#FF1493"

spinner:
  thinking_verbs: ["jacking in", "decrypting", "uploading"]
  wings:
    - ["⟨⚡", "⚡⟩"]

branding:
  agent_name: "Cyber Agent"
  response_label: " ⚡ Cyber "

tool_prefix: "┊"

6.4 多配置系统（Profiles）

支持多配置文件，用于隔离不同环境：

# 列出所有配置
hermes profile list

# 创建新配置
hermes profile create work

# 切换配置
hermes -p work

# 导出/导入
hermes profile export work.zip
hermes profile import work.zip

目录结构

~/.hermes/
├── config.yaml              # 默认配置
├── profiles/
│   ├── work/
│   │   └── config.yaml
│   ├── dev/
│   │   └── config.yaml
│   └── experiment/
│       └── config.yaml
├── skills/                  # 技能
├── memories/                # 记忆
└── .env                     # API 密钥

第七章 消息网关

Gateway 系统（gateway/run.py，7905 行）支持 15+ 消息平台统一接入。

7.1 网关架构

GatewayRunner (gateway/run.py)
    │
    ├── connect()           # 连接所有平台
    ├── disconnect()        # 断开所有平台
    │
    └── tick()              # 每 60s 被调用一次
        │
        ├── get_due_jobs()  # 获取到期定时任务
        │
        └── platform_adapters[]
            │
            ├── TelegramAdapter
            ├── DiscordAdapter
            ├── SlackAdapter
            └── ... (15+ 适配器)

基类 BasePlatformAdapter

class BasePlatformAdapter:
    """所有平台适配器的基类"""
    
    def __init__(self, config: dict):
        self.config = config
        self._connected = False
    
    # 生命周期
    async def connect(self) -> None:
        """连接平台"""
        raise NotImplementedError
    
    async def disconnect(self) -> None:
        """断开连接"""
        raise NotImplementedError
    
    # 消息处理
    async def handle_message(self, event: MessageEvent) -> None:
        """
        处理收到的消息
        → 授权检查 → 命令拦截 → 会话管理 → AIAgent → 发送响应
        """
        raise NotImplementedError
    
    # 发送消息
    async def send(self, chat_id: str, content: str, 
                   reply_to: str = None, metadata: dict = None) -> SendResult:
        """发送消息"""
        raise NotImplementedError
    
    # 媒体发送
    async def send_image(self, chat_id: str, image_path: str) -> SendResult: ...
    async def send_voice(self, chat_id: str, audio_path: str) -> SendResult: ...
    async def send_document(self, chat_id: str, file_path: str) -> SendResult: ...

7.2 会话管理

会话键生成

# SessionKey 格式
platform:chat_type:chat_id:thread_id:user_id

# 示例
telegram:dm:123456789::987654321           # Telegram 私聊
discord:channel:111222333:444555:666777888  # Discord 频道
slack:channel:C0123:D4567:U7890             # Slack 频道

# chat_type: dm / channel / group / thread

SessionStore

class SessionStore:
    """基于 SQLite 的会话存储"""
    
    def create_session(self, source: SessionSource, **kwargs) -> str:
        """创建新会话"""
        session_id = generate_session_id()
        
        self.db.execute("""
            INSERT INTO sessions (id, source, user_id, model, ...)
            VALUES (?, ?, ?, ?, ...)
        """, (session_id, source.platform, source.user_id, ...))
        
        return session_id
    
    def get_or_create_session(self, source: SessionSource) -> str:
        """获取或创建会话"""
        key = build_session_key(source)
        
        existing = self.db.execute(
            "SELECT id FROM sessions WHERE key = ? AND ended_at IS NULL",
            (key,)
        ).fetchone()
        
        if existing:
            return existing[0]
        
        return self.create_session(source)

7.3 平台适配器

Telegram 适配器要点

class TelegramAdapter(BasePlatformAdapter):
    """Telegram Bot API 适配器"""
    
    async def handle_message(self, update: Update) -> None:
        """处理 Telegram 更新"""
        
        # 1. 消息类型过滤
        if update.message:
            text = update.message.text
            chat_id = update.message.chat.id
            user_id = update.message.from_user.id
        elif update.callback_query:
            # 按钮点击
            ...
        
        # 2. 构建 MessageEvent
        event = MessageEvent(
            text=text,
            message_type=MessageType.TEXT,
            source=SessionSource(
                platform="telegram",
                chat_id=str(chat_id),
                user_id=str(user_id),
                chat_type="dm" if update.message.chat.type == "private" else "channel"
            )
        )
        
        # 3. 发送到网关
        await self.gateway.handle_message(event)

Discord 适配器要点

class DiscordAdapter(BasePlatformAdapter):
    """Discord Bot API 适配器"""
    
    def __init__(self, config: dict):
        super().__init__(config)
        self.intents = Intents.default()
        self.intents.message_content = True  # 需要这个权限
        self.client = discord.Client(intents=self.intents)
    
    async def handle_message(self, message: discord.Message) -> None:
        """处理 Discord 消息"""
        
        # 忽略 Bot 消息
        if message.author.bot:
            return
        
        # 检查是否是我们自己的消息
        if message.author.id == self.client.user.id:
            return
        
        # 检查 @mention 或 DM
        if isinstance(message.channel, discord.DMChannel):
            chat_type = "dm"
        else:
            chat_type = "channel"
        
        event = MessageEvent(
            text=message.content,
            source=SessionSource(
                platform="discord",
                chat_id=str(message.channel.id),
                user_id=str(message.author.id),
                thread_id=str(message.id) if message.thread else None,
                chat_type=chat_type
            )
        )

7.4 消息投递

发送流程

Agent 响应
    │
    ▼
┌─────────────────────────────────────────────────────────────────┐
│  DeliveryRouter._route()                                         │
│                                                                 │
│  1. 文本消息 → format_message() → truncate_message()           │
│  2. 媒体消息 → cache_media() → 选择 send_image/voice/...        │
│  3. 格式化 → 平台特定格式（MarkdownV2 / mrkdwn / ...）          │
└─────────────────────────────────────────────────────────────────┘
    │
    ▼
┌─────────────────────────────────────────────────────────────────┐
│  平台适配器.send()                                                │
│                                                                 │
│  1. 消息分块（超过最大长度）                                      │
│  2. 批量合并（0.6s 窗口）                                        │
│  3. 重试机制                                                     │
└─────────────────────────────────────────────────────────────────┘

文本格式化

# Telegram 使用 MarkdownV2
def format_telegram(self, content: str) -> str:
    """转换为 Telegram MarkdownV2 格式"""
    
    # Telegram 不支持 ``` 代码块，需要转换
    content = re.sub(r'```(\w+)?\n', '```\n', content)
    
    # 转义特殊字符
    special_chars = ['_', '*', '[', ']', '(', ')', '~', '`', '>', '#', '+', '-', '=', '|', '{', '}', '.', '!']
    for char in special_chars:
        content = content.replace(char, f'\\{char}')
    
    return content

# Slack 使用 mrkdwn
def format_slack(self, content: str) -> str:
    """转换为 Slack mrkdwn 格式"""
    
    #  格式
    content = re.sub(r'\[([^\]]+)\]\(([^)]+)\)', r'<\2|\1>', content)
    
    # *bold* → 
    content = re.sub(r'\*([^*]+)\*', r'*\1*', content)
    
    return content

Cron 结果投递

# Cron 任务完成后自动投递
async def _deliver_result(self, job_id: str, result: str, deliver: str):
    """投递 Cron 任务结果"""
    
    if deliver == "origin":
        # 投递到任务创建时的聊天
        session = self.session_store.get(job_id)
        adapter = self.platform_adapters[session.platform]
        await adapter.send(session.chat_id, result)
    
    elif deliver.startswith("telegram:"):
        # 指定 Telegram 聊天
        chat_id = deliver.split(":")[1]
        await self.telegram.send(chat_id, result)
    
    elif deliver == "silent":
        # 不投递，只保存
        pass
    
    else:
        # webhook 或其他
        await self._deliver_webhook(deliver, result)

第八章 高级特性

8.1 MCP (Model Context Protocol)

MCP 工具（tools/mcp_tool.py，2194 行）允许 Hermes 连接外部 MCP 服务器，扩展工具能力。

架构设计

hermes-agent
    │
    └── MCP Client (mcp_tool.py)
            │
            ├── _mcp_loop (daemon thread)
            │   └── asyncio event loop
            │
            └── MCPServerTask[]
                ├── StdioServer  (command + args)
                └── HTTP Server  (url)

连接 MCP 服务器

# 配置 ~/.hermes/config.yaml
mcp:
  servers:
    - name: filesystem
      type: stdio
      command: npx
      args: ["-y", "@modelcontextprotocol/server-filesystem", "/path"]
    
    - name: github
      type: http
      url: https://mcp.github.com/stdio

工具命名

# MCP 工具命名格式
mcp_{server_name}_{tool_name}

# 例如
mcp_filesystem_read_file
mcp_filesystem_write_file
mcp_github_create_issue

安全机制

# 凭证过滤
_CREDENTIAL_PATTERNS = [
    r'ghp_[a-zA-Z0-9]{36}',           # GitHub PAT
    r'sk-[a-zA-Z0-9]{48}',            # OpenAI Key
    r'sk-ant-[a-zA-Z0-9]{48}',        # Anthropic Key
    r'Bearer [a-zA-Z0-9.-_]+',         # Bearer Token
]

# 环境变量过滤（只传递安全变量）
_SAFE_ENV_VARS = {
    'PATH', 'HOME', 'USER', 'SHELL',
    'TMPDIR', 'TERM', 'LANG', 'LC_*'
}

def _build_safe_env(config_env: dict) -> dict:
    """构建安全的环境变量"""
    env = {k: v for k, v in os.environ.items() if k in _SAFE_ENV_VARS}
    env.update(config_env)  # 添加用户配置的环境变量
    return env

8.2 记忆系统

三层架构

MemoryManager (agent/memory_manager.py)
    │
    ├── BuiltinMemoryProvider (tools/memory_tool.py)
    │   ├── ~/.hermes/{profile}/memories/MEMORY.md
    │   └── ~/.hermes/{profile}/memories/USER.md
    │
    └── ExternalMemoryProvider[] (plugins/memory/{name}/)
        ├── honcho          # 用户画像
        ├── mem0           # 记忆存储
        ├── holographic    # 全息记忆
        ├── hindsight      # 经验回顾
        ├── byterover
        ├── retaindb
        └── supermemory

内置记忆格式

# ~/.hermes/default/memories/MEMORY.md
# Agent 个人笔记

§ 环境事实
- 服务器运行 Ubuntu 22.04
- Python 3.11+
- 网站部署在 /var/www/

§ 项目约定
- 配置文件使用 YAML
- 代码使用 Black 格式化
- 提交使用Conventional Commits

§ 工具习惯
- 使用 patch 而非 write_file 进行修改
- 优先使用 terminal 而非 subprocess

# ~/.hermes/default/memories/USER.md
# 用户画像

§ 偏好
- 中文交流
- 简洁回答
- 直接给出方案

§ 工作风格
- 不要重复任务
- 使用 Hermès 原生能力
- 自动运行优先

外部记忆 Provider

class MemoryProvider(ABC):
    """外部记忆 Provider 基类"""
    
    def initialize(self, config: dict) -> None:
        """初始化"""
        raise NotImplementedError
    
    def system_prompt_block(self) -> str:
        """返回插入系统提示的文本"""
        raise NotImplementedError
    
    def prefetch(self, messages: list) -> str:
        """在 API 调用前预取相关记忆"""
        raise NotImplementedError
    
    def sync_turn(self, role: str, content: str) -> None:
        """同步一轮对话到记忆"""
        raise NotImplementedError
    
    def shutdown(self) -> None:
        """关闭"""
        raise NotImplementedError

8.3 RL 训练

Hermes 内置 RL 研究支持，通过 tools/rl_training_tool.py（1396 行）与 Tinker-Atropos 集成。

核心概念

环境基类

class HermesAgentBaseEnv(BaseEnv):
    """Hermes Agent RL 环境基类"""
    
    def setup(self) -> None:
        """加载数据集、初始化环境"""
        raise NotImplementedError
    
    def get_next_item(self) -> dict:
        """获取下一个数据项（prompt）"""
        raise NotImplementedError
    
    def format_prompt(self, item: dict) -> str:
        """格式化 prompt"""
        raise NotImplementedError
    
    def compute_reward(self, trajectory: Trajectory) -> float:
        """
        计算奖励
        ToolContext 可访问所有工具调用
        """
        raise NotImplementedError
    
    def evaluate(self, num_episodes: int = 10) -> dict:
        """周期性评估"""
        raise NotImplementedError

工具函数

@registry.register(name="rl_list_environments", ...)
def _rl_list_environments(args, task_id=None):
    """列出可用 RL 环境"""
    ...

@registry.register(name="rl_select_environment", ...)
def _rl_select_environment(args, task_id=None):
    """选择并配置 RL 环境"""
    env_name = args["environment"]
    ...

@registry.register(name="rl_start_training", ...)
def _rl_start_training(args, task_id=None):
    """启动 RL 训练"""
    ...

@registry.register(name="rl_check_status", ...)
def _rl_check_status(args, task_id=None):
    """检查训练状态"""
    ...

@registry.register(name="rl_stop_training", ...)
def _rl_stop_training(args, task_id=None):
    """停止训练"""
    ...

8.4 委托代理 (Delegate)

委托工具（tools/delegate_tool.py）允许 Agent 创建子代理来处理子任务。

核心设计

父 Agent
    │
    ├── 创建子代理
    │   ├── 独立对话历史
    │   ├── 独立 task_id（独立 terminal session）
    │   ├── 受限 toolset
    │   └── 专注 system prompt
    │
    └── 阻塞等待所有子代理完成

委托限制

# 子代理永远无法使用的工具
DELEGATE_BLOCKED_TOOLS = frozenset({
    "delegate_task",   # 禁止嵌套委托
    "clarify",         # 禁止交互澄清
    "memory",          # 禁止直接记忆访问
    "send_message",    # 禁止直接发送消息
    "execute_code"     # 禁止代码执行
})

# 最大深度 = 2
# 父(0) → 子(1) → 孙辈被拒绝(2)

# 最大并发子代理 = 3
MAX_CONCURRENT_CHILDREN = 3

批量委托

# delegate_task 支持批量模式
delegate_task(
    tasks=[
        {"goal": "分析代码库 A", "toolsets": ["file", "terminal"]},
        {"goal": "分析代码库 B", "toolsets": ["file", "terminal"]},
        {"goal": "分析代码库 C", "toolsets": ["file", "terminal"]},
    ]
)
# 最多 3 个并发子代理

8.5 定时任务

Cron 系统（cron/）允许调度定时任务。

调度格式

创建任务

cronjob_tools(
    action="create",
    prompt="抓取 GitHub Trending 并更新网站",
    schedule="0 6 * * *",           # 每天 6 点
    name="github-trending-update",
    skills=["github-trending-update"],
    deliver="origin"                # 投递到创建时的聊天
)

Cron 环境特殊处理

# Cron 任务禁用以下功能
skip_memory = True              # 禁用用户画像
disabled_toolsets = [
    "cronjob",                   # 禁止创建定时任务
    "messaging",                 # 禁止直接发送消息
    "clarify"                    # 禁止交互澄清
]

# 超时处理
cron_timeout = 600  # 10 分钟不活跃则终止

防漂移机制

# 周期性任务在执行前推进 next_run_at
# 防止进程崩溃后重复执行

# 例如：every 2h 任务
# 14:00 执行 → next_run_at = 16:00
# 如果 15:00 进程崩溃 → 16:00 执行（不重复）