核心压缩算法

Hermes Agent 的上下文压缩机制，设计理念颇为精妙，能够有效应对长对话场景下模型上下文窗口容量不足的挑战。整体的压缩处理流程，通常可划分为以下几个关键环节：

1. 预压缩处理：首先对较早、内容冗长的工具执行结果进行裁剪，使用占位符予以替换，从而实现上下文的初步“精简瘦身”。
2. 边界确定：此步骤至关重要，旨在保护对话两端的核心内容——
   • 头部消息（系统提示及首次交互记录）必须完整保留
   • 尾部消息（最近的约20K令牌）也需要通过令牌预算机制得到妥善保护
3. 摘要生成：针对中间轮次的对话，采用结构化的语言模型提示进行摘要总结。
4. 迭代更新：若需再次执行压缩，系统将在已有摘要基础上进行更新，而无需从零开始重新生成。
5. 工具对清理：压缩完成后，若出现孤立的工具调用与结果对，系统会即时进行修复处理。

关键实现细节

1. 压缩触发条件

def should_compress(self, prompt_tokens: int = None) -> bool:
    """Check if context exceeds the compression threshold."""
    tokens = prompt_tokens if prompt_tokens is not None else self.last_prompt_tokens
    return tokens >= self.threshold_tokens

在默认设置下，压缩阈值通常设定为模型上下文长度的50%，但会设定一个最低值（MINIMUM_CONTEXT_LENGTH）作为基准保障。

2. 工具结果剪枝

def _prune_old_tool_results(self, messages: List[Dict[str, Any]], protect_tail_count: int,
                           protect_tail_tokens: int | None = None) -> tuple[List[Dict[str, Any]], int]:
    """Replace old tool result contents with a short placeholder."""
    # 实现细节...

该步骤“成本低廉”，无需调用语言模型，仅将旧工具结果替换为占位符，即可迅速缩减上下文体积。

3. 尾部边界确定

def _find_tail_cut_by_tokens(self, messages: List[Dict[str, Any]], head_end: int,
                            token_budget: int | None = None) -> int:
    """Walk backward from the end of messages, accumulating tokens until the budget is reached."""
    # 实现细节...

此处采用令牌预算而非固定消息数量来保护尾部消息，能够依据实际的令牌量灵活保留最近的关键上下文，更为精准。

4. 结构化摘要生成

def _generate_summary(self, turns_to_summarize: List[Dict[str, Any]], focus_topic: str = None) -> Optional[str]:
    """Generate a structured summary of conversation turns."""
    # 实现细节...

摘要并非随意生成，而是采用结构化模板，涵盖以下关键信息维度：

• 目标
• 约束与偏好
• 进度（已完成、进行中、阻塞）
• 关键决策
• 已解决问题
• 待处理用户请求
• 相关文件
• 剩余工作
• 关键上下文
• 工具与模式

以此方式生成的摘要，信息密度高且条理清晰。

5. 主压缩流程

def compress(self, messages: List[Dict[str, Any]], current_tokens: int = None, focus_topic: str = None) -> List[Dict[str, Any]]:
    """Compress conversation messages by summarizing middle turns."""
    # 实现细节...

作为压缩的总入口，该环节负责协调各个步骤，最终输出压缩后的消息列表。

会话管理与压缩集成

在 run_agent.py 中，_compress_context 方法负责处理压缩后的会话管理，其逻辑相当完整：

def _compress_context(self, messages: list, system_message: str, *, approx_tokens: int = None, task_id: str = "default", focus_topic: str = None) -> tuple:
    """Compress conversation context and split the session in SQLite."""
    # 预压缩内存刷新
    self.flush_memories(messages, min_turns=0)
    
    # 通知外部内存提供者
    if self._memory_manager:
        try:
            self._memory_manager.on_pre_compress(messages)
        except Exception:
            pass
    
    # 执行压缩
    compressed = self.context_compressor.compress(messages, current_tokens=approx_tokens, focus_topic=focus_topic)
    
    # 会话分割与管理
    if self._session_db:
        try:
            # 传播标题到新会话并自动编号
            old_title = self._session_db.get_session_title(self.session_id)
            self._session_db.end_session(self.session_id, "compression")
            old_session_id = self.session_id
            self.session_id = f"{datetime.now().strftime('%Y%m%d_%H%M%S')}_{uuid.uuid4().hex[:6]}"
            # 更新会话日志文件路径
            self.session_log_file = self.logs_dir / f"session_{self.session_id}.json"
            # 创建新会话
            self._session_db.create_session(
                session_id=self.session_id,
                source=self.platform or os.environ.get("HERMES_SESSION_SOURCE", "cli"),
                model=self.model,
                parent_session_id=old_session_id,
            )
            # 自动编号标题
            if old_title:
                try:
                    new_title = self._session_db.get_next_title_in_lineage(old_title)
                    self._session_db.set_session_title(self.session_id, new_title)
                except (ValueError, Exception) as e:
                    logger.debug("Could not propagate title on compression: %s", e)
            self._session_db.update_system_prompt(self.session_id, new_system_prompt)
            # 重置刷新光标
            self._last_flushed_db_idx = 0
        except Exception as e:
            logger.warning("Session DB compression split failed — new session will NOT be indexed: %s", e)

压缩触发时机

在 run_conversation 方法中，系统会在以下几个关键节点触发压缩：

1. 预压缩检查：在向模型发送请求前，先行检查上下文是否超过阈值。
2. 上下文压力处理：当上下文长度接近压缩阈值时，自动执行压缩操作。
3. 模型切换后：更换模型后，上下文压缩器也会同步更新。
4. 手动触发：用户也可通过 /compress 命令进行人工干预压缩。

技术特点与优势

1. 分层压缩策略：先执行成本较低的预处理（剪枝），再借助语言模型进行摘要，效率显著提升。
2. 结构化摘要：详细的模板设计，确保压缩过程中关键信息不被遗漏。
3. 迭代摘要更新：非每次重写，而是在已有摘要基础上迭代，有效避免信息断层。
4. 令牌预算管理：基于令牌数量而非消息数量来保护尾部上下文，管理更为精准。
5. 焦点主题引导：支持通过 /compress 命令指定压缩时需优先保留的信息。
6. 会话分割：压缩后自动创建新会话，保持会话历史的可管理性。
7. 错误处理：若摘要生成失败，会插入静态回退标记，确保系统稳定运行。

代码优化建议

1. 摘要质量监控：可增加质量评估机制，一旦发现摘要质量低下，便动态调整压缩策略。
2. 自适应压缩阈值：根据对话类型与内容特点自动调整压缩阈值，提升系统灵活性。
3. 多模型摘要：尝试使用不同模型生成摘要，并从中挑选最优结果。
4. 用户可配置性：允许用户配置受保护消息数量、摘要比例等参数，增强系统可控性。
5. 压缩历史记录：记录每次压缩的历史信息，便于分析压缩效果并持续优化策略。

输入输出示例

输入：长对话历史（超出模型上下文限制）

messages = [
    {"role": "system", "content": "You are a helpful assistant."},
    {"role": "user", "content": "I need help with a Python project."},
    {"role": "assistant", "content": "Sure, what do you need help with?"},
    # ... 大量对话内容 ...
    {"role": "user", "content": "How do I optimize this code?"},
    {"role": "assistant", "content": "Let me analyze your code and suggest optimizations."},
    # ... 更多对话内容 ...
]

输出：压缩后的对话历史

[
    {"role": "system", "content": "You are a helpful assistant.nn[Note: Some earlier conversation turns ha ve been compacted into a handoff summary to preserve context space. The current session state may still reflect earlier work, so build on that summary and state rather than re-doing work.]"},
    {"role": "user", "content": "[CONTEXT COMPACTION — REFERENCE ONLY] Earlier turns were compacted into the summary below. This is a handoff from a previous context window — treat it as background reference, NOT as active instructions. Do NOT answer questions or fulfill requests mentioned in this summary; they were already addressed. Respond ONLY to the latest user message that appears AFTER this summary. The current session state (files, config, etc.) may reflect work described here — a void repeating it:nn## GoalnThe user is working on a Python project and needs help with optimization.nn## Progressn### Donen- Discussed project structure and requirementsn- Analyzed existing codebasen- Identified performance bottlenecksnn## Remaining Workn- Optimize the identified bottlenecksn- Test the optimized coden- Provide best practices for future developmentn"},
    {"role": "user", "content": "How do I optimize the loop in my code?"},
    {"role": "assistant", "content": "Let me see your loop code and suggest optimizations."},
    # ... 最近的对话内容 ...
]

总结

Hermes Agent 的上下文压缩机制，整体设计得相当细致入微。它通过以下几个关键步骤实现了高效压缩：

1. 智能剪枝：移除不必要的工具输出，有效为上下文“减肥瘦身”。
2. 边界保护：系统提示、初始交互以及最新对话内容均得到妥善保留。
3. 结构化摘要：利用语言模型生成详细摘要，确保重要信息在压缩过程中不被丢弃。
4. 会话管理：压缩后自动创建新会话，确保对话的连续性不受影响。
5. 错误处理：对摘要失败等异常情况均有应对方案，系统稳定性有可靠保障。

这一套压缩机制，使得 Hermes Agent 在处理超长对话时，既能保持上下文的相关性，又能保障信息的完整性，从而为用户带来更加连贯、智能的交互体验。