插件开发

插件架构

Hermes Agent 采用基于插件的架构，支持组件的动态加载和卸载。每个插件都是一个独立的模块，与核心代理集成。

┌─────────────────────────────────────────────────────────────┐ │ Hermes Agent Core │ ├─────────────────────────────────────────────────────────────┤ │ ┌─────────────────────────────────────────────────────────┐│ │ │ Plugin Registry ││ │ │ ┌───────────┐ ┌───────────┐ ┌───────────┐ ┌──────────┐ ││ │ │ │ memory │ │context_eng│ │ model │ │ kanban │ ││ │ │ │ plugins │ │ _engine │ │_providers │ │ obs. │ ││ │ │ └───────────┘ └───────────┘ └───────────┘ └──────────┘ ││ │ └─────────────────────────────────────────────────────────┘│ │ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │ │ │Tool Registry │ │ Session │ │ Gateway │ │ │ │ │ │ Manager │ │ │ │ │ └──────────────┘ └──────────────┘ └──────────────┘ │ └─────────────────────────────────────────────────────────────┘

核心组件

插件注册表

中央注册表，管理插件生命周期：发现、加载、初始化和卸载。

插件加载器

动态导入系统，在启动时从 plugins/ 目录加载插件。

钩子系统

允许插件钩入代理事件：on_message、on_tool_call、on_response 等。

配置管理器

通过 YAML/JSON 处理插件配置，支持验证和热重载。

插件类型

Hermes Agent 支持五种主要插件类别，每个类别扩展系统的不同方面：

memory

记忆插件

实现对话历史记录的持久化存储后端。默认：SQLite + FTS5。示例：Redis 记忆、向量存储记忆。

context_engine

上下文引擎插件

控制上下文的构建和压缩方式。处理轨迹压缩、RAG 检索和上下文窗口管理。

model-providers

模型提供者插件

添加对新 LLM 后端的支持。实现 Provider 接口以连接任何 API 兼容的模型服务。

kanban

看板插件

任务管理集成。连接项目管理工具，如 Trello、Notion、Linear 或自定义看板系统。

observability

可观测性插件

监控、日志记录和指标。与 Prometheus、Grafana、OpenTelemetry 或自定义分析管道集成。

注册表模式

插件使用 Python 的入口点机制和中央注册表进行自动发现。每个插件通过标准化清单声明其类型和功能。

插件清单 (plugin.json)

{
  "name": "my-memory-plugin",
  "version": "1.0.0",
  "type": "memory",
  "entry": "my_memory_plugin:main",
  "description": "Redis-backed memory storage",
  "config_schema": {
    "redis_url": {"type": "string", "required": true},
    "ttl": {"type": "integer", "default": 3600}
  }
}
            

注册表 API

from astrbot.core.registry import PluginRegistry

registry = PluginRegistry.get_instance()

# Get all plugins of a specific type
memory_plugins = registry.get_plugins_by_type("memory")

# Get plugin by name
plugin = registry.get_plugin("my-memory-plugin")

# Register a new plugin
registry.register(plugin_instance, metadata)
            

编写插件

以下是使用 Redis 作为后端创建自定义记忆插件的完整示例：

1

创建插件结构

在 plugins/ 下创建新目录，包含插件文件。

文件：plugins/redis_memory/init.py

"""
Redis Memory Plugin - Hermes Agent
Provides Redis-backed persistent memory storage.
"""

import json
import redis
from typing import List, Optional, Dict, Any
from datetime import timedelta

from astrbot.core.plugin import (
    BasePlugin,
    plugin_manager,
    PluginMeta
)
from astrbot.core.memory import BaseMemory


@plugin_manager.register
class RedisMemoryPlugin(BasePlugin):
    """Redis-backed memory implementation for Hermes Agent."""
    
    meta = PluginMeta(
        name="redis_memory",
        version="1.0.0",
        description="Redis-backed memory storage with TTL support",
        author="Your Name"
    )
    
    def __init__(self, config: Dict[str, Any]):
        super().__init__(config)
        self.redis_url = config.get("redis_url", "redis://localhost:6379/0")
        self.ttl = config.get("ttl", 3600)
        self._client = None
    
    def initialize(self):
        """Initialize Redis connection."""
        self._client = redis.from_url(self.redis_url, decode_responses=True)
        self.log.info(f"Redis Memory initialized: {self.redis_url}")
    
    def store(self, session_id: str, messages: List[Dict]):
        """Store messages for a session with TTL."""
        key = f"memory:{session_id}"
        self._client.setex(
            key,
            timedelta(seconds=self.ttl),
            json.dumps(messages)
        )
    
    def retrieve(self, session_id: str, limit: int = 100) -> List[Dict]:
        """Retrieve messages for a session."""
        key = f"memory:{session_id}"
        data = self._client.get(key)
        if data:
            messages = json.loads(data)
            return messages[-limit:]
        return []
    
    def search(self, query: str, session_id: Optional[str] = None) -> List[Dict]:
        """Full-text search across memories."""
        # Redis search implementation
        pattern = f"memory:{session_id or '*'}"
        keys = self._client.keys(pattern)
        results = []
        for key in keys:
            messages = json.loads(self._client.get(key) or "[]")
            for msg in messages:
                if query.lower() in str(msg.get("content", "")).lower():
                    results.append(msg)
        return results
    
    def clear(self, session_id: str):
        """Clear memory for a session."""
        self._client.delete(f"memory:{session_id}")
    
    def teardown(self):
        """Cleanup resources."""
        if self._client:
            self._client.close()


# Plugin entry point
def main(config: Dict[str, Any]) -> RedisMemoryPlugin:
    """Entry point for the plugin system."""
    return RedisMemoryPlugin(config)
            

2

注册入口点

添加到您的 pyproject.toml 或 setup.py：

[project.entry-points."astrbot.plugins"]
redis_memory = "redis_memory:main"
            

3

在 astrbot_config.yml 中配置

启用并配置您的插件：

plugins:
  redis_memory:
    enabled: true
    redis_url: "redis://localhost:6379/0"
    ttl: 7200  # 2 hour session TTL
            

插件生命周期

了解插件生命周期有助于确保正确的资源管理：

发现

插件加载器在启动时扫描 plugins/ 目录和入口点。

加载

插件模块被导入，__init__ 被调用并传入配置。

初始化

initialize() 被调用。建立连接、加载资源。

活跃状态

插件钩入事件并响应代理请求。

清理

在关闭或卸载时调用 teardown()。清理资源。

热重载

插件可通过 API 调用在代理运行时重新加载。

钩子参考

插件可以注册钩子来拦截和处理事件：

钩子名称	描述
`on_message`	拦截传入的用户消息
`on_tool_call`	处理工具调用事件
`on_response`	在传递前处理代理响应
`on_session_start`	会话初始化钩子
`on_session_end`	会话清理钩子
`on_error`	全局错误处理钩子