windlass.providers.vectordb.chroma

chroma

ChromaDB vector store.

Chroma is the pragmatic middle ground: persistent on disk with no server to run, or client/server when you need it, with native metadata filtering. Windlass uses Chroma purely as storage — embeddings are always computed by the configured Windlass embedder, so switching stores never silently switches embedding models.

Install with::

pip install "windlass[chroma]"

Example

from windlass import Windlass # doctest: +SKIP store = Windlass.vectordb("chroma", persist_path="./chroma") # doctest: +SKIP

ChromaVectorStore

ChromaVectorStore(
    *,
    collection: str = "windlass",
    dimensions: int | None = None,
    metric: str = "cosine",
    persist_path: str | None = None,
    host: str | None = None,
    port: int = 8000,
    client: Any = None,
    **config: Any
)

Bases: VectorStore

Vector store backed by ChromaDB.

Parameters:

Name	Type	Description	Default
collection	str	Chroma collection name.	'windlass'
dimensions	int | None	Vector length. Chroma infers it, so this is informational.	None
metric	str	cosine, dot or euclidean. Set at collection creation and immutable afterwards.	'cosine'
persist_path	str | None	Directory for the embedded persistent client. When omitted an ephemeral in-memory client is used.	None
host	str | None	Server host for client/server mode. Takes precedence over persist_path.	None
port	int	Server port.	8000
client	Any	Pass a pre-configured chromadb client to use instead.	None
**config	Any	Forwarded to :class:~windlass.interfaces.vectordb.VectorStore.	{}

Raises:

Type	Description
MissingDependencyError	When chromadb is not installed.
ProviderError	When the client or collection cannot be created.

Note

Chroma stores metadata values as scalars only. Windlass JSON-encodes lists and dicts on write and decodes them on read, so nested metadata round-trips correctly.

Source code in src\windlass\providers\vectordb\chroma.py

def __init__(
    self,
    *,
    collection: str = "windlass",
    dimensions: int | None = None,
    metric: str = "cosine",
    persist_path: str | None = None,
    host: str | None = None,
    port: int = 8000,
    client: Any = None,
    **config: Any,
) -> None:
    super().__init__(
        collection=collection,
        dimensions=dimensions,
        metric=metric,
        persist_path=persist_path,
        **config,
    )
    chromadb = require("chromadb", extra="chroma", feature="The ChromaDB vector store")
    self._chromadb = chromadb
    try:
        if client is not None:
            self._client = client
        elif host:
            self._client = chromadb.HttpClient(host=host, port=port)
        elif persist_path:
            self._client = chromadb.PersistentClient(path=str(persist_path))
        else:
            self._client = chromadb.EphemeralClient()

        self._collection = self._client.get_or_create_collection(
            name=collection,
            metadata={"hnsw:space": _space(metric)},
        )
    except Exception as exc:
        raise ProviderError(
            f"Could not open the Chroma collection {collection!r}: {exc}",
            provider="chroma",
            hint="Check the persist_path is writable, or that the server is reachable.",
            original=exc,
        ) from exc

native

native() -> Any

Return the underlying chromadb collection (Level 3 access).

Source code in src\windlass\providers\vectordb\chroma.py

def native(self) -> Any:
    """Return the underlying ``chromadb`` collection (Level 3 access)."""
    return self._collection

aadd async

aadd(chunks: Sequence[Chunk]) -> int

Upsert chunks into the collection.

Parameters:

Name	Type	Description	Default
chunks	Sequence[Chunk]	Chunks with embeddings set.	required

Returns:

Type	Description
int	How many chunks were written.

Raises:

Type	Description
ConfigurationError	If a chunk has no embedding.
ProviderError	When Chroma rejects the batch.

Source code in src\windlass\providers\vectordb\chroma.py

async def aadd(self, chunks: Sequence[Chunk]) -> int:
    """Upsert chunks into the collection.

    Args:
        chunks: Chunks with embeddings set.

    Returns:
        How many chunks were written.

    Raises:
        ConfigurationError: If a chunk has no embedding.
        ProviderError: When Chroma rejects the batch.
    """
    if not chunks:
        return 0
    self.validate_embeddings(chunks)

    def _upsert() -> int:
        self._collection.upsert(
            ids=[c.id for c in chunks],
            embeddings=[list(c.embedding or []) for c in chunks],
            documents=[c.content for c in chunks],
            metadatas=[_encode_metadata(c) for c in chunks],
        )
        return len(chunks)

    try:
        return await to_thread(_upsert)
    except Exception as exc:
        raise ProviderError(
            f"Chroma rejected a batch of {len(chunks)} chunks: {exc}",
            provider="chroma",
            original=exc,
        ) from exc

adelete async

adelete(
    ids: Sequence[str] | None = None, *, filters: MetadataFilter | None = None
) -> int

Delete by id or metadata filter.

Parameters:

Name	Type	Description	Default
ids	Sequence[str] | None	Chunk ids to remove.	None
filters	MetadataFilter | None	Metadata constraints, translated to Chroma's where syntax.	None

Returns:

Type	Description
int	How many chunks were removed.

Raises:

Type	Description
ValueError	When neither argument is supplied.

Source code in src\windlass\providers\vectordb\chroma.py

async def adelete(
    self, ids: Sequence[str] | None = None, *, filters: MetadataFilter | None = None
) -> int:
    """Delete by id or metadata filter.

    Args:
        ids: Chunk ids to remove.
        filters: Metadata constraints, translated to Chroma's ``where`` syntax.

    Returns:
        How many chunks were removed.

    Raises:
        ValueError: When neither argument is supplied.
    """
    if ids is None and filters is None:
        raise ValueError("Pass ids or filters; use clear() to empty the collection.")

    def _delete() -> int:
        before = self._collection.count()
        kwargs: dict[str, Any] = {}
        if ids:
            kwargs["ids"] = list(ids)
        if filters:
            kwargs["where"] = _to_where(filters)
        self._collection.delete(**kwargs)
        return max(0, before - self._collection.count())

    return await to_thread(_delete)

aclear async

aclear() -> None

Delete and recreate the collection.

Source code in src\windlass\providers\vectordb\chroma.py

async def aclear(self) -> None:
    """Delete and recreate the collection."""

    def _reset() -> None:
        self._client.delete_collection(self.collection)
        self._collection = self._client.get_or_create_collection(
            name=self.collection, metadata={"hnsw:space": _space(self.metric)}
        )

    await to_thread(_reset)

asearch async

asearch(
    vector: Sequence[float],
    k: int = 5,
    *,
    filters: MetadataFilter | None = None,
    **kwargs: Any
) -> list[ScoredChunk]

Query the collection.

Parameters:

Name	Type	Description	Default
vector	Sequence[float]	The query embedding.	required
k	int	How many results to return.	5
filters	MetadataFilter | None	Metadata constraints, pushed down to Chroma.	None
**kwargs	Any	Extra keyword arguments for collection.query.	{}

Returns:

Type	Description
list[ScoredChunk]	Ranked hits. Chroma returns distances; they are converted to
list[ScoredChunk]	similarity so higher is always better.

Source code in src\windlass\providers\vectordb\chroma.py

async def asearch(
    self,
    vector: Sequence[float],
    k: int = 5,
    *,
    filters: MetadataFilter | None = None,
    **kwargs: Any,
) -> list[ScoredChunk]:
    """Query the collection.

    Args:
        vector: The query embedding.
        k: How many results to return.
        filters: Metadata constraints, pushed down to Chroma.
        **kwargs: Extra keyword arguments for ``collection.query``.

    Returns:
        Ranked hits. Chroma returns distances; they are converted to
        similarity so higher is always better.
    """

    def _query() -> dict[str, Any]:
        payload: dict[str, Any] = {
            "query_embeddings": [list(vector)],
            "n_results": max(1, k),
            "include": ["documents", "metadatas", "distances"],
        }
        if filters:
            payload["where"] = _to_where(filters)
        payload.update(kwargs)
        return self._collection.query(**payload)

    try:
        result = await to_thread(_query)
    except Exception as exc:
        raise ProviderError(
            f"Chroma query failed: {exc}", provider="chroma", original=exc
        ) from exc

    ids = (result.get("ids") or [[]])[0]
    documents = (result.get("documents") or [[]])[0]
    metadatas = (result.get("metadatas") or [[]])[0]
    distances = (result.get("distances") or [[]])[0]

    hits: list[ScoredChunk] = []
    for cid, text, meta, distance in zip(ids, documents, metadatas, distances, strict=False):
        chunk = _decode_chunk(cid, text or "", meta or {})
        hits.append(
            ScoredChunk(
                chunk=chunk,
                score=_to_similarity(float(distance), self.metric),
                retriever=self.name,
            )
        )
    return self.rank(hits)

aget async

aget(ids: Sequence[str]) -> list[Chunk]

Fetch chunks by id.

Source code in src\windlass\providers\vectordb\chroma.py

async def aget(self, ids: Sequence[str]) -> list[Chunk]:
    """Fetch chunks by id."""
    if not ids:
        return []

    def _get() -> dict[str, Any]:
        return self._collection.get(ids=list(ids), include=["documents", "metadatas"])

    result = await to_thread(_get)
    return [
        _decode_chunk(cid, text or "", meta or {})
        for cid, text, meta in zip(
            result.get("ids") or [],
            result.get("documents") or [],
            result.get("metadatas") or [],
            strict=False,
        )
    ]

acount async

acount() -> int

Return how many chunks the collection holds.

Source code in src\windlass\providers\vectordb\chroma.py

async def acount(self) -> int:
    """Return how many chunks the collection holds."""
    return int(await to_thread(self._collection.count))

all_chunks

all_chunks() -> list[Chunk]

Return every stored chunk, for building a companion lexical index.

Note

Loads the whole collection into memory. Fine for the tens of thousands of chunks a hybrid retriever needs; not something to call on a multi-million-row collection.

Source code in src\windlass\providers\vectordb\chroma.py

def all_chunks(self) -> list[Chunk]:
    """Return every stored chunk, for building a companion lexical index.

    Note:
        Loads the whole collection into memory. Fine for the tens of
        thousands of chunks a hybrid retriever needs; not something to call
        on a multi-million-row collection.
    """
    result = self._collection.get(include=["documents", "metadatas"])
    return [
        _decode_chunk(cid, text or "", meta or {})
        for cid, text, meta in zip(
            result.get("ids") or [],
            result.get("documents") or [],
            result.get("metadatas") or [],
            strict=False,
        )
    ]