H繁中版
文件開發者指南context engine plugin
<!-- Source: https://hermesbible.com/docs/developer-guide/context-engine-plugin -->

建立 Context Engine 外掛

Context engine 外掛用替代策略取代內建的 ContextCompressor 來管理對話上下文。例如,一個無損上下文管理(LCM)引擎,它建構知識 DAG 而非有損的摘要化。

運作原理

代理程式的上下文管理建構在 ContextEngine ABC(agent/context_engine.py)之上。內建的 ContextCompressor 是預設實作。外掛引擎必須實作相同的介面。

同一時間只能有一個 context engine 啟動。選擇由設定驅動:

# config.yaml
context:
  engine: "compressor"    # 內建預設
  engine: "lcm"           # 啟動名為 "lcm" 的外掛引擎

外掛引擎永遠不會自動啟動 — 使用者必須明確設定 context.engine 為外掛的名稱。

目錄結構

每個 context engine 位於 plugins/context_engine/<name>/

plugins/context_engine/lcm/
├── __init__.py      # 匯出 ContextEngine 子類別
├── plugin.yaml      # 中繼資料(名稱、描述、版本)
└── ...              # 引擎所需的其他模組

ContextEngine ABC

你的引擎必須實作以下必要方法:

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()重設 token 計數器你有需要清除的每個 session 狀態
update_model(model, context_length, ...)更新 context_length + threshold你需要在模型切換時重新計算預算
get_tool_schemas()回傳 []你的引擎提供代理程式可呼叫的工具(例如 lcm_grep
handle_tool_call(name, args, **kwargs)回傳錯誤 JSON你實作了工具處理器
should_compress_preflight(messages)回傳 False你可以進行低成本的 API 呼叫前預估
get_status()標準 token/threshold 字典你有自訂指標要暴露

引擎工具

Context engine 可以暴露代理程式直接呼叫的工具。從 get_tool_schemas() 回傳 schema,並在 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}"})

引擎工具會在啟動時注入到代理程式的工具列表中並自動分派 — 不需要 registry 註冊。

註冊方式

透過目錄(建議)

將你的引擎放在 plugins/context_engine/<name>/ 中。__init__.py 必須匯出一個 ContextEngine 子類別。發現系統會自動找到並實例化它。

透過通用外掛系統

通用外掛也可以註冊 context engine:

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() — Session 邊界(CLI 退出、/reset、閘道器過期)

on_session_reset()/new/reset 時呼叫,用於清除每個 session 的狀態而不需完整關閉。

設定

使用者透過 hermes plugins → Provider Plugins → Context Engine 選擇你的引擎,或直接編輯 config.yaml

context:
  engine: "lcm"   # 必須與你引擎的 name 屬性相符

compression 設定區塊(compression.thresholdcompression.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

另請參閱



Cron Internals