Skip to content

Commit e27477f

Browse files
committed
feat(agent-memory): Phase 6 — consolidate()
- consolidate() summarizes old/low-importance memories into one durable source:'summary' memory via a caller-supplied summarize(items) callback - Select candidates by scope/tags + olderThanSeconds + maxImportance; write the summary before deleting sources so a failure never destroys memories without leaving the replacement - deleteSources defaults true; summaryImportance defaults 0.7 - Require at least one selection criterion to prevent whole-store consolidation (mirrors forgetByScope's mass-delete guard) - Fetch only the fields parseMemoryItem needs (no vector blobs) on the candidate scan - Add ConsolidateOptions/ConsolidateResult types
1 parent a820313 commit e27477f

4 files changed

Lines changed: 297 additions & 0 deletions

File tree

packages/agent-memory/src/MemoryStore.ts

Lines changed: 93 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -6,8 +6,11 @@ import { parseMemoryItem } from './parseMemoryItem';
66
import { compositeScore, similarityFromDistance, type RecallWeights } from './compositeScore';
77
import { selectEvictions, type EvictionCandidate } from './selectEvictions';
88
import type {
9+
ConsolidateOptions,
10+
ConsolidateResult,
911
EmbedFn,
1012
MemoryHit,
13+
MemoryItem,
1114
MemoryScope,
1215
MemoryStoreClient,
1316
RecallOptions,
@@ -22,6 +25,8 @@ const RECALL_OVERFETCH = 4;
2225
const FORGET_BATCH_SIZE = 500;
2326
const FORGET_MAX_BATCHES = 10000;
2427
const EVICTION_SCAN_LIMIT = 10000;
28+
const CONSOLIDATE_SCAN_LIMIT = 10000;
29+
const DEFAULT_SUMMARY_IMPORTANCE = 0.7;
2530

2631
export interface MemoryStoreOptions {
2732
client: MemoryStoreClient;
@@ -200,6 +205,77 @@ export class MemoryStore {
200205
return id;
201206
}
202207

208+
async consolidate(options: ConsolidateOptions): Promise<ConsolidateResult> {
209+
const now = Date.now();
210+
const tags = options.tags ?? [];
211+
const scope: MemoryScope = {
212+
threadId: options.threadId,
213+
agentId: options.agentId,
214+
namespace: options.namespace,
215+
};
216+
217+
const hasCriteria =
218+
scope.threadId !== undefined ||
219+
scope.agentId !== undefined ||
220+
scope.namespace !== undefined ||
221+
tags.length > 0 ||
222+
options.olderThanSeconds !== undefined ||
223+
options.maxImportance !== undefined;
224+
if (!hasCriteria) {
225+
throw new Error(
226+
'consolidate requires a scope, tags, olderThanSeconds, or maxImportance to select candidates',
227+
);
228+
}
229+
230+
const raw = await this.client.call(
231+
'FT.SEARCH',
232+
`${this.name}:mem:idx`,
233+
buildScopeFilter(scope, tags),
234+
'RETURN',
235+
'10',
236+
'content',
237+
'importance',
238+
'tags',
239+
'created_at',
240+
'last_accessed_at',
241+
'access_count',
242+
'source',
243+
'threadId',
244+
'agentId',
245+
'namespace',
246+
'LIMIT',
247+
'0',
248+
String(CONSOLIDATE_SCAN_LIMIT),
249+
'DIALECT',
250+
'2',
251+
);
252+
const candidates = parseFtSearchResponse(raw)
253+
.map((hit) => parseMemoryItem(this.name, hit))
254+
.filter((item) => isConsolidationCandidate(item, options, now));
255+
256+
if (candidates.length === 0) {
257+
return { consolidated: 0, created: [], deleted: 0 };
258+
}
259+
260+
// Write the summary before deleting sources so a failure can never destroy
261+
// memories without leaving their consolidated replacement behind.
262+
const summary = await options.summarize(candidates);
263+
const summaryId = await this.remember(summary, {
264+
...scope,
265+
tags,
266+
source: 'summary',
267+
importance: options.summaryImportance ?? DEFAULT_SUMMARY_IMPORTANCE,
268+
});
269+
270+
let deleted = 0;
271+
if (options.deleteSources !== false) {
272+
const keys = candidates.map((item) => `${this.name}:mem:${item.id}`);
273+
deleted = Number(await this.client.call('DEL', ...keys));
274+
}
275+
276+
return { consolidated: candidates.length, created: [summaryId], deleted };
277+
}
278+
203279
private async writeRecord(key: string, fields: (string | Buffer)[], ttl?: number): Promise<void> {
204280
if (ttl === undefined || ttl <= 0) {
205281
await this.client.call('HSET', key, ...fields);
@@ -303,3 +379,20 @@ function ftSearchTotal(raw: unknown): number {
303379
const total = typeof raw[0] === 'string' ? parseInt(raw[0], 10) : Number(raw[0]);
304380
return Number.isFinite(total) && total > 0 ? total : 0;
305381
}
382+
383+
function isConsolidationCandidate(
384+
item: MemoryItem,
385+
options: ConsolidateOptions,
386+
now: number,
387+
): boolean {
388+
if (options.maxImportance !== undefined && item.importance > options.maxImportance) {
389+
return false;
390+
}
391+
if (options.olderThanSeconds !== undefined) {
392+
const ageSeconds = (now - item.createdAt) / 1000;
393+
if (ageSeconds < options.olderThanSeconds) {
394+
return false;
395+
}
396+
}
397+
return true;
398+
}
Lines changed: 187 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,187 @@
1+
import { describe, it, expect, vi } from 'vitest';
2+
import { MemoryStore } from '../MemoryStore';
3+
import { fakeEmbed } from './helpers/fakeEmbed';
4+
import { mockClient } from './helpers/mockClient';
5+
6+
const now = Date.now();
7+
8+
interface HitSpec {
9+
importance: number;
10+
ageSeconds: number;
11+
source?: string;
12+
}
13+
14+
function itemHit(id: string, spec: HitSpec): [string, string[]] {
15+
const created = now - spec.ageSeconds * 1000;
16+
const fields: Record<string, string> = {
17+
content: `c-${id}`,
18+
importance: String(spec.importance),
19+
created_at: String(created),
20+
last_accessed_at: String(created),
21+
access_count: '0',
22+
};
23+
if (spec.source !== undefined) {
24+
fields.source = spec.source;
25+
}
26+
const flat: string[] = [];
27+
for (const [field, value] of Object.entries(fields)) {
28+
flat.push(field, value);
29+
}
30+
return [`mem:mem:${id}`, flat];
31+
}
32+
33+
function searchReply(hits: Array<[string, string[]]>): unknown[] {
34+
const out: unknown[] = [String(hits.length)];
35+
for (const [key, flat] of hits) {
36+
out.push(key, flat);
37+
}
38+
return out;
39+
}
40+
41+
function consolidatingClient(hits: Array<[string, string[]]>) {
42+
return mockClient((command, ...args) => {
43+
if (command === 'FT.SEARCH') {
44+
return searchReply(hits);
45+
}
46+
if (command === 'DEL') {
47+
return args.length;
48+
}
49+
return 'OK';
50+
});
51+
}
52+
53+
function fieldValue(call: unknown[] | undefined, field: string): string | undefined {
54+
if (!call) {
55+
return undefined;
56+
}
57+
const idx = call.indexOf(field);
58+
return idx >= 0 ? (call[idx + 1] as string) : undefined;
59+
}
60+
61+
describe('MemoryStore.consolidate', () => {
62+
it('summarizes matching candidates, writes a summary, deletes sources, returns counts', async () => {
63+
const summarize = vi.fn(async (items) => `summary of ${items.length}`);
64+
const client = consolidatingClient([
65+
itemHit('a', { importance: 0.2, ageSeconds: 100000 }),
66+
itemHit('b', { importance: 0.3, ageSeconds: 200000 }),
67+
]);
68+
const store = new MemoryStore({ client, name: 'mem', embedFn: fakeEmbed(8) });
69+
70+
const result = await store.consolidate({
71+
namespace: 'u1',
72+
olderThanSeconds: 3600,
73+
maxImportance: 0.5,
74+
summarize,
75+
});
76+
77+
expect(summarize).toHaveBeenCalledTimes(1);
78+
expect(summarize.mock.calls[0][0].map((i: { id: string }) => i.id)).toEqual(['a', 'b']);
79+
expect(result.consolidated).toBe(2);
80+
expect(result.created).toHaveLength(1);
81+
expect(result.deleted).toBe(2);
82+
83+
const hset = client.call.mock.calls.find((c) => c[0] === 'HSET');
84+
expect(fieldValue(hset, 'content')).toBe('summary of 2');
85+
expect(fieldValue(hset, 'source')).toBe('summary');
86+
expect(hset?.[1]).toBe(`mem:mem:${result.created[0]}`);
87+
88+
const del = client.call.mock.calls.find((c) => c[0] === 'DEL');
89+
expect(del?.slice(1).sort()).toEqual(['mem:mem:a', 'mem:mem:b']);
90+
});
91+
92+
it('selects only candidates older than olderThanSeconds', async () => {
93+
const summarize = vi.fn(async () => 'summary');
94+
const client = consolidatingClient([
95+
itemHit('old', { importance: 0.2, ageSeconds: 100000 }),
96+
itemHit('recent', { importance: 0.2, ageSeconds: 10 }),
97+
]);
98+
const store = new MemoryStore({ client, name: 'mem', embedFn: fakeEmbed(8) });
99+
100+
const result = await store.consolidate({ olderThanSeconds: 3600, summarize });
101+
102+
expect(summarize.mock.calls[0][0].map((i: { id: string }) => i.id)).toEqual(['old']);
103+
expect(result.consolidated).toBe(1);
104+
const del = client.call.mock.calls.find((c) => c[0] === 'DEL');
105+
expect(del?.slice(1)).toEqual(['mem:mem:old']);
106+
});
107+
108+
it('selects only candidates at or below maxImportance', async () => {
109+
const summarize = vi.fn(async () => 'summary');
110+
const client = consolidatingClient([
111+
itemHit('low', { importance: 0.2, ageSeconds: 100000 }),
112+
itemHit('high', { importance: 0.9, ageSeconds: 100000 }),
113+
]);
114+
const store = new MemoryStore({ client, name: 'mem', embedFn: fakeEmbed(8) });
115+
116+
const result = await store.consolidate({ maxImportance: 0.5, summarize });
117+
118+
expect(summarize.mock.calls[0][0].map((i: { id: string }) => i.id)).toEqual(['low']);
119+
expect(result.consolidated).toBe(1);
120+
});
121+
122+
it('writes the summary scoped to the request at summaryImportance', async () => {
123+
const summarize = vi.fn(async () => 'merged');
124+
const client = consolidatingClient([itemHit('a', { importance: 0.1, ageSeconds: 100000 })]);
125+
const store = new MemoryStore({ client, name: 'mem', embedFn: fakeEmbed(8) });
126+
127+
await store.consolidate({ namespace: 'u1', summarize, summaryImportance: 0.9 });
128+
129+
const hset = client.call.mock.calls.find((c) => c[0] === 'HSET');
130+
expect(fieldValue(hset, 'importance')).toBe('0.9');
131+
expect(fieldValue(hset, 'namespace')).toBe('u1');
132+
expect(fieldValue(hset, 'source')).toBe('summary');
133+
});
134+
135+
it('keeps sources when deleteSources is false', async () => {
136+
const summarize = vi.fn(async () => 'summary');
137+
const client = consolidatingClient([
138+
itemHit('a', { importance: 0.2, ageSeconds: 100000 }),
139+
itemHit('b', { importance: 0.2, ageSeconds: 100000 }),
140+
]);
141+
const store = new MemoryStore({ client, name: 'mem', embedFn: fakeEmbed(8) });
142+
143+
const result = await store.consolidate({
144+
summarize,
145+
deleteSources: false,
146+
olderThanSeconds: 3600,
147+
});
148+
149+
expect(result.consolidated).toBe(2);
150+
expect(result.created).toHaveLength(1);
151+
expect(result.deleted).toBe(0);
152+
expect(client.call.mock.calls.some((c) => c[0] === 'DEL')).toBe(false);
153+
});
154+
155+
it('returns zeros and does not summarize or write when nothing matches', async () => {
156+
const summarize = vi.fn(async () => 'summary');
157+
const client = consolidatingClient([]);
158+
const store = new MemoryStore({ client, name: 'mem', embedFn: fakeEmbed(8) });
159+
160+
const result = await store.consolidate({ olderThanSeconds: 3600, summarize });
161+
162+
expect(summarize).not.toHaveBeenCalled();
163+
expect(result).toEqual({ consolidated: 0, created: [], deleted: 0 });
164+
expect(client.call.mock.calls.some((c) => c[0] === 'HSET')).toBe(false);
165+
expect(client.call.mock.calls.some((c) => c[0] === 'DEL')).toBe(false);
166+
});
167+
168+
it('throws when given no scope, tags, or selection criteria (prevents whole-store consolidation)', async () => {
169+
const summarize = vi.fn(async () => 'summary');
170+
const store = new MemoryStore({ client: mockClient(), name: 'mem', embedFn: fakeEmbed(8) });
171+
172+
await expect(store.consolidate({ summarize })).rejects.toThrow(/scope|criteria/i);
173+
expect(summarize).not.toHaveBeenCalled();
174+
});
175+
176+
it('defaults summaryImportance to 0.7 and deletes sources by default', async () => {
177+
const summarize = vi.fn(async () => 'summary');
178+
const client = consolidatingClient([itemHit('a', { importance: 0.2, ageSeconds: 100000 })]);
179+
const store = new MemoryStore({ client, name: 'mem', embedFn: fakeEmbed(8) });
180+
181+
const result = await store.consolidate({ summarize, olderThanSeconds: 3600 });
182+
183+
const hset = client.call.mock.calls.find((c) => c[0] === 'HSET');
184+
expect(fieldValue(hset, 'importance')).toBe('0.7');
185+
expect(result.deleted).toBe(1);
186+
});
187+
});

packages/agent-memory/src/index.ts

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -10,6 +10,8 @@ export type {
1010
MemoryItem,
1111
RecallOptions,
1212
MemoryHit,
13+
ConsolidateOptions,
14+
ConsolidateResult,
1315
} from './types';
1416
export { compositeScore, similarityFromDistance } from './compositeScore';
1517
export type { RecallWeights, CompositeScoreParams } from './compositeScore';

packages/agent-memory/src/types.ts

Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -43,3 +43,18 @@ export interface MemoryHit {
4343
similarity: number;
4444
score: number;
4545
}
46+
47+
export interface ConsolidateOptions extends MemoryScope {
48+
olderThanSeconds?: number;
49+
maxImportance?: number;
50+
summarize: (items: MemoryItem[]) => Promise<string>;
51+
deleteSources?: boolean;
52+
summaryImportance?: number;
53+
tags?: string[];
54+
}
55+
56+
export interface ConsolidateResult {
57+
consolidated: number;
58+
created: string[];
59+
deleted: number;
60+
}

0 commit comments

Comments
 (0)