Coverage for agentos/memory/summarizer.py: 34%
132 statements
« prev ^ index » next coverage.py v7.14.3, created at 2026-07-06 19:15 +0800
1"""
2AgentOS v0.60 Memory Summarizer — 上下文压缩与记忆管理。
3递归摘要 / 重要性评分 / 滑动窗口 / 混合记忆策略。
4"""
6from __future__ import annotations
8import math
9import time
10from collections.abc import Callable
11from dataclasses import dataclass, field
12from enum import StrEnum
15class MemoryType(StrEnum):
16 """记忆类型枚举。"""
18 EPISODIC = "episodic" # 对话片段
19 SEMANTIC = "semantic" # 知识点
20 PROCEDURAL = "procedural" # 操作步骤
21 WORKING = "working" # 当前上下文
24@dataclass
25class MemoryChunk:
26 """记忆块。"""
28 id: str
29 content: str
30 mtype: MemoryType = MemoryType.EPISODIC
31 timestamp: float = field(default_factory=time.time)
32 importance: float = 0.5 # 0~1
33 access_count: int = 0
34 token_estimate: int = 0
35 summary: str = ""
36 metadata: dict = field(default_factory=dict)
38 def __post_init__(self):
39 if self.token_estimate == 0:
40 self.token_estimate = max(1, len(self.content) // 3)
43class ImportanceScorer:
44 """多维度重要性评分。"""
46 WEIGHTS = {
47 "recency": 0.20, # 时间衰减
48 "access_frequency": 0.15, # 访问频率
49 "content_length": 0.10, # 内容长度（过短=噪音，适中=有用）
50 "keyword_density": 0.25, # 关键信息密度
51 "task_relevance": 0.30, # 任务相关性(外部传入)
52 }
54 _IMPORTANT_KEYWORDS = [
55 "error",
56 "exception",
57 "fail",
58 "critical",
59 "important",
60 "key",
61 "decision",
62 "conclusion",
63 "result",
64 "summary",
65 "must",
66 "urgent",
67 "deadline",
68 "blocker",
69 "fix",
70 ]
72 @classmethod
73 def score(
74 cls, chunk: MemoryChunk, task_relevance: float = 0.0, current_time: float | None = None
75 ) -> float:
76 now = current_time or time.time()
77 scores = {}
79 # 1. 时间衰减（指数衰减，半衰期24h）
80 age_hours = (now - chunk.timestamp) / 3600
81 scores["recency"] = math.exp(-age_hours * math.log(2) / 24)
83 # 2. 访问频率
84 scores["access_frequency"] = min(1.0, chunk.access_count / 10.0)
86 # 3. 内容长度评分（100~2000 token 最佳）
87 t = chunk.token_estimate
88 if t < 50:
89 scores["content_length"] = t / 50 * 0.3
90 elif t <= 2000:
91 scores["content_length"] = 1.0
92 else:
93 scores["content_length"] = max(0.1, 2000 / t)
95 # 4. 关键词密度
96 lowered = chunk.content.lower()
97 keyword_hits = sum(1 for kw in cls._IMPORTANT_KEYWORDS if kw in lowered)
98 scores["keyword_density"] = min(1.0, keyword_hits / 5.0)
100 # 5. 任务相关性
101 scores["task_relevance"] = task_relevance
103 total = sum(cls.WEIGHTS[k] * scores[k] for k in cls.WEIGHTS)
104 return round(min(1.0, max(0.0, total)), 4)
107class MemorySummarizer:
108 """记忆摘要器：递归压缩 + 重要性排序 + 滑动窗口裁剪。"""
110 def __init__(
111 self, max_context_tokens: int = 8000, summarizer_fn: Callable[[str], str] | None = None
112 ):
113 self.max_context_tokens = max_context_tokens
114 self._summarizer = summarizer_fn or self._default_summarizer
116 @staticmethod
117 def _default_summarizer(text: str) -> str:
118 """默认摘要器：提取首句 + 关键片段。"""
119 lines = [ln.strip() for ln in text.split("\n") if ln.strip()]
120 if len(lines) <= 3:
121 return " ".join(lines)
122 first = lines[0][:200]
123 # 截取中间代表性句子
124 mid = len(lines) // 2
125 snippet = lines[mid][:150] if mid < len(lines) else ""
126 return f"[{len(lines)}行] {first} ... {snippet}".strip()[:500]
128 # ── 递归摘要 ───────────────────────────────────────────────────────────
130 def recursive_summarize(
131 self, chunks: list[MemoryChunk], target_ratio: float = 0.3
132 ) -> list[MemoryChunk]:
133 """递归压缩：反复摘要直到总 token 数降至目标比例以下。"""
134 current = list(chunks)
135 total_tokens = sum(c.token_estimate for c in current)
136 target_tokens = int(self.max_context_tokens * target_ratio)
138 while total_tokens > target_tokens and len(current) > 1:
139 # 合并相邻 chunk 并摘要
140 merged: list[MemoryChunk] = []
141 for i in range(0, len(current) - 1, 2):
142 combined = current[i].content + "\n" + current[i + 1].content
143 summary = self._summarizer(combined)
144 merged.append(
145 MemoryChunk(
146 id=f"sum_{i}",
147 content=summary,
148 mtype=MemoryType.SEMANTIC,
149 importance=max(current[i].importance, current[i + 1].importance),
150 )
151 )
152 if len(current) % 2 == 1:
153 merged.append(current[-1])
154 current = merged
155 total_tokens = sum(c.token_estimate for c in current)
157 return current
159 # ── 重要性排序 ─────────────────────────────────────────────────────────
161 def rank_and_prune(self, chunks: list[MemoryChunk], max_chunks: int = 20) -> list[MemoryChunk]:
162 """按重要性排序并截断。"""
163 scored = [(ImportanceScorer.score(c), c) for c in chunks]
164 scored.sort(key=lambda x: x[0], reverse=True)
165 return [c for _, c in scored[:max_chunks]]
167 # ── 滑动窗口 ───────────────────────────────────────────────────────────
169 def sliding_window(self, chunks: list[MemoryChunk], window_size: int = 10) -> list[MemoryChunk]:
170 """最近N条记忆（按时间排序）。"""
171 sorted_chunks = sorted(chunks, key=lambda c: c.timestamp, reverse=True)
172 return sorted_chunks[:window_size]
174 # ── 混合策略 ───────────────────────────────────────────────────────────
176 def build_context(
177 self, chunks: list[MemoryChunk], strategy: str = "hybrid"
178 ) -> list[MemoryChunk]:
179 """构建上下文：混合策略 = 重要记忆 + 最近窗口。"""
180 if strategy == "recency":
181 return self.sliding_window(chunks, 15)
182 elif strategy == "importance":
183 return self.rank_and_prune(chunks, 15)
184 elif strategy == "hybrid":
185 recent = set(c.id for c in self.sliding_window(chunks, 7))
186 important = self.rank_and_prune(chunks, 15)
187 hybrid: list[MemoryChunk] = []
188 seen: set[str] = set()
189 for c in important:
190 if c.id not in seen:
191 hybrid.append(c)
192 seen.add(c.id)
193 for c in chunks:
194 if c.id in recent and c.id not in seen:
195 hybrid.append(c)
196 seen.add(c.id)
197 return hybrid
198 return chunks
200 def estimate_tokens(self, chunks: list[MemoryChunk]) -> int:
201 return sum(c.token_estimate for c in chunks)
204class ConversationMemory:
205 """对话记忆：按轮次组织，支持压缩与重置。"""
207 def __init__(self, max_turns: int = 50, summarizer: MemorySummarizer | None = None):
208 self.max_turns = max_turns
209 self.turns: list[MemoryChunk] = []
210 self.summarizer = summarizer or MemorySummarizer()
211 self._backup: list[MemoryChunk] = []
213 def add_turn(self, role: str, content: str, metadata: dict | None = None):
214 chunk = MemoryChunk(
215 id=f"turn_{len(self.turns)}",
216 content=f"[{role}] {content}",
217 mtype=MemoryType.EPISODIC,
218 importance=0.6 if role == "user" else 0.4,
219 metadata=metadata or {},
220 )
221 self.turns.append(chunk)
222 if len(self.turns) > self.max_turns:
223 self.compress()
225 def compress(self):
226 """压缩旧对话为摘要。"""
227 if len(self.turns) <= self.max_turns:
228 return
229 old_half = self.turns[: len(self.turns) // 2]
230 self._backup = old_half
231 compressed = self.summarizer.recursive_summarize(old_half, target_ratio=0.2)
232 self.turns = compressed + self.turns[len(self.turns) // 2 :]
234 def clear(self):
235 self.turns.clear()
236 self._backup.clear()
238 def restore(self):
239 """从备份恢复完整对话。"""
240 if self._backup:
241 self.turns = self._backup + self.turns
242 self._backup.clear()
244 @property
245 def total_tokens(self) -> int:
246 return self.summarizer.estimate_tokens(self.turns)