windlass.interfaces.vectordb

vectordb

The vector-store interface.

A vector store persists chunks together with their embeddings and answers nearest-neighbour queries. FAISS, Chroma, Pinecone and the built-in in-memory store all sit behind this one class, so moving from a laptop prototype to a managed index is a one-word change in the builder.

Implementers override four coroutines: :meth:VectorStore.aadd, :meth:VectorStore.asearch, :meth:VectorStore.adelete and :meth:VectorStore.acount.

Example

from windlass.providers.vectordb.memory import InMemoryVectorStore from windlass.core.types import Chunk store = InMemoryVectorStore(dimensions=3) store.add([Chunk(content="hi", embedding=[1.0, 0.0, 0.0])]) 1 store.search([1.0, 0.0, 0.0], k=1)[0].chunk.content 'hi'

VectorStore

VectorStore(
    *,
    collection: str = "windlass",
    dimensions: int | None = None,
    metric: str = "cosine",
    persist_path: str | None = None,
    name: str | None = None,
    **config: Any
)

Bases: Component

Abstract vector database.

Parameters:

Name	Type	Description	Default
collection	str	Name of the collection / index / namespace to use.	'windlass'
dimensions	int | None	Vector dimensionality. Required by stores that must create an index up front; inferred on first write by those that can.	None
metric	str	Similarity metric — cosine, dot or euclidean.	'cosine'
persist_path	str | None	Where an on-disk store should keep its data.	None
name	str | None	Component name for traces.	None
**config	Any	Store-specific options.	{}

Attributes:

Name	Type	Description
collection		The active collection name.
metric		The configured similarity metric.
supports_filters	bool	Whether metadata filtering is pushed down to the store rather than applied client-side.
supports_hybrid	bool	Whether the store has native sparse+dense search.

Example

Implementing a store means four methods::

class MyStore(VectorStore):
    provider_name = "mine"

    async def aadd(self, chunks): ...
    async def asearch(self, vector, k=5, filters=None): ...
    async def adelete(self, ids=None, filters=None): ...
    async def acount(self): ...

Source code in src\windlass\interfaces\vectordb.py

def __init__(
    self,
    *,
    collection: str = "windlass",
    dimensions: int | None = None,
    metric: str = "cosine",
    persist_path: str | None = None,
    name: str | None = None,
    **config: Any,
) -> None:
    if metric not in {"cosine", "dot", "euclidean", "l2"}:
        raise ConfigurationError(
            f"Unsupported metric {metric!r}.",
            hint="Use 'cosine', 'dot' or 'euclidean'.",
        )
    super().__init__(
        name=name or self.provider_name,
        collection=collection,
        dimensions=dimensions,
        metric=metric,
        persist_path=persist_path,
        **config,
    )
    self.collection = collection
    self.dimensions = dimensions
    self.metric = metric
    self.persist_path = persist_path

aadd abstractmethod async

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

Insert or update chunks.

Chunk ids are deterministic, so re-adding the same content must upsert rather than duplicate.

Parameters:

Name	Type	Description	Default
chunks	Sequence[Chunk]	Chunks with :attr:~windlass.core.types.Chunk.embedding set.	required

Returns:

Type	Description
int	How many chunks were written.

Raises:

Type	Description
ConfigurationError	If any chunk is missing its embedding.
ProviderError	For store-side failures.

Source code in src\windlass\interfaces\vectordb.py

@abc.abstractmethod
async def aadd(self, chunks: Sequence[Chunk]) -> int:
    """Insert or update chunks.

    Chunk ids are deterministic, so re-adding the same content must upsert
    rather than duplicate.

    Args:
        chunks: Chunks with :attr:`~windlass.core.types.Chunk.embedding` set.

    Returns:
        How many chunks were written.

    Raises:
        ConfigurationError: If any chunk is missing its embedding.
        ProviderError: For store-side failures.
    """

asearch abstractmethod async

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

Find the k nearest chunks to vector.

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. Stores without native filtering should call :meth:match_filters client-side.	None
**kwargs	Any	Store-specific search options.	{}

Returns:

Type	Description
list[ScoredChunk]	Hits sorted by descending score, each with rank set.

Source code in src\windlass\interfaces\vectordb.py

@abc.abstractmethod
async def asearch(
    self,
    vector: Sequence[float],
    k: int = 5,
    *,
    filters: MetadataFilter | None = None,
    **kwargs: Any,
) -> list[ScoredChunk]:
    """Find the ``k`` nearest chunks to ``vector``.

    Args:
        vector: The query embedding.
        k: How many results to return.
        filters: Metadata constraints. Stores without native filtering
            should call :meth:`match_filters` client-side.
        **kwargs: Store-specific search options.

    Returns:
        Hits sorted by descending score, each with ``rank`` set.
    """

adelete abstractmethod async

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

Delete chunks by id or by metadata filter.

Parameters:

Name	Type	Description	Default
ids	Sequence[str] | None	Chunk ids to remove.	None
filters	MetadataFilter | None	Metadata constraints selecting what to remove.	None

Returns:

Type	Description
int	How many chunks were deleted.

Raises:

Type	Description
ValueError	If neither ids nor filters is given — deleting the entire collection must go through :meth:aclear.

Source code in src\windlass\interfaces\vectordb.py

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

    Args:
        ids: Chunk ids to remove.
        filters: Metadata constraints selecting what to remove.

    Returns:
        How many chunks were deleted.

    Raises:
        ValueError: If neither ``ids`` nor ``filters`` is given — deleting
            the entire collection must go through :meth:`aclear`.
    """

acount abstractmethod async

acount() -> int

Return how many chunks the collection holds.

Source code in src\windlass\interfaces\vectordb.py

@abc.abstractmethod
async def acount(self) -> int:
    """Return how many chunks the collection holds."""

aget async

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

Fetch chunks by id.

The default returns []; stores that can look up by id should override it. Parent-child retrieval relies on this to expand a matched child into its parent.

Parameters:

Name	Type	Description	Default
ids	Sequence[str]	Chunk ids to fetch.	required

Returns:

Type	Description
list[Chunk]	The chunks that exist, in whatever order the store returns them.

Source code in src\windlass\interfaces\vectordb.py

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

    The default returns ``[]``; stores that can look up by id should
    override it. Parent-child retrieval relies on this to expand a matched
    child into its parent.

    Args:
        ids: Chunk ids to fetch.

    Returns:
        The chunks that exist, in whatever order the store returns them.
    """
    return []

aclear async

aclear() -> None

Remove every chunk from the collection.

The default deletes by listing ids, which is correct but slow; stores with a native truncate should override it.

Source code in src\windlass\interfaces\vectordb.py

async def aclear(self) -> None:
    """Remove every chunk from the collection.

    The default deletes by listing ids, which is correct but slow; stores
    with a native truncate should override it.
    """
    await self.adelete(filters={})

apersist async

apersist() -> None

Flush pending writes to durable storage.

A no-op for stores that write through.

Source code in src\windlass\interfaces\vectordb.py

async def apersist(self) -> None:
    """Flush pending writes to durable storage.

    A no-op for stores that write through.
    """

add

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

Blocking :meth:aadd.

Source code in src\windlass\interfaces\vectordb.py

def add(self, chunks: Sequence[Chunk]) -> int:
    """Blocking :meth:`aadd`."""
    return run_sync(self.aadd(chunks))

search

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

Blocking :meth:asearch.

Source code in src\windlass\interfaces\vectordb.py

def search(
    self,
    vector: Sequence[float],
    k: int = 5,
    *,
    filters: MetadataFilter | None = None,
    **kwargs: Any,
) -> list[ScoredChunk]:
    """Blocking :meth:`asearch`."""
    return run_sync(self.asearch(vector, k, filters=filters, **kwargs))

delete

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

Blocking :meth:adelete.

Source code in src\windlass\interfaces\vectordb.py

def delete(
    self, ids: Sequence[str] | None = None, *, filters: MetadataFilter | None = None
) -> int:
    """Blocking :meth:`adelete`."""
    return run_sync(self.adelete(ids, filters=filters))

count

count() -> int

Blocking :meth:acount.

Source code in src\windlass\interfaces\vectordb.py

def count(self) -> int:
    """Blocking :meth:`acount`."""
    return run_sync(self.acount())

get

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

Blocking :meth:aget.

Source code in src\windlass\interfaces\vectordb.py

def get(self, ids: Sequence[str]) -> list[Chunk]:
    """Blocking :meth:`aget`."""
    return run_sync(self.aget(ids))

clear

clear() -> None

Blocking :meth:aclear.

Source code in src\windlass\interfaces\vectordb.py

def clear(self) -> None:
    """Blocking :meth:`aclear`."""
    run_sync(self.aclear())

persist

persist() -> None

Blocking :meth:apersist.

Source code in src\windlass\interfaces\vectordb.py

def persist(self) -> None:
    """Blocking :meth:`apersist`."""
    run_sync(self.apersist())

validate_embeddings staticmethod

validate_embeddings(chunks: Sequence[Chunk]) -> None

Assert that every chunk carries an embedding.

Parameters:

Name	Type	Description	Default
chunks	Sequence[Chunk]	Chunks about to be written.	required

Raises:

Type	Description
ConfigurationError	Naming the first chunk that is missing one.

Source code in src\windlass\interfaces\vectordb.py

@staticmethod
def validate_embeddings(chunks: Sequence[Chunk]) -> None:
    """Assert that every chunk carries an embedding.

    Args:
        chunks: Chunks about to be written.

    Raises:
        ConfigurationError: Naming the first chunk that is missing one.
    """
    for chunk in chunks:
        if not chunk.embedding:
            raise ConfigurationError(
                f"Chunk {chunk.id!r} has no embedding.",
                hint="Embed chunks before adding them, or use Pipeline.ingest() "
                "which does it for you.",
                context={"chunk_id": chunk.id},
            )

match_filters staticmethod

match_filters(metadata: dict[str, Any], filters: MetadataFilter | None) -> bool

Evaluate a metadata filter client-side.

Supports plain equality plus the Mongo-style operators $eq, $ne, $gt, $gte, $lt, $lte, $in, $nin, $contains and $exists. Stores without native filtering use this so that filter semantics are identical across every backend.

Parameters:

Name	Type	Description	Default
metadata	dict[str, Any]	The chunk's metadata.	required
filters	MetadataFilter | None	The constraints. None or {} matches everything.	required

Returns:

Type	Description
bool	True when the metadata satisfies every constraint.

Example

VectorStore.match_filters({"year": 2024}, {"year": {"$gte": 2020}}) True VectorStore.match_filters({"tag": "a"}, {"tag": {"$in": ["b"]}}) False

Source code in src\windlass\interfaces\vectordb.py

@staticmethod
def match_filters(metadata: dict[str, Any], filters: MetadataFilter | None) -> bool:
    """Evaluate a metadata filter client-side.

    Supports plain equality plus the Mongo-style operators ``$eq``, ``$ne``,
    ``$gt``, ``$gte``, ``$lt``, ``$lte``, ``$in``, ``$nin``, ``$contains``
    and ``$exists``. Stores without native filtering use this so that filter
    semantics are identical across every backend.

    Args:
        metadata: The chunk's metadata.
        filters: The constraints. ``None`` or ``{}`` matches everything.

    Returns:
        True when the metadata satisfies every constraint.

    Example:
        >>> VectorStore.match_filters({"year": 2024}, {"year": {"$gte": 2020}})
        True
        >>> VectorStore.match_filters({"tag": "a"}, {"tag": {"$in": ["b"]}})
        False
    """
    if not filters:
        return True
    for key, condition in filters.items():
        value = metadata.get(key)
        if not isinstance(condition, dict):
            if value != condition:
                return False
            continue
        for op, operand in condition.items():
            if not _apply_operator(value, op, operand):
                return False
    return True

rank staticmethod

rank(hits: list[ScoredChunk]) -> list[ScoredChunk]

Sort hits by descending score and assign 1-based ranks.

Parameters:

Name	Type	Description	Default
hits	list[ScoredChunk]	Unordered hits.	required

Returns:

Type	Description
list[ScoredChunk]	The same objects, sorted and with rank populated.

Source code in src\windlass\interfaces\vectordb.py

@staticmethod
def rank(hits: list[ScoredChunk]) -> list[ScoredChunk]:
    """Sort hits by descending score and assign 1-based ranks.

    Args:
        hits: Unordered hits.

    Returns:
        The same objects, sorted and with ``rank`` populated.
    """
    hits.sort(key=lambda h: h.score, reverse=True)
    for position, hit in enumerate(hits, start=1):
        hit.rank = position
    return hits