windlass.providers.retrievers.contextual

contextual

Contextual retrieval and query transformation.

Two techniques that attack retrieval failure from opposite ends:

Contextual chunk enrichment (index time). A chunk that reads "revenue grew 3%" is unfindable — grew from what, for whom, when? Before embedding, a small model writes one situating sentence using the surrounding document, and that sentence is prepended. Anthropic's published results put the retrieval-failure reduction at roughly 35-50%.

Query transformation (query time). HyDE has the model write a hypothetical answer and embeds that instead of the question, because answers live closer to answers in embedding space than questions do. Multi-query expansion runs several rephrasings and fuses the results.

Both cost model calls. Enrichment pays once at ingestion; transformation pays on every query, so measure before enabling it in a latency-sensitive path.

Example

from windlass.providers.llm.fake import FakeLLM from windlass.providers.retrievers.bm25 import BM25Retriever from windlass.core.types import Chunk base = BM25Retriever() r = ContextualRetriever(retriever=base, llm=FakeLLM(responses=["Q3 revenue."])) r.index([Chunk(content="revenue grew 3%", metadata={"source": "10-K"})]) 1 "Q3 revenue." in base.chunks[next(iter(base.chunks))].content True

ContextualRetriever

ContextualRetriever(
    *,
    retriever: Retriever | None = None,
    llm: LLM | None = None,
    enrich: bool = True,
    transform: str = "none",
    n_queries: int = 3,
    context_window: int = 4000,
    top_k: int = 5,
    **config: Any
)

Bases: Retriever

Wraps another retriever with LLM-driven enrichment and query rewriting.

Parameters:

Name	Type	Description	Default
retriever	Retriever | None	The retriever doing the actual search. Required.	None
llm	LLM | None	Model used for enrichment and rewriting. Required. Use a small, cheap model — this is a summarisation task, not reasoning.	None
enrich	bool	Prepend a generated situating sentence to each chunk at index time.	True
transform	str	Query-time strategy — "none", "hyde" or "multi_query".	'none'
n_queries	int	How many rewrites to generate for multi_query.	3
context_window	int	Characters of surrounding document shown to the model when enriching. Larger gives better context and costs more.	4000
top_k	int	Default number of results.	5
**config	Any	Forwarded to :class:~windlass.interfaces.retriever.Retriever.	{}

Raises:

Type	Description
ConfigurationError	When retriever or llm is missing.
ValueError	For an unknown transform.

Performance

Enrichment issues one model call per chunk, run with bounded concurrency. For a 10k-chunk corpus that is a real but one-off cost; the embedder's cache makes re-ingestion cheap.

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

def __init__(
    self,
    *,
    retriever: Retriever | None = None,
    llm: LLM | None = None,
    enrich: bool = True,
    transform: str = "none",
    n_queries: int = 3,
    context_window: int = 4000,
    top_k: int = 5,
    **config: Any,
) -> None:
    if retriever is None:
        raise ConfigurationError(
            "Contextual retrieval wraps another retriever.",
            hint="Pass retriever=Windlass.retriever('vector', ...).",
        )
    if llm is None:
        raise ConfigurationError(
            "Contextual retrieval needs a model to write the context.",
            hint="Pass llm=Windlass.llm('openai', model='gpt-4o-mini').",
        )
    if transform not in {"none", "hyde", "multi_query"}:
        raise ValueError("transform must be 'none', 'hyde' or 'multi_query'")

    super().__init__(top_k=top_k, **config)
    self.retriever = retriever
    self.llm = llm
    self.enrich = enrich
    self.transform = transform
    self.n_queries = max(2, n_queries)
    self.context_window = context_window

native

native() -> Any

Return the wrapped retriever's native handle.

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

def native(self) -> Any:
    """Return the wrapped retriever's native handle."""
    return self.retriever.native()

aindex async

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

Enrich chunks and hand them to the wrapped retriever.

Parameters:

Name	Type	Description	Default
chunks	Sequence[Chunk]	Chunks to index.	required

Returns:

Type	Description
int	How many chunks were indexed.

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

async def aindex(self, chunks: Sequence[Chunk]) -> int:
    """Enrich chunks and hand them to the wrapped retriever.

    Args:
        chunks: Chunks to index.

    Returns:
        How many chunks were indexed.
    """
    if not chunks:
        return 0
    prepared = await self._enrich(list(chunks)) if self.enrich else list(chunks)
    return await self.retriever.aindex(prepared)

aretrieve_chunks async

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

Optionally rewrite the query, then delegate to the wrapped retriever.

Parameters:

Name	Type	Description	Default
query	str	The search query.	required
k	int	How many results to return.	required
filters	MetadataFilter | None	Metadata constraints.	None
**kwargs	Any	Forwarded to the wrapped retriever.	{}

Returns:

Type	Description
list[ScoredChunk]	Ranked hits.

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

async def aretrieve_chunks(
    self,
    query: str,
    k: int,
    *,
    filters: MetadataFilter | None = None,
    **kwargs: Any,
) -> list[ScoredChunk]:
    """Optionally rewrite the query, then delegate to the wrapped retriever.

    Args:
        query: The search query.
        k: How many results to return.
        filters: Metadata constraints.
        **kwargs: Forwarded to the wrapped retriever.

    Returns:
        Ranked hits.
    """
    if self.transform == "none":
        result = await self.retriever.aretrieve(query, k, filters=filters, **kwargs)
        return result.hits

    if self.transform == "hyde":
        probe = await self._hyde(query)
        result = await self.retriever.aretrieve(probe, k, filters=filters, **kwargs)
        return result.hits

    return await self._multi_query(query, k, filters, kwargs)