Context Engine Plugin 开发指南¶

ContextEnginePlugin (ABC)
  ├── bootstrap(state)       — 会话启动，初始化
  ├── ingest(messages, ctx)  — 逐轮消息摄入
  ├── assemble(ctx)          — 组装最终上下文
  ├── compact(messages, ctx) — 上下文压缩
  ├── on_compact_complete()  — 压缩完成回调
  └── on_session_end()       — 会话结束

ContextEnginePluginRegistry (Singleton)
  ├── register(plugin)
  ├── get(name) → plugin
  ├── get_default() → plugin
  └── list_plugins() → plugins

Builtin Plugins:
  ├── DefaultContextPlugin   — summarize + truncate
  └── DagCompressorPlugin    — 四层 DAG 压缩

# 查看可用插件
python -c "
from pyclaw.agents.context.plugins import ContextEnginePluginRegistry
for p in ContextEnginePluginRegistry.get_instance().list_plugins():
    print(f'{p[\"name\"]}: {p[\"description\"]}')
"

{
  context: {
    engine: {
      plugin: "dag-compressor",  // 使用 DAG 压缩插件
      compaction: {
        enabled: true,
        max_tokens: 128000,
        reserve_ratio: 0.3,
        strategy: "dag",  // dag | linear | hybrid
      },
      // DAG 层配置（仅 dag-compressor）
      layers: {
        critical: {
          max_tokens: 2000,
          strategy: "keep",
        },
        important: {
          max_tokens: 3000,
          strategy: "summarize",
        },
        contextual: {
          max_tokens: 2000,
          strategy: "extract_facts",
        },
        transient: {
          max_tokens: 0,
          strategy: "drop",
        },
      },
    },
  },
}

from pyclaw.agents.context.plugins import (
    ContextEnginePluginRegistry,
    DagCompressorPlugin,
    ContextState,
)

# 方式 1：注册并使用
registry = ContextEnginePluginRegistry.get_instance()
registry.register(DagCompressorPlugin(), set_default=True)

# 方式 2：直接使用
plugin = DagCompressorPlugin(total_max_tokens=128000)
state = ContextState(session_id="sess-1")

# 会话生命周期
await plugin.bootstrap(state)
await plugin.ingest(messages, state)
assembled = await plugin.assemble(state)
result = await plugin.compact(messages, state)

from pyclaw.agents.context.plugins.dag_compressor import _classify_message

# system → critical
_classify_message({"role": "system", "content": "You are a helpful assistant."})
# → "critical"

# tool_calls → important
_classify_message({"role": "assistant", "tool_calls": [{"name": "read_file"}]})
# → "important"

# tool result with data → important
_classify_message({"role": "tool", "content": "File contents: ..."})
# → "important"

# user instruction → critical
_classify_message({"role": "user", "content": "You must always respond in Chinese."})
# → "critical"

# short ack → transient (dropped)
_classify_message({"role": "assistant", "content": "ok"})
# → "transient"

from pyclaw.agents.context.plugins.dag_compressor import _extract_facts

messages = [
    {"role": "user", "content": "My email is alice@example.com. I prefer dark mode."},
    {"role": "assistant", "content": "We decided to use PostgreSQL for the database."},
]
facts = _extract_facts(messages)
# → "email: alice@example.com\npreference: I prefer dark mode\ndecision: decided to use PostgreSQL"

from pyclaw.agents.context.plugins.base import (
    ContextEnginePlugin,
    ContextState,
    CompactResult,
)

class MyPlugin(ContextEnginePlugin):
    @property
    def name(self) -> str:
        return "my-plugin"

    async def bootstrap(self, state: ContextState) -> None:
        """Session init — load persisted context, inject persona, etc."""
        pass

    async def ingest(self, messages: list[dict], state: ContextState) -> None:
        """Process new messages — tokenize, index, pre-process."""
        pass

    async def assemble(self, state: ContextState) -> list[dict]:
        """Assemble final context for LLM consumption."""
        return []  # return OpenAI-format message list

    async def compact(
        self,
        messages: list[dict],
        state: ContextState,
        *,
        max_tokens: int | None = None,
    ) -> CompactResult:
        """Compact messages when token budget exceeded."""
        return CompactResult(
            messages=messages,
            tokens_before=1000,
            tokens_after=800,
            tokens_saved=200,
        )

from pyclaw.agents.context.plugins import ContextEnginePluginRegistry

registry = ContextEnginePluginRegistry.get_instance()
registry.register(MyPlugin(), set_default=True)

# 使用插件
plugin = registry.get("my-plugin")
state = ContextState(session_id="s1")
await plugin.bootstrap(state)

async def on_compact_complete(self, result: CompactResult, state: ContextState) -> None:
    """Called after each compaction — persist summaries, cleanup."""
    await self._store.save_compaction_result(result)

async def on_session_end(self, state: ContextState, messages: list[dict]) -> None:
    """Called when session ends — flush caches, finalize state."""
    await self._store.close()

from pyclaw.agents.context.plugins import DefaultContextPlugin
from pyclaw.memory.context_store import ContextStore
from pathlib import Path

store = ContextStore(Path("data/context.db"))
await store.initialize()

class PersistentPlugin(DefaultContextPlugin):
    async def bootstrap(self, state: ContextState) -> None:
        await super().bootstrap(state)
        chunks = await store.get_chunks(state.session_id)
        for chunk in chunks:
            # 恢复压缩后的上下文
            self._load_chunk(chunk)

    async def on_compact_complete(self, result: CompactResult, state: ContextState) -> None:
        # 持久化压缩结果
        await store.save_chunk(
            session_id=state.session_id,
            chunk_id=f"compaction-{state.compaction_count}",
            content=result.summary or "",
        )