Notebooks · Reliability and control · 56

Episodic Memory Consolidation (v1.0.8)

Name: SHIPIT Agent
Author: SHIPIT

Turn long conversations into durable facts. Apply forgetting curves so old facts decay. Promote frequently-retrieved facts to a 'core memory' set always visible in the system prompt.

Open in Colab View on GitHub 5 code cells · 0 with saved output

Its imports and arguments are checked against shipit-agent 3.0. The outputs were saved when it was written, for an earlier release.

In [ ]

python

from pathlib import Path
import sys
ROOT = Path.cwd().resolve().parent if Path.cwd().name == 'notebooks' else Path.cwd().resolve()
if str(ROOT) not in sys.path:
    sys.path.insert(0, str(ROOT))

import json
import time
from shipit_agent.memory import (
    AgentMemory, MemoryConsolidator, SemanticMemory, InMemoryVectorStore,
)

1. Distill a conversation into facts

We use a tiny scripted LLM here so the notebook runs offline. Swap it for a real cheap model (build_llm_from_env('haiku')) and the rest is identical.

In [ ]

python

class CannedLLM:
    def __init__(self, response): self.response = response
    def complete(self, *, messages, **_):
        return self.response

fake_response = json.dumps({'facts': [
    {'text': 'User prefers concise answers, no preamble', 'category': 'preference', 'confidence': 0.9},
    {'text': 'Auth service uses Argon2 with a 12-byte salt', 'category': 'project', 'confidence': 0.95},
    {'text': 'Q3 release deadline is Oct 14', 'category': 'goal', 'confidence': 0.85},
    {'text': 'Alice owns the deploy pipeline', 'category': 'person', 'confidence': 0.8},
]})

memory = AgentMemory(knowledge=SemanticMemory(vector_store=InMemoryVectorStore()))
consolidator = MemoryConsolidator(llm=CannedLLM(fake_response), min_messages=2)

conversation = [
    {'role': 'user', 'content': 'Keep your answers concise — no preamble.'},
    {'role': 'assistant', 'content': 'Got it.'},
    {'role': 'user', 'content': 'How does our auth work?'},
    {'role': 'assistant', 'content': 'Argon2 with a 12-byte salt, then signed JWT.'},
    {'role': 'user', 'content': 'When is the Q3 release?'},
    {'role': 'assistant', 'content': 'Oct 14. Alice owns the deploy pipeline.'},
]

result = consolidator.consolidate(memory=memory, recent_messages=conversation)
for f in result.facts:
    print(f'[{f.category}] {f.text}  (conf {f.confidence:.2f})')

2. Apply decay (forgetting curve)

Pure local arithmetic — no LLM call. Run periodically (cron, or after each session).

In [ ]

python

# Simulate: pretend the first fact is already 30 days old
memory.knowledge.vector_store._entries[0]['metadata']['timestamp'] = time.time() - 30 * 86400

for entry in memory.knowledge.vector_store._entries:
    md = entry['metadata']
    age_days = (time.time() - md['timestamp']) / 86400
    print(f'before decay: strength={md["strength"]:.3f}  age={age_days:.1f}d  text={entry["text"][:50]}')

pruned = consolidator.decay(memory.knowledge, half_life_days=14, prune=False)
print(f'\npruned {pruned} facts (with prune=False, just lowering strengths)\n')

for entry in memory.knowledge.vector_store._entries:
    md = entry['metadata']
    age_days = (time.time() - md['timestamp']) / 86400
    print(f'after  decay: strength={md["strength"]:.3f}  age={age_days:.1f}d  text={entry["text"][:50]}')

3. Bump retrievals when facts are useful

When a search returns useful facts, tell the consolidator. Frequently-retrieved facts get promoted to core memory below.

In [ ]

python

consolidator.record_retrieval(memory.knowledge, [
    'User prefers concise answers, no preamble',
    'Auth service uses Argon2 with a 12-byte salt',
])
consolidator.record_retrieval(memory.knowledge, [
    'User prefers concise answers, no preamble',
])

for e in memory.knowledge.vector_store._entries:
    md = e['metadata']
    print(f'retrievals={md["retrievals"]}  text={e["text"][:60]}')

4. Pull the top-K core facts for the next system prompt

In [ ]

python

core = consolidator.core_memory(memory.knowledge, top_k=3)
for c in core:
    print(f'★ {c}')

# Inject into your next agent's system prompt
system_addition = 'Known facts about this user/project:\n' + '\n'.join(f'- {c}' for c in core)
print()
print('--- system prompt addition ---')
print(system_addition)

5. Lifecycle in your real agent

# Once per turn
core = consolidator.core_memory(memory.knowledge, top_k=5)
agent = Agent(llm=opus_llm, prompt=BASE + format_core(core))
result = agent.run(user_message)

# After every 10 turns
if turn % 10 == 0:
    consolidator.consolidate(memory=memory, recent_messages=memory.get_conversation_messages())

# Once a day (cron)
consolidator.decay(memory.knowledge, half_life_days=14)

Cost on Haiku: ~$0.0015 per consolidation. With 10-turn cadence, that's $0.00015 per turn overhead. The agent now remembers your project, your preferences, and your team across every restart.

Episodic Memory Consolidation (v1.0.8)

Turn long conversations into durable facts. Apply forgetting curves so old facts decay. Promote frequently-retrieved facts to a 'core memory' set always visible in the system prompt.

Its imports and arguments are checked against shipit-agent 3.0. The outputs were saved when it was written, for an earlier release.

In [ ]

python

from pathlib import Path
import sys
ROOT = Path.cwd().resolve().parent if Path.cwd().name == 'notebooks' else Path.cwd().resolve()
if str(ROOT) not in sys.path:
    sys.path.insert(0, str(ROOT))

import json
import time
from shipit_agent.memory import (
    AgentMemory, MemoryConsolidator, SemanticMemory, InMemoryVectorStore,
)

1. Distill a conversation into facts

We use a tiny scripted LLM here so the notebook runs offline. Swap it for a real cheap model (build_llm_from_env('haiku')) and the rest is identical.

In [ ]

python

class CannedLLM:
    def __init__(self, response): self.response = response
    def complete(self, *, messages, **_):
        return self.response

fake_response = json.dumps({'facts': [
    {'text': 'User prefers concise answers, no preamble', 'category': 'preference', 'confidence': 0.9},
    {'text': 'Auth service uses Argon2 with a 12-byte salt', 'category': 'project', 'confidence': 0.95},
    {'text': 'Q3 release deadline is Oct 14', 'category': 'goal', 'confidence': 0.85},
    {'text': 'Alice owns the deploy pipeline', 'category': 'person', 'confidence': 0.8},
]})

memory = AgentMemory(knowledge=SemanticMemory(vector_store=InMemoryVectorStore()))
consolidator = MemoryConsolidator(llm=CannedLLM(fake_response), min_messages=2)

conversation = [
    {'role': 'user', 'content': 'Keep your answers concise — no preamble.'},
    {'role': 'assistant', 'content': 'Got it.'},
    {'role': 'user', 'content': 'How does our auth work?'},
    {'role': 'assistant', 'content': 'Argon2 with a 12-byte salt, then signed JWT.'},
    {'role': 'user', 'content': 'When is the Q3 release?'},
    {'role': 'assistant', 'content': 'Oct 14. Alice owns the deploy pipeline.'},
]

result = consolidator.consolidate(memory=memory, recent_messages=conversation)
for f in result.facts:
    print(f'[{f.category}] {f.text}  (conf {f.confidence:.2f})')

2. Apply decay (forgetting curve)

Pure local arithmetic — no LLM call. Run periodically (cron, or after each session).

In [ ]

python

# Simulate: pretend the first fact is already 30 days old
memory.knowledge.vector_store._entries[0]['metadata']['timestamp'] = time.time() - 30 * 86400

for entry in memory.knowledge.vector_store._entries:
    md = entry['metadata']
    age_days = (time.time() - md['timestamp']) / 86400
    print(f'before decay: strength={md["strength"]:.3f}  age={age_days:.1f}d  text={entry["text"][:50]}')

pruned = consolidator.decay(memory.knowledge, half_life_days=14, prune=False)
print(f'\npruned {pruned} facts (with prune=False, just lowering strengths)\n')

for entry in memory.knowledge.vector_store._entries:
    md = entry['metadata']
    age_days = (time.time() - md['timestamp']) / 86400
    print(f'after  decay: strength={md["strength"]:.3f}  age={age_days:.1f}d  text={entry["text"][:50]}')

3. Bump retrievals when facts are useful

When a search returns useful facts, tell the consolidator. Frequently-retrieved facts get promoted to core memory below.

In [ ]

python

consolidator.record_retrieval(memory.knowledge, [
    'User prefers concise answers, no preamble',
    'Auth service uses Argon2 with a 12-byte salt',
])
consolidator.record_retrieval(memory.knowledge, [
    'User prefers concise answers, no preamble',
])

for e in memory.knowledge.vector_store._entries:
    md = e['metadata']
    print(f'retrievals={md["retrievals"]}  text={e["text"][:60]}')

4. Pull the top-K core facts for the next system prompt

In [ ]

python

core = consolidator.core_memory(memory.knowledge, top_k=3)
for c in core:
    print(f'★ {c}')

# Inject into your next agent's system prompt
system_addition = 'Known facts about this user/project:\n' + '\n'.join(f'- {c}' for c in core)
print()
print('--- system prompt addition ---')
print(system_addition)

5. Lifecycle in your real agent

# Once per turn
core = consolidator.core_memory(memory.knowledge, top_k=5)
agent = Agent(llm=opus_llm, prompt=BASE + format_core(core))
result = agent.run(user_message)

# After every 10 turns
if turn % 10 == 0:
    consolidator.consolidate(memory=memory, recent_messages=memory.get_conversation_messages())

# Once a day (cron)
consolidator.decay(memory.knowledge, half_life_days=14)

Cost on Haiku: ~$0.0015 per consolidation. With 10-turn cadence, that's $0.00015 per turn overhead. The agent now remembers your project, your preferences, and your team across every restart.