跳到主要內容

構建上下文引擎插件

上下文引擎插件會替換內置的 ContextCompressor，採用替代策略來管理對話上下文。例如，一種無損上下文管理（LCM）引擎，它構建知識有向無環圖（DAG）而非有損摘要。

工作原理​

Agent 的上下文管理基於 ContextEngine 抽象基類（agent/context_engine.py）。內置的 ContextCompressor 是默認實現。插件引擎必須實現相同的接口。

同一時間只能激活一個上下文引擎。選擇由配置驅動：

# config.yaml
context:
  engine: "compressor"    # 默認內置
  engine: "lcm"           # 激活名為“0”的插件引擎

插件引擎不會自動激活——用戶必須顯式將 context.engine 設置為插件名稱。

目錄結構​

每個上下文引擎位於 plugins/context_engine/<name>/ 目錄下：

plugins/context_engine/lcm/
├── __init__.py      # 導出 ContextEngine 子類
├── plugin.yaml      # 元數據（名稱、描述、版本）
└── ...              # 您的引擎需要的任何其他模塊

ContextEngine 抽象基類​

您的引擎必須實現以下必需方法：

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.
        """

引擎必須維護的類屬性​

Agent 會直接讀取這些屬性用於顯示和日誌記錄：

last_prompt_tokens: int = 0
last_completion_tokens: int = 0
last_total_tokens: int = 0
threshold_tokens: int = 0        # 當compression觸發時
context_length: int = 0          # model 的完整 context window
compression_count: int = 0       # compress() 運行了多少次

可選方法​

這些方法在 ABC 中已有合理默認實現。按需重寫：

方法默認行為重寫場景
on_session_start(session_id, **kwargs)空操作需要加載持久化狀態（DAG、數據庫）
on_session_end(session_id, messages)空操作需要刷新狀態、關閉連接
on_session_reset()重置 token 計數器有需要清除的會話級狀態
update_model(model, context_length, ...)更新 context_length + threshold模型切換時需要重新計算預算
get_tool_schemas()返回 []引擎提供 Agent 可調用的工具（如 lcm_grep）
handle_tool_call(name, args, **kwargs)返回錯誤 JSON實現工具處理器
should_compress_preflight(messages)返回 False可在 API 調用前進行低成本預估
get_status()返回標準的 token/threshold 字典有自定義指標需要暴露

引擎工具​

上下文引擎可以向 Agent 暴露直接調用的工具。通過 get_tool_schemas() 返回工具 schema，並在 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}"})

引擎工具會在 Agent 啟動時自動注入到工具列表中並被自動分發——無需註冊到註冊表。

註冊​

將您的引擎放置於 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)

on_session_reset() 在 /new 或 /reset 時被調用，用於清除會話級狀態，而無需完全關閉。

配置​

用戶通過 hermes plugins → Provider Plugins → Context Engine 選擇您的引擎，或通過編輯 config.yaml 進行配置：

context:
  engine: "lcm"   # 必須與您的引擎的名稱屬性匹配

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 合約測試套件。

參見​