| --- |
| sidebar_position: 9 |
| title: "Context Engine Plugins" |
| description: "How to build a context engine plugin that replaces the built-in ContextCompressor" |
| --- |
| |
| # Building a Context Engine Plugin |
|
|
| Context engine plugins replace the built-in `ContextCompressor` with an alternative strategy for managing conversation context. For example, a Lossless Context Management (LCM) engine that builds a knowledge DAG instead of lossy summarization. |
|
|
| ## How it works |
|
|
| The agent's context management is built on the `ContextEngine` ABC (`agent/context_engine.py`). The built-in `ContextCompressor` is the default implementation. Plugin engines must implement the same interface. |
|
|
| Only **one** context engine can be active at a time. Selection is config-driven: |
|
|
| ```yaml |
| # config.yaml |
| context: |
| engine: "compressor" # default built-in |
| engine: "lcm" # activates a plugin engine named "lcm" |
| ``` |
|
|
| Plugin engines are **never auto-activated** — the user must explicitly set `context.engine` to the plugin's name. |
|
|
| ## Directory structure |
|
|
| Each context engine lives in `plugins/context_engine/<name>/`: |
|
|
| ``` |
| plugins/context_engine/lcm/ |
| ├── __init__.py # exports the ContextEngine subclass |
| ├── plugin.yaml # metadata (name, description, version) |
| └── ... # any other modules your engine needs |
| ``` |
|
|
| ## The ContextEngine ABC |
|
|
| Your engine must implement these **required** methods: |
|
|
| ```python |
| from agent.context_engine import ContextEngine |
| |
| class LCMEngine(ContextEngine): |
| |
| @property |
| def name(self) -> str: |
| """Short identifier, e.g. 'lcm'. Must match config.yaml value.""" |
| return "lcm" |
| |
| def update_from_response(self, usage: dict) -> None: |
| """Called after every LLM call with the usage dict. |
| |
| Update self.last_prompt_tokens, self.last_completion_tokens, |
| self.last_total_tokens from the response. |
| """ |
| |
| def should_compress(self, prompt_tokens: int = None) -> bool: |
| """Return True if compaction should fire this turn.""" |
| |
| def compress(self, messages: list, current_tokens: int = None, |
| focus_topic: str = None) -> list: |
| """Compact the message list and return a new (possibly shorter) list. |
| |
| The returned list must be a valid OpenAI-format message sequence. |
| |
| ``focus_topic`` is an optional topic string from manual |
| ``/compress <focus>``; engines that support guided compression should |
| prioritise preserving information related to it, others may ignore it. |
| """ |
| ``` |
|
|
| ### Class attributes your engine must maintain |
|
|
| The agent reads these directly for display and logging: |
|
|
| ```python |
| last_prompt_tokens: int = 0 |
| last_completion_tokens: int = 0 |
| last_total_tokens: int = 0 |
| threshold_tokens: int = 0 # when compression triggers |
| context_length: int = 0 # model's full context window |
| compression_count: int = 0 # how many times compress() has run |
| ``` |
|
|
| ### Optional methods |
|
|
| These have sensible defaults in the ABC. Override as needed: |
|
|
| | Method | Default | Override when | |
| |--------|---------|--------------| |
| | `on_session_start(session_id, **kwargs)` | No-op | You need to load persisted state (DAG, DB) | |
| | `on_session_end(session_id, messages)` | No-op | You need to flush state, close connections | |
| | `on_session_reset()` | Resets token counters | You have per-session state to clear | |
| | `update_model(model, context_length, ...)` | Updates context_length + threshold | You need to recalculate budgets on model switch | |
| | `get_tool_schemas()` | Returns `[]` | Your engine provides agent-callable tools (e.g., `lcm_grep`) | |
| | `handle_tool_call(name, args, **kwargs)` | Returns error JSON | You implement tool handlers | |
| | `should_compress_preflight(messages)` | Returns `False` | You can do a cheap pre-API-call estimate | |
| | `get_status()` | Standard token/threshold dict | You have custom metrics to expose | |
| | `select_context(request_messages, *, conversation_messages, incoming_message, budget_tokens)` | Returns `None` (no-op) | You select/route which context enters **this** request (retrieval, topic routing) — see below | |
| | `on_turn_complete(messages, usage=None, **kwargs)` | No-op | You ingest/index/observe the finished turn — see below | |
|
|
| ## Per-turn context selection and observation |
|
|
| `compress()` answers "context is too long → make it shorter". Two optional, |
| no-op-default hooks cover the orthogonal *selection / observation* axis, so an |
| engine no longer has to force `should_compress()` to `True` and abuse |
| `compress()` as a per-turn callback: |
|
|
| ```python |
| def select_context(self, request_messages, *, conversation_messages=None, |
| incoming_message=None, budget_tokens=0): |
| """Choose/replace the context for THIS request, before dispatch. |
| |
| Return a new message list to use for this one provider call (retrieval, |
| topic routing, role/branch switching), or None to leave it unchanged. |
| Request-only: the persisted conversation history is never mutated. |
| """ |
| |
| def on_turn_complete(self, messages, usage=None, **kwargs): |
| """Observe a finished turn after the assistant/tool loop completes. |
| |
| Receives a shallow copy of the finalized transcript plus the turn's |
| canonical usage dict (or None if no provider response was reached), so the |
| engine can ingest/index/summarize for the next select_context(). The return |
| value is ignored. |
| """ |
| ``` |
|
|
| Contract: |
|
|
| - **No-op by default, fail-open.** Both default to `return None`. A missing hook, an exception, or an invalid return value leaves the request untouched — so a failing engine is never worse than not installing one. The host also identity-checks for the inherited ABC default and skips it entirely, so non-implementing engines (including the built-in compressor) pay no per-request work at all. |
| - **`select_context()` is request-only.** The returned list replaces the messages for a single provider call; persisted history is never written. Returning `None`, `[]`, a non-list, or a list containing non-dicts all fall open to the unmodified request. |
| - **Ordering / cache stability.** The hook runs **before** prompt cache-control and every request sanitizer, so (a) a replacement still passes the same validation as any request, and (b) the no-op default leaves the request byte-identical — prompt-cache behaviour is unchanged for non-implementing engines. An engine that replaces the list changes only its own cache prefix. Evaluated per provider request (re-runs on retries). |
| - **`on_turn_complete()`** is post-turn observation only; treat `messages` as read-only. **Coverage is best-effort:** it fires from the standard turn-finalization seam. Some abnormal early-return paths in the loop (e.g. a content-policy block or a provider terminal failure) persist and return without routing through finalization, so they do not currently emit this hook — treat it as a best-effort observation for completed turns, not a guaranteed callback for every early exit. Unifying all terminal paths behind one finalization seam is a separate follow-up. |
| |
| ### When to use these hooks — and when NOT to |
| |
| - **Implement `select_context()` only when your engine must *replace* the |
| per-request context** — retrieval-augmented selection, topic/branch routing, |
| role switching. It is the only verb that can swap which messages enter a |
| request: the `pre_llm_call` plugin hook is inject-only by documented design |
| (it appends to the user message and never rewrites the list, to preserve the |
| prompt-cache prefix). If you don't need replacement, don't implement it. |
| - **If your plugin only needs post-turn observation / ingestion** (indexing, |
| memory sync, analytics), implement a **memory provider** (`sync_turn()` — |
| see [Memory Provider Plugins](./memory-provider-plugin.md)) instead of a |
| context engine. A context engine takes ownership of the session's compaction |
| policy; a memory provider observes turns without owning anything. |
| `on_turn_complete()` exists as the observation mirror for engines that |
| *already* need `select_context()` — so the same component can learn from the |
| turn it just routed — not as a general-purpose turn callback. |
| - **Prompt-cache impact of a real `select_context()`.** A non-no-op selection |
| naturally changes the prompt-cache prefix for the turns where it changes the |
| selection — that request's prefix no longer matches the provider's cached |
| prefix, so those turns re-write cache instead of reading it. Engines should |
| return **stable selections when nothing has changed** (same object or an |
| equal list), and only reshape the context when the routing decision actually |
| differs; a selection that shuffles per turn silently forfeits cache reuse |
| every turn. |
| |
| ## Engine tools |
| |
| Context engines can expose tools the agent calls directly. Return schemas from `get_tool_schemas()` and handle calls in `handle_tool_call()`: |
| |
| ```python |
| def get_tool_schemas(self): |
| return [{ |
| "name": "lcm_grep", |
| "description": "Search the context knowledge graph", |
| "parameters": { |
| "type": "object", |
| "properties": { |
| "query": {"type": "string", "description": "Search query"} |
| }, |
| "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}"}) |
| ``` |
| |
| Engine tools are injected into the agent's tool list at startup and dispatched automatically — no registry registration needed. |
|
|
| ## Registration |
|
|
| ### Via directory (recommended) |
|
|
| Place your engine in `plugins/context_engine/<name>/`. The `__init__.py` must export a `ContextEngine` subclass. The discovery system finds and instantiates it automatically. |
|
|
| ### Via general plugin system |
|
|
| A general plugin can also register a context engine: |
|
|
| ```python |
| def register(ctx): |
| engine = LCMEngine(context_length=200000) |
| ctx.register_context_engine(engine) |
| ``` |
|
|
| Only one engine can be registered. A second plugin attempting to register is rejected with a warning. |
|
|
| ## Lifecycle |
|
|
| ``` |
| 1. Engine instantiated (plugin load or directory discovery) |
| 2. on_session_start() — conversation begins |
| 3. update_from_response() — after each API call |
| 4. should_compress() — checked each turn |
| 5. compress() — called when should_compress() returns True |
| 6. on_session_end() — session boundary (CLI exit, /reset, gateway shutdown) |
| ``` |
|
|
| `on_session_reset()` is called on `/new` or `/reset` to clear per-session state without a full shutdown. |
|
|
| ## Configuration |
|
|
| Users select your engine via `hermes plugins` → Provider Plugins → Context Engine, or by editing `config.yaml`: |
|
|
| ```yaml |
| context: |
| engine: "lcm" # must match your engine's name property |
| ``` |
|
|
| The `compression` config block (`compression.threshold`, `compression.protect_last_n`, etc.) is specific to the built-in `ContextCompressor`, with one explicit exception: `compression.model_thresholds` (per-model threshold overrides) is part of the context-engine contract. The host assigns the resolved map to `engine.model_thresholds` *before* the initial `update_model()` call, and the base-class `update_model()` applies it (longest substring match, falling back to the engine's configured threshold). Engines that override `update_model()` own their own compaction policy and may honor or ignore the map — `from agent.context_compressor import resolve_model_threshold` to reuse the same resolution logic. For everything else, your engine should define its own config format if needed, reading from `config.yaml` during initialization. |
|
|
| ## Testing |
|
|
| ```python |
| 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) |
| ``` |
|
|
| See `tests/agent/test_context_engine.py` for the full ABC contract test suite. |
|
|
| ## Thread safety |
|
|
| When `compression.context_timeout_seconds > 0` (the default), Hermes runs the |
| whole compression pass — including your engine's `compress()` and boundary |
| callbacks, and any memory provider's `on_pre_compress` / |
| `on_session_switch` — on a pooled daemon thread with a host-side timeout. |
| Your engine must therefore assume: |
|
|
| - Calls may arrive on an arbitrary pooled thread. Do not rely on thread |
| affinity or `threading.local` state shared with the conversation thread. |
| - The message list you receive is a private deep snapshot; mutating it in |
| place is allowed (legacy contract), but the mutation only becomes visible |
| if the pass commits. After a host timeout your still-running work is |
| discarded — never publish to external/durable state outside the commit. |
| - Passes for *different* sessions can run concurrently on pool siblings; a |
| single engine/provider instance shared across sessions must be thread-safe. |
|
|
| ## See also |
|
|
| - [Context Compression and Caching](/developer-guide/context-compression-and-caching) — how the built-in compressor works |
| - [Memory Provider Plugins](/developer-guide/memory-provider-plugin) — analogous single-select plugin system for memory |
| - [Plugins](/user-guide/features/plugins) — general plugin system overview |
|
|
