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.