Hermes Agent 插件系统与扩展机制

本文详解 Hermes Agent 的插件系统和扩展机制，这是构建企业级智能体的关键技术。

插件系统架构

插件类型

Hermes 支持三种插件类型：

1. 标准插件：在plugins/目录下的 Python 包
2. Pipeline 插件：通过tools.pipeline.*提供数据处理链
3. Custom 插件：用户自定义的plugins/custom/

插件加载机制

# cli.py 中的插件发现
from hermes_cli.plugins import discover_plugins

def _discover():
    from hermes_cli.plugins import (
        get_enabled_plugins,
        install_plugin_sources,
    )
    
    # 1. 检查项目级配置
    project_config = get_project_plugin_sources()
    # 2. 发现所有已安装的插件源
    plugin_sources = install_plugin_sources()
    # 3. 创建插件实例
    enabled_plugins = get_enabled_plugins(plugin_sources)
    return enabled_plugins

记忆提供商插件

Honcho 记忆提供商

honcho 是 Hermes 内置的记忆提供商，通过 archivy 后端实现：

# plugins/memory/honcho_plugin.py
class HonchoMemoryProvider:
    def __init__(self, config: dict):
        self.honcho = Archivy(
            url=config.get("HONCHO_URL"),
            api_token=config.get("HONCHO_TOKEN")
        )
    
    def prefetch(self, query: str, session_id: str = "") -> str:
        entries = self.honcho.search(query)
        # 编排 entries 为上下文格式
        return format_entries(entries)

Mem0 记忆提供商

基于向量数据库的记忆提供商：

# plugins/memory/mem0_plugin.py
class Mem0MemoryProvider:
    def __init__(self, config: dict):
        self.vector_db = Mem0VectorDB(
            database_url=config.get("MEM0_URL"),
            embedding_model=config.get("MEM0_EMBEDDING_MODEL"),
        )
    
    def prefetch(self, query: str, session_id: str = "") -> str:
        # 向量相似度搜索
        results = self.vector_db.search(query, top_k=5)
        return format_memories(results)

记忆管理器集成

# agent/memory_manager.py
class MemoryManager:
    def __init__(self):
        self._providers: List[MemoryProvider] = []
        self._has_external = False  # 最多一个外部提供商
    
    def add_provider(self, provider: MemoryProvider):
        is_builtin = provider.name == "builtin"
        if not is_builtin:
            if self._has_external:
                logger.warning("Only one external provider allowed")
                return
            self._has_external = True
        
        self._providers.append(provider)
        # 注册工具
        for schema in provider.get_tool_schemas():
            name = schema.get("name", "")
            if name and name not in self._tool_to_provider:
                self._tool_to_provider[name] = provider

模型提供商插件

模型提供商接口

# plugins/model-providers/provider_interface.py
class ModelProvider(ABC):
    @property
    @abstractmethod
    def name(self) -> str: ...
    
    @abstractmethod
    def get_client(self, provider_config: dict) -> "Client": ...
    
    @abstractmethod
    def supports_tool_calling(self) -> bool: ...
    
    @abstractmethod
    def supports_streaming(self) -> bool: ...
    
    @abstractmethod
    def get_available_models(self) -> List[str]: ...

内置提供商

Pipeline 插件

Pipeline API

Pipeline 允许在工具调用前后注入自定义逻辑：

# tools/pipeline/tool_test_pipeline.py
from hermes_cli.tools.pipeline import PipelineHook

class BeforeExecutionHook(PipelineHook):
    def on_pre_tool_call(self, tool_name: str, args: dict, ...) -> dict:
        # 在工具执行前运行
        if tool_name == "terminal":
            validate_dangerous_command(args.get("command"))
        return args

class AfterExecutionHook(PipelineHook):
    def on_post_tool_call(self, tool_name: str, result: dict, ...) -> dict:
        # 在工具执行后运行
        if tool_name == "execute_code":
            result = sanitize_output(result)
        return result

插件配置

配置文件位置

# ~/.hermes/config.yaml
plugins:
  # 全局启用/禁用
  enabled:
    - example_plugin
  disabled:
    - slow_plugin
  
  # 插件级配置
  example_plugin:
    api_key: "secret"
    timeout: 30
  
  # 记忆提供商配置
  memory:
    provider: "honcho"
    HONCHO_URL: "http://localhost:8000"
    HONCHO_TOKEN: "..."
  
  # 模型提供商配置
  model:
    provider: "anthropic"
    API_KEY: "sk-..."

Profile 级插件

# ~/.hermes/profiles/production.yaml
plugins:
  enabled:
    - telemetry
    - compliance

model:
  provider: "anthropic"
  model: "claude-opus-4-20250514"

扩展 Agent 功能

方法一：创建自定义工具

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

def check_fn() -> bool:
    return True

def handler(args: dict, **kwargs) -> str:
    return json.dumps({"result": "success"})

registry.register(
    name="my_tool",
    toolset="custom",
    schema={
        "name": "my_tool",
        "description": "Custom tool for my use case",
        "parameters": {...}
    },
    handler=handler,
    check_fn=check_fn,
)

方法二：创建记忆提供商

# plugins/memory/my_memory.py
from agent.memory_provider import MemoryProvider

class MyMemoryProvider(MemoryProvider):
    @property
    def name(self) -> str:
        return "my_memory"
    
    def prefetch(self, query: str, session_id: str = "") -> str:
        # 实现预取逻辑
        memories = fetch_from_database(query)
        return format_memories(memories)
    
    def sync_turn(self, user: str, assistant: str, session_id: str = ""):
        # 实现同步逻辑
        save_to_database(user, assistant, session_id)

方法三：创建 Pipeline 插件

# tools/pipeline/my_pipeline.py
from hermes_cli.tools.pipeline import PipelineHook

class MyBeforeHook(PipelineHook):
    def on_pre_tool_call(self, tool_name: str, args: dict, ...) -> dict:
        # 添加审计日志
        logger.info(f"Tool {tool_name} called with args")
        return args

多实例隔离

Profile 机制

# ~/.hermes/profiles/production.yaml
model:
  provider: "anthropic"
  model: "claude-opus-4-20250514"

gateway:
  platforms:
    - telegram

plugins:
  enabled:
    - compliance

# ~/.hermes/profiles/development.yaml
model:
  provider: "openai"
  model: "gpt-4o-mini"

gateway:
  platforms:
    - discord

plugins:
  enabled:
    - debug_mode

启动命令：

hermes -p production run
hermes -p development run

插件调试技巧

查看已加载插件和状态

hermes plugins list

在交互中测试

hermes
> plugins
> tools my_tool

安全考虑

1. 插件权限：自定义工具在独立的代码执行环境中运行
2. API Key 隔离：每个 Profile 的 API Key 独立
3. 工具命名空间：避免工具名冲突
4. 沙箱执行：execute_code工具在受限环境中运行

总结

Hermes 的插件系统提供了：

1. 运行时扩展：无需重启即可加载新功能
2. 多实例隔离：通过 Profile 机制实现环境隔离
3. 插件市场：支持第三方插件安装
4. 安全沙箱：代码执行工具的隔离执行环境