7 · 上下文引擎 ★

agent/context_engine.py，490 行。这个文件不做任何实际工作 —— 它只定义一份契约。但它是 Hermes 架构立场最集中的体现。

7.1 它定义的是什么

「只有一个引擎生效」这句话很重要 —— 它把上下文引擎归类为「互斥策略」而不是「可叠加能力」。第 9 章会讲这个区分为什么必须在插件系统层面就做出来。

7.2 生命周期

"""
Lifecycle:
  1. Engine is instantiated and registered (plugin register() or default)
  2. on_session_start() called when a conversation begins
  3. update_from_response() called after each API response with usage data
  4. should_compress() checked after each turn
  5. compress() called when should_compress() returns True
  6. on_session_end() called at real session boundaries (CLI exit, /reset,
     gateway session expiry) — NOT per-turn
"""

注意第 6 步那句「NOT per-turn」（不是每轮）。这是一个容易搞错的地方：

7.3 三个必须实现的方法

class ContextEngine(ABC):

    @property
    @abstractmethod
    def name(self) -> str:
        """Short identifier (e.g. 'compressor', 'lcm')."""

    @abstractmethod
    def update_from_response(self, usage: Dict[str, Any]) -> None:
        """Update tracked token usage from an API response."""

    @abstractmethod
    def should_compress(self, prompt_tokens: int = None) -> bool:
        """Return True if compaction should fire this turn."""

    @abstractmethod
    def compress(self, messages, current_tokens=None, focus_topic=None,
                 force=False, memory_context="") -> List[Dict[str, Any]]:
        """Compact the message list and return the new message list."""

用量字典的向后兼容设计

这是一份「接口演进」的教科书示范：旧字段永不删除（保证老引擎能跑），新字段可选（保证新引擎能用上更细的数据），而且在文档里明确写出兼容性契约。

compress 的四个可选参数各有用途

而且文档写明了参数演进的处理方式：「较老的引擎可以省略这个参数；宿主会按签名过滤掉不支持的可选参数。」—— 宿主用反射检查引擎方法的签名，只传它接受的参数。这样新增参数不会破坏老引擎。

7.4 最精辟的设计：select 和 compress 是两个正交动词

def select_context(
    self,
    request_messages: List[Dict[str, Any]],
    *,
    conversation_messages: List[Dict[str, Any]] = None,
    incoming_message: Dict[str, Any] = None,
    budget_tokens: int = 0,
) -> List[Dict[str, Any]]:
    """Optionally choose/replace the context for THIS request, pre-generation."""
    return None      # 默认空操作

这个接口是被真实的误用逼出来的

还原这个故事：

关键约束：只作用于本次请求

缓存契约写得比什么都清楚

7.5 后置观察钩子

def on_turn_complete(self, messages, usage: Dict[str, Any] = None, **kwargs) -> None:
    """Observe a finished user turn (post-turn ingestion / observation)."""
    return None

这是 select_context() 的对称面：选择发生在请求之前，观察发生在轮次之后。

一段诚实的覆盖范围说明

7.6 其他可选钩子

# 不调模型的确定性裁剪
def prune_tool_results_only(self, messages, current_tokens=None) -> tuple[List, int]:
    return messages, 0                     # 默认安全空操作

# 便宜的预检
def should_compress_preflight(self, messages) -> bool:
    return False
def should_defer_preflight_to_real_usage(self, rough_tokens: int) -> bool:
    return False

# 手动 /compress 的预检守卫
def has_content_to_compress(self, messages) -> bool:
    return True

# 会话生命周期
def on_session_start(self, session_id: str, **kwargs) -> None
def on_session_end(self, session_id: str, messages) -> None
def on_session_reset(self) -> None

# ★ 引擎可以自带工具
def get_tool_schemas(self) -> List[Dict[str, Any]]:
    return []
def handle_tool_call(self, name: str, args: Dict[str, Any], **kwargs) -> str

# 状态显示
def get_status(self) -> Dict[str, Any]

# 模型切换
def update_model(self, model, context_length, base_url="", api_key="",
                 provider="", api_mode="") -> None

不调模型的裁剪：为什么要单独一个钩子

关键在于「独立的低触发器」。使用 100 万 token 窗口的模型时，should_compress 可能几十轮都不触发 —— 但那些旧的工具输出每一轮都在被重发、每一轮都在花钱。所以需要一个成本导向的、和「会不会超窗口」无关的裁剪触发器。

引擎可以自带工具

get_tool_schemas() / handle_tool_call() 让引擎向模型暴露自己的工具。文档举的例子是：LCM 引擎可以提供 lcm_grep、lcm_describe、lcm_expand 这些工具 —— 也就是让模型能主动去搜索、描述、展开被折叠的上下文。

7.7 默认参数值

threshold_percent: float = 0.75      # 用到窗口的 75% 就开始压缩
protect_first_n:   int   = 3         # 开头保护 3 条（系统提示词之外）
protect_last_n:    int   = 6         # 结尾保护 6 条

emit_automatic_compaction_status: bool = True    # 自动压缩要不要通知用户

protect_first_n 的语义有一条演进说明：

7.8 用户可见状态的可控性

def automatic_compaction_status_message(engine, *, phase: str,
                                        default_message: str, **context) -> str | None:
    """Resolve host-visible status for an automatic compaction event.

    Engines can suppress routine automatic status with
    ``emit_automatic_compaction_status = False`` or customize it by defining
    ``get_automatic_compaction_status_message(...)``. Empty strings and
    ``None`` mean "do not emit a lifecycle status".
    """

这个设计考虑的是：不同引擎对「压缩」这件事的定位不同。

• 内置压缩器：压缩是大事（有损、不可逆），应该通知用户
• 某个检索型引擎：上下文重组是常规后台维护，每轮都在做，通知用户只会造成噪音

而且分得很细：「警告、错误、以及用户显式执行的手动命令，仍然会通知」 —— 只有「例行的自动成功」可以被静默。

7.9 内置实现的体量对比