🏠 Главная › developer guide › context engine plugin

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

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

Как это работает

Управление контекстом агента построено на ABC 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)
└──...              # любые другие модули, необходимые вашему движку

ABC 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()

Необязательные методы

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

Метод По умолчанию Переопределить, когда
on_session_start(session_id, **kwargs) Ничего не делаю Необходимо загрузить сохранённое состояние (DAG, БД)
on_session_end(session_id, messages) Ничего не делаю Нужно сбросить состояние, закрыть соединения
on_session_reset() Сбрасывает счётчики жетонов Есть нынешняя сессия, которую нужно сохранить
update_model(модель, context_length,...) Обновляет context_length + порог Необходимо пересчитать бюджеты при смене модели
get_tool_schemas() Возвращает [] Ваш движок предоставляет инструменты, вызываемые агентом (например, lcm_grep)
handle_tool_call(name, args, **kwargs) Возвращает JSON с ошибкой Вы реализуете обработчики инструментов
should_compress_preflight(сообщения) Возвращает 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"Неизвестный инструмент: {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 для проверки состояния сессии без полного завершения.

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

Пользователи выбирают ваш движок через hermesplugins → 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)

Полный набор контрактов ABC см. в tests/agent/test_context_engine.py.

См. также