Stores

Memory Stores

InMemoryStore

Reference store for tests and small single-process agents. search_by_embedding is a full-scan cosine over a lazily cached embedding matrix (rebuilt after save/update/delete). Suitable up to roughly thousands of vectors; for larger N or durability see Performance & Scaling.

emotional_memory.stores.in_memory.InMemoryStore()

Simple in-memory MemoryStore. Not persistent across restarts.

search_by_embedding uses brute-force cosine similarity with a cached embedding matrix rebuilt lazily after mutations — suitable for small datasets and testing. For production at larger scale, use a vector database (SQLiteStore, QdrantStore, ChromaStore).

Source code in src/emotional_memory/stores/in_memory.py

def __init__(self) -> None:
    self._store: dict[str, Memory] = {}
    self._ids: list[str] = []
    self._matrix: NDArray[np.float64] | None = None
    self._norms: NDArray[np.float64] | None = None
    self._dirty: bool = True

search_by_embedding(embedding: list[float], top_k: int) -> list[Memory]

Return top_k memories by cosine similarity to the query embedding.

Memories without an embedding are skipped. Uses a cached embedding matrix (rebuilt lazily after save/update/delete) so repeated searches avoid re-stacking all vectors on every call. A non-positive top_k returns an empty list.

Source code in src/emotional_memory/stores/in_memory.py

def search_by_embedding(self, embedding: list[float], top_k: int) -> list[Memory]:
    """Return top_k memories by cosine similarity to the query embedding.

    Memories without an embedding are skipped. Uses a cached embedding
    matrix (rebuilt lazily after save/update/delete) so repeated searches
    avoid re-stacking all vectors on every call. A non-positive ``top_k``
    returns an empty list.
    """
    if top_k <= 0:
        return []
    if self._dirty:
        self._rebuild_cache()

    if not self._ids or self._matrix is None or self._norms is None:
        return []

    query = np.asarray(embedding, dtype=np.float64)
    query_norm = float(np.linalg.norm(query))
    if query_norm == 0.0:
        # Preserve previous behaviour: first top_k embeddable memories
        return [self._store[mid] for mid in self._ids[:top_k]]

    with np.errstate(invalid="ignore", divide="ignore"):
        scores = (self._matrix @ query) / (self._norms * query_norm)
    scores = np.nan_to_num(scores, nan=0.0)

    n = len(self._ids)
    k = min(top_k, n)
    # np.argpartition is O(n) for finding top-k; sort only the k winners
    top_indices = np.argpartition(scores, n - k)[n - k :]
    top_indices = top_indices[np.argsort(scores[top_indices])[::-1]]
    return [self._store[self._ids[i]] for i in top_indices]

SQLiteStore

Note

Requires the sqlite extra: uv pip install "emotional-memory[sqlite]"

Databases written before v0.18

The sqlite-vec index is created with distance_metric=cosine so that ANN candidate prefiltering matches the cosine similarity used by retrieval scoring. Earlier versions used sqlite-vec's L2 default, which ranks differently whenever embeddings are not L2-normalised. Opening such a database emits a UserWarning; call store.rebuild_vector_index() once to migrate it (the vector table is derived data — embeddings are also stored in the memory rows, so nothing is lost).

emotional_memory.stores.sqlite.SQLiteStore(path: str | Path = ':memory:')

Persistent MemoryStore backed by SQLite with sqlite-vec vector search.

Parameters

path: File path for the SQLite database, or ":memory:" for an ephemeral in-memory database (useful in tests without writing to disk).

Source code in src/emotional_memory/stores/sqlite.py

def __init__(self, path: str | Path = ":memory:") -> None:
    self._path = str(path)
    sqlite_vec = _load_sqlite_vec()
    # check_same_thread=False is required when the store is used through
    # SyncToAsyncStore (asyncio.to_thread dispatches on arbitrary threads).
    # WAL mode allows concurrent readers without blocking writers.
    self._conn: sqlite3.Connection = sqlite3.connect(self._path, check_same_thread=False)
    self._conn.row_factory = sqlite3.Row
    self._conn.enable_load_extension(True)
    sqlite_vec.load(self._conn)
    self._conn.enable_load_extension(False)
    if self._path != ":memory:":
        self._conn.execute("PRAGMA journal_mode=WAL")
    self._conn.execute(_CREATE_MEMORIES)
    self._conn.commit()
    self._vec_ready = False
    self._vec_cosine = True
    self._dim: int = 0
    # Serialise all connection access across threads: a single shared
    # sqlite3.Connection is not safe for concurrent use even with
    # check_same_thread=False (Python's sqlite3 module leaves locking to
    # the caller).
    self._lock: threading.RLock = threading.RLock()
    self._init_vec_from_db()

close() -> None

Close the underlying database connection.

Source code in src/emotional_memory/stores/sqlite.py

def close(self) -> None:
    """Close the underlying database connection."""
    self._conn.close()

rebuild_vector_index() -> int

Recreate the vector index from the stored memories, using cosine ranking.

The vec table holds only derived data (every embedding is also inside the memories JSON blob), so it can be dropped and rebuilt safely. Use this to migrate a database created before the index switched from L2 to cosine ordering.

Returns:

Type	Description
int	Number of embeddings re-indexed.

Source code in src/emotional_memory/stores/sqlite.py

def rebuild_vector_index(self) -> int:
    """Recreate the vector index from the stored memories, using cosine ranking.

    The vec table holds only derived data (every embedding is also inside the
    ``memories`` JSON blob), so it can be dropped and rebuilt safely. Use this
    to migrate a database created before the index switched from L2 to cosine
    ordering.

    Returns:
        Number of embeddings re-indexed.
    """
    with self._lock:
        memories = [m for m in self.list_all() if m.embedding is not None]
        with self._conn:
            self._conn.execute("DROP TABLE IF EXISTS memory_vec")
        self._vec_ready = False
        self._dim = 0
        if not memories:
            return 0
        self._ensure_vec(memories[0])
        with self._conn:
            for memory in memories:
                if memory.embedding is None:  # pragma: no cover - filtered above
                    continue
                self._conn.execute(
                    "INSERT INTO memory_vec (id, embedding) VALUES (?, ?)",
                    (memory.id, _pack_embedding(memory.embedding)),
                )
    return len(memories)

search_by_embedding(embedding: list[float], top_k: int) -> list[Memory]

Return top_k memories by approximate cosine similarity (sqlite-vec).

Falls back to brute-force cosine scan when the vector table is not yet initialised (e.g. no memories with embeddings have been saved yet).

Zero-length vectors have no defined cosine distance and come back with a NULL distance, which SQLite sorts first; they are moved to the end so a degenerate embedding cannot outrank genuine matches — mirroring _math.cosine_similarity, which scores a zero vector 0.0.

Source code in src/emotional_memory/stores/sqlite.py

def search_by_embedding(self, embedding: list[float], top_k: int) -> list[Memory]:
    """Return top_k memories by approximate cosine similarity (sqlite-vec).

    Falls back to brute-force cosine scan when the vector table is not yet
    initialised (e.g. no memories with embeddings have been saved yet).

    Zero-length vectors have no defined cosine distance and come back with a
    NULL distance, which SQLite sorts *first*; they are moved to the end so a
    degenerate embedding cannot outrank genuine matches — mirroring
    ``_math.cosine_similarity``, which scores a zero vector 0.0.
    """
    if top_k <= 0:
        return []
    with self._lock:
        if not self._vec_ready:
            return self._brute_force_search(embedding, top_k)

        rows = self._conn.execute(
            """
            SELECT m.data AS data, v.distance AS distance
            FROM memory_vec v
            JOIN memories m ON m.id = v.id
            WHERE v.embedding MATCH ? AND k = ?
            ORDER BY distance
            """,
            (_pack_embedding(embedding), top_k),
        ).fetchall()
    ordered = [row for row in rows if row["distance"] is not None]
    ordered += [row for row in rows if row["distance"] is None]
    return [Memory.model_validate_json(row["data"]) for row in ordered]

QdrantStore

Note

Requires the qdrant extra: uv pip install "emotional-memory[qdrant]"

emotional_memory.stores.qdrant.QdrantStore(*, url: str | None = None, path: str | None = None, collection_name: str = 'emotional_memory', api_key: str | None = None)

MemoryStore backed by a Qdrant vector database.

Parameters

url: Qdrant server URL, e.g. "http://localhost:6333". Mutually exclusive with path. None (default) selects the ephemeral in-memory Qdrant instance, useful for tests and prototyping. path: Local filesystem path for a persistent on-disk Qdrant instance. Mutually exclusive with url. collection_name: Name of the Qdrant collection. Created lazily on first save(). api_key: Optional API key for Qdrant Cloud (ignored in local modes).

Source code in src/emotional_memory/stores/qdrant.py

def __init__(
    self,
    *,
    url: str | None = None,
    path: str | None = None,
    collection_name: str = "emotional_memory",
    api_key: str | None = None,
) -> None:
    if url is not None and path is not None:
        raise ValueError("QdrantStore: pass at most one of `url` or `path`, not both")
    if url is not None:
        self._client: QdrantClient = QdrantClient(url=url, api_key=api_key)
    elif path is not None:
        self._client = QdrantClient(path=path)
    else:
        self._client = QdrantClient(":memory:")
    self._collection_name = collection_name
    self._dim: int = 0
    self._collection_ready: bool = False
    self._init_collection_state()

close() -> None

Close the underlying Qdrant client connection.

Source code in src/emotional_memory/stores/qdrant.py

def close(self) -> None:
    """Close the underlying Qdrant client connection."""
    self._client.close()

ChromaStore

Note

Requires the HTTP-only chroma extra and a separately operated server: uv pip install "emotional-memory[chroma]". Pass host=; embedded and file-backed modes are intentionally unavailable.

emotional_memory.stores.chroma.ChromaStore(*, path: str | None = None, host: str | None = None, port: int = 8000, collection_name: str = 'emotional_memory')

MemoryStore backed by a Chroma vector database.

Parameters

path: Unsupported. Retained to provide a migration error for callers that used the former embedded persistent mode. host: Remote Chroma server hostname, e.g. "localhost". Required. port: Remote Chroma server port (default 8000). Only used when host is provided. collection_name: Name of the Chroma collection. Created lazily on first save().

Source code in src/emotional_memory/stores/chroma.py

def __init__(
    self,
    *,
    path: str | None = None,
    host: str | None = None,
    port: int = 8000,
    collection_name: str = "emotional_memory",
) -> None:
    try:
        import chromadb
    except ImportError as exc:
        raise ImportError(
            "the Chroma HTTP client is required for ChromaStore. "
            "Install with: pip install 'emotional-memory[chroma]'"
        ) from exc

    if path is not None:
        raise ValueError(
            "ChromaStore embedded persistence is unavailable for security reasons; "
            "run a patched Chroma server and pass host= instead"
        )
    if host is None:
        raise ValueError(
            "ChromaStore requires host= because the [chroma] extra installs only "
            "the HTTP client"
        )
    self._client: _ChromaClientAPI = chromadb.HttpClient(host=host, port=port)
    self._collection_name = collection_name
    self._collection: _ChromaCollection | None = None
    self._dim: int = 0
    self._init_collection_state()

close() -> None

Close the ChromaStore (Chroma clients manage their own lifecycle).

Source code in src/emotional_memory/stores/chroma.py

def close(self) -> None:
    """Close the ChromaStore (Chroma clients manage their own lifecycle)."""

Runtime mood/affect persistence is a separate protocol (AffectiveStateStore), not a MemoryStore. See State Stores.