Skip to content

symfonic.memory.backends.chroma

chroma

ChromaVectorBackend -- optional concrete VectorBackend using ChromaDB.

Requires the chroma optional dependency group: pip install symfonic-core[chroma]

All methods are async, wrapping synchronous ChromaDB calls with asyncio.to_thread for non-blocking operation.

ChromaVectorBackend 

ChromaVectorBackend(
    collection_name: str = "memory",
    persist_directory: str | None = None,
    embedding_provider: EmbeddingProvider | None = None,
)

VectorBackend implementation backed by ChromaDB.

Tenant isolation is enforced via metadata filtering on tenant_id. All synchronous ChromaDB operations are wrapped with asyncio.to_thread.

Source code in src/symfonic/memory/backends/chroma.py 
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
def __init__(
    self,
    collection_name: str = "memory",
    persist_directory: str | None = None,
    embedding_provider: EmbeddingProvider | None = None,
) -> None:
    self._embedding_provider = embedding_provider
    client_settings: dict[str, Any] = {}
    if persist_directory:
        client_settings["persist_directory"] = persist_directory

    self._client = chromadb.Client() if not persist_directory else (
        chromadb.PersistentClient(path=persist_directory)
    )
    self._collection = self._client.get_or_create_collection(
        name=collection_name,
        metadata={"hnsw:space": "cosine"},
    )

add async 

add(
    scope: TenantScope,
    ids: list[str],
    embeddings: list[list[float]],
    metadatas: list[dict[str, Any]],
    documents: list[str],
) -> None

Add vectors with tenant- and scope-path-scoped metadata.

Source code in src/symfonic/memory/backends/chroma.py 
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
async def add(
    self,
    scope: TenantScope,
    ids: list[str],
    embeddings: list[list[float]],
    metadatas: list[dict[str, Any]],
    documents: list[str],
) -> None:
    """Add vectors with tenant- and scope-path-scoped metadata."""
    scope_path = materialise_scope_path(scope)
    enriched_metadatas = [
        {**m, "tenant_id": scope.tenant_id, SCOPE_PATH_KEY: scope_path}
        for m in metadatas
    ]
    await asyncio.to_thread(
        self._collection.add,
        ids=ids,
        embeddings=embeddings,
        metadatas=enriched_metadatas,
        documents=documents,
    )

count async 

count(scope: TenantScope) -> int

Return the number of stored vectors for the tenant scope.

Without this method EpisodicLayer.count_events() falls back to 0, silently disabling consolidation phases gated on episodic volume for Chroma-backed deployments. collection.count() is not tenant-aware, so filter by tenant metadata. Round-2 fix.

Source code in src/symfonic/memory/backends/chroma.py 
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
async def count(self, scope: TenantScope) -> int:
    """Return the number of stored vectors for the tenant scope.

    Without this method EpisodicLayer.count_events() falls back to 0,
    silently disabling consolidation phases gated on episodic volume for
    Chroma-backed deployments. ``collection.count()`` is not tenant-aware,
    so filter by tenant metadata. Round-2 fix.
    """
    result = await asyncio.to_thread(
        self._collection.get,
        where={"tenant_id": scope.tenant_id},
        include=[],
    )
    ids = result.get("ids", []) if result else []
    return len(ids)

delete async 

delete(scope: TenantScope, ids: list[str]) -> None

Delete vectors by their IDs, scoped to the tenant.

Source code in src/symfonic/memory/backends/chroma.py 
138
139
140
141
142
143
144
async def delete(self, scope: TenantScope, ids: list[str]) -> None:
    """Delete vectors by their IDs, scoped to the tenant."""
    await asyncio.to_thread(
        self._collection.delete,
        ids=ids,
        where={"tenant_id": scope.tenant_id},
    )

search async 

search(
    scope: TenantScope,
    query_embedding: list[float],
    top_k: int = 5,
) -> list[dict[str, Any]]

Search for similar vectors within the tenant scope.

Source code in src/symfonic/memory/backends/chroma.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
async def search(
    self,
    scope: TenantScope,
    query_embedding: list[float],
    top_k: int = 5,
) -> list[dict[str, Any]]:
    """Search for similar vectors within the tenant scope."""
    results = await asyncio.to_thread(
        self._collection.query,
        query_embeddings=[query_embedding],
        n_results=top_k,
        where=_prefix_where(scope),
    )
    output: list[dict[str, Any]] = []
    if results and results.get("ids"):
        ids = results["ids"][0]
        distances = results.get("distances", [[]])[0]
        metadatas = results.get("metadatas", [[]])[0]
        documents = results.get("documents", [[]])[0]
        for i, doc_id in enumerate(ids):
            score = 1.0 - distances[i] if i < len(distances) else 0.0
            output.append({
                "id": doc_id,
                "score": score,
                "metadata": metadatas[i] if i < len(metadatas) else {},
                "document": documents[i] if i < len(documents) else "",
            })
    return output