Understanding AI Agent Memory Persistence
When you build an AI agent—whether it's a chatbot, an autonomous task runner, or a reasoning system powered by large language models—one of the first architectural decisions you face is how to handle memory persistence. Agents need to remember past interactions, context, decisions, and facts across sessions. Without persistence, every restart wipes the agent's mind clean, forcing it to start from scratch like a goldfish swimming circles in a bowl.
What is Agent Memory Persistence?
Agent memory persistence is the mechanism by which an AI agent stores and retrieves its conversational context, reasoning chains, tool call results, user preferences, and long-term knowledge between invocations. This isn't just about caching—it's about giving the agent a continuous identity and the ability to learn over time. Memory can take several forms:
- Short-term / working memory: The current conversation window, typically stored as a list of messages (system prompt, user turns, assistant responses, tool calls)
- Long-term episodic memory: Summaries or embeddings of past conversations, allowing retrieval of relevant historical context
- Semantic / factual memory: Key-value facts the agent has learned, user preferences, or domain knowledge
- Procedural memory: Learned workflows, decision trees, or patterns the agent has discovered
Why Memory Persistence Matters
Without persistence, every agent invocation is stateless and isolated. This means:
- No learning: The agent cannot improve from past mistakes or remember user feedback
- Repetitive interactions: Users must re-explain context every session, leading to poor UX
- Broken continuity: Multi-step tasks that span sessions (e.g., "research this topic over the next week and compile findings") become impossible
- Cost inefficiency: LLM tokens are wasted re-establishing context that could have been stored
- Limited autonomy: Truly autonomous agents need persistent memory to plan, reflect, and execute over long time horizons
Choosing the right persistence backend is a critical engineering decision that impacts latency, scalability, cost, and the sophistication of memory operations you can support.
Comparing the Three Storage Backends
Let's examine each option in detail: plain files, Redis, and PostgreSQL. Each occupies a distinct point on the spectrum of simplicity versus power.
File-Based Persistence: The Simplest Starting Point
File-based storage uses the local filesystem—JSON files, SQLite databases, or even plain text—to persist agent memory. It's the quickest to implement and requires zero infrastructure beyond the host machine's disk.
Strengths:
- Zero dependencies—no external services to manage
- Perfect for local development, single-user agents, or CLI tools
- Full serializability of complex Python objects via pickle or JSON
- SQLite offers surprising query power with no server process
- Easy to inspect and debug—just open the file
Weaknesses:
- No built-in concurrency safety—multiple processes writing to the same file will corrupt data
- No horizontal scaling—tied to a single machine
- Slower for high-frequency read/write patterns compared to in-memory stores
- No native pub/sub or real-time notifications
- Manual TTL/expiry management unless using SQLite triggers
Redis: Speed and Real-Time Patterns
Redis is an in-memory data structure store that persists to disk via snapshots (RDB) or an append-only log (AOF). It excels at low-latency operations and provides rich data structures: lists, hashes, sorted sets, streams, and more.
Strengths:
- Sub-millisecond latency for most operations
- Built-in TTL/expiry on any key—perfect for sliding conversation windows
- Pub/sub channels enable real-time agent communication across processes
- Atomic operations (LPUSH, RPOP, INCR) prevent race conditions
- Sorted sets enable time-based retrieval and priority queuing
- Redis Stack adds vector similarity search (RediSearch) for embedding-based memory
- Clustering and sentinel mode for high availability
Weaknesses:
- Data must fit in RAM, limiting total memory capacity by hardware cost
- Persistence is tunable but not ACID by default—you may lose recent writes on crash
- Query capabilities are limited compared to SQL (no joins, no aggregations)
- Requires managing another infrastructure component
PostgreSQL: Relational Power and ACID Guarantees
PostgreSQL is a battle-tested relational database with extensions that make it uniquely suited for AI agent memory: full-text search, trigram indexing, and the pgvector extension for vector similarity search.
Strengths:
- ACID transactions—never lose data, even on crash
- Rich SQL queries enable complex retrieval: filter by user, date range, metadata, and vector similarity simultaneously
pgvectorextension for embedding storage and ANN (approximate nearest neighbor) search- Massive storage capacity—terabytes of memory on cheap disk
- Proven tooling: backups, replication, migration, monitoring
- Role-based access control for multi-tenant agent systems
- Triggers and functions allow automated memory summarization pipelines
Weaknesses:
- Higher latency per operation compared to Redis (typically 1-5ms vs 0.1ms)
- Requires schema design upfront—less flexible for rapid prototyping
- Connection pooling and transaction management add complexity
- Heavier operational footprint than files or even Redis
Practical Implementation Examples
Below are complete, runnable examples for each backend. Each implements the same core interface for an AI agent's conversation memory: save_turn, get_history, and clear. The examples use Python and common libraries.
File-Based Memory with JSON
This implementation stores each conversation as a JSON file on disk. It's suitable for a single-user CLI agent or local development. The code handles file locking for basic concurrency safety using fcntl on Unix systems.
import json
import os
import fcntl
import time
from pathlib import Path
from typing import List, Dict, Optional
class JSONFileMemory:
"""
Persists agent conversation memory as JSON files on disk.
Each conversation session gets its own JSON file.
"""
def __init__(self, base_path: str = "./agent_memory", user_id: str = "default"):
self.base_path = Path(base_path)
self.base_path.mkdir(parents=True, exist_ok=True)
self.user_id = user_id
self.file_path = self.base_path / f"{user_id}_conversation.json"
def _read_file(self) -> List[Dict]:
"""Read the conversation file with a shared lock."""
if not self.file_path.exists():
return []
with open(self.file_path, 'r') as f:
fcntl.flock(f.fileno(), fcntl.LOCK_SH) # Shared lock for reading
try:
data = json.load(f)
finally:
fcntl.flock(f.fileno(), fcntl.LOCK_UN)
return data if isinstance(data, list) else []
def _write_file(self, data: List[Dict]) -> None:
"""Write the conversation file with an exclusive lock."""
# Use a temporary file for atomic writes
tmp_path = self.file_path.with_suffix('.tmp')
with open(tmp_path, 'w') as f:
fcntl.flock(f.fileno(), fcntl.LOCK_EX)
try:
json.dump(data, f, indent=2, default=str)
f.flush()
os.fsync(f.fileno())
finally:
fcntl.flock(f.fileno(), fcntl.LOCK_UN)
os.replace(tmp_path, self.file_path) # Atomic rename
def save_turn(self, role: str, content: str, metadata: Optional[Dict] = None) -> None:
"""Save a single conversation turn (user message, assistant response, or tool call)."""
history = self._read_file()
turn = {
"role": role,
"content": content,
"timestamp": time.time(),
"metadata": metadata or {}
}
history.append(turn)
self._write_file(history)
def get_history(self, limit: int = 50, since_timestamp: Optional[float] = None) -> List[Dict]:
"""Retrieve conversation history, optionally filtered by time."""
history = self._read_file()
if since_timestamp is not None:
history = [t for t in history if t.get("timestamp", 0) >= since_timestamp]
return history[-limit:]
def get_last_n_messages(self, n: int = 10) -> List[Dict]:
"""Convenience method to get the most recent messages."""
return self.get_history(limit=n)
def clear(self) -> None:
"""Clear all memory for this user."""
self._write_file([])
if self.file_path.exists():
self.file_path.unlink(missing_ok=True)
def get_token_count_estimate(self) -> int:
"""Rough estimate of total tokens in memory (4 chars ≈ 1 token)."""
history = self._read_file()
total_chars = sum(len(turn.get("content", "")) for turn in history)
return total_chars // 4
# Usage example
memory = JSONFileMemory(user_id="agent_42")
memory.save_turn("user", "What's the weather in Tokyo?")
memory.save_turn("assistant", "Let me check the weather API for Tokyo...",
metadata={"tool_calls": ["get_weather"]})
memory.save_turn("tool", "Temperature: 22°C, Humidity: 65%",
metadata={"tool_name": "get_weather"})
memory.save_turn("assistant", "The weather in Tokyo is 22°C with 65% humidity.")
recent = memory.get_last_n_messages(5)
for turn in recent:
print(f"[{turn['role']}] {turn['content'][:80]}...")File-Based Memory with SQLite
SQLite gives you a lightweight relational database in a single file. This approach supports multiple users, metadata queries, and is more robust under concurrent access than raw JSON files. SQLite's WAL (Write-Ahead Logging) mode allows concurrent readers while a writer is active.
import sqlite3
import time
import json
from typing import List, Dict, Optional
class SQLiteMemory:
"""
Agent memory backed by SQLite. Supports multiple users,
metadata queries, and concurrent access via WAL mode.
"""
def __init__(self, db_path: str = "./agent_memory/agent_memory.db"):
self.db_path = db_path
self.conn = sqlite3.connect(db_path, check_same_thread=False)
self.conn.execute("PRAGMA journal_mode=WAL;") # Better concurrency
self.conn.execute("PRAGMA synchronous=NORMAL;") # Faster writes
self._create_tables()
def _create_tables(self):
self.conn.execute("""
CREATE TABLE IF NOT EXISTS conversations (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id TEXT NOT NULL,
session_id TEXT NOT NULL,
role TEXT NOT NULL,
content TEXT NOT NULL,
metadata_json TEXT DEFAULT '{}',
created_at REAL NOT NULL,
INDEX idx_user_session (user_id, session_id, created_at)
)
""")
self.conn.execute("""
CREATE TABLE IF NOT EXISTS agent_facts (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id TEXT NOT NULL,
fact_key TEXT NOT NULL,
fact_value TEXT NOT NULL,
created_at REAL NOT NULL,
updated_at REAL NOT NULL,
UNIQUE(user_id, fact_key)
)
""")
self.conn.commit()
def save_turn(self, user_id: str, session_id: str, role: str,
content: str, metadata: Optional[Dict] = None) -> int:
"""Save a conversation turn and return its ID."""
cursor = self.conn.execute(
"""INSERT INTO conversations (user_id, session_id, role, content, metadata_json, created_at)
VALUES (?, ?, ?, ?, ?, ?)""",
(user_id, session_id, role, content,
json.dumps(metadata or {}), time.time())
)
self.conn.commit()
return cursor.lastrowid
def get_history(self, user_id: str, session_id: Optional[str] = None,
limit: int = 50, since_timestamp: Optional[float] = None) -> List[Dict]:
"""Retrieve conversation history with optional filters."""
query = "SELECT id, role, content, metadata_json, created_at FROM conversations WHERE user_id = ?"
params = [user_id]
if session_id is not None:
query += " AND session_id = ?"
params.append(session_id)
if since_timestamp is not None:
query += " AND created_at >= ?"
params.append(since_timestamp)
query += " ORDER BY created_at DESC LIMIT ?"
params.append(limit)
cursor = self.conn.execute(query, params)
rows = cursor.fetchall()
# Return in chronological order
results = []
for row in reversed(rows):
results.append({
"id": row[0],
"role": row[1],
"content": row[2],
"metadata": json.loads(row[3]),
"timestamp": row[4]
})
return results
def save_fact(self, user_id: str, key: str, value: str) -> None:
"""Persist a key-value fact about the user (upsert)."""
now = time.time()
self.conn.execute(
"""INSERT INTO agent_facts (user_id, fact_key, fact_value, created_at, updated_at)
VALUES (?, ?, ?, ?, ?)
ON CONFLICT(user_id, fact_key) DO UPDATE SET
fact_value = excluded.fact_value,
updated_at = excluded.updated_at""",
(user_id, key, value, now, now)
)
self.conn.commit()
def get_fact(self, user_id: str, key: str) -> Optional[str]:
"""Retrieve a specific fact."""
cursor = self.conn.execute(
"SELECT fact_value FROM agent_facts WHERE user_id = ? AND fact_key = ?",
(user_id, key)
)
row = cursor.fetchone()
return row[0] if row else None
def get_all_facts(self, user_id: str) -> Dict[str, str]:
"""Get all facts for a user."""
cursor = self.conn.execute(
"SELECT fact_key, fact_value FROM agent_facts WHERE user_id = ?",
(user_id,)
)
return {row[0]: row[1] for row in cursor.fetchall()}
def prune_old_conversations(self, max_age_days: int = 30) -> int:
"""Delete conversations older than specified days. Returns count of deleted rows."""
cutoff = time.time() - (max_age_days * 86400)
cursor = self.conn.execute(
"DELETE FROM conversations WHERE created_at < ?", (cutoff,)
)
self.conn.commit()
return cursor.rowcount
def close(self):
self.conn.close()
# Usage example
memory = SQLiteMemory()
memory.save_turn("user_42", "session_abc", "user", "Remember I love Python and sushi.")
memory.save_turn("user_42", "session_abc", "assistant",
"Got it! I'll remember that you love Python and sushi.",
metadata={"intent": "fact_storage"})
memory.save_fact("user_42", "favorite_language", "Python")
memory.save_fact("user_42", "favorite_food", "sushi")
history = memory.get_history("user_42", "session_abc", limit=10)
facts = memory.get_all_facts("user_42")
print(f"User facts: {facts}")
print(f"Last message: {history[-1]['content']}")Redis-Based Memory Persistence
Redis is ideal when you need extremely fast reads/writes, automatic TTL-based expiration, and real-time pub/sub. The example below uses Redis lists for conversation history and Redis hashes for key-value facts. It also demonstrates sorted sets for time-based retrieval.
import redis
import json
import time
from typing import List, Dict, Optional
from datetime import timedelta
class RedisMemory:
"""
Agent memory backed by Redis. Features TTL-based auto-expiry,
atomic list operations, and sorted set indexing.
"""
def __init__(self, host: str = "localhost", port: int = 6379,
db: int = 0, password: Optional[str] = None,
default_ttl_days: int = 7):
self.redis = redis.Redis(
host=host, port=port, db=db, password=password,
decode_responses=True # Automatically decode to strings
)
self.default_ttl = timedelta(days=default_ttl_days)
def _conversation_key(self, user_id: str, session_id: str) -> str:
return f"agent:conv:{user_id}:{session_id}"
def _facts_key(self, user_id: str) -> str:
return f"agent:facts:{user_id}"
def _timeline_key(self, user_id: str) -> str:
return f"agent:timeline:{user_id}"
def save_turn(self, user_id: str, session_id: str, role: str,
content: str, metadata: Optional[Dict] = None,
ttl: Optional[timedelta] = None) -> None:
"""Save a conversation turn. Uses pipeline for atomicity."""
conv_key = self._conversation_key(user_id, session_id)
timeline_key = self._timeline_key(user_id)
timestamp = time.time()
turn_data = json.dumps({
"role": role,
"content": content,
"metadata": metadata or {},
"timestamp": timestamp
})
pipe = self.redis.pipeline()
# Append to the conversation list (right push)
pipe.rpush(conv_key, turn_data)
# Add to the timeline sorted set for time-based retrieval
pipe.zadd(timeline_key, {f"{session_id}:{timestamp}": timestamp})
# Set TTL on the conversation key
ttl_delta = ttl or self.default_ttl
pipe.expire(conv_key, int(ttl_delta.total_seconds()))
pipe.execute()
def get_history(self, user_id: str, session_id: str,
limit: int = 50, start_index: int = 0) -> List[Dict]:
"""Retrieve conversation history as a list of turn dicts."""
conv_key = self._conversation_key(user_id, session_id)
# Get range from the list (negative indices for end of list)
# lrange returns [start, end] inclusive
raw_turns = self.redis.lrange(conv_key, start_index, start_index + limit - 1)
results = []
for raw in raw_turns:
turn = json.loads(raw)
results.append(turn)
return results
def get_last_n_messages(self, user_id: str, session_id: str, n: int = 10) -> List[Dict]:
"""Get the most recent N messages using negative indices."""
conv_key = self._conversation_key(user_id, session_id)
raw_turns = self.redis.lrange(conv_key, -n, -1)
return [json.loads(raw) for raw in raw_turns]
def save_fact(self, user_id: str, key: str, value: str) -> None:
"""Store a fact in a Redis hash."""
facts_key = self._facts_key(user_id)
self.redis.hset(facts_key, key, value)
def get_fact(self, user_id: str, key: str) -> Optional[str]:
"""Get a single fact."""
return self.redis.hget(self._facts_key(user_id), key)
def get_all_facts(self, user_id: str) -> Dict[str, str]:
"""Get all facts for a user."""
return self.redis.hgetall(self._facts_key(user_id))
def delete_fact(self, user_id: str, key: str) -> None:
"""Remove a fact."""
self.redis.hdel(self._facts_key(user_id), key)
def get_active_sessions(self, user_id: str,
since_seconds: int = 3600) -> List[str]:
"""Get sessions active within the last N seconds using sorted set."""
timeline_key = self._timeline_key(user_id)
cutoff = time.time() - since_seconds
# zrangebyscore returns entries with scores between min and max
session_refs = self.redis.zrangebyscore(timeline_key, cutoff, float('inf'))
# Extract unique session IDs
sessions = set()
for ref in session_refs:
session_id = ref.split(":")[0]
sessions.add(session_id)
return list(sessions)
def trim_conversation(self, user_id: str, session_id: str,
max_turns: int = 100) -> int:
"""Trim conversation to max_turns, removing oldest entries. Returns count removed."""
conv_key = self._conversation_key(user_id, session_id)
current_length = self.redis.llen(conv_key)
if current_length > max_turns:
excess = current_length - max_turns
# LTRIM keeps only elements in the specified range
self.redis.ltrim(conv_key, excess, -1)
return excess
return 0
def clear_session(self, user_id: str, session_id: str) -> None:
"""Completely remove a session's conversation data."""
conv_key = self._conversation_key(user_id, session_id)
self.redis.delete(conv_key)
def search_memory(self, user_id: str, keyword: str,
limit: int = 20) -> List[Dict]:
"""Basic keyword search across all sessions (CPU-intensive, use RediSearch for production)."""
results = []
# Scan for conversation keys for this user
pattern = f"agent:conv:{user_id}:*"
cursor = 0
while True:
cursor, keys = self.redis.scan(cursor, match=pattern, count=100)
for key in keys:
turns = self.redis.lrange(key, 0, -1)
for turn_raw in turns:
if keyword.lower() in turn_raw.lower():
results.append(json.loads(turn_raw))
if len(results) >= limit:
return results[:limit]
if cursor == 0:
break
return results[:limit]
# Usage example
memory = RedisMemory(host="localhost", port=6379)
user_id = "agent_42"
session_id = "session_xyz"
# Save conversation turns
memory.save_turn(user_id, session_id, "user", "I'm planning a trip to Japan next spring.")
memory.save_turn(user_id, session_id, "assistant",
"Japan in spring is beautiful! Cherry blossoms will be blooming.",
metadata={"topic": "travel", "season": "spring"})
memory.save_turn(user_id, session_id, "user", "Great! What cities should I visit?")
memory.save_turn(user_id, session_id, "assistant",
"I recommend Tokyo, Kyoto, and Osaka. Let me save this preference.",
metadata={"recommendations": ["Tokyo", "Kyoto", "Osaka"]})
# Store facts
memory.save_fact(user_id, "destination", "Japan")
memory.save_fact(user_id, "season", "spring")
memory.save_fact(user_id, "cities_of_interest", "Tokyo, Kyoto, Osaka")
# Retrieve and display
history = memory.get_last_n_messages(user_id, session_id, n=5)
facts = memory.get_all_facts(user_id)
print(f"Stored facts: {facts}")
print(f"\nLast 3 messages:")
for turn in history[-3:]:
print(f" [{turn['role']}] {turn['content'][:60]}...")
# Search for travel-related memories
travel_memories = memory.search_memory(user_id, "Japan", limit=5)
print(f"\nFound {len(travel_memories)} memories mentioning Japan")PostgreSQL-Based Memory Persistence
PostgreSQL is the go-to choice for production systems requiring ACID guarantees, complex queries, and vector search. This example uses psycopg2 for connection management and includes the pgvector extension for embedding-based semantic memory retrieval.
import psycopg2
import psycopg2.pool
import json
import time
from typing import List, Dict, Optional, Tuple
from contextlib import contextmanager
class PostgresMemory:
"""
Agent memory backed by PostgreSQL with pgvector support.
Provides ACID transactions, rich queries, and vector similarity search.
Prerequisites:
CREATE EXTENSION IF NOT EXISTS vector;
"""
def __init__(self, dsn: str = "postgresql://postgres:postgres@localhost:5432/agent_memory",
min_connections: int = 2, max_connections: int = 10):
self.pool = psycopg2.pool.ThreadedConnectionPool(
min_connections, max_connections, dsn
)
self._init_schema()
@contextmanager
def _get_conn(self):
"""Get a connection from the pool with automatic cleanup."""
conn = self.pool.getconn()
try:
yield conn
conn.commit()
except Exception:
conn.rollback()
raise
finally:
self.pool.putconn(conn)
def _init_schema(self):
"""Create tables if they don't exist."""
with self._get_conn() as conn:
with conn.cursor() as cur:
cur.execute("""
CREATE EXTENSION IF NOT EXISTS vector;
CREATE TABLE IF NOT EXISTS conversations (
id BIGSERIAL PRIMARY KEY,
user_id TEXT NOT NULL,
session_id TEXT NOT NULL,
role TEXT NOT NULL CHECK (role IN ('system', 'user', 'assistant', 'tool')),
content TEXT NOT NULL,
metadata JSONB DEFAULT '{}',
embedding vector(1536),
created_at TIMESTAMPTZ DEFAULT NOW()
);
CREATE INDEX IF NOT EXISTS idx_conv_user_session
ON conversations (user_id, session_id, created_at DESC);
CREATE INDEX IF NOT EXISTS idx_conv_embedding
ON conversations USING ivfflat (embedding vector_cosine_ops)
WITH (lists = 100);
CREATE TABLE IF NOT EXISTS agent_facts (
id BIGSERIAL PRIMARY KEY,
user_id TEXT NOT NULL,
fact_key TEXT NOT NULL,
fact_value TEXT NOT NULL,
metadata JSONB DEFAULT '{}',
created_at TIMESTAMPTZ DEFAULT NOW(),
updated_at TIMESTAMPTZ DEFAULT NOW(),
UNIQUE (user_id, fact_key)
);
CREATE INDEX IF NOT EXISTS idx_facts_user ON agent_facts (user_id);
CREATE TABLE IF NOT EXISTS memory_summaries (
id BIGSERIAL PRIMARY KEY,
user_id TEXT NOT NULL,
session_id TEXT,
summary_text TEXT NOT NULL,
source_range_start BIGINT,
source_range_end BIGINT,
embedding vector(1536),
created_at TIMESTAMPTZ DEFAULT NOW()
);
""")
def save_turn(self, user_id: str, session_id: str, role: str,
content: str, metadata: Optional[Dict] = None,
embedding: Optional[List[float]] = None) -> int:
"""Save a conversation turn. Returns the row ID."""
with self._get_conn() as conn:
with conn.cursor() as cur:
cur.execute("""
INSERT INTO conversations (user_id, session_id, role, content, metadata, embedding)
VALUES (%s, %s, %s, %s, %s, %s)
RETURNING id
""", (
user_id, session_id, role, content,
json.dumps(metadata or {}),
embedding # List of floats, pgvector handles conversion
))
row_id = cur.fetchone()[0]
return row_id
def save_turn_batch(self, turns: List[Tuple[str, str, str, str, Optional[Dict], Optional[List[float]]]]) -> List[int]:
"""Batch insert multiple turns efficiently. Returns list of IDs."""
with self._get_conn() as conn:
with conn.cursor() as cur:
ids = []
for user_id, session_id, role, content, metadata, embedding in turns:
cur.execute("""
INSERT INTO conversations (user_id, session_id, role, content, metadata, embedding)
VALUES (%s, %s, %s, %s, %s, %s)
RETURNING id
""", (user_id, session_id, role, content,
json.dumps(metadata or {}), embedding))
ids.append(cur.fetchone()[0])
return ids
def get_history(self, user_id: str, session_id: Optional[str] = None,
limit: int = 50, offset: int = 0,
since_timestamp: Optional[float] = None,
include_embeddings: bool = False) -> List[Dict]:
"""Retrieve conversation history with flexible filtering."""
with self._get_conn() as conn:
with conn.cursor() as cur:
columns = "id, role, content, metadata, created_at"
if include_embeddings:
columns += ", embedding::text"
query = f"SELECT {columns} FROM conversations WHERE user_id = %s"
params = [user_id]
if session_id is not None:
query += " AND session_id = %s"
params.append(session_id)
if since_timestamp is not None:
query += " AND EXTRACT(EPOCH FROM created_at) >= %s"
params.append(since_timestamp)
query += " ORDER BY created_at ASC LIMIT %s OFFSET %s"
params.extend([limit, offset])
cur.execute(query, params)
rows = cur.fetchall()
results = []
for row in rows:
result = {
"id": row[0],
"role": row[1],
"content": row[2],
"metadata": row[3] if isinstance(row[3], dict) else json.loads(row[3] or '{}'),
"timestamp": row[4].isoformat() if hasattr(row[4], 'isoformat') else str(row[4])
}
results.append(result)
return results
def search_semantic(self, user_id: str, query_embedding: List[float],
top_k: int = 10, min_similarity: float = 0.7) -> List[Dict]:
"""
Perform vector similarity search on conversation embeddings.
Returns most relevant turns with similarity scores.
"""
with self._get_conn() as conn:
with conn.cursor() as cur:
cur.execute("""
SELECT id, role, content, metadata, created_at,
1 - (embedding <=> %s::vector) AS similarity
FROM conversations
WHERE user_id = %s
AND embedding IS NOT NULL
AND 1 - (embedding <=> %s::vector) >= %s
ORDER BY embedding <=> %s::vector
LIMIT %s
""", (query_embedding, user_id, query_embedding,
min_similarity, query_embedding, top_k))
rows = cur.fetchall()
results = []
for row in rows:
results.append({
"id": row[0],
"role": row[1],
"content": row[2],
"metadata": row[3] if isinstance(row[3], dict) else json.loads(row[3] or '{}'),
"similarity": round(float(row[5]), 4)
})
return results
def save_fact(self, user_id: str, key: str, value: str,
metadata: Optional[Dict] = None) -> None:
"""Upsert a fact (insert or update)."""
with self._get_conn() as conn:
with conn.cursor() as cur:
cur.execute("""
INSERT INTO agent_facts (user_id, fact_key, fact_value, metadata, updated_at)
VALUES (%s, %s, %s, %s, NOW())
ON CONFLICT (user_id, fact_key) DO UPDATE SET
fact_value = EXCLUDED.fact_value,
metadata = EXCLUDED.metadata,
updated_at = NOW()
""", (user_id, key, value, json.dumps(metadata or {})))
def get_fact(self, user_id: str, key: str) -> Optional[Dict]:
"""Get a fact with metadata."""
with self._get_conn() as conn: