windlass.providers.retrievers.vector

vector

Dense vector retrieval.

The standard semantic search path: embed the query with the same model that embedded the corpus, ask the vector store for nearest neighbours, return them.

Two refinements are built in:

• MMR (diversity > 0) trades a little relevance for coverage, so you stop getting five near-identical chunks from the same page.
• Parent expansion swaps matched child chunks for their larger parents when the pipeline uses parent-child chunking.

Example

from windlass.providers.embeddings.hash import HashEmbedder from windlass.providers.vectordb.memory import InMemoryVectorStore from windlass.core.types import Chunk emb = HashEmbedder(dimensions=128) store = InMemoryVectorStore(dimensions=128) chunks = [Chunk(content=t) for t in ["kittens purr", "engines roar"]] for c, v in zip(chunks, emb.embed([c.content for c in chunks])): ... c.embedding = v _ = store.add(chunks) r = VectorRetriever(embedder=emb, vectorstore=store) r.retrieve("kittens").hits[0].chunk.content 'kittens purr'

VectorRetriever

VectorRetriever(
    *,
    embedder: Embedder | None = None,
    vectorstore: VectorStore | None = None,
    top_k: int = 5,
    diversity: float = 0.0,
    fetch_k: int | None = None,
    expand_parents: bool = False,
    parent_source: Any = None,
    **config: Any
)

Bases: Retriever

Dense retrieval over a vector store.

Parameters:

Name	Type	Description	Default
embedder	Embedder | None	Model used to embed the query. Must be the same model that embedded the corpus.	None
vectorstore	VectorStore | None	Where the vectors live.	None
top_k	int	Default number of results.	5
diversity	float	MMR strength in [0, 1]. 0 disables MMR (plain similarity); 0.3 is a good default when your corpus has near-duplicate passages.	0.0
fetch_k	int | None	Candidates fetched before MMR narrows them. Defaults to 4 * top_k when MMR is on.	None
expand_parents	bool	Swap matched child chunks for their parents. Set automatically by the RAG builder when parent-child chunking is used.	False
parent_source	Any	Object exposing expand(chunks) — normally the :class:~windlass.providers.chunkers.hierarchical.ParentChildChunker.	None
**config	Any	Forwarded to :class:~windlass.interfaces.retriever.Retriever.	{}

Raises:

Type	Description
ConfigurationError	When the embedder or vector store is missing.

Performance

One embedding call plus one store query per retrieval. Enabling MMR fetches fetch_k candidates and does O(fetch_k²) similarity work in-process — negligible for the usual fetch_k ≤ 100.

Source code in src\windlass\providers\retrievers\vector.py

def __init__(
    self,
    *,
    embedder: Embedder | None = None,
    vectorstore: VectorStore | None = None,
    top_k: int = 5,
    diversity: float = 0.0,
    fetch_k: int | None = None,
    expand_parents: bool = False,
    parent_source: Any = None,
    **config: Any,
) -> None:
    if embedder is None:
        raise ConfigurationError(
            "Vector retrieval needs an embedding model.",
            hint="Pass embedder=..., or build the retriever through "
            "Windlass.rag(), which wires it for you.",
        )
    if vectorstore is None:
        raise ConfigurationError(
            "Vector retrieval needs a vector store.",
            hint="Pass vectorstore=..., or build the retriever through Windlass.rag().",
        )
    if not 0.0 <= diversity <= 1.0:
        raise ValueError("diversity must be between 0 and 1")

    super().__init__(
        top_k=top_k,
        fetch_k=fetch_k or (top_k * 4 if diversity else top_k),
        **config,
    )
    self.embedder = embedder
    self.vectorstore = vectorstore
    self.diversity = diversity
    self.expand_parents = expand_parents
    self.parent_source = parent_source

native

native() -> Any

Return the underlying vector store's native handle.

Source code in src\windlass\providers\retrievers\vector.py

def native(self) -> Any:
    """Return the underlying vector store's native handle."""
    return self.vectorstore.native()

aindex async

aindex(chunks: Any) -> int

Embed and write chunks to the store.

Chunks that already carry an embedding are written as-is, so this is cheap to call on an already-embedded batch.

Parameters:

Name	Type	Description	Default
chunks	Any	Chunks to index.	required

Returns:

Type	Description
int	How many chunks were written.

Source code in src\windlass\providers\retrievers\vector.py

async def aindex(self, chunks: Any) -> int:
    """Embed and write chunks to the store.

    Chunks that already carry an embedding are written as-is, so this is
    cheap to call on an already-embedded batch.

    Args:
        chunks: Chunks to index.

    Returns:
        How many chunks were written.
    """
    items = list(chunks)
    if not items:
        return 0
    missing = [c for c in items if not c.embedding]
    if missing:
        vectors = await self.embedder.aembed([c.content for c in missing])
        for chunk, vector in zip(missing, vectors, strict=True):
            chunk.embedding = vector
    return await self.vectorstore.aadd(items)

aretrieve_chunks async

aretrieve_chunks(
    query: str, k: int, *, filters: MetadataFilter | None = None, **kwargs: Any
) -> list[ScoredChunk]

Embed the query and search the store.

Parameters:

Name	Type	Description	Default
query	str	The search query.	required
k	int	How many results to return.	required
filters	MetadataFilter | None	Metadata constraints, pushed down to the store.	None
**kwargs	Any	Forwarded to the store's search method.	{}

Returns:

Type	Description
list[ScoredChunk]	Ranked hits.

Raises:

Type	Description
RetrievalError	When embedding or the store query fails.

Source code in src\windlass\providers\retrievers\vector.py

async def aretrieve_chunks(
    self,
    query: str,
    k: int,
    *,
    filters: MetadataFilter | None = None,
    **kwargs: Any,
) -> list[ScoredChunk]:
    """Embed the query and search the store.

    Args:
        query: The search query.
        k: How many results to return.
        filters: Metadata constraints, pushed down to the store.
        **kwargs: Forwarded to the store's search method.

    Returns:
        Ranked hits.

    Raises:
        RetrievalError: When embedding or the store query fails.
    """
    try:
        vector = await self.embedder.aembed_query(query)
    except Exception as exc:
        raise RetrievalError(
            f"Could not embed the query: {exc}",
            hint="Check that the embedding provider is reachable and configured.",
        ) from exc

    fetch = max(k, self.fetch_k) if self.diversity else k
    try:
        hits = await self.vectorstore.asearch(vector, fetch, filters=filters, **kwargs)
    except Exception as exc:
        raise RetrievalError(f"Vector store query failed: {exc}") from exc

    if self.diversity and len(hits) > k:
        hits = self._diversify(vector, hits, k)
    else:
        hits = hits[:k]

    if self.expand_parents:
        hits = self._expand(hits)
    return hits