Hermes agent

Создание плагина движка контекста

Плагины движка контекста заменяют встроенный ContextCompressor другой стратегией управления контекстом диалога. Например, движок Lossless Context Management (LCM) строит граф знаний (DAG) вместо сжатия с потерями.

Принцип работы

Управление контекстом агента построено на абстрактном классе ContextEngine (agent/context_engine.py). Реализация по умолчанию — встроенный ContextCompressor. Плагины должны реализовать тот же интерфейс.

Одновременно активен только один движок контекста. Выбор задаётся через конфигурацию:

# config.yaml
context:
  engine: "compressor"    # встроенный по умолчанию
  engine: "lcm"           # активирует плагинный движок с именем "lcm"

Сами по себе плагинные движки никогда не включаются — пользователь должен явно задать context.engine с именем плагина.

Структура каталогов

Каждый движок контекста размещается в plugins/context_engine/<name>/:

plugins/context_engine/lcm/
├── __init__.py      # экспортирует подкласс ContextEngine
├── plugin.yaml      # метаданные (name, description, version)
└── ...              # прочие модули вашего движка

Абстрактный класс ContextEngine

Ваш движок должен реализовать следующие обязательные методы:

from agent.context_engine import ContextEngine

class LCMEngine(ContextEngine):

    @property
    def name(self) -> str:
        """Краткий идентификатор, например 'lcm'. Должен совпадать со значением в config.yaml."""
        return "lcm"

    def update_from_response(self, usage: dict) -> None:
        """Вызывается после каждого обращения к LLM со словарём usage.

        Обновите self.last_prompt_tokens, self.last_completion_tokens,
        self.last_total_tokens из ответа.
        """

    def should_compress(self, prompt_tokens: int = None) -> bool:
        """Возвращает True, если в этом ходу нужно запустить сжатие."""

    def compress(self, messages: list, current_tokens: int = None,
                 focus_topic: str = None) -> list:
        """Сжимает список сообщений и возвращает новый (возможно, укороченный) список.

        Возвращаемый список должен быть валидной последовательностью сообщений формата OpenAI.

        ``focus_topic`` — опциональная тема из ручной команды
        ``/compress <focus>``; движки с поддержкой управляемого сжатия должны
        приоритизировать сохранение связанной с ней информации, остальные могут игнорировать.
        """

Атрибуты класса, которые движок обязан поддерживать

Агент читает их напрямую для отображения и логирования:

last_prompt_tokens: int = 0
last_completion_tokens: int = 0
last_total_tokens: int = 0
threshold_tokens: int = 0        # порог, при котором срабатывает сжатие
context_length: int = 0          # полное контекстное окно модели
compression_count: int = 0       # сколько раз был вызван compress()

Опциональные методы

В абстрактном классе для них есть разумные реализации по умолчанию. Переопределяйте по необходимости:

МетодПо умолчаниюКогда переопределять
on_session_start(session_id, **kwargs)Пустая операцияНужно загрузить сохранённое состояние (DAG, БД)
on_session_end(session_id, messages)Пустая операцияНужно сбросить состояние, закрыть соединения
on_session_reset()Сбрасывает счётчики токеновЕсть состояние уровня сессии для очистки
update_model(model, context_length, ...)Обновляет context_length и порогНужно пересчитать бюджеты при смене модели
get_tool_schemas()Возвращает []Движок предоставляет инструменты для вызова агентом (например, lcm_grep)
handle_tool_call(name, args, **kwargs)Возвращает JSON с ошибкойРеализуете собственные обработчики инструментов
should_compress_preflight(messages)Возвращает FalseМожно дать дешёвую оценку до вызова API
get_status()Стандартный словарь токенов и порогаЕсть нестандартные метрики для отображения

Инструменты движка

Движок контекста может предоставлять инструменты, которые агент вызывает напрямую. Возвращайте схемы из get_tool_schemas() и обрабатывайте вызовы в handle_tool_call():

def get_tool_schemas(self):
    return [{
        "name": "lcm_grep",
        "description": "Поиск по графу знаний контекста",
        "parameters": {
            "type": "object",
            "properties": {
                "query": {"type": "string", "description": "Поисковый запрос"}
            },
            "required": ["query"],
        },
    }]

def handle_tool_call(self, name, args, **kwargs):
    if name == "lcm_grep":
        results = self._search_dag(args["query"])
        return json.dumps({"results": results})
    return json.dumps({"error": f"Unknown tool: {name}"})

Инструменты движка добавляются в набор инструментов агента при запуске, и вызовы к ним направляются автоматически — отдельно регистрировать их не нужно.

Регистрация

Через каталог (рекомендуется)

Разместите движок в plugins/context_engine/<name>/. В __init__.py должен экспортироваться подкласс ContextEngine. Система обнаружения найдёт его и создаст экземпляр сама.

Через общую систему плагинов

Обычный плагин тоже может зарегистрировать движок контекста:

def register(ctx):
    engine = LCMEngine(context_length=200000)
    ctx.register_context_engine(engine)

Зарегистрировать можно только один движок. Если зарегистрировать свой движок попробует второй плагин, такая попытка отклоняется с предупреждением.

Жизненный цикл

1. Создание экземпляра движка (загрузка плагина или автообнаружение из каталога)
2. on_session_start() — начало диалога
3. update_from_response() — после каждого вызова API
4. should_compress() — проверка на каждом ходу
5. compress() — вызывается, когда should_compress() вернул True
6. on_session_end() — граница сессии (выход из CLI, /reset, истечение шлюза)

on_session_reset() вызывается при /new или /reset — очищает состояние уровня сессии без полной остановки.

Конфигурация

Пользователь выбирает движок через hermes plugins → Provider Plugins → Context Engine либо правкой config.yaml:

context:
  engine: "lcm"   # должно совпадать со свойством name вашего движка

Блок compression в конфигурации (compression.threshold, compression.protect_last_n и т. д.) относится только к встроенному ContextCompressor. Если вашему движку нужны свои параметры — задайте собственный формат и читайте их из config.yaml при инициализации.

Тестирование

from agent.context_engine import ContextEngine

def test_engine_satisfies_abc():
    engine = YourEngine(context_length=200000)
    assert isinstance(engine, ContextEngine)
    assert engine.name == "your-name"

def test_compress_returns_valid_messages():
    engine = YourEngine(context_length=200000)
    msgs = [{"role": "user", "content": "hello"}]
    result = engine.compress(msgs)
    assert isinstance(result, list)
    assert all("role" in m for m in result)

Полный набор тестов контракта абстрактного класса — в tests/agent/test_context_engine.py.

См. также

⌕ ESC