Build a Self-Organizing Agent Memory System

Build a Self-Organizing Agent Memory System

An agent needs two kinds of memory. A central knowledge store holds the shared, canonical facts an application depends on; but each agent also needs its own local, episodic memory to keep context across a long-running task. This guide builds that second layer — a self-organizing store that turns raw interactions into durable, reusable knowledge instead of accumulating chat logs.

A second, lighter approach. This is the fuller, runnable implementation, built on SQLite with scene consolidation and lexical (full-text) search. For a DSPy plus vector-database variant, better suited to semantic recall over large stores, see How to Build a Custom LLM Memory Layer.

Design

The system separates two concerns: a worker agent answers the user, while a dedicated memory manager extracts, compresses, and organizes what happened. Storage is SQLite, interactions are grouped into scenes, and each scene is periodically consolidated into a stable summary. That separation is the point — the agent never has to reason about how memory is kept.

1. Core runtime

Import the libraries, collect the API key at run time, initialize the model client, and define one helper that standardizes every model call.

import sqlite3
import json
import re
from datetime import datetime
from typing import List, Dict
from getpass import getpass
from openai import OpenAI


OPENAI_API_KEY = getpass("Enter your OpenAI API key: ").strip()
client = OpenAI(api_key=OPENAI_API_KEY)


def llm(prompt, temperature=0.1, max_tokens=500):
   return client.chat.completions.create(
       model="gpt-4o-mini",
       messages=[{"role": "user", "content": prompt}],
       temperature=temperature,
       max_tokens=max_tokens
   ).choices[0].message.content.strip()

2. The memory schema

Three tables persist memory across interactions: mem_cells for atomic units, mem_scenes for higher-level summaries, and an FTS5 virtual table (mem_cells_fts) for symbolic, full-text retrieval.

class MemoryDB:
   def __init__(self):
       self.db = sqlite3.connect(":memory:")
       self.db.row_factory = sqlite3.Row
       self._init_schema()


   def _init_schema(self):
       self.db.execute("""
       CREATE TABLE mem_cells (
           id INTEGER PRIMARY KEY,
           scene TEXT,
           cell_type TEXT,
           salience REAL,
           content TEXT,
           created_at TEXT
       )
       """)


       self.db.execute("""
       CREATE TABLE mem_scenes (
           scene TEXT PRIMARY KEY,
           summary TEXT,
           updated_at TEXT
       )
       """)


       self.db.execute("""
       CREATE VIRTUAL TABLE mem_cells_fts
       USING fts5(content, scene, cell_type)
       """)


   def insert_cell(self, cell):
       self.db.execute(
           "INSERT INTO mem_cells VALUES(NULL,?,?,?,?,?)",
           (
               cell["scene"],
               cell["cell_type"],
               cell["salience"],
               json.dumps(cell["content"]),
               datetime.utcnow().isoformat()
           )
       )
       self.db.execute(
           "INSERT INTO mem_cells_fts VALUES(?,?,?)",
           (
               json.dumps(cell["content"]),
               cell["scene"],
               cell["cell_type"]
           )
       )
       self.db.commit()

3. Retrieval and scene logic

Full-text search over the cells, with the user query sanitized into tokens and a salience-ranked fallback when there are no lexical matches.

   def get_scene(self, scene):
       return self.db.execute(
           "SELECT * FROM mem_scenes WHERE scene=?", (scene,)
       ).fetchone()


   def upsert_scene(self, scene, summary):
       self.db.execute("""
       INSERT INTO mem_scenes VALUES(?,?,?)
       ON CONFLICT(scene) DO UPDATE SET
           summary=excluded.summary,
           updated_at=excluded.updated_at
       """, (scene, summary, datetime.utcnow().isoformat()))
       self.db.commit()


   def retrieve_scene_context(self, query, limit=6):
       tokens = re.findall(r"[a-zA-Z0-9]+", query)
       if not tokens:
           return []


       fts_query = " OR ".join(tokens)


       rows = self.db.execute("""
       SELECT scene, content FROM mem_cells_fts
       WHERE mem_cells_fts MATCH ?
       LIMIT ?
       """, (fts_query, limit)).fetchall()


       if not rows:
           rows = self.db.execute("""
           SELECT scene, content FROM mem_cells
           ORDER BY salience DESC
           LIMIT ?
           """, (limit,)).fetchall()


       return rows


   def retrieve_scene_summary(self, scene):
       row = self.get_scene(scene)
       return row["summary"] if row else ""

4. The memory manager

This component turns interactions into structured cells, stores them, and periodically consolidates each scene into a stable, reusable summary.

class MemoryManager:
   def __init__(self, db: MemoryDB):
       self.db = db


   def extract_cells(self, user, assistant) -> List[Dict]:
       prompt = f"""
Convert this interaction into structured memory cells.


Return JSON array with objects containing:
- scene
- cell_type (fact, plan, preference, decision, task, risk)
- salience (0-1)
- content (compressed, factual)


User: {user}
Assistant: {assistant}
"""
       raw = llm(prompt)
       raw = re.sub(r"```json|```", "", raw)


       try:
           cells = json.loads(raw)
           return cells if isinstance(cells, list) else []
       except Exception:
           return []


   def consolidate_scene(self, scene):
       rows = self.db.db.execute(
           "SELECT content FROM mem_cells WHERE scene=? ORDER BY salience DESC",
           (scene,)
       ).fetchall()


       if not rows:
           return


       cells = [json.loads(r["content"]) for r in rows]


       prompt = f"""
Summarize this memory scene in under 100 words.
Keep it stable and reusable for future reasoning.


Cells:
{cells}
"""
       summary = llm(prompt, temperature=0.05)
       self.db.upsert_scene(scene, summary)


   def update(self, user, assistant):
       cells = self.extract_cells(user, assistant)


       for cell in cells:
           self.db.insert_cell(cell)


       for scene in set(c["scene"] for c in cells):
           self.consolidate_scene(scene)

5. The worker agent

The agent reasons while staying memory-aware: it recalls relevant scenes, assembles their summaries into context, answers, and then hands the exchange back to the memory manager.

class WorkerAgent:
   def __init__(self, db: MemoryDB, mem_manager: MemoryManager):
       self.db = db
       self.mem_manager = mem_manager


   def answer(self, user_input):
       recalled = self.db.retrieve_scene_context(user_input)
       scenes = set(r["scene"] for r in recalled)


       summaries = "\n".join(
           f"[{scene}]\n{self.db.retrieve_scene_summary(scene)}"
           for scene in scenes
       )


       prompt = f"""
You are an intelligent agent with long-term memory.


Relevant memory:
{summaries}


User: {user_input}
"""
       assistant_reply = llm(prompt)
       self.mem_manager.update(user_input, assistant_reply)
       return assistant_reply


# --- Execution ---

db = MemoryDB()
memory_manager = MemoryManager(db)
agent = WorkerAgent(db, memory_manager)


print(agent.answer("We are building an agent that remembers projects long term."))
print(agent.answer("It should organize conversations into topics automatically."))
print(agent.answer("This memory system should support future reasoning."))


for row in db.db.execute("SELECT * FROM mem_scenes"):
   print(dict(row))

Summary

The agent curates its own memory, turning past interactions into stable, reusable knowledge rather than ephemeral logs. Consolidation and selective recall let that memory evolve, which supports more consistent, grounded reasoning across sessions — without the context bloat of replaying raw history.

See also

This entry was posted in . Bookmark the permalink.
Adam Bernard