Structured External Memory
Not all agent memory fits neatly into conversation buffers or semantic vector stores. Some information is inherently structured: user preferences stored as key-value pairs, task state as a state machine, knowledge graphs of entity relationships, or episodic memories with timestamps and metadata. Structured external memory gives agents access to typed, queryable data stores that complement unstructured vector memory.
Types of Structured Memory
Entity Memory: Tracks facts about specific entities (users, products, concepts):
User: Alice
- Preferred language: Python
- Timezone: PST
- Active project: Kafka pipeline
- Last interaction: 2025-01-15
Episodic Memory: Timestamped records of past interactions:
2025-01-14 14:32: Alice asked about Kafka consumer groups
2025-01-15 09:15: Helped Alice debug partition rebalancing
2025-01-15 15:00: Alice successfully fixed the consumer issue
State Memory: Current status of ongoing tasks:
Task: Deploy new API version
- Status: In progress
- Step: 3/7 (Running integration tests)
- Started: 2025-01-15 10:00
- Last updated: 2025-01-15 11:30
Entity Memory with SQLite
import sqlite3
import json
from datetime import datetime
from typing import Any, Optional
class EntityMemoryStore:
"""
SQLite-backed entity memory for storing structured facts about entities.
Stores typed key-value facts with timestamps, enabling the agent to track
evolving information about users, projects, and domain entities.
"""
def __init__(self, db_path: str = "agent_memory.db"):
self.conn = sqlite3.connect(db_path, check_same_thread=False)
self._init_schema()
def _init_schema(self) -> None:
self.conn.execute("""
CREATE TABLE IF NOT EXISTS entity_facts (
id INTEGER PRIMARY KEY AUTOINCREMENT,
entity_type TEXT NOT NULL,
entity_id TEXT NOT NULL,
fact_key TEXT NOT NULL,
fact_value TEXT NOT NULL,
confidence REAL DEFAULT 1.0,
source TEXT,
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL,
UNIQUE(entity_type, entity_id, fact_key)
)
""")
self.conn.execute("""
CREATE INDEX IF NOT EXISTS idx_entity
ON entity_facts(entity_type, entity_id)
""")
self.conn.commit()
def upsert_fact(
self,
entity_type: str,
entity_id: str,
fact_key: str,
fact_value: Any,
confidence: float = 1.0,
source: str = "conversation",
) -> None:
"""Store or update a fact about an entity."""
now = datetime.utcnow().isoformat()
value_json = json.dumps(fact_value)
self.conn.execute("""
INSERT INTO entity_facts
(entity_type, entity_id, fact_key, fact_value, confidence, source, created_at, updated_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(entity_type, entity_id, fact_key) DO UPDATE SET
fact_value = excluded.fact_value,
confidence = excluded.confidence,
source = excluded.source,
updated_at = excluded.updated_at
""", (entity_type, entity_id, fact_key, value_json, confidence, source, now, now))
self.conn.commit()
def get_entity(self, entity_type: str, entity_id: str) -> dict[str, Any]:
"""Get all facts about an entity as a dictionary."""
cursor = self.conn.execute("""
SELECT fact_key, fact_value, confidence, updated_at
FROM entity_facts
WHERE entity_type = ? AND entity_id = ?
ORDER BY updated_at DESC
""", (entity_type, entity_id))
return {
row[0]: {
"value": json.loads(row[1]),
"confidence": row[2],
"updated_at": row[3],
}
for row in cursor.fetchall()
}
def format_for_context(self, entity_type: str, entity_id: str) -> str:
"""Format entity facts as a string for injection into prompts."""
facts = self.get_entity(entity_type, entity_id)
if not facts:
return f"No stored information about {entity_type} '{entity_id}'."
lines = [f"Known facts about {entity_type} '{entity_id}':"]
for key, data in facts.items():
value = data["value"]
lines.append(f" - {key}: {value}")
return "\n".join(lines)
# Usage with an agent
memory_store = EntityMemoryStore()
# Store facts extracted from conversation
memory_store.upsert_fact("user", "alice", "preferred_language", "Python")
memory_store.upsert_fact("user", "alice", "timezone", "America/Los_Angeles")
memory_store.upsert_fact("user", "alice", "active_project", "Kafka data pipeline")
memory_store.upsert_fact("user", "alice", "expertise_level", "senior")
# Retrieve and inject into prompt
context = memory_store.format_for_context("user", "alice")
print(context)
Episodic Memory with Timestamps
class EpisodicMemoryStore:
"""Stores time-stamped episodes of agent interactions."""
def __init__(self, db_path: str = "agent_memory.db"):
self.conn = sqlite3.connect(db_path, check_same_thread=False)
self._init_schema()
def _init_schema(self) -> None:
self.conn.execute("""
CREATE TABLE IF NOT EXISTS episodes (
id INTEGER PRIMARY KEY AUTOINCREMENT,
session_id TEXT,
user_id TEXT,
summary TEXT NOT NULL,
outcome TEXT,
tags TEXT, -- JSON array of tags
created_at TEXT NOT NULL
)
""")
self.conn.commit()
def record_episode(
self,
summary: str,
user_id: str = None,
session_id: str = None,
outcome: str = None,
tags: list[str] = None,
) -> int:
cursor = self.conn.execute("""
INSERT INTO episodes (session_id, user_id, summary, outcome, tags, created_at)
VALUES (?, ?, ?, ?, ?, ?)
""", (session_id, user_id, summary, outcome, json.dumps(tags or []), datetime.utcnow().isoformat()))
self.conn.commit()
return cursor.lastrowid
def get_recent_episodes(self, user_id: str, limit: int = 5) -> list[dict]:
cursor = self.conn.execute("""
SELECT summary, outcome, tags, created_at
FROM episodes
WHERE user_id = ?
ORDER BY created_at DESC
LIMIT ?
""", (user_id, limit))
return [
{"summary": row[0], "outcome": row[1], "tags": json.loads(row[2]), "when": row[3]}
for row in cursor.fetchall()
]
Integrating Structured Memory with Tool Use
from langchain_core.tools import tool
entity_store = EntityMemoryStore()
@tool
def update_user_preference(user_id: str, preference_key: str, preference_value: str) -> str:
"""Update a stored preference or fact about the current user.
Use when the user explicitly states a preference, corrects stored information,
or shares new facts about themselves that should be remembered.
Args:
user_id: The user's unique identifier
preference_key: The preference name (e.g., 'communication_style', 'expertise_level')
preference_value: The preference value
"""
entity_store.upsert_fact("user", user_id, preference_key, preference_value)
return f"Updated: {preference_key} = {preference_value} for user {user_id}"
@tool
def get_user_context(user_id: str) -> str:
"""Retrieve all stored information about a user for context.
Use at the start of a conversation to load relevant user context.
"""
return entity_store.format_for_context("user", user_id)
Structured external memory complements vector search: use structured memory for typed facts you know you'll query directly, and vector search for semantic retrieval of unstructured information. Together they give agents the full range of human-like memory capabilities.