上下文引擎插件

上下文引擎插件用管理对话上下文的替代策略替换内置的 ContextCompressor 。例如，无损上下文管理（LCM）引擎构建知识DAG而不是有损摘要。

它是如何工作的

代理的上下文管理构建在 ContextEngine ABC (agent/context_engine.py) 之上。内置 ContextCompressor 是默认实现。插件引擎必须实现相同的接口。

一次只能有一个上下文引擎处于活动状态。选择是配置驱动的：

# config.yaml
context:
  engine: "compressor"    # default built-in
  engine: "lcm"           # activates a plugin engine named "lcm"

插件引擎永远不会自动激活 - 用户必须显式将 context.engine 设置为插件的名称。

目录结构

每个上下文引擎都位于 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

ContextEngine ABC

您的引擎必须实现这些必需的方法：

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) -> list:
        """Compact the message list and return a new (possibly shorter) list.

        The returned list must be a valid OpenAI-format message sequence.
        """

你的引擎必须维护的类属性

代理直接读取这些内容以进行显示和记录：

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

可选方法

这些在 ABC 中有合理的默认值。根据需要覆盖：

方法	默认	覆盖时
on_session_start(session_id, **kwargs)	on_session_start(session_id, **kwargs)无操作	您需要加载持久状态（DAG、DB）
on_session_end(session_id, messages)	on_session_end(session_id, messages)无操作	您需要刷新状态，关闭连接
on_session_reset()	on_session_reset()重置令牌计数器	您需要清除每个会话的状态
update_model(model, context_length, ...)	update_model(model, context_length, ...)更新 context_length + 阈值	您需要重新计算型号切换预算
get_tool_schemas()	get_tool_schemas()返回 []	您的引擎提供代理可调用工具（例如 lcm_grep）
handle_tool_call(name, args, **kwargs)	handle_tool_call(name, args, **kwargs)返回错误 JSON	您实施工具处理程序
should_compress_preflight(messages)	should_compress_preflight(messages)返回 False	您可以进行廉价的 API 调用前估算
get_status()	get_status()标准令牌/阈值字典	您有要公开的自定义指标

引擎工具

上下文引擎可以公开代理直接调用的工具。从 get_tool_schemas() 返回模式并处理 handle_tool_call() 中的调用：

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}"})

引擎工具在启动时注入到代理的工具列表中并自动调度 - 无需注册表注册。

登记

通过目录（推荐）

将您的引擎置于 plugins/context_engine/<name>/ 中。 __init__.py 必须导出 ContextEngine 子类。发现系统自动找到并实例化它。

通过通用插件系统

通用插件还可以注册上下文引擎：

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

只能注册一台引擎。尝试注册的第二个插件被拒绝并出现警告。

生命周期

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 expiry)

在 /new 或 /reset 上调用 on_session_reset() 以清除每个会话状态，而无需完全关闭。

配置

用户通过 hermes plugins → Provider Plugins → Context Engine 选择您的引擎，或者通过编辑 config.yaml：

context:
  engine: "lcm"   # must match your engine's name property

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 了解完整的 ABC 合约测试套件。

另请参阅

• Context Compression and Caching — 内置压缩机如何工作
• Memory Provider Plugins — 类似的内存单选插件系统
• Plugins — 一般插件系统概述