概述

上下文压缩是 Hermes Agent 的核心特性之一，用于在对话历史超过模型上下文窗口限制时，自动压缩中间轮次，保留关键信息的同时释放上下文空间。

---

压缩触发条件

1. 预检压缩（Preflight Compression）

在 run_conversation() 开始主循环之前：

# 位置: run_agent.py 第 7563-7613 行
if (
    self.compression_enabled
    and len(messages) > self.context_compressor.protect_first_n
                        + self.context_compressor.protect_last_n + 1
):
    # 计算请求 token 数（包括工具 schema）
    _preflight_tokens = estimate_request_tokens_rough(
        messages,
        system_prompt=active_system_prompt or "",
        tools=self.tools or None,
    )
    
    # 超过阈值（默认 50% 上下文窗口）
    if _preflight_tokens >= self.context_compressor.threshold_tokens:
        # 执行压缩
        messages, active_system_prompt = self._compress_context(
            messages, system_message, approx_tokens=_preflight_tokens,
            task_id=effective_task_id,
        )

触发条件:

• 对话历史超过 protect_first_n + protect_last_n + 1 条消息
• 估算 token 数 ≥ context_length * threshold_percent（默认 50%）

2. 循环中压缩

在工具执行后的循环中：

# 位置: run_agent.py 第 9546-9556 行
if self.compression_enabled and _compressor.should_compress(_real_tokens):
    self._safe_print("  ⟳ compacting context…")
    messages, active_system_prompt = self._compress_context(
        messages, system_message,
        approx_tokens=self.context_compressor.last_prompt_tokens,
        task_id=effective_task_id,
    )

触发条件:

• prompt_tokens >= threshold_tokens（通过 should_compress() 判断）

3. 错误恢复压缩

遇到特定错误时触发：

3.1 上下文溢出错误

# 位置: run_agent.py 第 8755-8881 行
is_context_length_error = (
    classified.reason == FailoverReason.context_overflow
)

if is_context_length_error:
    # 降低上下文限制
    new_ctx = get_next_probe_tier(old_ctx)
    compressor.update_model(
        model=self.model,
        context_length=new_ctx,
        ...
    )
    # 执行压缩
    messages, active_system_prompt = self._compress_context(
        messages, system_message, approx_tokens=approx_tokens,
        task_id=effective_task_id,
    )

3.2 Payload Too Large 错误（HTTP 413）

# 位置: run_agent.py 第 8703-8749 行
is_payload_too_large = (
    classified.reason == FailoverReason.payload_too_large
)

if is_payload_too_large:
    messages, active_system_prompt = self._compress_context(
        messages, system_message, approx_tokens=approx_tokens,
        task_id=effective_task_id,
    )

3.3 Anthropic 长上下文层级限制

# 位置: run_agent.py 第 8631-8680 行
if classified.reason == FailoverReason.long_context_tier:
    # 降低上下文限制到 200k
    _reduced_ctx = 200000
    compressor.update_model(
        model=self.model,
        context_length=_reduced_ctx,
        ...
    )
    # 执行压缩
    messages, active_system_prompt = self._compress_context(
        messages, system_message, approx_tokens=approx_tokens,
        task_id=effective_task_id,
    )

---

压缩算法详解

核心入口：compress() 方法

位置：agent/context_compressor.py 第 623-766 行

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

    Algorithm:
      1. Prune old tool results (cheap pre-pass, no LLM call)
      2. Protect head messages (system prompt + first exchange)
      3. Find tail boundary by token budget (~20K tokens of recent context)
      4. Summarize middle turns with structured LLM prompt
      5. On re-compression, iteratively update the previous summary
    """

实例演示

假设一个长对话场景：

# 初始状态
messages = [
    {"role": "system", "content": "You are Hermes..."},           # 消息 0
    {"role": "user", "content": "Fix the bug in app.py"},         # 消息 1
    {"role": "assistant", "content": "I'll help you...", "tool_calls": [...]},  # 消息 2
    {"role": "tool", "content": "[file content]", ...},           # 消息 3
    {"role": "assistant", "content": "...", "tool_calls": [...]}, # 消息 4
    {"role": "tool", "content": "[large output]", ...},           # 消息 5
    # ... 更多中间轮次 ...
    {"role": "assistant", "content": "...", "tool_calls": [...]}, # 消息 48
    {"role": "tool", "content": "[result]", ...},                 # 消息 49
    {"role": "assistant", "content": "I've fixed the bug..."},    # 消息 50
]

# 模型上下文窗口: 200,000 tokens
# 当前估算: 150,000 tokens
# 阈值: 100,000 tokens (50%)
# → 触发压缩

---

Phase 1: 修剪旧工具结果（Tool Result Pruning）

目的: 快速释放空间，无需 LLM 调用

# 位置: agent/context_compressor.py 第 166-221 行
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."""
    
    # 计算保护边界（按 token 预算）
    accumulated = 0
    boundary = len(result)
    for i in range(len(result) - 1, -1, -1):
        msg = result[i]
        msg_tokens = len(msg.get("content", "")) // 4 + 10
        if accumulated + msg_tokens > protect_tail_tokens:
            boundary = i
            break
        accumulated += msg_tokens
    
    # 替换边界之前的工具结果
    for i in range(prune_boundary):
        msg = result[i]
        if msg.get("role") == "tool":
            content = msg.get("content", "")
            if len(content) > 200:
                result[i] = {**msg, "content": "[Old tool output cleared to save context space]"}
                pruned += 1

实例结果:

# 压缩前（消息 5，工具结果有 5000 字符）
{"role": "tool", "content": "def main():\n    print('Hello')\n    # ... 5000 chars ...", "tool_call_id": "call_abc"}

# 压缩后
{"role": "tool", "content": "[Old tool output cleared to save context space]", "tool_call_id": "call_abc"}

节省: 每个大工具结果约 1000+ tokens

---

Phase 2: 确定压缩边界

2.1 保护头部消息

# 位置: agent/context_compressor.py 第 657-659 行
compress_start = self.protect_first_n  # 默认 3
compress_start = self._align_boundary_forward(messages, compress_start)

边界对齐: 避免在工具调用/结果组中间切割

def _align_boundary_forward(self, messages: List[Dict[str, Any]], idx: int) -> int:
    """Push a compress-start boundary forward past any orphan tool results."""
    while idx < len(messages) and messages[idx].get("role") == "tool":
        idx += 1
    return idx

实例:

# protect_first_n = 3
messages = [
    {"role": "system", ...},    # 消息 0 ✓ 保护
    {"role": "user", ...},      # 消息 1 ✓ 保护
    {"role": "assistant", ...}, # 消息 2 ✓ 保护
    {"role": "tool", ...},      # 消息 3 - 工具结果
    {"role": "assistant", ...}, # 消息 4 - 压缩起点
]

# compress_start = 3，但消息 3 是 tool
# _align_boundary_forward 将其推进到 4

2.2 保护尾部消息（Token 预算）

# 位置: agent/context_compressor.py 第 561-617 行
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, accumulating tokens until budget is reached."""
    
    # 默认预算 = summary_target_ratio * threshold_tokens
    # 例: 0.20 * 100,000 = 20,000 tokens
    
    accumulated = 0
    cut_idx = n
    
    for i in range(n - 1, head_end - 1, -1):
        msg = messages[i]
        msg_tokens = len(content) // 4 + 10
        # 包含工具调用参数
        for tc in msg.get("tool_calls") or []:
            args = tc.get("function", {}).get("arguments", "")
            msg_tokens += len(args) // 4
        
        # 超过软上限（1.5x 预算）则停止
        if accumulated + msg_tokens > soft_ceiling and (n - i) >= min_tail:
            break
        accumulated += msg_tokens
        cut_idx = i
    
    # 对齐到工具组边界
    cut_idx = self._align_boundary_backward(messages, cut_idx)
    return cut_idx

最终边界:

# 完整的保护策略
head_messages = messages[0:compress_start]           # 0-3: 系统提示 + 首次交互
middle_messages = messages[compress_start:compress_end]  # 4-34: 待压缩
tail_messages = messages[compress_end:]              # 35-50: 最近上下文

---

Phase 3: 生成结构化摘要

3.1 序列化中间轮次

# 位置: agent/context_compressor.py 第 247-296 行
def _serialize_for_summary(self, turns: List[Dict[str, Any]]) -> str:
    """Serialize conversation turns into labeled text for the summarizer."""
    
    parts = []
    for msg in turns:
        role = msg.get("role", "unknown")
        content = msg.get("content") or ""
        
        # 工具结果: 保留足够内容
        if role == "tool":
            if len(content) > 6000:
                content = content[:4000] + "\n...[truncated]...\n" + content[-1500:]
            parts.append(f"[TOOL RESULT {tool_id}]: {content}")
        
        # 助手消息: 包含工具调用参数
        elif role == "assistant":
            tool_calls = msg.get("tool_calls", [])
            if tool_calls:
                tc_parts = []
                for tc in tool_calls:
                    name = tc.get("function", {}).get("name", "?")
                    args = tc.get("function", {}).get("arguments", "")
                    if len(args) > 1500:
                        args = args[:1200] + "..."
                    tc_parts.append(f"  {name}({args})")
                content += "\n[Tool calls:\n" + "\n".join(tc_parts) + "\n]"
            parts.append(f"[ASSISTANT]: {content}")
        
        # 用户消息
        else:
            parts.append(f"[{role.upper()}]: {content}")
    
    return "\n\n".join(parts)

实例输出:

[USER]: Fix the bug in app.py where the login fails

[ASSISTANT]: I'll analyze the login issue. Let me read the file first.
[Tool calls:
  read_file(path="app.py")
]

[TOOL RESULT call_abc]: from flask import Flask, request, jsonify
from auth import authenticate

app = Flask(__name__)

@app.route('/login', methods=['POST'])
def login():
    data = request.get_json()
    username = data.get('username')
    password = data.get('password')
    # Bug: missing validation
    result = authenticate(username, password)
    return jsonify(result)
...[truncated]...

[ASSISTANT]: I found the issue. The authenticate function doesn't validate input.
[Tool calls:
  patch(path="app.py", old_string="result = authenticate(username, password)", new_string="if not username or not password:\n    return jsonify({'error': 'Missing credentials'}), 400\nresult = authenticate(username, password)")
]

[TOOL RESULT call_def]: Successfully patched app.py

[USER]: Now add logging

[ASSISTANT]: I'll add logging to track login attempts.
[Tool calls:
  patch(path="app.py", old_string="...", new_string="import logging\nlogging.basicConfig(level=logging.INFO)\n...")
]
...

3.2 计算摘要预算

# 位置: agent/context_compressor.py 第 227-236 行
def _compute_summary_budget(self, turns_to_summarize: List[Dict[str, Any]]) -> int:
    """Scale summary token budget with the amount of content being compressed."""
    
    content_tokens = estimate_messages_tokens_rough(turns_to_summarize)
    budget = int(content_tokens * 0.20)  # 20% of original content
    return max(2000, min(budget, 12000))  # 2K-12K tokens

实例:

# 中间轮次: 31 条消息，约 80,000 tokens
# 摘要预算 = 80,000 * 0.20 = 16,000
# 上限 = 12,000 tokens
# → 最终预算: 12,000 tokens

3.3 调用 LLM 生成摘要

首次压缩

# 位置: agent/context_compressor.py 第 365-404 行
prompt = f"""Create a structured handoff summary for a later assistant that will continue this conversation after earlier turns are compacted.

TURNS TO SUMMARIZE:
{content_to_summarize}

Use this exact structure:

## Goal
[What the user is trying to accomplish]

## Constraints & Preferences
[User preferences, coding style, constraints, important decisions]

## Progress
### Done
[Completed work — include specific file paths, commands run, results obtained]
### In Progress
[Work currently underway]
### Blocked
[Any blockers or issues encountered]

## Key Decisions
[Important technical decisions and why they were made]

## Relevant Files
[Files read, modified, or created — with brief note on each]

## Next Steps
[What needs to happen next to continue the work]

## Critical Context
[Any specific values, error messages, configuration details, or data that would be lost without explicit preservation]

## Tools & Patterns
[Which tools were used, how they were used effectively, and any tool-specific discoveries]

Target ~12000 tokens. Be specific — include file paths, command outputs, error messages, and concrete values rather than vague descriptions.
"""

迭代更新（后续压缩）

# 位置: agent/context_compressor.py 第 320-363 行
if self._previous_summary:
    prompt = f"""You are updating a context compaction summary. A previous compaction produced the summary below. New conversation turns have occurred since then and need to be incorporated.

PREVIOUS SUMMARY:
{self._previous_summary}

NEW TURNS TO INCORPORATE:
{content_to_summarize}

Update the summary using this exact structure. PRESERVE all existing information that is still relevant. ADD new progress. Move items from "In Progress" to "Done" when completed. Remove information only if it is clearly obsolete.

## Goal
[What the user is trying to accomplish — preserve from previous summary, update if goal evolved]

## Constraints & Preferences
[User preferences, coding style, constraints, important decisions — accumulate across compactions]

## Progress
### Done
[Completed work — include specific file paths, commands run, results obtained]
### In Progress
[Work currently underway]
### Blocked
[Any blockers or issues encountered]
...
"""

3.4 摘要示例输出

[CONTEXT COMPACTION] Earlier turns in this conversation were compacted to save context space. The summary below describes work that was already completed, and the current session state may still reflect that work.

## Goal
Fix a login bug in app.py and add logging for authentication attempts.

## Constraints & Preferences
- Python/Flask backend
- User prefers defensive programming with input validation
- Logging should be at INFO level

## Progress
### Done
- Identified missing input validation in `/login` endpoint (app.py:10-14)
- Patched authenticate call to validate username/password before authentication
- Added `import logging` and configured basic logging
- Added logging statement before authentication attempt

### In Progress
- Testing the login endpoint with various inputs
- Adding rate limiting for login attempts

### Blocked
None currently

## Key Decisions
- Chose to validate at the endpoint level rather than in `authenticate()` to keep authentication logic pure
- Used basic logging instead of structured logging to minimize dependencies

## Relevant Files
- `app.py` - Main Flask application, login endpoint modified
- `auth.py` - Authentication module (read only, not modified)

## Next Steps
1. Test login with missing credentials (should return 400)
2. Test login with valid credentials (should succeed)
3. Verify logging output in console
4. Consider adding rate limiting for brute-force protection

## Critical Context
- Bug location: `/login` endpoint in app.py
- Error: authenticate() was called with potentially None values
- Fix: Added validation before calling authenticate
- Logging format: `logging.info(f"Login attempt for user: {username}")`

## Tools & Patterns
- `read_file` used to examine app.py structure
- `patch` tool successful for surgical edits (preferred over write_file for small changes)
- Truncation limit: content > 6000 chars truncated to 4000 head + 1500 tail

---

Phase 4: 组装压缩后的消息列表

# 位置: agent/context_compressor.py 第 694-753 行
compressed = []

# 1. 添加头部消息
for i in range(compress_start):
    msg = messages[i].copy()
    # 首次压缩时在系统提示中添加提示
    if i == 0 and msg.get("role") == "system" and self.compression_count == 0:
        msg["content"] = (
            (msg.get("content") or "")
            + "\n\n[Note: Some earlier conversation turns have 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.]"
        )
    compressed.append(msg)

# 2. 插入摘要消息（选择合适的角色）
if last_head_role in ("assistant", "tool"):
    summary_role = "user"
else:
    summary_role = "assistant"

compressed.append({"role": summary_role, "content": summary})

# 3. 添加尾部消息
for i in range(compress_end, n_messages):
    compressed.append(messages[i].copy())

# 4. 清理孤立工具调用/结果
compressed = self._sanitize_tool_pairs(compressed)

角色选择逻辑

目的: 保持消息角色交替规则

# 示例场景
head = [
    {"role": "system", ...},      # 位置 0
    {"role": "user", ...},        # 位置 1
    {"role": "assistant", ...},   # 位置 2 (last_head_role = "assistant")
]

# 方案 1: summary_role = "user"
compressed = [
    {"role": "system", ...},
    {"role": "user", ...},
    {"role": "assistant", ...},
    {"role": "user", "content": summary},  # ✓ 不与 head 冲突
]

tail = [
    {"role": "assistant", ...},   # 位置 4 (first_tail_role = "assistant")
    {"role": "tool", ...},
]

# 最终结果: system → user → assistant → user → assistant → tool ✓

# 如果 first_tail_role == "user"，会尝试翻转
if summary_role == first_tail_role:
    flipped = "assistant" if summary_role == "user" else "user"
    if flipped != last_head_role:
        summary_role = flipped

---

Phase 5: 清理孤立工具调用/结果

问题场景

压缩可能破坏工具调用/结果的配对：

# 压缩前
[
    {"role": "assistant", "tool_calls": [{"id": "call_abc", ...}]},
    {"role": "tool", "tool_call_id": "call_abc", "content": "result"},
    {"role": "assistant", "tool_calls": [{"id": "call_def", ...}]},  # 被压缩
    {"role": "tool", "tool_call_id": "call_def", "content": "result"},  # 孤立
]

# 压缩后（孤立结果）
[
    {"role": "assistant", "tool_calls": [{"id": "call_abc", ...}]},
    {"role": "tool", "tool_call_id": "call_abc", "content": "result"},
    {"role": "tool", "tool_call_id": "call_def", "content": "result"},  # 孤立！
]

解决方案

# 位置: agent/context_compressor.py 第 463-521 行
def _sanitize_tool_pairs(self, messages: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
    """Fix orphaned tool_call / tool_result pairs after compression."""
    
    # 1. 收集所有存活的 tool_call_id
    surviving_call_ids: set = set()
    for msg in messages:
        if msg.get("role") == "assistant":
            for tc in msg.get("tool_calls") or []:
                surviving_call_ids.add(tc.get("id"))
    
    # 2. 收集所有工具结果的 call_id
    result_call_ids: set = set()
    for msg in messages:
        if msg.get("role") == "tool":
            result_call_ids.add(msg.get("tool_call_id"))
    
    # 3. 移除孤立的工具结果
    orphaned_results = result_call_ids - surviving_call_ids
    if orphaned_results:
        messages = [
            m for m in messages
            if not (m.get("role") == "tool" and m.get("tool_call_id") in orphaned_results)
        ]
    
    # 4. 为孤立的工具调用添加存根结果
    missing_results = surviving_call_ids - result_call_ids
    if missing_results:
        patched = []
        for msg in messages:
            patched.append(msg)
            if msg.get("role") == "assistant":
                for tc in msg.get("tool_calls") or []:
                    cid = tc.get("id")
                    if cid in missing_results:
                        patched.append({
                            "role": "tool",
                            "content": "[Result from earlier conversation — see context summary above]",
                            "tool_call_id": cid,
                        })
        messages = patched
    
    return messages

实例:

# 压缩前（中间轮次有工具调用）
[
    {"role": "system", ...},
    {"role": "user", "content": "Fix the bug"},
    {"role": "assistant", "tool_calls": [{"id": "call_1", "name": "read_file"}]},
    {"role": "tool", "tool_call_id": "call_1", "content": "file content..."},
    {"role": "assistant", "tool_calls": [{"id": "call_2", "name": "patch"}]},  # 压缩
    {"role": "tool", "tool_call_id": "call_2", "content": "success"},          # 压缩
    {"role": "assistant", "content": "Fixed!"},
    {"role": "user", "content": "Add logging"},
    {"role": "assistant", "tool_calls": [{"id": "call_3", "name": "patch"}]},
    {"role": "tool", "tool_call_id": "call_3", "content": "success"},
]

# 压缩后（插入摘要，工具调用 call_2 被移除）
[
    {"role": "system", ...},
    {"role": "user", "content": "Fix the bug"},
    {"role": "assistant", "tool_calls": [{"id": "call_1", "name": "read_file"}]},
    {"role": "tool", "tool_call_id": "call_1", "content": "[Old tool output cleared...]"},
    {"role": "user", "content": "[CONTEXT SUMMARY]..."},
    {"role": "assistant", "content": "Fixed!"},
    {"role": "user", "content": "Add logging"},
    {"role": "assistant", "tool_calls": [{"id": "call_3", "name": "patch"}]},
    {"role": "tool", "tool_call_id": "call_3", "content": "success"},
]

---

会话持久化处理

压缩会创建新的会话 ID：

# 位置: run_agent.py 第 6394-6420 行
if self._session_db:
    # 1. 结束旧会话
    old_title = self._session_db.get_session_title(self.session_id)
    self._session_db.end_session(self.session_id, "compression")
    
    # 2. 创建新会话
    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 "cli",
        model=self.model,
        parent_session_id=old_session_id,  # 链接到父会话
    )
    
    # 3. 自动编号标题
    if old_title:
        new_title = self._session_db.get_next_title_in_lineage(old_title)
        self._session_db.set_session_title(self.session_id, new_title)

会话链示例:

session_20250101_abc123  (原始会话)
    ↓ compression
session_20250102_def456  (第一次压缩)
    ↓ compression
session_20250103_ghi789  (第二次压缩)

---

压缩后的内存刷新

在压缩前，允许模型保存重要记忆：

# 位置: run_agent.py 第 6374-6382 行
# Pre-compression memory flush
self.flush_memories(messages, min_turns=0)

# Notify external memory provider
if self._memory_manager:
    try:
        self._memory_manager.on_pre_compress(messages)
    except Exception:
        pass

---

完整实例流程

初始状态

Session: session_20250101_abc123
Model: claude-3-opus (200K context)
Messages: 52 条
Estimated tokens: 150,000
Threshold: 100,000 (50%)

执行过程

1. 预检压缩检查
   ✓ messages > protect_first_n + protect_last_n + 1 (52 > 3 + 20 + 1)
   ✓ estimated_tokens >= threshold (150K >= 100K)
   → 触发压缩

2. Phase 1: 修剪旧工具结果
   - 扫描消息 0-32 的工具结果
   - 替换 12 个超过 200 字符的工具结果
   - 节省约 8,000 tokens

3. Phase 2: 确定压缩边界
   - compress_start = 4 (保护系统提示 + 首次交互)
   - tail_token_budget = 20,000
   - 从消息 52 向前累积: 500 + 300 + 1500 + ... = 21,000
   - compress_end = 35 (保护最近 17 条消息)
   - 待压缩: 消息 4-34 (31 条)

4. Phase 3: 生成摘要
   - 序列化消息 4-34
   - 计算预算: 80,000 * 0.20 = 16,000 → 上限 12,000
   - 调用辅助 LLM (gpt-4o-mini)
   - 生成结构化摘要 (约 2,500 tokens)

5. Phase 4: 组装压缩消息
   - 保留消息 0-3
   - 插入摘要消息 (role="user")
   - 保留消息 35-51
   - 总计: 4 + 1 + 17 = 22 条消息

6. Phase 5: 清理孤立工具对
   - 移除 3 个孤立的工具结果
   - 添加 1 个存根工具结果
   - 最终: 20 条消息

7. 会话持久化
   - 结束 session_20250101_abc123 (reason="compression")
   - 创建 session_20250102_def456 (parent=abc123)
   - 标题: "Fix login bug #2"

8. 内存刷新
   - 执行 flush_memories() 保存重要记忆到 MEMORY.md

最终状态

Session: session_20250102_def456
Messages: 20 条
Estimated tokens: 45,000 (节省 105,000 tokens)
Compression count: 1

---

压缩质量保障

重复压缩警告

# 位置: run_agent.py 第 6422-6429 行
_cc = self.context_compressor.compression_count
if _cc >= 2:
    self._vprint(
        "⚠️  Session compressed 2 times — "
        "accuracy may degrade. Consider /new to start fresh.",
        force=True,
    )

摘要失败降级

# 位置: agent/context_compressor.py 第 706-717 行
if not summary:
    n_dropped = compress_end - compress_start
    summary = (
        f"[CONTEXT COMPACTION] Summary generation was unavailable. "
        f"{n_dropped} conversation turns were removed to free context space "
        f"but could not be summarized. The removed turns contained earlier "
        f"work in this session. Continue based on the recent messages below "
        f"and the current state of any files or resources."
    )

冷却期机制

# 位置: agent/context_compressor.py 第 309-315 行
if now < self._summary_failure_cooldown_until:
    logger.debug(
        "Skipping context summary during cooldown (%.0fs remaining)",
        self._summary_failure_cooldown_until - now,
    )
    return None

---

总结

Hermes Agent 的上下文压缩是一个精心设计的多阶段过程：

1. 智能触发: 多种触发条件确保在需要时及时压缩
2. 分层保护: 头部（系统提示 + 首次交互）+ 尾部（最近 ~20K tokens）
3. 两阶段优化: 快速修剪 + LLM 摘要
4. 结构化摘要: 保留目标、进度、决策、文件等关键信息
5. 迭代更新: 后续压缩时更新而非重建摘要
6. 完整性保障: 清理孤立工具对，维护消息交替规则
7. 会话连续性: 创建新会话并链接到父会话

这种设计确保了在长对话中既能有效利用有限的上下文窗口，又能最大程度地保留关键信息，让 Agent 能够持续工作而不重复已完成任务。