Skip to content

Commit ece2ea6

Browse files
AurelioloAureliolo
committed
fix(memory): address 28 review findings from 13 agents and 2 external reviewers
Source fixes: - Use NotBlankStr | None for DualModeConfig.summarization_model (#1) - Parallelize LLM calls with asyncio.TaskGroup in summarize_batch and _build_content (#2) - Remove dead-code guards in _build_anchors (#3) - Narrow except Exception to re-raise non-retryable ProviderErrors (#4) - Fix double-logging on abstractive fallback (#9) - Remove unnecessary import builtins (#10) - Preserve key-value pairs verbatim in extractive mode (#5) - Emit extracted facts one per line (#6) - Strengthen ConsolidationResult validator with cross-field checks (#7) - Check _backend.delete() return value in _process_group (#8) - Fix mode_map type to dict[NotBlankStr, ArchivalMode] (#11) - Move tie-breaking comment to _determine_group_mode (#12) - Fix misleading DualModeConfig docstring (#13) - Add missing mkdocstrings entries for retention/archival/simple_strategy (#14) - Use O(M) lookup dict in _archive_entries (#15) - Document 1000-entry query limit in run_consolidation docstring (#16) - Add Raises section to AbstractiveSummarizer docstring (#17) Test fixes: - Fix imports in test_density.py to module level (#18) - Strengthen fallback assertion to verify exact content (#19) - Use exact call counts for summarizer/extractor (#20) - Add tests: blank model rejection, MemoryError/RecursionError propagation (#21, #22) - Add tests: validator rejects invalid archival state (#23) - Add tests: 50/50 tie-breaking, None relevance handling (#24, #25) - Assert actual preserved facts in extractive tests (#26) - Prove archival index keyed by original_id not position (#27) - Add test: empty string classifies as SPARSE (#28)
1 parent 1830b89 commit ece2ea6

14 files changed

Lines changed: 334 additions & 75 deletions

File tree

‎docs/api/memory.md‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -77,6 +77,12 @@ Persistent agent memory — protocol, retrieval pipeline, shared org memory, con
7777

7878
::: synthorg.memory.consolidation.service
7979

80+
::: synthorg.memory.consolidation.retention
81+
82+
::: synthorg.memory.consolidation.archival
83+
84+
::: synthorg.memory.consolidation.simple_strategy
85+
8086
::: synthorg.memory.consolidation.density
8187

8288
::: synthorg.memory.consolidation.extractive

‎src/synthorg/memory/consolidation/abstractive.py‎

Lines changed: 43 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -5,7 +5,7 @@
55
if the LLM call fails.
66
"""
77

8-
import builtins
8+
import asyncio
99

1010
from synthorg.core.types import NotBlankStr # noqa: TC001
1111
from synthorg.memory.models import MemoryEntry # noqa: TC001
@@ -15,6 +15,7 @@
1515
DUAL_MODE_ABSTRACTIVE_SUMMARY,
1616
)
1717
from synthorg.providers.enums import MessageRole
18+
from synthorg.providers.errors import ProviderError
1819
from synthorg.providers.models import ChatMessage, CompletionConfig
1920
from synthorg.providers.protocol import CompletionProvider # noqa: TC001
2021

@@ -41,20 +42,23 @@ class AbstractiveSummarizer:
4142
4243
Uses a ``CompletionProvider`` to generate concise summaries of
4344
conversational/narrative memory content. Falls back to truncation
44-
if the LLM call fails.
45+
if the LLM call fails with a retryable error.
4546
4647
Args:
4748
provider: Completion provider for LLM calls.
4849
model: Model identifier to use for summarization.
4950
max_summary_tokens: Maximum tokens for the summary response.
5051
temperature: Sampling temperature for summarization.
52+
53+
Raises:
54+
ValueError: If ``model`` is empty or whitespace-only.
5155
"""
5256

5357
def __init__(
5458
self,
5559
*,
5660
provider: CompletionProvider,
57-
model: str,
61+
model: NotBlankStr,
5862
max_summary_tokens: int = 200,
5963
temperature: float = 0.3,
6064
) -> None:
@@ -71,8 +75,9 @@ def __init__(
7175
async def summarize(self, content: str) -> str:
7276
"""Generate an abstractive summary of the given content.
7377
74-
Falls back to truncation if the LLM call fails or returns
75-
empty content.
78+
Falls back to truncation if the LLM call fails with a
79+
retryable error or returns empty content. Non-retryable
80+
provider errors (authentication, invalid model) propagate.
7681
7782
Args:
7883
content: The sparse/conversational text to summarize.
@@ -98,42 +103,63 @@ async def summarize(self, content: str) -> str:
98103
model=self._model,
99104
)
100105
return response.content.strip()
101-
except builtins.MemoryError, RecursionError:
106+
except MemoryError, RecursionError:
102107
raise
108+
except ProviderError as exc:
109+
if not exc.is_retryable:
110+
raise
111+
logger.warning(
112+
DUAL_MODE_ABSTRACTIVE_FALLBACK,
113+
content_length=len(content),
114+
error=str(exc),
115+
error_type=type(exc).__name__,
116+
)
117+
return _truncate_fallback(content)
103118
except Exception as exc:
104119
logger.warning(
105120
DUAL_MODE_ABSTRACTIVE_FALLBACK,
106121
content_length=len(content),
107122
error=str(exc),
108123
error_type=type(exc).__name__,
109124
)
125+
return _truncate_fallback(content)
110126

111-
# Fallback: truncation
127+
# Fallback: empty/whitespace-only LLM response
112128
logger.debug(
113129
DUAL_MODE_ABSTRACTIVE_FALLBACK,
114130
content_length=len(content),
115-
reason="empty_or_failed",
131+
reason="empty_response",
116132
)
117133
return _truncate_fallback(content)
118134

119135
async def summarize_batch(
120136
self,
121137
entries: tuple[MemoryEntry, ...],
122138
) -> tuple[tuple[NotBlankStr, str], ...]:
123-
"""Summarize multiple entries.
139+
"""Summarize multiple entries concurrently.
124140
125-
Each entry is summarized independently. Failures for
126-
individual entries fall back to truncation without aborting
127-
the batch.
141+
Each entry is summarized independently via ``asyncio.TaskGroup``.
142+
Failures for individual entries fall back to truncation without
143+
aborting the batch.
128144
129145
Args:
130146
entries: Memory entries to summarize.
131147
132148
Returns:
133149
Tuple of ``(entry_id, summary)`` pairs in input order.
134150
"""
135-
results: list[tuple[NotBlankStr, str]] = []
136-
for entry in entries:
137-
summary = await self.summarize(entry.content)
138-
results.append((entry.id, summary))
139-
return tuple(results)
151+
if not entries:
152+
return ()
153+
154+
results: dict[NotBlankStr, str] = {}
155+
async with asyncio.TaskGroup() as tg:
156+
tasks: dict[NotBlankStr, asyncio.Task[str]] = {}
157+
for entry in entries:
158+
tasks[entry.id] = tg.create_task(
159+
self.summarize(entry.content),
160+
)
161+
162+
for entry_id, task in tasks.items():
163+
results[entry_id] = task.result()
164+
165+
return tuple((entry.id, results[entry.id]) for entry in entries)

‎src/synthorg/memory/consolidation/config.py‎

Lines changed: 5 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,7 @@
99
from pydantic import BaseModel, ConfigDict, Field, model_validator
1010

1111
from synthorg.core.enums import ConsolidationInterval
12+
from synthorg.core.types import NotBlankStr # noqa: TC001
1213
from synthorg.memory.consolidation.models import RetentionRule # noqa: TC001
1314

1415

@@ -57,7 +58,7 @@ class DualModeConfig(BaseModel):
5758
5859
Attributes:
5960
enabled: Whether dual-mode density classification is active.
60-
When ``False``, all entries use abstractive mode.
61+
When ``False``, the dual-mode strategy is not used.
6162
dense_threshold: Density score threshold for DENSE classification
6263
(0.0 = classify everything as dense, 1.0 = everything sparse).
6364
summarization_model: Model ID for abstractive summarization.
@@ -80,8 +81,8 @@ class DualModeConfig(BaseModel):
8081
le=1.0,
8182
description="Density score threshold for DENSE classification",
8283
)
83-
summarization_model: str = Field(
84-
default="",
84+
summarization_model: NotBlankStr | None = Field(
85+
default=None,
8586
description="Model ID for abstractive summarization",
8687
)
8788
max_summary_tokens: int = Field(
@@ -106,7 +107,7 @@ class DualModeConfig(BaseModel):
106107
@model_validator(mode="after")
107108
def _validate_model_when_enabled(self) -> Self:
108109
"""Require a summarization model when dual-mode is enabled."""
109-
if self.enabled and not self.summarization_model.strip():
110+
if self.enabled and self.summarization_model is None:
110111
msg = "summarization_model must be non-blank when dual-mode is enabled"
111112
raise ValueError(msg)
112113
return self

‎src/synthorg/memory/consolidation/dual_mode_strategy.py‎

Lines changed: 22 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,7 @@
55
preservation (dense) accordingly.
66
"""
77

8+
import asyncio
89
from itertools import groupby
910
from operator import attrgetter
1011

@@ -179,8 +180,6 @@ async def _process_group(
179180
group_mode = self._determine_group_mode(group_tuple)
180181
_, to_remove = self._select_entries(group_tuple)
181182

182-
# Tie-breaking note: 50/50 dense/sparse splits default to
183-
# ABSTRACTIVE (strict > comparison), which is the safer mode.
184183
logger.debug(
185184
DUAL_MODE_GROUP_CLASSIFIED,
186185
agent_id=agent_id,
@@ -203,7 +202,16 @@ async def _process_group(
203202
removed_ids: list[NotBlankStr] = []
204203
assignments: list[ArchivalModeAssignment] = []
205204
for entry in to_remove:
206-
await self._backend.delete(agent_id, entry.id)
205+
deleted = await self._backend.delete(agent_id, entry.id)
206+
if not deleted:
207+
logger.warning(
208+
DUAL_MODE_GROUP_CLASSIFIED,
209+
agent_id=agent_id,
210+
category=category.value,
211+
reason="delete_not_found",
212+
entry_id=entry.id,
213+
)
214+
continue
207215
removed_ids.append(entry.id)
208216
assignments.append(
209217
ArchivalModeAssignment(
@@ -229,6 +237,8 @@ def _determine_group_mode(
229237
dense_count = sum(
230238
1 for _, density in classified if density == ContentDensity.DENSE
231239
)
240+
# Tie-breaking: 50/50 dense/sparse splits default to
241+
# ABSTRACTIVE (strict > comparison), which is the safer mode.
232242
is_majority_dense = dense_count > len(classified) / 2
233243
return (
234244
ArchivalMode.EXTRACTIVE if is_majority_dense else ArchivalMode.ABSTRACTIVE
@@ -273,11 +283,13 @@ async def _build_content(
273283
Returns:
274284
Consolidated content text.
275285
"""
276-
parts: list[str] = []
277-
for entry in entries:
278-
if mode == ArchivalMode.EXTRACTIVE:
279-
parts.append(self._extractor.extract(entry.content))
280-
else:
281-
summary = await self._summarizer.summarize(entry.content)
282-
parts.append(summary)
286+
if mode == ArchivalMode.EXTRACTIVE:
287+
parts = [self._extractor.extract(e.content) for e in entries]
288+
else:
289+
async with asyncio.TaskGroup() as tg:
290+
tasks = [
291+
tg.create_task(self._summarizer.summarize(e.content))
292+
for e in entries
293+
]
294+
parts = [t.result() for t in tasks]
283295
return "\n---\n".join(parts)

‎src/synthorg/memory/consolidation/extractive.py‎

Lines changed: 7 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -29,7 +29,7 @@
2929
_VERSION_PATTERN = re.compile(r"\bv?\d+\.\d+\.\d+(?:\.\d+)?\b")
3030

3131
_KEY_VALUE_PATTERN = re.compile(
32-
r"^\s*([\w.-]+)\s*[:=]\s*(\S.*)$",
32+
r"^\s*([\w.-]+\s*[:=]\s*.*)$",
3333
re.MULTILINE,
3434
)
3535

@@ -52,9 +52,8 @@ def _extract_versions(text: str) -> list[str]:
5252

5353

5454
def _extract_key_values(text: str) -> list[str]:
55-
"""Extract key=value and key: value pairs from text."""
56-
matches = _KEY_VALUE_PATTERN.findall(text)
57-
return [f"{k}={v.strip()}" for k, v in matches]
55+
"""Extract key-value pairs from text, preserving original form."""
56+
return [m.strip() for m in _KEY_VALUE_PATTERN.findall(text)]
5857

5958

6059
def _build_anchors(
@@ -75,9 +74,7 @@ def _build_anchors(
7574
if text_len <= anchor_length:
7675
return text, "", ""
7776

78-
start = text[:anchor_length]
79-
if text_len > anchor_length:
80-
start += "..."
77+
start = text[:anchor_length] + "..."
8178

8279
mid_offset = max(0, (text_len - anchor_length) // 2)
8380
mid = text[mid_offset : mid_offset + anchor_length]
@@ -86,9 +83,7 @@ def _build_anchors(
8683
if mid_offset + anchor_length < text_len:
8784
mid += "..."
8885

89-
end = text[-anchor_length:]
90-
if text_len > anchor_length:
91-
end = "..." + end
86+
end = "..." + text[-anchor_length:]
9287

9388
return start, mid, end
9489

@@ -152,7 +147,8 @@ def extract(self, content: str) -> str:
152147

153148
lines = ["[Extractive preservation]"]
154149
if facts:
155-
lines.append(f"Key facts: {', '.join(facts)}")
150+
lines.append("Key facts:")
151+
lines.extend(f"- {fact}" for fact in facts)
156152
lines.append(f"[START] {start}")
157153
if mid:
158154
lines.append(f"[MID] {mid}")

‎src/synthorg/memory/consolidation/models.py‎

Lines changed: 26 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -116,14 +116,38 @@ class ConsolidationResult(BaseModel):
116116
)
117117

118118
@model_validator(mode="after")
119-
def _validate_index_count(self) -> Self:
120-
"""Ensure archival index does not exceed archived count."""
119+
def _validate_archival_consistency(self) -> Self:
120+
"""Ensure archival fields are internally consistent."""
121+
if self.archived_count > self.consolidated_count:
122+
msg = (
123+
f"archived_count ({self.archived_count}) must not exceed "
124+
f"consolidated_count ({self.consolidated_count})"
125+
)
126+
raise ValueError(msg)
121127
if len(self.archival_index) > self.archived_count:
122128
msg = (
123129
f"archival_index length ({len(self.archival_index)}) "
124130
f"must not exceed archived_count ({self.archived_count})"
125131
)
126132
raise ValueError(msg)
133+
if len(self.mode_assignments) > len(self.removed_ids):
134+
msg = (
135+
f"mode_assignments length ({len(self.mode_assignments)}) "
136+
f"must not exceed removed_ids length "
137+
f"({len(self.removed_ids)})"
138+
)
139+
raise ValueError(msg)
140+
removed_set = set(self.removed_ids)
141+
for idx_entry in self.archival_index:
142+
if idx_entry.original_id not in removed_set:
143+
msg = (
144+
f"archival_index entry '{idx_entry.original_id}' not in removed_ids"
145+
)
146+
raise ValueError(msg)
147+
index_ids = [e.original_id for e in self.archival_index]
148+
if len(index_ids) != len(set(index_ids)):
149+
msg = "archival_index contains duplicate original_ids"
150+
raise ValueError(msg)
127151
return self
128152

129153
@computed_field # type: ignore[prop-decorator]

‎src/synthorg/memory/consolidation/service.py‎

Lines changed: 12 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -78,10 +78,10 @@ async def run_consolidation(
7878
) -> ConsolidationResult:
7979
"""Run memory consolidation for an agent.
8080
81-
Retrieves memories, applies the consolidation strategy, and
82-
archives removed entries if archival is configured and enabled.
83-
Per-entry archival failures are logged and skipped — they do
84-
not abort the entire archival batch.
81+
Retrieves up to 1000 entries per invocation and applies the
82+
consolidation strategy, then archives removed entries if archival
83+
is configured and enabled. Per-entry archival failures are
84+
logged and skipped — they do not abort the entire batch.
8585
8686
Args:
8787
agent_id: Agent whose memories to consolidate.
@@ -260,14 +260,17 @@ async def _archive_entries(
260260
if self._archival_store is None:
261261
return 0, ()
262262

263-
mode_map = {a.original_id: a.mode for a in mode_assignments}
264-
removed_set = set(removed_ids)
263+
mode_map: dict[NotBlankStr, ArchivalMode] = {
264+
a.original_id: a.mode for a in mode_assignments
265+
}
266+
entry_map = {entry.id: entry for entry in all_entries}
265267
now = datetime.now(UTC)
266268
archived = 0
267269
index_entries: list[ArchivalIndexEntry] = []
268270

269-
for entry in all_entries:
270-
if entry.id not in removed_set:
271+
for removed_id in removed_ids:
272+
entry = entry_map.get(removed_id)
273+
if entry is None:
271274
continue
272275
success, idx = await self._archive_single_entry(
273276
entry,
@@ -294,7 +297,7 @@ async def _archive_single_entry(
294297
self,
295298
entry: MemoryEntry,
296299
agent_id: NotBlankStr,
297-
mode_map: dict[str, ArchivalMode],
300+
mode_map: dict[NotBlankStr, ArchivalMode],
298301
now: datetime,
299302
) -> tuple[bool, ArchivalIndexEntry | None]:
300303
"""Archive a single entry to cold storage.

0 commit comments

Comments
 (0)