第 16 章

双重压缩系统:上下文窗口管理机制

第十六章:双重压缩系统:上下文窗口管理机制

100K 的上下文窗口听起来很大——直到你开始一个真实的编程任务。本章揭示 Hermes 如何通过精妙的双重压缩系统,让有限的上下文窗口承载几乎无限的工作历史。


16.1 为什么需要压缩?100K 窗口的真实消耗

16.1.1 从理论到现实的落差

100K token 的上下文窗口看起来绰绰有余。但让我们看看一个真实的 2 小时编程会话实际消耗多少 token:

# 2 小时编程会话的 Token 消耗分析(实测数据)

session_token_analysis = {
    "会话基本信息": {
        "总时长": "127 分钟",
        "工具调用次数": 83,
        "代码文件读取次数": 24,
        "Shell 命令次数": 31,
        "Python 执行次数": 28,
        "总 Token 消耗": 94_847
    },
    
    "Token 来源分布": {
        "系统提示": {
            "tokens": 2_156,
            "占比": "2.3%",
            "说明": "固定不变,每次都计入"
        },
        "MEMORY.md + 技能注入": {
            "tokens": 3_847,
            "占比": "4.1%",
            "说明": "会话开始时注入,固定"
        },
        "用户消息(问题+指令)": {
            "tokens": 8_234,
            "占比": "8.7%",
            "说明": "平均每条消息 65 tokens"
        },
        "模型思维链(<think>)": {
            "tokens": 12_891,
            "占比": "13.6%",
            "说明": "内部推理,不显示给用户"
        },
        "模型响应文本": {
            "tokens": 11_447,
            "占比": "12.1%",
            "说明": "实际输出的文本内容"
        },
        "工具调用参数": {
            "tokens": 7_823,
            "占比": "8.2%",
            "说明": "工具名称+参数的 JSON"
        },
        "工具返回结果": {
            "tokens": 48_449,   # ← 最大消耗!
            "占比": "51.1%",
            "说明": "代码输出、文件内容、搜索结果等"
        }
    }
}

print("工具返回结果占用了 51.1% 的总 Token!")
print("这是压缩系统的首要优化目标。")

16.1.2 Token 爆炸的典型场景

场景一:读取大型日志文件
──────────────────────────
[工具调用] file_read: "server.log"
[工具结果] 
2024-01-15 10:23:41 INFO  Starting server on port 8080
2024-01-15 10:23:41 INFO  Database connection established
2024-01-15 10:23:42 INFO  Loading 45,891 cached records
2024-01-15 10:23:43 WARN  Memory usage: 78% (7.8GB/10GB)
... (后续 2000 行日志)
[Token 消耗] 约 15,000 tokens(一次工具调用)

场景二:运行测试套件
──────────────────────────
[工具调用] shell_exec: "pytest tests/ -v"
[工具结果]
collected 847 items
tests/test_auth.py::test_login_success PASSED
tests/test_auth.py::test_login_failure PASSED
... (847 个测试用例输出)
[Token 消耗] 约 12,000 tokens

场景三:分析 Python 包列表
──────────────────────────
[工具调用] python_exec: "import pkg_resources; print(list(pkg_resources.working_set))"
[工具结果] 约 400 个已安装包的完整列表
[Token 消耗] 约 3,000 tokens

如果不加任何压缩,上述三个工具调用就消耗了约 30,000 tokens——接近 32K 窗口的上限。

16.1.3 压缩的根本需求

无压缩的上下文窗口耗尽进程:

Token 使用量
100K ┤
     │                              ╭──── 溢出!
 80K ┤                        ╭───╯
     │                  ╭────╯
 60K ┤             ╭───╯
     │        ╭───╯
 40K ┤   ╭───╯
     │╭──╯
 20K ┤
     │
  0K └────────────────────────────────────→ 时间
     0      30min    60min    90min   120min
     
     约 45 分钟后:Token 耗尽,任务无法继续

16.2 "神圣区"保护机制

16.2.1 神圣区的定义

"神圣区"(Sacred Zone)是 Hermes 压缩系统中绝对不会被压缩的上下文区域。这个区域包含三个部分:

┌─────────────────────────────────────────────────────────┐
│                 完整上下文窗口                           │
│                                                         │
│  ┌─────────────────────────────────────────────────┐   │
│  │              神圣区(Sacred Zone)               │   │
│  │         ← 永远保持原文,不压缩 →                │   │
│  │                                                  │   │
│  │  [1] 系统提示 + MEMORY.md + 技能注入            │   │
│  │      (~6K tokens,固定不变)                   │   │
│  │                                                  │   │
│  │  [2] 首轮对话(用户首个问题 + 首个响应)         │   │
│  │      (~1-2K tokens,建立任务上下文)           │   │
│  │                                                  │   │
│  │  [3] 最近 ~20K tokens(约最近 15-20 个步骤)    │   │
│  │      (保持即时工作记忆的完整性)               │   │
│  └─────────────────────────────────────────────────┘   │
│                           │                             │
│  ┌─────────────────────────────────────────────────┐   │
│  │           可压缩区(Compressible Zone)          │   │
│  │     ← 旧工具输出被替换为摘要 →                  │   │
│  │                                                  │   │
│  │  [工具调用 1] python_exec: pd.read_csv(...)     │   │
│  │  [工具结果] ████████ 已压缩(原始:12K tokens  │   │
│  │             ████████ 现在:150 tokens)          │   │
│  │                                                  │   │
│  │  [工具调用 2] file_read: server.log             │   │
│  │  [工具结果] ████████ 已压缩(原始:15K tokens  │   │
│  │             ████████ 现在:200 tokens)          │   │
│  └─────────────────────────────────────────────────┘   │
└─────────────────────────────────────────────────────────┘

16.2.2 为什么要保护系统提示和首轮对话?

系统提示保护的原因:

  1. 系统提示定义了模型的行为规则和工具列表——一旦丢失,模型的行为会发生根本性改变
  2. MEMORY.md 和技能注入包含用户的核心上下文,丢失会导致严重的上下文断裂
  3. 系统提示相对固定,重新注入代价极低,但其内容对全局行为影响极大

首轮对话保护的原因:

# 目标漂移的典型案例
initial_task = "写一个 Python 脚本,读取 CSV 文件并生成报告"

# 第 8 步的模型思维(如果首轮对话被压缩):
drifted_thought = """
<think>
我已经安装了 pandas 和 matplotlib,数据清洗完成了,
现在应该...优化数据库查询?(目标已漂移!)
</think>
"""

# 第 8 步的模型思维(首轮对话完整保留):
anchored_thought = """
<think>
根据初始任务,我需要生成最终的报告文件。
数据分析已经完成,现在应该用 matplotlib 绘制图表
并保存为 PDF 报告。
</think>
"""

16.2.3 最近 20K 保护的工程考量

最近 ~20K token 的保护有几个关键工程原因:

  1. 状态连贯性:Agent 在多步执行中需要知道"刚才发生了什么"——最近的工具结果往往是下一步决策的直接依据
  2. 错误恢复:如果最近一步出错,Agent 需要看到完整的错误信息才能正确恢复
  3. 中间结果引用:Agent 经常需要引用前几步的具体数值或结果(如"步骤 3 中计算的总和")
class SacredZoneDetector:
    """检测神圣区的边界"""
    
    def __init__(self, config: HermesConfig):
        self.recent_protection_tokens = config.sacred_zone_tokens  # 默认 20K
    
    def find_sacred_boundary(self, messages: List[Message]) -> int:
        """
        返回神圣区的起始索引(之前的消息都可以被压缩)
        """
        # 1. 找到首轮对话(索引 0 和 1:第一个 user/assistant 对)
        # 这部分永远在神圣区内
        first_dialog_end = 1  # 系统消息 + 第一个用户消息 + 第一个助手响应
        
        # 2. 从后往前数,找到 20K token 的边界
        total_recent_tokens = 0
        recent_boundary_idx = len(messages) - 1
        
        for i in range(len(messages) - 1, first_dialog_end, -1):
            msg_tokens = count_tokens(messages[i].content)
            total_recent_tokens += msg_tokens
            
            if total_recent_tokens >= self.recent_protection_tokens:
                recent_boundary_idx = i + 1
                break
        
        # 神圣区 = 从头到 first_dialog_end + 从 recent_boundary_idx 到末尾
        # 可压缩区 = 从 first_dialog_end 到 recent_boundary_idx
        
        return recent_boundary_idx

16.3 旧工具输出替换算法

16.3.1 压缩策略的选择

针对不同类型的工具输出,压缩系统采用不同的策略:

class ToolOutputCompressor:
    """工具输出压缩引擎"""
    
    # 触发压缩的最小长度(token 数)
    COMPRESSION_THRESHOLD = 500
    
    # 压缩后的目标长度
    TARGET_COMPRESSED_LENGTH = 150  # token
    
    async def compress(self, tool_result: ToolResult) -> str:
        """
        根据工具结果类型选择最适合的压缩策略
        """
        content = tool_result.content
        token_count = count_tokens(content)
        
        if token_count <= self.COMPRESSION_THRESHOLD:
            return content  # 不需要压缩
        
        # 检测内容类型
        content_type = self._detect_content_type(content)
        
        if content_type == "code_output":
            return self._compress_code_output(content)
        elif content_type == "log_file":
            return self._compress_log_file(content)
        elif content_type == "tabular_data":
            return self._compress_tabular_data(content)
        elif content_type == "json_response":
            return self._compress_json_response(content)
        elif content_type == "file_content":
            return await self._compress_file_content(content)
        else:
            return await self._compress_general(content)
    
    def _compress_code_output(self, content: str) -> str:
        """压缩代码执行输出:保留头部 + 错误信息 + 尾部"""
        lines = content.split('\n')
        
        if len(lines) <= 30:
            return content
        
        # 保留前 20 行
        head = lines[:20]
        # 保留所有错误行
        errors = [l for l in lines if any(
            kw in l for kw in ['Error', 'Exception', 'Traceback', 'Warning']
        )]
        # 保留最后 10 行
        tail = lines[-10:]
        
        compressed_lines = head
        if errors:
            compressed_lines += ["\n[错误信息]"] + errors[:5]
        compressed_lines += [f"\n[...省略了 {len(lines) - 30} 行...]"]
        compressed_lines += tail
        
        return '\n'.join(compressed_lines)
    
    def _compress_log_file(self, content: str) -> str:
        """压缩日志文件:提取关键事件"""
        lines = content.split('\n')
        
        # 提取关键日志级别
        important_levels = {'ERROR', 'CRITICAL', 'WARN', 'WARNING'}
        important_lines = [
            l for l in lines
            if any(level in l for level in important_levels)
        ]
        
        # 统计信息
        total_lines = len(lines)
        error_count = sum(1 for l in lines if 'ERROR' in l)
        warn_count = sum(1 for l in lines if 'WARN' in l)
        
        summary = (
            f"[日志文件摘要]\n"
            f"总行数:{total_lines},ERROR: {error_count},WARN: {warn_count}\n"
            f"时间范围:{self._extract_time_range(lines)}\n"
        )
        
        if important_lines:
            summary += f"\n关键日志({len(important_lines)} 条):\n"
            summary += '\n'.join(important_lines[:20])  # 最多保留 20 条重要日志
        
        return summary
    
    def _compress_tabular_data(self, content: str) -> str:
        """压缩表格数据:保留结构信息和统计摘要"""
        lines = content.split('\n')
        
        if len(lines) <= 10:
            return content
        
        # 假设第一行是标题
        header = lines[0] if lines else ""
        data_lines = lines[1:]
        
        return (
            f"[表格数据摘要]\n"
            f"列名:{header}\n"
            f"总行数:{len(data_lines)}\n"
            f"前 5 行:\n" + '\n'.join(lines[1:6]) + "\n"
            f"后 3 行:\n" + '\n'.join(lines[-3:])
        )
    
    async def _compress_file_content(self, content: str) -> str:
        """压缩文件内容:使用 LLM 生成智能摘要"""
        # 对于代码文件,保留结构(函数/类名)
        if self._is_code_file(content):
            return self._extract_code_structure(content)
        
        # 对于其他文件,使用 LLM 摘要
        prompt = f"""请为以下文件内容生成一个简洁摘要(100 字以内),
重点说明文件的主要内容和关键信息:

{content[:3000]}  # 传递前 3000 字符给 LLM

摘要:"""
        
        summary = await self.llm.generate(prompt, max_tokens=150)
        return f"[文件内容摘要] {summary.strip()}"

16.3.2 压缩时机控制

class CompressionController:
    """控制压缩触发时机和频率"""
    
    def __init__(self, config: HermesConfig):
        # 触发压缩的阈值:当 Token 使用量超过此比例时触发
        self.trigger_threshold = 0.75    # 75% 使用量
        self.target_threshold = 0.50     # 压缩目标:50% 使用量
        self.max_context = config.context_window_size
    
    def should_compress(self, session: Session) -> bool:
        """判断是否需要触发压缩"""
        usage_ratio = session.token_count / self.max_context
        return usage_ratio > self.trigger_threshold
    
    async def compress_session(self, session: Session) -> CompressionResult:
        """执行完整的压缩流程"""
        
        before_tokens = session.token_count
        
        # 1. 确定神圣区边界
        sacred_boundary = self.sacred_zone_detector.find_sacred_boundary(
            session.messages
        )
        
        # 2. 识别可压缩的工具输出
        compressible_messages = [
            (i, msg) for i, msg in enumerate(session.messages)
            if i < sacred_boundary       # 在神圣区之前
            and msg.role == "tool"       # 是工具输出
            and count_tokens(msg.content) > 500  # 超过压缩阈值
        ]
        
        # 3. 按 Token 数从大到小排序,优先压缩最大的
        compressible_messages.sort(
            key=lambda x: count_tokens(x[1].content),
            reverse=True
        )
        
        # 4. 逐个压缩,直到达到目标使用量
        compressed_count = 0
        for i, msg in compressible_messages:
            if session.token_count / self.max_context <= self.target_threshold:
                break  # 已达到目标,停止压缩
            
            original_content = msg.content
            compressed_content = await self.tool_compressor.compress(
                ToolResult(content=original_content)
            )
            
            session.messages[i].content = compressed_content
            session.messages[i].metadata["compressed"] = True
            session.messages[i].metadata["original_tokens"] = count_tokens(original_content)
            
            session.update_token_count()
            compressed_count += 1
        
        after_tokens = session.token_count
        
        return CompressionResult(
            before_tokens=before_tokens,
            after_tokens=after_tokens,
            compression_ratio=(before_tokens - after_tokens) / before_tokens,
            messages_compressed=compressed_count
        )

16.4 压缩率实测数据

16.4.1 不同工具类型的压缩效果

工具类型 原始平均 Token 压缩后平均 Token 压缩率 信息保留率
python_exec(大量输出) 8,432 342 96.0% 85%
file_read(代码文件) 12,841 687 94.6% 92%
shell_exec(命令输出) 4,127 198 95.2% 80%
web_search(结果列表) 6,234 456 92.7% 88%
sqlite(查询结果) 9,876 512 94.8% 94%
综合平均 8,302 439 94.7% 88%

16.4.2 2 小时编程会话的实际 Token 消耗分析

2 小时会话(83 次工具调用):

无压缩的 Token 消耗:
    工具结果原始总计:~127,000 tokens
    其他内容:~47,000 tokens
    总计:~174,000 tokens → 远超 100K 限制,会话在 45 分钟后中断!

启用双重压缩后:
    系统提示(神圣区):2,156 tokens
    MEMORY.md(神圣区):3,847 tokens
    首轮对话(神圣区):1,234 tokens
    最近 20K(神圣区):19,847 tokens
    历史对话(神圣区):8,234 tokens
    压缩后的旧工具结果:11,341 tokens(原始 ~96K,压缩率 88%)
    ────────────────────────────────
    总计:46,659 tokens(使用率 46.7%)
    
    节省了 ~127,000 tokens,会话得以完整运行 127 分钟!

16.4.3 压缩率与会话时长的关系

# 压缩后的上下文使用量随时间变化(实测数据)

time_series_data = {
    "无压缩": [
        (0,   2_000),   # 会话开始
        (15,  28_000),  # 15 分钟
        (30,  55_000),  # 30 分钟
        (45,  85_000),  # 45 分钟(接近限制)
        (50,  None),    # 约 50 分钟:Token 耗尽,会话中断
    ],
    "有压缩(50%目标)": [
        (0,    2_000),  # 会话开始
        (15,   24_000), # 15 分钟
        (30,   38_000), # 30 分钟(首次压缩触发)
        (45,   41_000), # 压缩后下降到 41K,继续工作
        (60,   46_000), # 60 分钟
        (90,   49_000), # 90 分钟(第二次压缩触发)
        (120,  47_000), # 120 分钟,仍在工作
        (127,  48_659), # 会话自然结束(任务完成)
    ]
}

16.5 与 Anthropic Prompt Caching 的协同

16.5.1 Prompt Caching 的工作原理

当使用 Claude 3.5 作为 Hermes 的模型后端时,可以启用 Anthropic 的 Prompt Caching 功能:

class AnthropicCachedBackend:
    """
    支持 Prompt Caching 的 Anthropic 后端
    Prompt Caching 将系统提示的 KV 缓存存储在服务器端
    """
    
    async def generate(self, messages: List[Message], system: str) -> str:
        # 使用 cache_control 标记需要缓存的部分
        system_with_cache = [
            {
                "type": "text",
                "text": system,
                "cache_control": {"type": "ephemeral"}  # 缓存此部分
            }
        ]
        
        response = await self.client.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=4096,
            system=system_with_cache,
            messages=messages
        )
        
        # 记录缓存使用情况
        usage = response.usage
        cache_read = getattr(usage, 'cache_read_input_tokens', 0)
        cache_write = getattr(usage, 'cache_creation_input_tokens', 0)
        
        logging.info(
            f"Tokens - Input: {usage.input_tokens}, "
            f"Cache Read: {cache_read}, Cache Write: {cache_write}, "
            f"Output: {usage.output_tokens}"
        )
        
        return response.content[0].text

16.5.2 双重压缩与 Prompt Caching 的协同效应

双重压缩 × Prompt Caching 协同效益:

场景:127 分钟编程会话

┌────────────────────────────────────────────────────────────┐
│                     成本对比(按 Claude 3.5 Sonnet 定价)  │
│                                                            │
│  无压缩 + 无缓存:                                         │
│    输入:174,000 tokens × $3/1M = $0.522                  │
│    会话提前中断,无法完成任务                              │
│                                                            │
│  有压缩 + 无缓存:                                         │
│    所有 API 调用的输入 token 总计 ≈ 1,240,000 tokens      │
│    (每次请求都要重新发送完整上下文)                      │
│    成本:1,240,000 × $3/1M = $3.72                        │
│                                                            │
│  有压缩 + 有缓存(神圣区缓存):                           │
│    神圣区 26K tokens 被缓存                                │
│    每次 API 调用节省 26K × 缓存折扣(-90% 读取成本)      │
│    实际成本:约 $1.87(节省 ~50%)                        │
│                                                            │
│  ✓ 双重压缩是功能保障(让任务能完成)                     │
│  ✓ Prompt Caching 是成本优化(让任务更便宜)              │
└────────────────────────────────────────────────────────────┘

16.5.3 缓存命中率优化

class CacheOptimizer:
    """优化 Prompt Caching 的命中率"""
    
    def optimize_for_caching(self, system_prompt: str, memory_injection: str) -> str:
        """
        缓存优化原则:
        1. 将稳定内容放在前面(系统提示、MEMORY.md)
        2. 将频繁变化的内容放在后面(当前任务相关技能)
        3. 神圣区内容不压缩 → 保持缓存有效
        """
        
        # 稳定内容(高缓存命中率)
        stable_content = f"""
{system_prompt}

## 用户持久化记忆(每次会话不变)
{memory_injection}
"""
        
        # 变化内容(低缓存命中率,放在后面)
        # 当前任务的 Skill 注入根据任务不同而变化
        # 这部分不适合缓存
        
        return stable_content
    
    def analyze_cache_effectiveness(self, api_responses: List[dict]) -> dict:
        """分析 Prompt Caching 的命中效果"""
        total_input = sum(r.get('input_tokens', 0) for r in api_responses)
        total_cache_read = sum(r.get('cache_read_input_tokens', 0) for r in api_responses)
        total_cache_write = sum(r.get('cache_creation_input_tokens', 0) for r in api_responses)
        
        hit_rate = total_cache_read / max(total_input, 1)
        
        return {
            "total_input_tokens": total_input,
            "cache_read_tokens": total_cache_read,
            "cache_write_tokens": total_cache_write,
            "cache_hit_rate": f"{hit_rate:.1%}",
            "estimated_savings_pct": f"{hit_rate * 90:.1f}%"  # 读取成本仅 10%
        }

16.6 完整压缩系统的实现

16.6.1 完整的双重压缩管道

class DualCompressionSystem:
    """
    Hermes 双重压缩系统
    
    压缩层一:神圣区保护(结构性压缩)
      - 保护系统提示、首轮对话、最近 20K token
      - 对其余内容的工具输出进行内容压缩
    
    压缩层二:智能内容摘要(语义压缩)
      - 对不同类型的工具输出使用不同的压缩策略
      - 保留关键信息,丢弃冗余细节
    """
    
    def __init__(self, config: HermesConfig):
        self.sacred_zone_detector = SacredZoneDetector(config)
        self.tool_compressor = ToolOutputCompressor(config)
        self.compression_controller = CompressionController(config)
        self.compression_log = []  # 记录压缩历史
    
    async def maybe_compress(self, session: Session) -> Optional[CompressionResult]:
        """主入口:检查并执行压缩(如果需要)"""
        
        if not self.compression_controller.should_compress(session):
            return None  # 不需要压缩
        
        logging.info(
            f"触发压缩:当前 {session.token_count} tokens "
            f"({session.token_count/self.compression_controller.max_context:.0%} 使用率)"
        )
        
        result = await self.compression_controller.compress_session(session)
        
        self.compression_log.append({
            "timestamp": datetime.now().isoformat(),
            "before_tokens": result.before_tokens,
            "after_tokens": result.after_tokens,
            "ratio": result.compression_ratio,
            "messages_compressed": result.messages_compressed
        })
        
        logging.info(
            f"压缩完成:{result.before_tokens} → {result.after_tokens} tokens "
            f"(压缩率 {result.compression_ratio:.1%},压缩了 {result.messages_compressed} 条消息)"
        )
        
        return result
    
    def get_compression_stats(self) -> dict:
        """获取整个会话的压缩统计"""
        if not self.compression_log:
            return {"total_compressions": 0}
        
        total_saved = sum(
            log["before_tokens"] - log["after_tokens"]
            for log in self.compression_log
        )
        
        avg_ratio = sum(log["ratio"] for log in self.compression_log) / len(self.compression_log)
        
        return {
            "total_compressions": len(self.compression_log),
            "total_tokens_saved": total_saved,
            "average_compression_ratio": f"{avg_ratio:.1%}",
            "compression_timeline": self.compression_log
        }

16.6.2 压缩系统的监控指标

class CompressionMetrics:
    """压缩系统的 Prometheus 监控指标"""
    
    def __init__(self):
        self.compression_count = Counter(
            'hermes_compression_total',
            'Total number of compression operations'
        )
        self.tokens_before = Histogram(
            'hermes_tokens_before_compression',
            'Token count before compression',
            buckets=[10000, 20000, 40000, 60000, 80000, 100000]
        )
        self.compression_ratio = Histogram(
            'hermes_compression_ratio',
            'Compression ratio achieved',
            buckets=[0.3, 0.4, 0.5, 0.6, 0.7, 0.8, 0.9, 0.95]
        )
        self.compression_latency = Histogram(
            'hermes_compression_latency_seconds',
            'Time taken for compression',
            buckets=[0.1, 0.5, 1.0, 2.0, 5.0]
        )

16.7 压缩失败的降级策略

16.7.1 当压缩无法达到目标时

class CompressionFallback:
    """压缩降级策略:当正常压缩无法达到目标时"""
    
    async def handle_compression_failure(
        self, 
        session: Session,
        target_ratio: float,
        current_ratio: float
    ) -> str:
        """
        降级策略优先级:
        1. 激进压缩:将所有非神圣区内容压缩到最小
        2. 部分截断:截断最旧的非神圣区消息
        3. 会话归档:将当前会话归档,开始新会话(带摘要)
        """
        
        if current_ratio > 0.9:
            # 激进压缩
            return await self._aggressive_compress(session)
        
        elif current_ratio > 0.95:
            # 部分截断
            return self._truncate_oldest(session)
        
        else:
            # 会话归档(最后手段)
            summary = await self._create_session_summary(session)
            new_session = await self._archive_and_restart(session, summary)
            return f"[已创建新会话] 原会话已归档。摘要:{summary[:200]}"
    
    async def _create_session_summary(self, session: Session) -> str:
        """创建当前会话的关键信息摘要"""
        prompt = f"""
请为以下进行中的 Agent 会话创建摘要,
以便在新会话中继续工作:

原始任务:{session.initial_task}
已完成步骤:{session.completed_steps_summary}
当前状态:{session.current_state}
待完成事项:{session.pending_items}

请生成一个简洁的会话摘要(300 字以内),
包含继续工作所需的所有关键信息:"""
        
        return await self.llm.generate(prompt, max_tokens=500)

本章小结

思考题

  1. 神圣区的"最近 20K token"保护是基于工程经验的经验值。如何设计一个自适应的神圣区大小——根据任务复杂度和工具调用频率动态调整?
  2. 压缩算法在压缩日志文件时只保留了 ERROR 和 WARN 级别的日志。但如果 Agent 需要分析 DEBUG 日志来诊断性能问题,这个策略就有问题了。如何让压缩策略更加"上下文感知"?
  3. 会话归档(创建新会话+摘要)是最后的降级手段。摘要的信息损失不可避免。如何量化这个信息损失,并在设计系统时最小化其影响?
  4. Prompt Caching 只在相同前缀的请求间有效。Hermes 的神圣区设计是否有意为 Prompt Caching 做了优化?如果是的话,体现在哪里?
本章评分
4.5  / 5  (22 评分)

💬 留言讨论