MiniCode 项目详解8：上下文压缩

零、先定位这套代码在整条链路中的位置agent_loop.py 主循环: step_start: context_cybernetics.run_cycle(messages) ← interview5.md 讲过了控制论决策层 └─ ContextCompactor.process_request() ← 本篇主角压缩执行层 LLM 调用: model.next(messages) ← 使用压缩后的消息 如果 API 报错 prompt too long: ContextCompactor.reactive_recover() ← 也是本篇的 ReactiveCompactEngine与控制论的关系控制论context_cybernetics.py决定什么时候压、压多少Compactor 决定用什么技术压。本篇只讲后者。一、为什么需要这套系统LLM 的上下文窗口有物理上限Claude 200K tokens。长任务中对话历史不断增长迟早要删东西。朴素做法满了就截断最旧的消息。朴素做法的问题截断丢掉了用户最初的需求帮我把整个项目的异常处理重构一遍截断了 LLM 已经读取过的文件内容再读一遍浪费 token截断了工具执行结果可能包含关键错误信息截断的时机是满了才截没有提前量MiniCode 的做法一个7 阶段压缩流水线——先做轻量级优化不动消息结构不够再做重量级压缩摘要截断最后还有错误恢复兜底。一句话能不删就不删能少删就少删要删就好好删。二、整体架构ContextCompactor.process_request() ← 统一入口 │ ├─ 阶段1: 重建活跃上下文从上次压缩边界开始 ├─ 阶段2: ToolResultBudgetManager ← 超大工具结果写盘 ├─ 阶段3: ReadDedupManager ← 重复文件读取去重 ├─ 阶段4: MicrocompactEngine ← 时间触发清理旧工具结果 ├─ 阶段5: AutoCompactDispatcher ← 85% 高水位检查 │ ├─ 优先: SessionMemoryCompact ← 用已有记忆做摘要 │ └─ 兜底: FullCompact ← 结构化摘要 截断 └─ 阶段7: ReactiveCompactEngine ← API 报错后的恢复不在正常路径中三、核心数据结构class CompactTrigger(str, Enum): MANUAL manual # 手动触发用户命令 AUTO auto # 自动触发高水位 REACTIVE reactive # 被动触发API 报错后 MICROCOMPACT_TIME microcompact_time # 时间触发 MICROCOMPACT_CACHED microcompact_cached # 缓存触发 class CompactStrategy(str, Enum): TOOL_BUDGET tool_budget # 大工具结果写盘 READ_DEDUP read_dedup # 重复读取去重 MICROCOMPACT microcompact # 时间清理 SESSION_MEMORY session_memory # 用记忆做摘要 FULL full # 完整压缩结构化摘要 REACTIVE reactive # 错误恢复激进截断 PARTIAL partial # 部分压缩 dataclass class CompactionResult: success: bool strategy: CompactStrategy trigger: CompactTrigger messages: list[dict] # 压缩后的消息列表 boundary: CompactBoundary | None # 压缩边界标记 tokens_freed: int summary_text: str property def effective(self) - bool: return self.success and self.tokens_freed 0 # 必须真省了 tokenCompactBoundary压缩边界标记——记录在对话历史的哪里做了压缩dataclass class CompactBoundary: trigger: CompactTrigger strategy: CompactStrategy timestamp: float tokens_before: int # 压缩前 token 数 tokens_after: int # 压缩后 token 数 messages_removed: int # 移除了多少条消息 preserved_segment: tuple[int, int] | None # (start, end) 保留的消息索引范围为什么需要 boundary下一次压缩时只需要看 boundary之后的消息——boundary 之前的内容已经被摘要替代了不需要再处理。相当于在对话历史中插入了一个快照点。四、阶段1重建活跃上下文 — 从上次快照点继续4.1 做什么压缩不是每次都从头压缩整个消息列表。上一次压缩时生成的CompactBoundary是一个快照点——boundary 之前的内容已经被摘要替代boundary 之后的才是活跃上下文。process_request()的第一步是找到最近的 boundary只处理 boundary 之后的消息。boundary 本身以_compact_boundary: True标记的形式嵌在消息列表中# 压缩后注入的 system 消息 { role: system, content: [Context compacted at 14:32:15 — Full Compact]\n... _compact_boundary: True, # ← 标记这是一个压缩边界 }4.2 不在代码中直接体现但在设计意图中阶段 1 没有对应的 Engine 类——它是一个隐式的逻辑上一次压缩后的消息列表: [system_msg_0] ← 原始 system prompt [user_msg_1] ← boundary 之前已被摘要替代不需要再处理 [assistant_msg_2] ... [system_msg_boundary] ← _compact_boundaryTrue快照点 [user_msg_N] ← boundary 之后 活跃上下文需要处理 [assistant_msg_N1] ...process_request()收到messages参数时调用方context_cybernetics.py已经传入了完整的消息列表。Compactor 不做切分——它假设调用方已经把 boundary 之前的内容处理好。Compactor 的职责是从当前消息列表出发做优化不管历史。4.3 为什么阶段 3Read Dedup也不在process_request()流水线中Read Dedup去重依赖正在被读取的文件内容 文件路径——这只有工具调度器在工具执行时才知道。压缩流水线看到的已经是注入上下文的 tool_result 消息没有原始文件路径和内容哈希。所以ReadDedupManager的调用点在工具调度层agent_loop.py中工具执行时不在process_request()里。process_request()只持有ReadDedupManager实例供调度器查询不在流水线中主动调它。真实的调用链agent_loop.py: 工具执行 read_file(main.py) → 读文件拿到内容 → ReadDedupManager.register_read(main.py, content, msg_index) → 如果是重复读取 → 返回 stub 文本替代完整内容 → 将 stub 或完整内容注入 messages五、阶段2ToolResultBudgetManager — 大工具结果写盘5.1 做什么read_file返回一个 8000 行的文件grep输出 5000 条匹配这些工具结果直接塞进上下文大量消耗 token。策略超过 4000 字符的工具结果 → 写盘上下文里只留一个预览桩。5.2 核心逻辑# context_compactor.py:176-209 def check_and_replace(self, messages): for i, msg in enumerate(modified): if msg.get(role) ! tool_result: continue content msg.get(content, ) if len(content) 4000: # 小结果不动 continue # 写盘原子写入tempfile os.replace path self._results_dir / f{tool_name}_{index}_{timestamp}.txt # 写入格式JSON header ---CONTENT--- 原始内容 # 上下文里只留预览 preview self._generate_preview(content, tool_name, path) modified[i] {**msg, content: preview, _persisted_path: str(path)}5.3 预览桩长什么样# context_compactor.py:249-270 def _generate_preview(self, content, tool_name, path): lines content.splitlines() head_lines lines[:8] # 前 8 行 tail_lines lines[-3:] if len(lines) 12 else [] # 后 3 行 return f [Tool result persisted to disk — {len(content)} chars] Tool: {tool_name} Path: {path.name} --- Preview (first/last lines) --- {head_lines} ... (XXX lines omitted) ... {tail_lines} [:500] # 最多 500 字符效果一个 80K 字符的文件内容 → 一个 500 字符的预览桩。LLM 看预览知道文件大概长什么样如果需要完整内容可以用工具再读磁盘文件。5.4 关键细节原子写入tempfile.mkstemp()os.replace()——跟 Memory 模块的_atomic_write()同款模式写到一半崩溃不会损坏已有文件存储位置.mini-code-tool-results/目录项目根目录下只针对tool_result不处理 user/assistant 消息六、阶段3ReadDedupManager — 重复文件读取去重6.1 做什么同一个文件被 LLM 反复读取每轮都在读main.py看有没有变化。每次读取都往上下文里塞完整文件内容——大量冗余。策略MD5 比对——同样的文件路径 同样的内容哈希 不重复注入返回一个桩。6.2 核心逻辑# context_compactor.py:284-333 class ReadDedupManager: def __init__(self): self._entries: dict[str, ReadDedupEntry] {} # file_path → entry def register_read(self, file_path, content, message_index) - bool: content_hash hashlib.md5(content.encode()).hexdigest() existing self._entries.get(file_path) if existing and existing.content_hash content_hash: return False # 重复读取 self._entries[file_path] ReadDedupEntry( file_pathfile_path, content_hashcontent_hash, timestamptime.time(), message_indexmessage_index, # 记下原始内容在哪条消息 ) return True # 新读取或内容变了 def get_stub(self, file_path) - str: entry self._entries.get(file_path) return ( f[Read deduplicated: {file_path}]\n fFile unchanged since last read. fThe content from the earlier Read tool_result fin this conversation is still current — refer to that instead.\n f(Original content at message index {entry.message_index}) )6.3 关键细节路径内容双重比对同一路径但内容变了被编辑过了→ 不算重复重新注入文件被编辑后自动失效外部调用invalidate(file_path)清除缓存——写操作后文件内容变了下次读必须重新注入引用原始位置桩里写了message_indexLLM 知道去第几条消息找完整内容MD5 使用usedforsecurityFalse明确声明不做安全用途纯内容比对七、阶段4MicrocompactEngine — 时间触发清理7.1 做什么工具执行结果是临时数据——旧的文件读取结果大概率已经过期。超过 1 小时的对话里文件可能已经被编辑过多次LLM 需要的是最新内容而非历史快照。策略每隔一段时间默认 1 小时清理旧的工具结果——只保留最近 N 个默认 5 个其余替换为占位符。7.2 核心逻辑# context_compactor.py:347-429 class MicrocompactEngine: def __init__(self, configNone): self._state config or MicrocompactState() # 默认参数 # time_based_interval 3600.0 ← 1 小时间隔 # keep_recent_tool_results 5 ← 保留最近 5 个工具结果 def run_time_based_microcompact(self, messages, nowNone): now now or time.time() elapsed now - self._state.last_time_based_compact if elapsed self._state.time_based_interval: return # 没到时间跳过 # 找到所有 tool_result排除已经被写盘的和已经被清理的 tool_results [(i, m) for i, m in enumerate(messages) if m.get(role) tool_result and not m.get(content, ).startswith([Tool result persisted) and not m.get(content, ).startswith([Old tool result)] if len(tool_results) 5: return # 还没到 5 个不清 # 保留最后 5 个其余替换为占位符 keep_indices {idx for idx, _ in tool_results[-5:]} for idx, msg in tool_results: if idx in keep_indices: continue modified[idx] { **msg, content: [Old tool result content cleared by time-based microcompact], _microcompacted: True, }{**msg, ...}的含义这是 Python 的字典解包dictionary unpacking。{**msg, content: xxx}创建一个新字典包含msg的所有键值对然后覆盖content字段为新值。例如# msg 原始内容 msg {role: tool_result, content: 8000行文件内容..., toolName: read_file} # {**msg, content: [Old tool result...]} 等价于 new_msg {role: tool_result, content: [Old tool result...], toolName: read_file}所以替换为占位符就是把消息字典里的content字段从原始文件内容替换为一个占位符字符串role、toolName等字段保持不变。7.3 为什么叫 micro compact因为它不删消息、不改结构、不做摘要——只是把旧工具结果的内容替换为占位符。消息还在消息结构不变token 数少了。是一个零风险的轻量优化。7.4 和 ToolResultBudgetManager 的区别ToolResultBudgetManagerMicrocompactEngine触发条件立即结果 4000 字符定时每 1 小时处理对象单个大工具结果一批旧工具结果操作写盘 预览桩可恢复替换为占位符不可恢复目的防止超大结果撑爆上下文清理不再需要的旧数据八、阶段5-6AutoCompactDispatcher — 高水位自动压缩8.1 做什么前面三个阶段都是轻量级优化——不动消息结构只优化工具结果。但当上下文持续增长轻量优化不够时需要真正删除并摘要中间部分的消息。策略当 token 使用量超过 85% 窗口上限时触发。优先尝试会话记忆压缩轻失败则兜底完整压缩重。8.2 触发条件# context_compactor.py:630-642 def should_trigger(self, messages, token_usageNone): if not self._config.enabled: return False if self.is_tripped: # 断路器跳了 return False usage token_usage or sum(self._estimate(m) for m in messages) return usage self.threshold_tokens # 85% 窗口大小断路器Circuit Breaker连续 3 次压缩失败 →is_tripped True→ 不再触发压缩。防止压缩→失败→再压缩→再失败的死循环。8.3 两级策略Session Memory → Full Compact# context_compactor.py:644-681 def dispatch(self, messages, token_usageNone, force_fullFalse): # 优先尝试 Session Memory Compact if not force_full: sm_result self._session_memory_engine.try_session_memory_compact( messages, self._context_window, self._estimate, self._config ) if sm_result and sm_result.effective: # 成功 真省了 token return sm_result # 兜底Full Compact return self._run_full_compact(messages, usage)九、SessionMemoryCompactEngine — 用记忆做摘要9.1 核心思想Memory 模块已经维护了项目知识决策、约定、模式。压缩时直接用它作为摘要基础——不需要再调 LLM 生成摘要。输入当前消息列表MemoryManager 提供的项目记忆最多 6000 tokens输出一条 system 消息包含记忆摘要 统计保留的尾部消息最近几轮对话原样保留9.2 完整流程# context_compactor.py:453-547 def try_session_memory_compact(self, messages, context_window, estimate_fn, config): # 步骤1获取记忆上下文作为摘要基础 memory_context self._memory.get_relevant_context(max_tokens6000) if not memory_context.strip(): return None # 没有记忆 → 退化为 Full Compact # 步骤2从消息列表尾部往前扫描确定保留尾部的切点 tail_tokens 0 tail_start len(non_system) for i in range(len(non_system) - 1, -1, -1): msg_tokens estimate(non_system[i]) if tail_tokens msg_tokens 40000 and len(non_system) - i 5: tail_start i 1 break tail_tokens msg_tokens # 步骤3切点调整——不能切断 tool_use/tool_result 对 tail_start self._adjust_for_tool_pair(non_system, tail_start) # 步骤4构建压缩后的消息 compacted [{ role: system, content: ( f[Context compacted via Session Memory]\n fMessages removed: {tail_start}. Tokens before: ~{before}\n\n f## Project Memory Context\n\n{memory_context}\n\n f--- Recent conversation continues below --- ), }] compacted.extend(non_system[tail_start:]) # 尾部最近消息原样保留 # 步骤5检查压缩效果——没省 5% 以上视为失败 if boundary.tokens_after boundary.tokens_before * 0.95: return None9.3 切点调整不能切断工具调用对staticmethod def _adjust_for_tool_pair(messages, cut_point): # 情况1切点后面的 tool_result它的 tool_use 在切点前面 # → 切点后移把这对包含进来 for i in range(cut_point, len(messages)): if messages[i].get(role) tool_result: # 检查对应的 tool_use 是否在切点之前 has_match_before any( msg.get(role) assistant and tool_use in str(msg.get(content)) for msg in messages[max(0, cut_point-10):cut_point] ) if has_match_before: cut_point i 1 # 下移包含这对 # 情况2切点前面的 tool_use它的 tool_result 在切点后面 # → 切点前移包含这对 for i in range(cut_point - 1, max(0, cut_point - 10), -1): msg messages[i] if tool_use in str(msg.get(content)): has_result_after any( m.get(role) tool_result for m in messages[cut_point:] ) if has_result_after: cut_point min(cut_point, i) # 上移包含这对 return max(0, cut_point)为什么这个很重要如果切断了tool_use和tool_result的配对LLM 会看到有一个工具返回结果但没有对应的调用或者有一个工具调用但没有结果——这会让 LLM 困惑甚至报错。9.4 三个关键参数参数默认值含义TAIL_MIN_TOKENS10000尾部至少保留 10K tokensTAIL_MAX_TOKENS40000尾部最多保留 40K tokensTAIL_MIN_MESSAGES5尾部至少保留 5 条消息十、FullCompact — 结构化摘要无需 LLM10.1 当 Session Memory 不可用时MemoryManager 为 None 或者没有相关记忆 →try_session_memory_compact()返回 None → 触发 Full Compact。10.2 Full Compact 的摘要生成不需要 LLM——纯规则提取# context_compactor.py:749-799 def _generate_structured_summary(self, messages): parts [### Summary of conversation so far:\n] # 1. 提取用户话题取每段 user 消息的前 100 字符 user_topics [] for msg in messages: if msg[role] user and len(msg[content]) 10: user_topics.append(msg[content][:100]) parts.append(**Topics discussed:**\n) for t in user_topics[:8]: parts.append(f- {t}) # 2. 提取使用的工具 tool_calls_made set() for msg in messages: if msg[role] assistant and isinstance(msg[content], list): for block in msg[content]: if block.get(type) tool_use: tool_calls_made.add(block[name]) if file_path in block.get(input, {}): files_mentioned.add(block[input][file_path]) parts.append(f**Tools used:** {, .join(sorted(tool_calls_made))}) # 3. 提取涉及的文件 parts.append(f**Files touched:** {, .join(sorted(files_mentioned)[:10])}) # 4. 提取错误 errors_seen [] for msg in messages: if msg.get(isError): errors_seen.append(msg[content][:80]) parts.append(**Errors encountered:**\n) for e in errors_seen[:3]: parts.append(f- {e}) parts.append(\n*Continue from where we left off.*) return \n.join(parts)最终产出的 system 消息[Context compacted at 14:32:15 — Full Compact] Original: ~85000 tokens, 142 messages ## Conversation Summary ### Summary of conversation so far: **Topics discussed:** - 帮我把项目的异常处理重构一遍 - 好的我先看一下当前的异常处理情况 - ... **Tools used:** read_file, write_file, grep_files **Files touched:** minicode/agent_loop.py, minicode/memory.py *Continue from where we left off.*尾部保留len(non_system) // 3条最近消息最多min_keep_messages条原样保留。十一、阶段7ReactiveCompactEngine — 错误恢复11.1 做什么前面的阶段都属于预防——在上下文还没超限时提前优化。但万一还是超了API 返回 prompt too long需要紧急处理。策略3 次重试逐步激进11.2 核心流程# context_compactor.py:855-905 def try_recover_from_overflow(self, messages, error_message): self._recovery_attempts 1 if self._recovery_attempts 3: return None # 3 次重试都失败 → 放弃 # 尝试1Force Full Compact正常压缩流程force_fullTrue if self._auto_compact: result self._auto_compact.dispatch(messages, force_fullTrue) # 检查压缩后是否低于窗口的 87.3%97% * 0.9 result_usage sum(self._estimate(m) for m in result.messages) if result_usage self._auto_compact.blocking_limit * 0.9: self._recovery_attempts 0 # 成功 → 重置计数器 return result # 尝试2-NAggressive Truncate激进截断 return self._aggressive_truncate(messages)11.3 渐进式激进截断# context_compactor.py:907-944 def _aggressive_truncate(self, messages): # 保留比例随重试次数递减 keep_ratio 0.4 - (self._recovery_attempts * 0.1) # 第1次重试: 40% # 第2次重试: 30% # 第3次重试: 20% # 最少保留 15% 或 3 条消息 keep_count max(3, int(len(non_system) * max(keep_ratio, 0.15))) truncated list(system_msgs) truncated.append({ role: system, content: f[Context aggressively truncated for recovery — fattempt {self._recovery_attempts}] fEarlier conversation was removed to fit context limits., }) truncated.extend(non_system[-keep_count:]) # 只保留最后 N 条三级恢复层次层次触发策略保真度正常路径85% 阈值先 Session Memory 后 Full Compact高有摘要Reactive 第1次API 报错Force Full Compact高Reactive 第2次仍失败保留 40%中Reactive 第3次仍失败保留 30%低Reactive 第4次放弃return None → 错误抛给用户—Force Full Compact 和 §十的 FullCompact 是同一个底层方法——_run_full_compact()。force_fullTrue只是跳过了 Session Memory 的尝试def dispatch(self, messages, force_fullFalse): if not force_full: # 正常路径 sm_result self._session_memory_engine.try_session_memory_compact(...) if sm_result and sm_result.effective: return sm_result # Session Memory 成功 → 返回 return self._run_full_compact(messages, usage) # force_full 或 SM 失败 → 同一个兜底区别只在于到达路径正常路径先试轻的紧急恢复直接上重的。十二、统一入口ContextCompactor.process_request()12.1 完整流水线# context_compactor.py:993-1054 def process_request(self, messages, *, enable_tool_budgetTrue, enable_read_dedupTrue, enable_microcompactTrue, enable_auto_compactTrue): current list(messages) total_freed 0 steps_taken [] # 阶段2: Tool Result Budget超大结果写盘 if enable_tool_budget: current, budget_saved self._tool_budget.check_and_replace(current) if budget_saved 0: total_freed budget_saved # 阶段3: Read Dedup在工具调用结果处理时使用这里只追踪状态 # 实际的去重在工具调度层完成 # 阶段4: Microcompact时间清理 if enable_microcompact: mc_result self._microcompact.run_time_based_microcompact(current) if mc_result.effective: current mc_result.messages total_freed mc_result.tokens_freed # 阶段56: Auto Compact高水位检查 → Session Memory / Full if enable_auto_compact and self._auto_compact.should_trigger(current): ac_result self._auto_compact.dispatch(current) if ac_result.effective: current ac_result.messages total_freed ac_result.tokens_freed return CompactionResult( successtotal_freed 0, messagescurrent, tokens_freedtotal_freed, )源码中为什么没有阶段1process_request()收到messages时调用方context_cybernetics.py已经把 boundary 之前的内容处理好了。Compactor 假设传入的就是活跃上下文boundary 之后的部分切分是上层的职责不是 Compactor 的职责。阶段1 是设计层面的概念不是代码层面的一个步骤。12.2 设计要点各阶段独立可开关enable_tool_budgetFalse→ 跳过阶段2适合短任务累加 token 释放量每个阶段释放的 token 都加到total_freed最终返回一个总结果串行执行不并行每个阶段依赖前一阶段的输出消息列表——压缩是累进的阶段3Read Dedup不在 process_request 中实现去重发生在工具调用时调度器拿到文件读取结果后先调register_read检查是否重复不在压缩流水线中。这里只提供ReadDedupManager实例供调度器调用为什么 Read Dedup 不在流水线中因为去重依赖正在被读取的文件内容——这只有工具调度器知道。压缩流水线看到的是已经注入上下文的 tool_result没有原始文件路径和内容哈希。十三、Token 估算整个压缩系统依赖 token 估算来判断当前用了多少 context。MiniCode 不调 API 做精确计数太慢、太贵而是用启发式估算在context_manager.py中def estimate_tokens(text: str) - int: cjk_chars sum(1 for c in text if 一 c 鿿) ascii_chars len(text) - cjk_chars return int(cjk_chars / 1.5 ascii_chars / 4)字符类型估算比率原因CJK中日韩1.5 字符/token一个汉字在大多数 tokenizer 中占 1-2 个 tokenASCII4 字符/token英文单词平均 4-5 个字母一个常见词 ≈ 1 token注意这个估算不是精确值——不同模型Claude vs GPT的 tokenizer 不同。但在压缩场景中我们只需要是否接近窗口上限的二元判断±10% 的误差可以接受。同一文件中的estimate_message_tokens()对消息做逐条估算带 LRU 缓存128 条。消息内容不变时不需要重复计算。十四、与 Memory 模块的关系SessionMemoryCompactEngine 是唯一直接依赖 Memory 模块的压缩路径MemoryManager.get_relevant_context(max_tokens6000) └─ 返回项目记忆摘要文本 └─ 用作 compaction 的摘要基础流程上 1.agent_loop.py初始化时把memory_manager传给ContextCompactor2.压缩触发时SessionMemoryCompactEngine查询 Memory 获取项目当前状态3.记忆内容被注入压缩后的 system 消息为什么记忆能做摘要基础Memory 模块维护了项目的决策、约定、已发现模式——这些正是压缩后最不该丢的信息。用户最初的需求重构异常处理可能已经在 Memory 中存为user_intent模式的条目了。十五、完整数据流agent_loop.py 每步调用: context_cybernetics.run_cycle(messages) ← 控制论决定是否压缩 │ └─ ContextCompactor.process_request() ← 执行层实际压缩 │ ├─ ToolResultBudgetManager │ 大工具结果(4000字符) → 写盘 .mini-code-tool-results/xxx.txt │ 上下文里保留 500 字符预览桩 │ ├─ MicrocompactEngine │ 超过 1 小时 超过 5 个工具结果 → 清理旧的保留最近 5 个 │ ├─ AutoCompactDispatcher │ │ usage 85% 窗口? │ │ │ ├─ YES → SessionMemoryCompactEngine │ │ Memory.get_relevant_context() → 记忆摘要 │ │ 保留尾部 10k-40k tokens tool_use/tool_result 配对保护 │ │ 省了 ≥ 5%? → 成功 │ │ │ └─ 失败 / Memory 不可用 → FullCompact │ 规则提取摘要话题、工具、文件、错误 │ 保留尾部 1/3 最近消息 │ └─→ CompactionResult(messages..., tokens_freed...) │ └─ effective? → messages 被替换原地修改API 报错路径model.next(messages) → API Error: prompt too long │ └─ ContextCompactor.reactive_recover() │ ├─ 第1次: Force Full Compact → 低于窗口 87.3%? → 成功 ├─ 第2次: 保留 40% 消息 → 成功? ├─ 第3次: 保留 30% 消息 → 成功? └─ 第4次: 放弃 → return None → 错误抛给用户十六、要点问题答案要点压缩系统的职责控制论决定何时压、压多少Compactor 决定用什么技术压为什么不直接截断截断丢信息——先用轻量级优化写盘/去重/清理尽量避免删消息7 阶段是哪 7 个1.重建上下文 → 2.Tool Budget → 3.Read Dedup → 4.Microcompact → 5.高水位检查 → 6.Session Memory/Full → 7.Reactive 恢复ToolResultBudgetManager超大工具结果(4000字符)写盘上下文留 500 字预览桩。原子写入ReadDedupManagerMD5 比对路径内容重复读取不重复注入。写操作后 invalidate 失效MicrocompactEngine每 1 小时清理旧工具结果保留最近 5 个替换为占位符不删消息、不改结构85% 高水位触发token 使用量 ≥ 窗口 85% 时触发 AutoCompactSessionMemoryCompact用 Memory 模块的项目记忆做摘要基础保留尾部 10k-40k tokens切点配对保护不能切断 tool_use/tool_result 对——否则 LLM 困惑FullCompact纯规则提取摘要话题/工具/文件/错误不需要 LLMCircuit Breaker连续 3 次压缩失败 → 不再触发防止死循环ReactiveCompactEngine3 级重试Force Full → 保留 40% → 保留 30% → 放弃Aggressive Truncatekeep_ratio 随重试递减0.4 → 0.3 → 0.2最少保留 15% 或 3 条Token 估算CJK: 1.5 字符/tokenASCII: 4 字符/token。LRU 缓存 128 条消息的结果compact boundary标记压缩点——下次压缩只处理 boundary 之后的消息CompactionResult.effectivesuccess and tokens_freed 0——必须真省了 tokenRead Dedup 为什么不在流水线中去重依赖正在被读取的文件内容只有工具调度器知道。流水线看到的是已注入的 tool_result十七、简要描述上下文压缩不能简单截断——那会丢掉用户最初的意图、已经探索过的代码结构、甚至截断未闭合的工具调用对。我设计了一个 7 阶段压缩流水线先用轻量级优化回收 token大工具结果写盘、重复文件读取去重、定时清理旧结果不够再触发高水位压缩优先用已有记忆做摘要、兜底结构化摘要最后还有 API 报错后的 3 级渐进恢复。关键设计点不丢信息写盘保留完整内容去重保留原始引用记忆做摘要保留项目知识保护工具调用对切点调整确保不切断 tool_use/tool_result 的配对断路器连续 3 次压缩失败后不再触发防止死循环纯规则实现不依赖额外 LLM 调用来做压缩——token 估算、去重、摘要提取全是确定性的