Build an AI Agent Memory System with Daily Consolidation
My AI agent kept forgetting everything.
Every morning, I’d start a conversation and it would ask the same questions. “What project are you working on?” “What’s your preferred coding style?” “Which database are you using?”
I got tired of repeating myself. So I built a memory system that works like human memory: daily logs, nightly consolidation, and long-term knowledge storage.
After three weeks running 24/7 on a Raspberry Pi 4, my agent now remembers our conversations, learns my preferences, and stores important facts permanently.
The Problem: AI Agents Have Goldfish Memory
Most AI agents start fresh every session. The context window fills up, older messages get dropped, and everything learned is forgotten.
I tried the built-in memory options in LangChain:
┌─────────────────────────────────────────────────────────────┐│ Standard LangChain Memory Options ││ ││ ConversationBufferMemory ││ └─ Stores EVERYTHING until context window fills ││ └─ Problem: Gets expensive, hits token limits ││ ││ ConversationBufferWindowMemory ││ └─ Keeps only last N messages ││ └─ Problem: Forgets older important context ││ ││ ConversationSummaryMemory ││ └─ Summarizes old conversations ││ └─ Problem: Summary alone loses details ││ ││ Vector Store Memory ││ └─ Semantic search over embeddings ││ └─ Problem: Needs external service, complex setup ││ │└─────────────────────────────────────────────────────────────┘None of these gave me what I wanted: a system that stores raw interactions, summarizes them periodically, and extracts long-term knowledge.
The Solution: Three-Layer Memory Architecture
I modeled the system after human memory:
┌─────────────────────────────────────────────┐│ User Interactions / Agent Tasks │└──────────────────┬──────────────────────────┘ │ ▼┌─────────────────────────────────────────────┐│ Layer 1: Daily Memory (Raw Logs) ││ - Timestamped interactions ││ - Conversation transcripts ││ - Agent decisions and reasoning ││ Storage: SQLite table: daily_memories │└──────────────────┬──────────────────────────┘ │ │ Daily Consolidation Job │ (Cron at midnight) ▼┌─────────────────────────────────────────────┐│ Layer 2: Consolidated Memory (Summary) ││ - Key insights from daily interactions ││ - Important facts extracted ││ - Compressed representation ││ Storage: SQLite table: consolidated_memories│└──────────────────┬──────────────────────────┘ │ │ Knowledge Extraction ▼┌─────────────────────────────────────────────┐│ Layer 3: Long-term Memory (Knowledge) ││ - Structured facts (entity-relationship) ││ - User preferences and patterns ││ - Cross-session knowledge graph ││ Storage: SQLite tables: facts, entities │└─────────────────────────────────────────────┘Why SQLite? It’s serverless, runs anywhere, and perfect for a Raspberry Pi. No Docker containers, no external services, just a single file I can back up with cp.
Step 1: Design the SQLite Schema
I created three core tables for the three memory layers:
-- Daily memory: Raw interaction logsCREATE TABLE daily_memories ( id INTEGER PRIMARY KEY AUTOINCREMENT, agent_id TEXT NOT NULL, session_id TEXT NOT NULL, timestamp DATETIME DEFAULT CURRENT_TIMESTAMP, interaction_type TEXT, role TEXT, content TEXT NOT NULL, context JSON, importance_score REAL DEFAULT 0.5, consolidated BOOLEAN DEFAULT FALSE, created_at DATETIME DEFAULT CURRENT_TIMESTAMP);
-- Indexes for performanceCREATE INDEX idx_daily_agent ON daily_memories(agent_id);CREATE INDEX idx_daily_timestamp ON daily_memories(timestamp);CREATE INDEX idx_daily_unconsolidated ON daily_memories(consolidated, timestamp);
-- Consolidated memory: Summarized daily insightsCREATE TABLE consolidated_memories ( id INTEGER PRIMARY KEY AUTOINCREMENT, agent_id TEXT NOT NULL, date DATE NOT NULL, summary TEXT NOT NULL, key_facts JSON, entities JSON, topics JSON, created_at DATETIME DEFAULT CURRENT_TIMESTAMP, UNIQUE(agent_id, date));
-- Long-term memory: Structured knowledge baseCREATE TABLE facts ( id INTEGER PRIMARY KEY AUTOINCREMENT, agent_id TEXT NOT NULL, entity_type TEXT, entity_name TEXT, fact_type TEXT, fact_content TEXT NOT NULL, confidence REAL DEFAULT 1.0, source TEXT, created_at DATETIME DEFAULT CURRENT_TIMESTAMP, last_accessed DATETIME, access_count INTEGER DEFAULT 0);The importance_score column lets me prioritize what to consolidate first. The consolidated flag tracks which memories have been processed.
Step 2: Build the Memory System Core
Here’s the Python implementation:
import sqlite3import jsonfrom datetime import datetime, date, timedeltafrom typing import List, Dict, Optional, Anyfrom dataclasses import dataclassfrom pathlib import Pathimport logging
@dataclassclass DailyMemory: """Daily interaction memory""" agent_id: str content: str timestamp: datetime session_id: str interaction_type: str role: str context: Dict[str, Any] importance_score: float = 0.5
@dataclassclass ConsolidatedMemory: """Consolidated summary memory""" agent_id: str date: date summary: str key_facts: List[str] entities: List[str] topics: List[str]
class AgentMemorySystem: """ Three-layer memory system for AI agents with daily consolidation. Designed for Raspberry Pi and resource-constrained environments. """
def __init__(self, db_path: str = "agent_memory.db", agent_id: str = "default"): self.db_path = Path(db_path) self.agent_id = agent_id self.logger = logging.getLogger(f"AgentMemory-{agent_id}") self._init_database()
def _init_database(self): """Initialize SQLite database with schema""" with sqlite3.connect(self.db_path) as conn: conn.executescript(""" CREATE TABLE IF NOT EXISTS daily_memories ( id INTEGER PRIMARY KEY AUTOINCREMENT, agent_id TEXT NOT NULL, session_id TEXT NOT NULL, timestamp DATETIME DEFAULT CURRENT_TIMESTAMP, interaction_type TEXT, role TEXT, content TEXT NOT NULL, context JSON, importance_score REAL DEFAULT 0.5, consolidated BOOLEAN DEFAULT FALSE, created_at DATETIME DEFAULT CURRENT_TIMESTAMP );
CREATE TABLE IF NOT EXISTS consolidated_memories ( id INTEGER PRIMARY KEY AUTOINCREMENT, agent_id TEXT NOT NULL, date DATE NOT NULL, summary TEXT NOT NULL, key_facts JSON, entities JSON, topics JSON, created_at DATETIME DEFAULT CURRENT_TIMESTAMP, UNIQUE(agent_id, date) );
CREATE TABLE IF NOT EXISTS facts ( id INTEGER PRIMARY KEY AUTOINCREMENT, agent_id TEXT NOT NULL, entity_type TEXT, entity_name TEXT, fact_type TEXT, fact_content TEXT NOT NULL, confidence REAL DEFAULT 1.0, source TEXT, created_at DATETIME DEFAULT CURRENT_TIMESTAMP, last_accessed DATETIME, access_count INTEGER DEFAULT 0 );
CREATE INDEX IF NOT EXISTS idx_daily_agent ON daily_memories(agent_id); CREATE INDEX IF NOT EXISTS idx_daily_timestamp ON daily_memories(timestamp); CREATE INDEX IF NOT EXISTS idx_daily_unconsolidated ON daily_memories(consolidated, timestamp); """) self.logger.info(f"Database initialized at {self.db_path}")
def add_daily_memory( self, content: str, session_id: str, interaction_type: str = "conversation", role: str = "assistant", context: Optional[Dict] = None, importance_score: float = 0.5 ) -> int: """Store a new daily memory""" with sqlite3.connect(self.db_path) as conn: cursor = conn.execute(""" INSERT INTO daily_memories (agent_id, session_id, interaction_type, role, content, context, importance_score) VALUES (?, ?, ?, ?, ?, ?, ?) """, ( self.agent_id, session_id, interaction_type, role, content, json.dumps(context or {}), importance_score )) memory_id = cursor.lastrowid conn.commit()
self.logger.debug(f"Added daily memory {memory_id}") return memory_id
def get_daily_memories( self, days: int = 1, unconsolidated_only: bool = False, limit: int = 100 ) -> List[Dict]: """Retrieve recent daily memories""" query = """ SELECT * FROM daily_memories WHERE agent_id = ? AND timestamp > datetime('now', ?) """ params = [self.agent_id, f'-{days} days']
if unconsolidated_only: query += " AND consolidated = FALSE"
query += " ORDER BY timestamp DESC LIMIT ?" params.append(limit)
with sqlite3.connect(self.db_path) as conn: conn.row_factory = sqlite3.Row rows = conn.execute(query, params).fetchall() return [dict(row) for row in rows]
def add_fact( self, entity_type: str, entity_name: str, fact_type: str, fact_content: str, confidence: float = 1.0, source: str = "conversation" ) -> int: """Add a fact to long-term memory""" with sqlite3.connect(self.db_path) as conn: cursor = conn.execute(""" INSERT OR REPLACE INTO facts (agent_id, entity_type, entity_name, fact_type, fact_content, confidence, source) VALUES (?, ?, ?, ?, ?, ?, ?) """, ( self.agent_id, entity_type, entity_name, fact_type, fact_content, confidence, source )) fact_id = cursor.lastrowid conn.commit()
self.logger.debug(f"Added fact: {entity_name} - {fact_content}") return fact_id
def get_relevant_facts( self, query: str, entity_types: Optional[List[str]] = None, limit: int = 10 ) -> List[Dict]: """Retrieve facts relevant to a query""" base_query = """ SELECT * FROM facts WHERE agent_id = ? AND fact_content LIKE ? """ params = [self.agent_id, f'%{query}%']
if entity_types: placeholders = ','.join('?' * len(entity_types)) base_query += f" AND entity_type IN ({placeholders})" params.extend(entity_types)
base_query += " ORDER BY confidence DESC, access_count DESC LIMIT ?" params.append(limit)
with sqlite3.connect(self.db_path) as conn: conn.row_factory = sqlite3.Row rows = conn.execute(base_query, params).fetchall() return [dict(row) for row in rows]
def get_context_for_query( self, query: str, include_daily: bool = True, include_consolidated: bool = True, include_facts: bool = True ) -> str: """Build comprehensive context from all memory layers for a query.""" context_parts = []
if include_facts: facts = self.get_relevant_facts(query, limit=5) if facts: fact_text = "\n".join([ f"- {f['entity_name']}: {f['fact_content']}" for f in facts ]) context_parts.append(f"Known Facts:\n{fact_text}")
if include_consolidated: with sqlite3.connect(self.db_path) as conn: conn.row_factory = sqlite3.Row rows = conn.execute(""" SELECT * FROM consolidated_memories WHERE agent_id = ? ORDER BY date DESC LIMIT 3 """, (self.agent_id,)).fetchall()
if rows: consolidated_text = "\n\n".join([ f"{row['date']}: {row['summary']}" for row in rows ]) context_parts.append(f"Recent History:\n{consolidated_text}")
if include_daily: daily = self.get_daily_memories(days=1, limit=10) if daily: daily_text = "\n".join([ f"[{m['timestamp']}] {m['role']}: {m['content']}" for m in daily ]) context_parts.append(f"Recent Interactions:\n{daily_text}")
return "\n\n".join(context_parts)
def get_memory_stats(self) -> Dict[str, int]: """Get statistics about memory usage""" with sqlite3.connect(self.db_path) as conn: stats = {}
stats['total_daily_memories'] = conn.execute( "SELECT COUNT(*) FROM daily_memories WHERE agent_id = ?", (self.agent_id,) ).fetchone()[0]
stats['unconsolidated_memories'] = conn.execute( "SELECT COUNT(*) FROM daily_memories WHERE agent_id = ? AND consolidated = FALSE", (self.agent_id,) ).fetchone()[0]
stats['total_facts'] = conn.execute( "SELECT COUNT(*) FROM facts WHERE agent_id = ?", (self.agent_id,) ).fetchone()[0]
return statsStep 3: Implement Daily Consolidation
The consolidation process runs at midnight, summarizes the day’s interactions, and extracts facts:
#!/usr/bin/env python3"""Daily memory consolidation cron job.Run at midnight: 0 0 * * * /path/to/consolidation_job.py"""
import sysimport loggingfrom datetime import date, timedeltafrom langchain_openai import ChatOpenAIfrom agent_memory import AgentMemorySystem
logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler('/var/log/agent_memory_consolidation.log'), logging.StreamHandler(sys.stdout) ])
logger = logging.getLogger(__name__)
def consolidate_daily_memories(memory_system, llm_client, target_date=None): """Consolidate daily memories into summarized form.""" if target_date is None: target_date = date.today() - timedelta(days=1)
# Fetch unconsolidated memories memories = memory_system.get_daily_memories( days=1, unconsolidated_only=True )
if not memories: logger.info(f"No memories to consolidate for {target_date}") return None
# Prepare memories for summarization memory_texts = [ f"[{m['timestamp']}] {m['role']}: {m['content']}" for m in memories ]
# Consolidate using LLM summary_prompt = f""" Analyze the following daily interactions and create a consolidated summary.
Daily Interactions: {chr(10).join(memory_texts)}
Extract and return JSON with: 1. summary: A concise paragraph summarizing the day's key events 2. key_facts: Array of important facts learned or confirmed 3. entities: Array of people, projects, or concepts mentioned 4. topics: Array of main topics discussed
Format: JSON only, no markdown. """
response = llm_client.invoke(summary_prompt) consolidation_data = json.loads(response.content)
# Store consolidated memory import sqlite3 import json
with sqlite3.connect(memory_system.db_path) as conn: conn.execute(""" INSERT OR REPLACE INTO consolidated_memories (agent_id, date, summary, key_facts, entities, topics) VALUES (?, ?, ?, ?, ?, ?) """, ( memory_system.agent_id, target_date.isoformat(), consolidation_data['summary'], json.dumps(consolidation_data['key_facts']), json.dumps(consolidation_data['entities']), json.dumps(consolidation_data['topics']) ))
# Mark daily memories as consolidated memory_ids = [m['id'] for m in memories] conn.execute(""" UPDATE daily_memories SET consolidated = TRUE WHERE id IN ({}) """.format(','.join('?' * len(memory_ids))), memory_ids)
conn.commit()
logger.info(f"Consolidated {len(memories)} memories for {target_date}")
return consolidation_data
def main(): """Run daily consolidation for all agents""" # Use a cost-effective model for summarization llm = ChatOpenAI( model="gpt-4o-mini", temperature=0.3 )
# Or use local model for zero cost # from langchain_ollama import ChatOllama # llm = ChatOllama(model="llama3.2:3b", temperature=0.3)
agents = ["main", "researcher", "writer", "coder", "memory_manager"]
for agent_id in agents: try: memory = AgentMemorySystem( db_path=f"/home/openclaw/memories/{agent_id}.db", agent_id=agent_id )
yesterday = date.today() - timedelta(days=1) result = consolidate_daily_memories(memory, llm, yesterday)
if result: logger.info(f"Consolidated {agent_id}: {len(result['key_facts'])} facts")
# Extract facts from consolidated memory for fact in result['key_facts']: memory.add_fact( entity_type="knowledge", entity_name="general", fact_type="learned", fact_content=fact, source="consolidation" )
# Log stats stats = memory.get_memory_stats() logger.info(f"{agent_id} stats: {stats}")
except Exception as e: logger.error(f"Consolidation failed for {agent_id}: {e}") continue
if __name__ == "__main__": main()Step 4: Set Up the Cron Job
I added this to my crontab on the Raspberry Pi:
# Edit crontabcrontab -e
# Add daily consolidation at midnight0 0 * * * /usr/bin/python3 /home/openclaw/scripts/consolidation_job.py
# Add monitoring every 5 minutes*/5 * * * * /usr/bin/python3 /home/openclaw/scripts/monitor_memory.pyThe monitoring script checks for issues:
#!/usr/bin/env python3import psutilimport sqlite3from pathlib import Pathimport jsonfrom datetime import datetime
def check_system_resources(): """Check Raspberry Pi resource usage""" cpu_percent = psutil.cpu_percent(interval=1) memory = psutil.virtual_memory() disk = psutil.disk_usage('/')
return { 'cpu_percent': cpu_percent, 'memory_percent': memory.percent, 'memory_available_gb': memory.available / (1024**3), 'disk_percent': disk.percent, }
def check_consolidation_status(db_path: str): """Check if consolidation is keeping up""" with sqlite3.connect(db_path) as conn: unconsolidated = conn.execute(""" SELECT COUNT(*) FROM daily_memories WHERE consolidated = FALSE AND timestamp > datetime('now', '-1 day') """).fetchone()[0]
return { 'unconsolidated_last_24h': unconsolidated, 'status': 'healthy' if unconsolidated < 100 else 'backlog' }
def main(): report = { 'timestamp': datetime.now().isoformat(), 'system': check_system_resources(), 'agents': {} }
agents = ['main', 'researcher', 'writer', 'coder', 'memory_manager'] base_path = Path('/home/openclaw/memories')
for agent_id in agents: db_path = base_path / f"{agent_id}.db" if db_path.exists(): report['agents'][agent_id] = check_consolidation_status(str(db_path))
print(json.dumps(report, indent=2))
if __name__ == "__main__": main()Step 5: Optimize for Raspberry Pi
The Raspberry Pi has limited resources, so I applied SQLite optimizations:
class OptimizedMemorySystem(AgentMemorySystem): """Memory system optimized for Raspberry Pi resource constraints."""
def __init__(self, db_path: str = "agent_memory.db", agent_id: str = "default"): super().__init__(db_path, agent_id) self._optimize_database()
def _optimize_database(self): """Apply SQLite optimizations for Raspberry Pi""" import sqlite3 with sqlite3.connect(self.db_path) as conn: # Enable WAL mode for better concurrency conn.execute("PRAGMA journal_mode=WAL")
# Increase cache size (negative = KB) conn.execute("PRAGMA cache_size=-64000") # 64MB cache
# Memory-mapped I/O conn.execute("PRAGMA mmap_size=268435456") # 256MB
# Synchronous mode (balance between safety and speed) conn.execute("PRAGMA synchronous=NORMAL")
# Temp store in memory conn.execute("PRAGMA temp_store=MEMORY")
conn.commit()
self.logger.info("Database optimizations applied")
def cleanup_aggressive(self): """More aggressive cleanup for Raspberry Pi.""" import sqlite3 with sqlite3.connect(self.db_path) as conn: # Delete old consolidated daily memories (keep 7 days) conn.execute(""" DELETE FROM daily_memories WHERE agent_id = ? AND consolidated = TRUE AND timestamp < datetime('now', '-7 days') """, (self.agent_id,))
# Vacuum to reclaim space conn.execute("PRAGMA incremental_vacuum(1024)")
conn.commit()Step 6: Integrate with LangChain Agents
To use the memory system with a LangChain agent:
from langchain.agents import AgentExecutor, create_openai_functions_agentfrom langchain_openai import ChatOpenAIfrom langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholderfrom langchain_core.tools import Toolfrom datetime import datetime
class MemoryEnabledAgent: """LangChain agent with integrated memory system."""
def __init__( self, agent_id: str, llm: ChatOpenAI, memory_system: AgentMemorySystem, tools: list = None ): self.agent_id = agent_id self.llm = llm self.memory = memory_system self.session_id = f"session_{datetime.now().strftime('%Y%m%d_%H%M%S')}" self.agent = self._create_agent(tools or [])
def _create_agent(self, tools: list) -> AgentExecutor: """Create LangChain agent with memory tools""" memory_tool = Tool( name="search_memory", description="Search long-term memory for relevant information", func=self._search_memory ) tools.append(memory_tool)
prompt = ChatPromptTemplate.from_messages([ ("system", """You are an AI assistant with persistent memory.
You have access to:1. Long-term memory: Facts and knowledge from past interactions2. Recent context: Consolidated summaries of recent conversations
Use the search_memory tool to retrieve relevant information.
Current context from memory:{memory_context}"""), MessagesPlaceholder(variable_name="chat_history"), ("human", "{input}"), MessagesPlaceholder(variable_name="agent_scratchpad"), ])
agent = create_openai_functions_agent(self.llm, tools, prompt) return AgentExecutor(agent=agent, tools=tools, verbose=True)
def _search_memory(self, query: str) -> str: """Search memory for relevant information""" facts = self.memory.get_relevant_facts(query, limit=5) if facts: return "\n".join([f"- {f['fact_content']}" for f in facts]) return "No relevant memories found."
async def chat(self, user_input: str) -> str: """Chat with agent, automatically storing and retrieving memory.""" memory_context = self.memory.get_context_for_query(user_input)
# Store user input self.memory.add_daily_memory( content=user_input, session_id=self.session_id, interaction_type='conversation', role='user' )
# Get response response = await self.agent.ainvoke({ "input": user_input, "memory_context": memory_context, "chat_history": [] })
# Store assistant response self.memory.add_daily_memory( content=response['output'], session_id=self.session_id, interaction_type='conversation', role='assistant' )
return response['output']Step 7: Quick Start Example
Here’s a complete working example:
from agent_memory import AgentMemorySystemfrom langchain_openai import ChatOpenAIfrom datetime import datetime
def quick_start(): # 1. Initialize memory system memory = AgentMemorySystem( db_path="my_agent.db", agent_id="main" )
# 2. Setup LLM client llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.3)
# 3. Add interactions session_id = f"session_{datetime.now().strftime('%Y%m%d_%H%M%S')}"
memory.add_daily_memory( content="I'm working on a Python project for AI agents", session_id=session_id, role="user" )
memory.add_daily_memory( content="That's great! I can help you with AI agent development.", session_id=session_id, role="assistant" )
# 4. Query memory context = memory.get_context_for_query("Python project") print("Memory context:", context)
# 5. Add a fact memory.add_fact( entity_type="project", entity_name="ai-agent", fact_type="technology", fact_content="Built with Python and LangChain", source="conversation" )
# 6. Check stats stats = memory.get_memory_stats() print(f"Memory stats: {stats}")
if __name__ == "__main__": quick_start()Results After Three Weeks
My Raspberry Pi 4 (8GB) has been running this system for three weeks straight:
- Memory stored: 2,847 daily memories across 5 agents
- Facts extracted: 342 long-term facts
- Database size: 12 MB total
- CPU usage: ~2% average during consolidation
- Memory usage: 340 MB for the database cache
The agent now remembers my coding preferences, project details, and even which coffee shop I work from on Tuesdays.
Key Decisions I Made
SQLite over Vector DBs: I considered Chroma and Pinecone for similarity search. But for an agent that primarily needs structured knowledge retrieval, SQLite is simpler, cheaper, and fast enough.
Cron over Real-time: I tried real-time consolidation after every 50 messages. It worked, but it felt wasteful. Daily consolidation at midnight is more efficient and matches how human memory works.
Separate Databases per Agent: My main agent has 5 subagents. Each gets its own SQLite file. This keeps queries fast and lets me share specific memories between agents when needed.
When to Consider Alternatives
- Over 100,000 memories: Consider PostgreSQL or a vector database
- Need semantic search: Add sentence embeddings with a small local model
- Multi-server deployment: Move to PostgreSQL with proper replication
Quick Checklist
- SQLite database with three-layer schema
- Daily memory storage function
- Consolidation script with LLM summarization
- Cron job running at midnight
- Monitoring script for health checks
- Raspberry Pi optimizations applied
- LangChain integration for agent use
Final Words + More Resources
My intention with this article was to help others share my knowledge and experience. If you want to contact me, you can contact by email: Email me
Here are also the most important links from this article along with some further resources that will help you in this scope:
Oh, and if you found these resources useful, don’t forget to support me by starring the repo on GitHub!
Comments