Coverage for agentos/memory/compressor.py: 29%

1""" 

2四级上下文压缩 — Claude Code核心工程洞察。 

3基因来源: Claude Code的s06 Context Compression 

4""" 

5 

6from __future__ import annotations 

7 

8from agentos.core.context import AgentContext, Message 

9 

10 

11class ContextCompressor: 

12 """ 

13 四级上下文压缩 — 不是一刀切截断，而是分层渐进式压缩。 

14 

15 L1 滑动窗口: 保留最近N轮对话 

16 L2 工具结果摘要: 长输出→结构化摘要 

17 L3 语义压缩: LLM压缩历史并保留关键信息 

18 L4 文件系统卸载: 关键信息写入磁盘按需读取 

19 """ 

20 

21 def __init__( 

22 self, 

23 window_size: int = 50, 

24 max_tool_output_chars: int = 2000, 

25 target_tokens: int = 100_000, 

26 ): 

27 self.window_size = window_size 

28 self.max_tool_output_chars = max_tool_output_chars 

29 self.target_tokens = target_tokens 

30 

31 def compress(self, context: AgentContext) -> AgentContext: 

32 """渐进式压缩上下文到合理大小。""" 

33 

34 # L1: 滑动窗口 

35 if len(context.messages) > self.window_size: 

36 context.messages = self._apply_sliding_window(context.messages) 

37 

38 # L2: 工具结果摘要 

39 context.messages = self._summarize_tool_outputs(context.messages) 

40 

41 # L3: 语义压缩（需要调用LLM，留接口） 

42 if self._estimate_tokens(context) > self.target_tokens: 

43 context = self._semantic_compress_stub(context) 

44 

45 return context 

46 

47 def _apply_sliding_window(self, messages: list[Message]) -> list[Message]: 

48 """L1: 保留system消息 + 最近N条消息。""" 

49 system_msgs = [m for m in messages if m.role == "system"] 

50 others = [m for m in messages if m.role != "system"] 

51 return system_msgs + others[-self.window_size:] 

52 

53 def _summarize_tool_outputs(self, messages: list[Message]) -> list[Message]: 

54 """L2: 截断过长的工具输出。""" 

55 for msg in messages: 

56 if msg.role == "tool" and len(msg.content) > self.max_tool_output_chars: 

57 truncated = msg.content[: self.max_tool_output_chars] 

58 msg.content = ( 

59 truncated 

60 + f"\n... [truncated {len(msg.content) - self.max_tool_output_chars} chars]" 

61 ) 

62 return messages 

63 

64 def _semantic_compress_stub(self, context: AgentContext) -> AgentContext: 

65 """L3: 语义压缩（当前为stub，实际使用时需调用轻量模型）。""" 

66 # TODO: 调用轻量模型压缩历史消息 

67 return context 

68 

69 def _estimate_tokens(self, context: AgentContext) -> int: 

70 """粗略估算token数。""" 

71 total = 0 

72 for msg in context.messages: 

73 total += len(msg.content) // 4 

74 return total