Hermes Agent 核心组件与工具系统

本文深入解析 Hermes Agent 的核心组件设计和工具系统的实现机制。

核心组件详解

AIAgent 类架构

AIAgent 是 Hermes 的核心编排类，负责管理整个智能体的生命周期。其设计采用前后端分离原则：

# run_agent.py 中的 AIAgent 核心流程
class AIAgent:
    def run_conversation(self, user_message: str, ...):
        while (api_call_count < self.max_iterations and 
               self.iteration_budget.remaining > 0) or self._budget_grace_call:
            if self._interrupt_requested:
                break
            response = client.chat.completions.create(
                model=model, 
                messages=messages, 
                tools=tool_schemas
            )
            if response.tool_calls:
                for tool_call in response.tool_calls:
                    result = handle_function_call(tool_call.name, tool_call.args, task_id)
                    messages.append(tool_result_message(result))
                api_call_count += 1
            else:
                return response.content

关键设计决策

1. 同步工具调用循环：

   • 工具调用在同步环境下执行，但工具内部可以是异步的
   • 使用 _run_async() 函数处理 async/sync 桥接
   • 持久化事件循环避免 "Event loop is closed" 错误

2. 预算控制：

   • IterationBudget 类跟踪剩余迭代次数
   • 支持优雅降级（grace call）防止在边界条件下提前中止

3. 中断处理：

   • 线程安全的中断标志 self._interrupt_requested
   • 网关环境下的双重消息队列保护

工具注册与发现机制

工具注册流程

工具系统采用零配置自动发现模式。每个工具文件在导入时自动注册：

# tools/example_tool.py
from tools.registry import registry

def check_requirements() -> bool:
    return bool(os.getenv("API_KEY"))

def example_tool(param: str, task_id: str = None) -> str:
    return json.dumps({"success": True, "data": "..."})

registry.register(
    name="example_tool",
    toolset="example",
    schema={
        "name": "example_tool",
        "description": "Tool description",
        "parameters": {...}
    },
    handler=lambda args, **kw: example_tool(param=args.get("param"), task_id=kw.get("task_id")),
    check_fn=check_requirements,
    requires_env=["API_KEY"],
)

注册中心架构

# tools/registry.py
class ToolRegistry:
    def register(self, name, toolset, schema, handler, check_fn, requires_env):
        # 存储工具定义
        self._tools[name] = {
            "schema": schema,
            "handler": handler,
            "toolset": toolset,
            "check_fn": check_fn,
            "requires_env": requires_env or []
        }
    
    def get_definitions(self, tool_names, quiet=False):
        # 返回可用的工具 schema，只包含 check_fn 通过的工具

工具集系统 (Toolsets)

工具集是 Hermes 中的核心资源隔离机制。每个平台可以配置不同的工具集：

# toolsets.py
_HERMES_CORE_TOOLS = [
    "web_search", "web_extract",           # Web 工具
    "terminal", "process",                 # 终端工具
    "read_file", "write_file", "patch",    # 文件工具
    "vision_analyze",                      # 视觉工具
    "browser_navigate",                    # 浏览器工具
    "todo", "memory",                      # 记忆工具
    "session_search",                      # 会话搜索
    "cronjob",                             # 定时任务
    "delegate_task",                       # 委派工具
]

TOOLSETS = {
    "hermes-cli": {
        "description": "Full interactive CLI toolset",
        "tools": _HERMES_CORE_TOOLS,
    },
    "hermes-telegram": {
        "description": "Telegram bot toolset",
        "tools": _HERMES_CORE_TOOLS,
    },
    # ... 平台特定的工具集
}

工具集解析过程

# model_tools.py 中的工具解析
def _compute_tool_definitions(enabled_toolsets, disabled_toolsets, quiet_mode):
    tools_to_include = set()
    
    # 1. 根据启用的工具集收集工具名
    for toolset_name in enabled_toolsets:
        resolved = resolve_toolset(toolset_name)
        tools_to_include.update(resolved)
    
    # 2. 从注册中心获取具体的工具定义
    filtered_tools = registry.get_definitions(tools_to_include, quiet=quiet_mode)
    
    # 3. 动态调整工具 schema（如 discord 动态权限）
    # 4. 移除跨工具引用（避免模型幻觉）
    
    return filtered_tools

工具调用深入

异步处理机制

工具系统支持异步工具，通过 _run_async() 实现 async/sync 桥接：

# agent/utils.py
def _run_async(coro):
    try:
        loop = asyncio.get_running_loop()
    except RuntimeError:
        loop = None
    
    if loop and loop.is_running():
        # 在已有事件循环中运行，使用线程池
        with ThreadPoolExecutor(max_workers=1) as pool:
            future = pool.submit(lambda: asyncio.run(coro))
            return future.result(timeout=300)
    else:
        # 使用主线程的持久循环
        return _get_tool_loop().run_until_complete(coro)

工具结果处理

工具结果经过多层处理：

1. 结果存储：大结果自动保存到临时文件，避免 token 溢出
2. 结果压缩：通过 enforce_turn_budget 控制结果大小
3. 结果分类：区分文件修改操作、会话相关操作等

# tools/tool_result_storage.py
def maybe_persist_tool_result(result: dict, max_size: int = 10000) -> dict:
    result_size = len(json.dumps(result))
    if result_size > max_size:
        # 保存到临时文件，返回引用
        file_path = persist_large_result(result)
        return {"persisted": True, "file_path": file_path}
    return {"persisted": False, "data": result}

记忆系统架构

MemoryManager 设计

记忆系统采用 多提供商编排 模式：

# agent/memory_manager.py
class MemoryManager:
    def __init__(self):
        self._providers: List[MemoryProvider] = []
        self._tool_to_provider: Dict[str, MemoryProvider] = {}
    
    def add_provider(self, provider: MemoryProvider):
        # 只允许一个外部记忆提供商
        if provider.name != "builtin" and self._has_external:
            logger.warning("Rejected: only one external provider allowed")
            return
        self._providers.append(provider)

记忆提供商接口

# agent/memory_provider.py
class MemoryProvider(ABC):
    @property
    @abstractmethod
    def name(self) -> str: ...
    
    @abstractmethod
    def prefetch(self, query: str, session_id: str = "") -> str: ...
    
    @abstractmethod
    def sync_turn(self, user: str, assistant: str, session_id: str = ""): ...
    
    @abstractmethod
    def handle_tool_call(self, tool_name: str, args: dict, **kwargs) -> str: ...
    
    @abstractmethod
    def get_tool_schemas(self) -> List[dict]: ...

记忆注入机制

记忆内容在对话开始前注入：

# run_agent.py
def run_conversation(self, user_message):
    # 1. 记忆预取
    context = self._memory_manager.prefetch_all(user_message)
    if context:
        messages.insert(1, {"role": "user", "content": build_memory_context_block(context)})
    
    # 2. 执行对话循环
    # ...
    
    # 3. 记忆同步
    self._memory_manager.sync_all(user_message, final_response)

会话状态管理

SessionDB 架构

# hermes_state.py
class SessionDB:
    def __init__(self, db_path: Path = None):
        self.db_path = db_path or get_hermes_home() / "state.db"
        self._conn = sqlite3.connect(
            str(self.db_path),
            check_same_thread=False,
            timeout=1.0,  # 短超时，应用层 retry
            isolation_level=None,  # 手动事务管理
        )
        apply_wal_with_fallback(self._conn, db_label="state.db")

写入优化

采用 BEGIN IMMEDIATE + jitter retry 策略：

def _execute_write(self, fn):
    last_err = None
    for attempt in range(self._WRITE_MAX_RETRIES):  # 15次重试
        try:
            with self._lock:
                self._conn.execute("BEGIN IMMEDIATE")  # 立即获取写锁
                result = fn(self._conn)
                self._conn.commit()
            return result
        except sqlite3.OperationalError as exc:
            if "locked" in str(exc).lower():
                time.sleep(random.uniform(0.020, 0.150))  # 随机退避
                continue
            raise

技能系统 (Skills)

SKILL.md 标准格式

---
name: example-skill
description: Demonstrates a specific capability in under 60 characters.
version: 1.0.0
author: Human Name <github>
license: MIT
metadata:
  hermes:
    tags: [tag1, tag2]
    category: devops
    related_skills: [other-skill]
    config:
      EXAMPLE_API_KEY:
        description: API key for the service
        prompt: Example Service API Key
        url: https://example.com/apikeys
---

# Example Skill

Brief intro...

## When to Use
...

## How to Run
Use the `terminal` tool to...

技能加载流程

1. 静态加载：CLI/Gateway 启动时扫描 ~/.hermes/skills/
2. 动态注入：通过 skill_commands.py 扫描并注入系统提示
3. 缓存控制：支持延迟失效（--now 标志）

总结

Hermes 的核心组件设计秉承以下原则：

1. 解耦：每个组件职责单一，接口清晰
2. 扩展性：插件系统支持运行时扩展
3. 企业级：内置限流、重试、并发控制
4. 开发友好：自动发现、热加载、清晰的错误处理