The short version: give your agent's long-term memory exactly two methods — remember(content, embedding, metadata) and recall(embedding, k, where) — and make every vector store hide behind them. Store the same shape back from all three. Then the store you picked on day one becomes a swap, not a rewrite: when sqlite-vec's brute-force search stops being fast enough, you point the same agent at LanceDB or Qdrant by changing one line. The interface is the investment. The store is a detail.
This is the piece our sqlite-vec vs LanceDB vs Qdrant comparison ended on — "build so the memory layer is a swap, not a rewrite." Here's the actual code.
Why the interface, before the store#
If you build alone, you will pick the store that's easiest today — for most agents that's sqlite-vec: one .db file, zero infrastructure, semantic recall in the same file as your conversation log. That's the right call. It's also the store you'll outgrow, because it does brute-force KNN only — every query is a linear scan, fast to roughly a few hundred thousand vectors and then not.
The mistake isn't picking sqlite-vec. It's calling its SDK directly from your agent — serialize_float32 here, a raw SQL string there — so that outgrowing it means editing every call site. Put two methods in front of it and the migration is one new file.
The interface: two methods, one result shape#
Define the contract first, in terms of what the agent needs, not what any store offers:
from dataclasses import dataclass, field
from typing import Protocol
@dataclass
class Memory:
id: str | int
content: str
score: float # higher = closer, roughly 0..1 — same for every store
metadata: dict = field(default_factory=dict)
class MemoryStore(Protocol):
def remember(self, content: str, embedding: list[float],
*, metadata: dict | None = None) -> None: ...
def recall(self, embedding: list[float],
*, k: int = 5, where: dict | None = None) -> list[Memory]: ...
Two decisions in that snippet do all the work. First, recall() returns a list of Memory — a plain shape with a normalized score where higher always means closer — so calling code ranks and thresholds one way regardless of backend. Second, where is a plain dict of equality filters ({"user_id": 42}), not a store-specific object, so the filter doesn't leak the backend. Keep both honest and the swap holds.
You supply the embeddings yourself (embed() below is your model — pick that first, because re-embedding a million memories is the expensive part). The store only ever sees vectors and metadata.
Adapter 1 — sqlite-vec (memory in a file)#
sqlite-vec wants a serialized float32 blob and does KNN with a MATCH plus a k = ? constraint. It returns an L2 distance (smaller is closer), so the adapter converts that to our "higher is closer" score.
import sqlite3, sqlite_vec
from sqlite_vec import serialize_float32
class SqliteVecStore:
def __init__(self, path="memory.db", dim=384):
self.db = sqlite3.connect(path)
self.db.enable_load_extension(True)
sqlite_vec.load(self.db)
self.db.enable_load_extension(False)
self.db.execute(f"""
CREATE VIRTUAL TABLE IF NOT EXISTS memories USING vec0(
memory_id INTEGER PRIMARY KEY,
user_id INTEGER,
embedding FLOAT[{dim}],
+content TEXT
)""")
def remember(self, content, embedding, *, metadata=None):
m = metadata or {}
self.db.execute(
"INSERT INTO memories(embedding, user_id, content) VALUES (?, ?, ?)",
[serialize_float32(embedding), m.get("user_id"), content])
self.db.commit()
def recall(self, embedding, *, k=5, where=None):
cond = "AND user_id = ?" if where and "user_id" in where else ""
params = [serialize_float32(embedding)]
if cond: params.append(where["user_id"])
params.append(k)
rows = self.db.execute(f"""
SELECT memory_id, content, distance FROM memories
WHERE embedding MATCH ? {cond} AND k = ?
ORDER BY distance""", params).fetchall()
return [Memory(id=r[0], content=r[1], score=1/(1+r[2])) for r in rows]
That 1/(1+distance) is a deliberate choice: a monotonic map from "0 distance" to "score 1," so bigger is always better. It isn't a cosine similarity — it's a ranking-preserving normalization, and that's all recall() promises.
Adapter 2 — LanceDB (memory in a library)#
LanceDB takes a plain list, filters with a SQL-style .where() string, and also returns a distance (_distance). Same normalization, different plumbing:
import lancedb
class LanceDBStore:
def __init__(self, path="./memory.lance"):
db = lancedb.connect(path)
self.tbl = (db.open_table("memories")
if "memories" in db.table_names()
else db.create_table("memories", schema={
"vector": "float32[384]", "user_id": "int64", "content": "string"}))
def remember(self, content, embedding, *, metadata=None):
m = metadata or {}
self.tbl.add([{"vector": embedding, "user_id": m.get("user_id"), "content": content}])
def recall(self, embedding, *, k=5, where=None):
q = self.tbl.search(embedding)
if where and "user_id" in where:
q = q.where(f"user_id = {int(where['user_id'])}")
return [Memory(id=r.get("user_id"), content=r["content"], score=1/(1+r["_distance"]),
metadata={"user_id": r.get("user_id")})
for r in q.limit(k).to_list()]
Note where the abstraction earns its keep: the calling code is identical to the sqlite-vec version, but here you'd build an ANN index (tbl.create_index(...)) once the table is large — an operational detail the agent never sees.
Adapter 3 — Qdrant (memory in a service)#
Qdrant is the odd one out twice over: you build a PointStruct to insert, a Filter object to filter, and it returns a **similarity score where larger is closer** — the opposite direction from the other two. The adapter absorbs all of it:
from qdrant_client import QdrantClient, models
class QdrantStore:
def __init__(self, url="http://localhost:6333", dim=384):
self.c = QdrantClient(url=url)
if not self.c.collection_exists("memories"):
self.c.create_collection("memories",
vectors_config=models.VectorParams(size=dim, distance=models.Distance.COSINE))
self._id = 0
def remember(self, content, embedding, *, metadata=None):
self._id += 1
self.c.upsert("memories", points=[models.PointStruct(
id=self._id, vector=embedding,
payload={**(metadata or {}), "content": content})])
def recall(self, embedding, *, k=5, where=None):
flt = None
if where and "user_id" in where:
flt = models.Filter(must=[models.FieldCondition(
key="user_id", match=models.MatchValue(value=where["user_id"]))])
hits = self.c.query_points("memories", query=embedding,
query_filter=flt, limit=k).points
# cosine score is -1..1; rescale to 0..1 so "higher = closer" holds like the others
return [Memory(id=h.id, content=h.payload["content"],
score=(h.score + 1) / 2,
metadata={k2: v for k2, v in h.payload.items() if k2 != "content"})
for h in hits]
Because we configured Distance.COSINE, h.score runs -1..1; (score + 1) / 2 puts it in the same 0..1 frame as the other two. Change the metric and you change this one line — inside the adapter, where it belongs.
The payoff: swap in one line#
Every adapter satisfies the same MemoryStore protocol, so the agent is written once:
def build_memory() -> MemoryStore:
return SqliteVecStore() # ← the only line that changes
# return LanceDBStore() # embedded, at scale, with ANN
# return QdrantStore() # shared service, heavy filtering, ops
mem = build_memory()
mem.remember("prefers terse answers", embed("prefers terse answers"), metadata={"user_id": 42})
hits = mem.recall(embed("how should I reply to this user?"), k=5, where={"user_id": 42})
Start on SqliteVecStore. The day brute-force latency stops being fast enough, uncomment LanceDBStore (stay embedded) or QdrantStore (go to a service), migrate the data once, and ship. The agent code doesn't move.
Keep it honest with one shared test#
"Swappable" is a claim until you prove it. Write the test against the interface and run it against all three — this is what catches the day a store's default metric isn't what you assumed:
import pytest
@pytest.mark.parametrize("factory", [SqliteVecStore, LanceDBStore, QdrantStore])
def test_recall_ranks_the_right_memory_first(factory):
mem = factory()
for text in ["prefers dark mode", "lives in Lisbon", "ships on Fridays"]:
mem.remember(text, embed(text), metadata={"user_id": 1})
hits = mem.recall(embed("what UI theme does the user like?"), k=3, where={"user_id": 1})
assert hits[0].content == "prefers dark mode" # same top result everywhere
assert all(0 <= h.score <= 1 for h in hits) # same normalized frame
assert hits == sorted(hits, key=lambda h: -h.score) # higher = closer, always
Three assertions, three backends, one contract. That parametrized test is the difference between a memory layer you can move and one you only hope you can.
The one-line takeaway#
Don't choose the perfect store. Choose a two-method interface — remember() and recall() returning a normalized Memory — and let sqlite-vec, LanceDB, and Qdrant compete behind it. The right store on day one is almost always a single file; the right architecture is the one that lets you leave it without a rewrite. If you're still deciding which tier of memory you even need before you get here, start with the three kinds of agent memory — this layer is only the long-term one.



