windlass.rag

rag

Retrieval-augmented generation.

Start with :func:windlass.Windlass.rag, which returns a :class:~windlass.rag.builder.RAGBuilder::

from windlass import Windlass

rag = Windlass.rag().chunker("semantic").retriever("hybrid")
rag.ingest("./docs")
print(rag.ask("How does authentication work?"))

The pieces:

• :class:~windlass.rag.builder.RAGBuilder — the fluent API, and the place where cross-component wiring is resolved.
• :class:~windlass.rag.pipeline.RAGPipeline — the assembled pipeline that does the work.
• :func:~windlass.rag.loading.loader_for — format auto-detection.

RAGBuilder

RAGBuilder(container: Container | None = None)

Fluent builder for a :class:~windlass.rag.pipeline.RAGPipeline.

Every configuration method returns self, and the pipeline methods (:meth:ingest, :meth:ask, :meth:search, ...) are forwarded, so a builder is a usable pipeline — no explicit .build() required.

Parameters:

Name	Type	Description	Default
container	Container | None	Dependency container. Defaults to a child of the process root, so application-wide bindings are inherited.	None

Attributes:

Name	Type	Description
container		The container backing this builder.

Example

Level 1 — defaults all the way down, no dependencies, no API keys::

rag = Windlass.rag()
rag.ingest_text("Windlass unifies the AI ecosystem.")
print(rag.ask("What does Windlass do?"))

Level 2 — configured::

rag = (Windlass.rag()
       .llm("openai", model="gpt-4o-mini", temperature=0)
       .embedding("huggingface", model="BAAI/bge-small-en-v1.5")
       .chunker("recursive", chunk_size=800, overlap=120)
       .vectordb("faiss", persist_path="./index")
       .retriever("hybrid", weights=[0.6, 0.4])
       .top_k(8))

Level 3 — reach past the abstraction::

index = rag.native_store()      # the raw faiss.Index
client = rag.native_llm()       # the raw openai.AsyncOpenAI

Source code in src\windlass\rag\builder.py

def __init__(self, container: Container | None = None) -> None:
    self.container = container or root_container().scope()
    self._specs: dict[str, tuple[Any, dict[str, Any]]] = {}
    self._options: dict[str, Any] = {
        "prompt": DEFAULT_PROMPT,
        "top_k": 5,
        "max_context_tokens": 6000,
        "include_sources": True,
        "answer_without_context": False,
        "batch_size": 64,
    }
    self._score_threshold: float | None = None
    self._pipeline: RAGPipeline | None = None

pipeline property

pipeline: RAGPipeline

The built pipeline, constructing it on first access.

llm

llm(spec: Any = None, /, **config: Any) -> Self

Choose the generation model.

Parameters:

Name	Type	Description	Default
spec	Any	Registry name ("openai", "anthropic", "ollama", "fake"), an :class:~windlass.interfaces.llm.LLM instance, or a factory.	None
**config	Any	Provider options — model, temperature, max_tokens, api_key, base_url, ...	{}

Returns:

Type	Description
Self	self.

Example

from windlass import Windlass _ = Windlass.rag().llm("fake", responses=["hi"])

Source code in src\windlass\rag\builder.py

def llm(self, spec: Any = None, /, **config: Any) -> Self:
    """Choose the generation model.

    Args:
        spec: Registry name (``"openai"``, ``"anthropic"``, ``"ollama"``,
            ``"fake"``), an :class:`~windlass.interfaces.llm.LLM` instance, or
            a factory.
        **config: Provider options — ``model``, ``temperature``,
            ``max_tokens``, ``api_key``, ``base_url``, ...

    Returns:
        ``self``.

    Example:
        >>> from windlass import Windlass
        >>> _ = Windlass.rag().llm("fake", responses=["hi"])
    """
    return self._set("llm", spec, config)

embedding

embedding(spec: Any = None, /, **config: Any) -> Self

Choose the embedding model.

The same model must embed your corpus and your queries — Windlass uses this one for both, which is why swapping it means re-ingesting.

Parameters:

Name	Type	Description	Default
spec	Any	Registry name ("huggingface", "openai", "hash"), an :class:~windlass.interfaces.embedding.Embedder, or a factory.	None
**config	Any	Provider options — model, batch_size, device, ...	{}

Returns:

Type	Description
Self	self.

Source code in src\windlass\rag\builder.py

def embedding(self, spec: Any = None, /, **config: Any) -> Self:
    """Choose the embedding model.

    The same model must embed your corpus and your queries — Windlass uses
    this one for both, which is why swapping it means re-ingesting.

    Args:
        spec: Registry name (``"huggingface"``, ``"openai"``, ``"hash"``),
            an :class:`~windlass.interfaces.embedding.Embedder`, or a factory.
        **config: Provider options — ``model``, ``batch_size``, ``device``, ...

    Returns:
        ``self``.
    """
    return self._set("embedding", spec, config)

vectordb

vectordb(spec: Any = None, /, **config: Any) -> Self

Choose the vector store.

Parameters:

Name	Type	Description	Default
spec	Any	Registry name ("memory", "faiss", "chroma", "pinecone"), a :class:~windlass.interfaces.vectordb.VectorStore, or a factory.	None
**config	Any	Store options — collection, persist_path, metric, namespace, ... dimensions is filled in from the embedding model when you omit it.	{}

Returns:

Type	Description
Self	self.

Source code in src\windlass\rag\builder.py

def vectordb(self, spec: Any = None, /, **config: Any) -> Self:
    """Choose the vector store.

    Args:
        spec: Registry name (``"memory"``, ``"faiss"``, ``"chroma"``,
            ``"pinecone"``), a :class:`~windlass.interfaces.vectordb.VectorStore`,
            or a factory.
        **config: Store options — ``collection``, ``persist_path``,
            ``metric``, ``namespace``, ... ``dimensions`` is filled in from
            the embedding model when you omit it.

    Returns:
        ``self``.
    """
    return self._set("vectordb", spec, config)

chunker

chunker(spec: Any = None, /, **config: Any) -> Self

Choose the chunking strategy.

The highest-leverage knob in a RAG system: it decides what the retriever can find and what the model gets to read.

Parameters:

Name	Type	Description	Default
spec	Any	Registry name ("recursive", "semantic", "token", "markdown", "code", "parent_child"), a :class:~windlass.interfaces.chunker.Chunker, or a factory.	None
**config	Any	Strategy options — chunk_size, overlap, threshold, parent_size, ...	{}

Returns:

Type	Description
Self	self.

Example

from windlass import Windlass _ = Windlass.rag().chunker("recursive", chunk_size=1000, overlap=200)

Source code in src\windlass\rag\builder.py

def chunker(self, spec: Any = None, /, **config: Any) -> Self:
    """Choose the chunking strategy.

    The highest-leverage knob in a RAG system: it decides what the retriever
    can find and what the model gets to read.

    Args:
        spec: Registry name (``"recursive"``, ``"semantic"``, ``"token"``,
            ``"markdown"``, ``"code"``, ``"parent_child"``), a
            :class:`~windlass.interfaces.chunker.Chunker`, or a factory.
        **config: Strategy options — ``chunk_size``, ``overlap``,
            ``threshold``, ``parent_size``, ...

    Returns:
        ``self``.

    Example:
        >>> from windlass import Windlass
        >>> _ = Windlass.rag().chunker("recursive", chunk_size=1000, overlap=200)
    """
    return self._set("chunker", spec, config)

retriever

retriever(spec: Any = None, /, **config: Any) -> Self

Choose the retrieval strategy.

Parameters:

Name	Type	Description	Default
spec	Any	Registry name ("vector", "bm25", "hybrid", "contextual"), a :class:~windlass.interfaces.retriever.Retriever, or a factory.	None
**config	Any	Strategy options — top_k, diversity, weights, rerank, ...	{}

Returns:

Type	Description
Self	self.

Note

"hybrid" builds a dense leg plus a BM25 leg over the same corpus automatically. Pass retrievers=[...] to fuse your own set.

Source code in src\windlass\rag\builder.py

def retriever(self, spec: Any = None, /, **config: Any) -> Self:
    """Choose the retrieval strategy.

    Args:
        spec: Registry name (``"vector"``, ``"bm25"``, ``"hybrid"``,
            ``"contextual"``), a :class:`~windlass.interfaces.retriever.Retriever`,
            or a factory.
        **config: Strategy options — ``top_k``, ``diversity``, ``weights``,
            ``rerank``, ...

    Returns:
        ``self``.

    Note:
        ``"hybrid"`` builds a dense leg plus a BM25 leg over the same corpus
        automatically. Pass ``retrievers=[...]`` to fuse your own set.
    """
    return self._set("retriever", spec, config)

loader

loader(spec: Any = None, /, **config: Any) -> Self

Pin a document loader.

Parameters:

Name	Type	Description	Default
spec	Any	Registry name ("pdf", "docx", "web", ...), a :class:~windlass.interfaces.loader.Loader, or a factory.	None
**config	Any	Loader options.	{}

Returns:

Type	Description
Self	self.

Note

Optional. With no loader configured, Windlass picks one per file from the extension, which is what you want for a mixed corpus.

Source code in src\windlass\rag\builder.py

def loader(self, spec: Any = None, /, **config: Any) -> Self:
    """Pin a document loader.

    Args:
        spec: Registry name (``"pdf"``, ``"docx"``, ``"web"``, ...), a
            :class:`~windlass.interfaces.loader.Loader`, or a factory.
        **config: Loader options.

    Returns:
        ``self``.

    Note:
        Optional. With no loader configured, Windlass picks one per file from
        the extension, which is what you want for a mixed corpus.
    """
    return self._set("loader", spec, config)

preprocessor

preprocessor(spec: Any = None, /, **config: Any) -> Self

Add a preprocessing stage.

Called repeatedly, stages compose into a chain in call order.

Parameters:

Name	Type	Description	Default
spec	Any	Registry name ("clean", "pii", "dedup", "language", "metadata", "tables", "ocr"), a :class:~windlass.interfaces.preprocessor.Preprocessor, or a factory. Omit it for the default cleaner.	None
**config	Any	Preprocessor options.	{}

Returns:

Type	Description
Self	self.

Example

from windlass import Windlass _ = (Windlass.rag() ... .preprocessor("clean", min_length=50) ... .preprocessor("dedup", threshold=0.95) ... .preprocessor("pii", action="redact"))

Source code in src\windlass\rag\builder.py

def preprocessor(self, spec: Any = None, /, **config: Any) -> Self:
    """Add a preprocessing stage.

    Called repeatedly, stages compose into a chain in call order.

    Args:
        spec: Registry name (``"clean"``, ``"pii"``, ``"dedup"``,
            ``"language"``, ``"metadata"``, ``"tables"``, ``"ocr"``), a
            :class:`~windlass.interfaces.preprocessor.Preprocessor`, or a
            factory. Omit it for the default cleaner.
        **config: Preprocessor options.

    Returns:
        ``self``.

    Example:
        >>> from windlass import Windlass
        >>> _ = (Windlass.rag()
        ...      .preprocessor("clean", min_length=50)
        ...      .preprocessor("dedup", threshold=0.95)
        ...      .preprocessor("pii", action="redact"))
    """
    chain = self._specs.setdefault("preprocessor_chain", ([], {}))[0]
    chain.append((spec if spec is not None else "clean", config))
    return self

memory

memory(spec: Any = None, /, **config: Any) -> Self

Attach conversation memory, making ask multi-turn aware.

Parameters:

Name	Type	Description	Default
spec	Any	Registry name ("window", "buffer", "summary", "vector"), a :class:~windlass.interfaces.memory.Memory, or a factory. Omit it for a sliding window.	None
**config	Any	Memory options.	{}

Returns:

Type	Description
Self	self.

Source code in src\windlass\rag\builder.py

def memory(self, spec: Any = None, /, **config: Any) -> Self:
    """Attach conversation memory, making ``ask`` multi-turn aware.

    Args:
        spec: Registry name (``"window"``, ``"buffer"``, ``"summary"``,
            ``"vector"``), a :class:`~windlass.interfaces.memory.Memory`, or a
            factory. Omit it for a sliding window.
        **config: Memory options.

    Returns:
        ``self``.
    """
    return self._set("memory", spec if spec is not None else "window", config)

guardrails

guardrails(spec: Any = None, /, **config: Any) -> Self

Enable input and output guardrails.

Parameters:

Name	Type	Description	Default
spec	Any	Registry name ("rules", "nemo"), a :class:~windlass.interfaces.guardrail.Guardrail, or a factory. Omit it for the dependency-free rule guardrail.	None
**config	Any	Policy options — pii, injection, banned_words, on_violation, ...	{}

Returns:

Type	Description
Self	self.

Example

from windlass import Windlass _ = Windlass.rag().guardrails(pii=True, on_violation="redact")

Source code in src\windlass\rag\builder.py

def guardrails(self, spec: Any = None, /, **config: Any) -> Self:
    """Enable input and output guardrails.

    Args:
        spec: Registry name (``"rules"``, ``"nemo"``), a
            :class:`~windlass.interfaces.guardrail.Guardrail`, or a factory.
            Omit it for the dependency-free rule guardrail.
        **config: Policy options — ``pii``, ``injection``, ``banned_words``,
            ``on_violation``, ...

    Returns:
        ``self``.

    Example:
        >>> from windlass import Windlass
        >>> _ = Windlass.rag().guardrails(pii=True, on_violation="redact")
    """
    return self._set("guardrail", spec if spec is not None else "rules", config)

observe

observe(spec: Any = None, /, **config: Any) -> Self

Enable tracing.

Parameters:

Name	Type	Description	Default
spec	Any	Registry name ("console", "langfuse", "langsmith", "memory"), a :class:~windlass.interfaces.tracer.Tracer, or a factory. Omit it for console tracing.	None
**config	Any	Tracer options — project, tags, show_io, ...	{}

Returns:

Type	Description
Self	self.

Source code in src\windlass\rag\builder.py

def observe(self, spec: Any = None, /, **config: Any) -> Self:
    """Enable tracing.

    Args:
        spec: Registry name (``"console"``, ``"langfuse"``, ``"langsmith"``,
            ``"memory"``), a :class:`~windlass.interfaces.tracer.Tracer`, or a
            factory. Omit it for console tracing.
        **config: Tracer options — ``project``, ``tags``, ``show_io``, ...

    Returns:
        ``self``.
    """
    return self._set("tracer", spec if spec is not None else "console", config)

prompt

prompt(template: str) -> Self

Override the answer prompt template.

Parameters:

Name	Type	Description	Default
template	str	A format string containing {context} and {question}.	required

Returns:

Type	Description
Self	self.

Raises:

Type	Description
ConfigurationError	When a placeholder is missing.

Source code in src\windlass\rag\builder.py

def prompt(self, template: str) -> Self:
    """Override the answer prompt template.

    Args:
        template: A format string containing ``{context}`` and ``{question}``.

    Returns:
        ``self``.

    Raises:
        ConfigurationError: When a placeholder is missing.
    """
    if "{context}" not in template or "{question}" not in template:
        raise ConfigurationError("The prompt template must contain {context} and {question}.")
    self._options["prompt"] = template
    return self._invalidate()

top_k

top_k(k: int) -> Self

Set how many chunks to retrieve per question.

Parameters:

Name	Type	Description	Default
k	int	Number of chunks. Must be positive.	required

Returns:

Type	Description
Self	self.

Raises:

Type	Description
ValueError	If k is not positive.

Source code in src\windlass\rag\builder.py

def top_k(self, k: int) -> Self:
    """Set how many chunks to retrieve per question.

    Args:
        k: Number of chunks. Must be positive.

    Returns:
        ``self``.

    Raises:
        ValueError: If ``k`` is not positive.
    """
    if k <= 0:
        raise ValueError("top_k must be positive")
    self._options["top_k"] = k
    return self._invalidate()

max_context_tokens

max_context_tokens(tokens: int) -> Self

Cap how much retrieved context reaches the prompt.

Parameters:

Name	Type	Description	Default
tokens	int	Token ceiling. Lower-ranked chunks are dropped first.	required

Returns:

Type	Description
Self	self.

Source code in src\windlass\rag\builder.py

def max_context_tokens(self, tokens: int) -> Self:
    """Cap how much retrieved context reaches the prompt.

    Args:
        tokens: Token ceiling. Lower-ranked chunks are dropped first.

    Returns:
        ``self``.
    """
    self._options["max_context_tokens"] = tokens
    return self._invalidate()

strict

strict(enabled: bool = True) -> Self

Refuse to answer when retrieval finds nothing.

Removes the model's opportunity to fill an empty context with plausible invention, which is more effective than asking it not to.

Parameters:

Name	Type	Description	Default
enabled	bool	True to refuse, False to answer from parametric knowledge.	True

Returns:

Type	Description
Self	self.

Note

Dense retrieval returns the nearest vectors, however far away they are — so it is almost never empty, and strict mode alone will rarely fire. Pair it with :meth:min_score to make "found nothing relevant" mean what you expect:

Windlass.rag().strict().min_score(0.25)

Source code in src\windlass\rag\builder.py

def strict(self, enabled: bool = True) -> Self:
    """Refuse to answer when retrieval finds nothing.

    Removes the model's opportunity to fill an empty context with plausible
    invention, which is more effective than asking it not to.

    Args:
        enabled: True to refuse, False to answer from parametric knowledge.

    Returns:
        ``self``.

    Note:
        Dense retrieval returns the *nearest* vectors, however far away they
        are — so it is almost never empty, and strict mode alone will rarely
        fire. Pair it with :meth:`min_score` to make "found nothing relevant"
        mean what you expect:

        ```python
        Windlass.rag().strict().min_score(0.25)
        ```
    """
    self._options["answer_without_context"] = not enabled
    return self._invalidate()

min_score

min_score(threshold: float | None) -> Self

Discard retrieval hits scoring below threshold.

This is what turns strict mode into a real relevance gate. Without it, dense retrieval hands back its k nearest neighbours no matter how unrelated they are, and the model is asked to answer from noise.

Parameters:

Name	Type	Description	Default
threshold	float | None	Minimum score to keep, or None to keep everything. The right value depends on the retriever: cosine similarity is bounded in [-1, 1] (0.2-0.4 is a common floor), BM25 is unbounded, and hybrid fusion produces small reciprocal-rank scores. Measure on your own corpus rather than guessing.	required

Returns:

Type	Description
Self	self.

Example

from windlass import Windlass rag = Windlass.rag().retriever("vector").strict().min_score(0.3) rag.ingest_text("Paris is the capital of France.") 1 rag.ask("Explain quantum chromodynamics.").metadata["no_context"] True

Source code in src\windlass\rag\builder.py

def min_score(self, threshold: float | None) -> Self:
    """Discard retrieval hits scoring below ``threshold``.

    This is what turns strict mode into a real relevance gate. Without it,
    dense retrieval hands back its ``k`` nearest neighbours no matter how
    unrelated they are, and the model is asked to answer from noise.

    Args:
        threshold: Minimum score to keep, or ``None`` to keep everything.
            The right value depends on the retriever: cosine similarity is
            bounded in ``[-1, 1]`` (0.2-0.4 is a common floor), BM25 is
            unbounded, and hybrid fusion produces small reciprocal-rank
            scores. Measure on your own corpus rather than guessing.

    Returns:
        ``self``.

    Example:
        >>> from windlass import Windlass
        >>> rag = Windlass.rag().retriever("vector").strict().min_score(0.3)
        >>> rag.ingest_text("Paris is the capital of France.")
        1
        >>> rag.ask("Explain quantum chromodynamics.").metadata["no_context"]
        True
    """
    self._score_threshold = threshold
    return self._invalidate()

batch_size

batch_size(size: int) -> Self

Set the ingestion batch size.

Parameters:

Name	Type	Description	Default
size	int	Chunks embedded and indexed per batch.	required

Returns:

Type	Description
Self	self.

Source code in src\windlass\rag\builder.py

def batch_size(self, size: int) -> Self:
    """Set the ingestion batch size.

    Args:
        size: Chunks embedded and indexed per batch.

    Returns:
        ``self``.
    """
    self._options["batch_size"] = size
    return self._invalidate()

bind

bind(key: str, factory: Any) -> Self

Bind an arbitrary dependency into this builder's container.

The escape hatch for wiring something Windlass has no named slot for.

Parameters:

Name	Type	Description	Default
key	str	Binding key.	required
factory	Any	A factory callable, or a ready-made instance.	required

Returns:

Type	Description
Self	self.

Source code in src\windlass\rag\builder.py

def bind(self, key: str, factory: Any) -> Self:
    """Bind an arbitrary dependency into this builder's container.

    The escape hatch for wiring something Windlass has no named slot for.

    Args:
        key: Binding key.
        factory: A factory callable, or a ready-made instance.

    Returns:
        ``self``.
    """
    if callable(factory):
        self.container.bind(key, factory)
    else:
        self.container.bind_instance(key, factory)
    return self._invalidate()

build

build() -> RAGPipeline

Resolve every component and construct the pipeline.

Called automatically on first use. Call it explicitly when you want construction errors (a missing API key, an uninstalled extra) to surface at startup rather than on the first request.

Returns:

Type	Description
RAGPipeline	The assembled :class:~windlass.rag.pipeline.RAGPipeline.

Raises:

Type	Description
ConfigurationError	When a component cannot be constructed.
MissingDependencyError	When a chosen provider's extra is not installed.

Example

from windlass import Windlass pipeline = Windlass.rag().build() pipeline.top_k 5

Source code in src\windlass\rag\builder.py

def build(self) -> RAGPipeline:
    """Resolve every component and construct the pipeline.

    Called automatically on first use. Call it explicitly when you want
    construction errors (a missing API key, an uninstalled extra) to surface
    at startup rather than on the first request.

    Returns:
        The assembled :class:`~windlass.rag.pipeline.RAGPipeline`.

    Raises:
        ConfigurationError: When a component cannot be constructed.
        MissingDependencyError: When a chosen provider's extra is not installed.

    Example:
        >>> from windlass import Windlass
        >>> pipeline = Windlass.rag().build()
        >>> pipeline.top_k
        5
    """
    if self._pipeline is not None:
        return self._pipeline

    from windlass.core.config import settings

    cfg = settings()

    llm = self._resolve("llm", cfg.default_llm)
    embedder = self._resolve("embedding", cfg.default_embedding)
    vectorstore = self._resolve_store(embedder, cfg.default_vectordb)
    chunker = self._resolve_chunker(embedder, cfg.default_chunker)
    retriever = self._resolve_retriever(
        embedder, vectorstore, llm, chunker, cfg.default_retriever
    )

    self._pipeline = RAGPipeline(
        llm=llm,
        embedder=embedder,
        vectorstore=vectorstore,
        retriever=retriever,
        chunker=chunker,
        loader=self._resolve_optional("loader"),
        preprocessor=self._resolve_preprocessors(llm),
        guardrail=self._resolve_optional("guardrail"),
        memory=self._resolve_memory(embedder),
        tracer=self._resolve_optional("tracer"),
        **self._options,
    )
    return self._pipeline

ingest

ingest(source: Any, **kwargs: Any) -> int

Ingest a source. See :meth:~windlass.rag.pipeline.RAGPipeline.ingest.

Source code in src\windlass\rag\builder.py

def ingest(self, source: Any, **kwargs: Any) -> int:
    """Ingest a source. See :meth:`~windlass.rag.pipeline.RAGPipeline.ingest`."""
    return self.build().ingest(source, **kwargs)

aingest async

aingest(source: Any, **kwargs: Any) -> int

Async :meth:ingest.

Source code in src\windlass\rag\builder.py

async def aingest(self, source: Any, **kwargs: Any) -> int:
    """Async :meth:`ingest`."""
    return await self.build().aingest(source, **kwargs)

ingest_text

ingest_text(text: str, **kwargs: Any) -> int

Ingest a raw string. See :meth:~windlass.rag.pipeline.RAGPipeline.ingest_text.

Source code in src\windlass\rag\builder.py

def ingest_text(self, text: str, **kwargs: Any) -> int:
    """Ingest a raw string. See :meth:`~windlass.rag.pipeline.RAGPipeline.ingest_text`."""
    return self.build().ingest_text(text, **kwargs)

aingest_text async

aingest_text(text: str, **kwargs: Any) -> int

Async :meth:ingest_text.

Source code in src\windlass\rag\builder.py

async def aingest_text(self, text: str, **kwargs: Any) -> int:
    """Async :meth:`ingest_text`."""
    return await self.build().aingest_text(text, **kwargs)

ingest_documents

ingest_documents(documents: Sequence[Any]) -> int

Ingest pre-loaded documents.

Source code in src\windlass\rag\builder.py

def ingest_documents(self, documents: Sequence[Any]) -> int:
    """Ingest pre-loaded documents."""
    return self.build().ingest_documents(documents)

ask

ask(question: str, **kwargs: Any) -> RAGAnswer

Answer a question. See :meth:~windlass.rag.pipeline.RAGPipeline.ask.

Source code in src\windlass\rag\builder.py

def ask(self, question: str, **kwargs: Any) -> RAGAnswer:
    """Answer a question. See :meth:`~windlass.rag.pipeline.RAGPipeline.ask`."""
    return self.build().ask(question, **kwargs)

aask async

aask(question: str, **kwargs: Any) -> RAGAnswer

Async :meth:ask.

Source code in src\windlass\rag\builder.py

async def aask(self, question: str, **kwargs: Any) -> RAGAnswer:
    """Async :meth:`ask`."""
    return await self.build().aask(question, **kwargs)

stream_ask

stream_ask(question: str, **kwargs: Any) -> Iterator[str]

Stream an answer. See :meth:~windlass.rag.pipeline.RAGPipeline.stream_ask.

Source code in src\windlass\rag\builder.py

def stream_ask(self, question: str, **kwargs: Any) -> Iterator[str]:
    """Stream an answer. See :meth:`~windlass.rag.pipeline.RAGPipeline.stream_ask`."""
    return self.build().stream_ask(question, **kwargs)

astream_ask

astream_ask(question: str, **kwargs: Any) -> AsyncIterator[str]

Async :meth:stream_ask.

Source code in src\windlass\rag\builder.py

def astream_ask(self, question: str, **kwargs: Any) -> AsyncIterator[str]:
    """Async :meth:`stream_ask`."""
    return self.build().astream_ask(question, **kwargs)

search

search(
    query: str, k: int | None = None, *, filters: MetadataFilter | None = None
) -> SearchResult

Retrieve without generating. See :meth:~windlass.rag.pipeline.RAGPipeline.search.

Source code in src\windlass\rag\builder.py

def search(
    self, query: str, k: int | None = None, *, filters: MetadataFilter | None = None
) -> SearchResult:
    """Retrieve without generating. See :meth:`~windlass.rag.pipeline.RAGPipeline.search`."""
    return self.build().search(query, k, filters=filters)

asearch async

asearch(
    query: str, k: int | None = None, *, filters: MetadataFilter | None = None
) -> SearchResult

Async :meth:search.

Source code in src\windlass\rag\builder.py

async def asearch(
    self, query: str, k: int | None = None, *, filters: MetadataFilter | None = None
) -> SearchResult:
    """Async :meth:`search`."""
    return await self.build().asearch(query, k, filters=filters)

evaluate

evaluate(questions: Sequence[Any], **kwargs: Any) -> Any

Evaluate the pipeline. See :meth:~windlass.rag.pipeline.RAGPipeline.evaluate.

Source code in src\windlass\rag\builder.py

def evaluate(self, questions: Sequence[Any], **kwargs: Any) -> Any:
    """Evaluate the pipeline. See :meth:`~windlass.rag.pipeline.RAGPipeline.evaluate`."""
    return self.build().evaluate(questions, **kwargs)

aevaluate async

aevaluate(questions: Sequence[Any], **kwargs: Any) -> Any

Async :meth:evaluate.

Source code in src\windlass\rag\builder.py

async def aevaluate(self, questions: Sequence[Any], **kwargs: Any) -> Any:
    """Async :meth:`evaluate`."""
    return await self.build().aevaluate(questions, **kwargs)

count

count() -> int

Return how many chunks are indexed.

Source code in src\windlass\rag\builder.py

def count(self) -> int:
    """Return how many chunks are indexed."""
    return self.build().count()

save

save(directory: str) -> Any

Persist the index. See :meth:~windlass.rag.pipeline.RAGPipeline.save.

Source code in src\windlass\rag\builder.py

def save(self, directory: str) -> Any:
    """Persist the index. See :meth:`~windlass.rag.pipeline.RAGPipeline.save`."""
    return self.build().save(directory)

load

load(directory: str) -> Self

Restore a persisted index and return self.

Source code in src\windlass\rag\builder.py

def load(self, directory: str) -> Self:
    """Restore a persisted index and return ``self``."""
    self.build().load(directory)
    return self

close

close() -> None

Release every component's resources.

Source code in src\windlass\rag\builder.py

def close(self) -> None:
    """Release every component's resources."""
    if self._pipeline is not None:
        self._pipeline.close()

native_llm

native_llm() -> Any

Return the model provider's own client.

Source code in src\windlass\rag\builder.py

def native_llm(self) -> Any:
    """Return the model provider's own client."""
    return self.build().native_llm()

native_store

native_store() -> Any

Return the vector store's own handle.

Source code in src\windlass\rag\builder.py

def native_store(self) -> Any:
    """Return the vector store's own handle."""
    return self.build().native_store()

native_retriever

native_retriever() -> Any

Return the retriever's underlying object.

Source code in src\windlass\rag\builder.py

def native_retriever(self) -> Any:
    """Return the retriever's underlying object."""
    return self.build().native_retriever()

describe

describe() -> dict[str, Any]

Return a JSON-safe description of the wired pipeline.

Source code in src\windlass\rag\builder.py

def describe(self) -> dict[str, Any]:
    """Return a JSON-safe description of the wired pipeline."""
    return self.build().describe()

AutoLoader

AutoLoader(
    *,
    metadata: dict[str, Any] | None = None,
    recursive: bool = True,
    on_error: str = "skip",
    loader_config: dict[str, dict[str, Any]] | None = None,
    **config: Any
)

Bases: Loader

Dispatches each source to the loader that claims it.

This is what makes ingest('./docs') handle a folder of mixed formats. Files with no matching loader are skipped with a warning rather than aborting the run, so one unreadable file cannot lose a large ingestion.

Parameters:

Name	Type	Description	Default
metadata	dict[str, Any] | None	Metadata attached to every produced document.	None
recursive	bool	Walk directories recursively.	True
on_error	str	"skip" (default) or "raise".	'skip'
loader_config	dict[str, dict[str, Any]] | None	Per-loader keyword arguments, keyed by loader name, e.g. {"pdf": {"per_page": False}}.	None
**config	Any	Forwarded to :class:~windlass.interfaces.loader.Loader.	{}

Example

import tempfile, pathlib folder = pathlib.Path(tempfile.mkdtemp()) _ = (folder / "a.txt").write_text("plain") _ = (folder / "b.md").write_text("# heading") sorted(d.metadata["extension"] for d in AutoLoader().load(folder)) ['.md', '.txt']

Source code in src\windlass\rag\loading.py

def __init__(
    self,
    *,
    metadata: dict[str, Any] | None = None,
    recursive: bool = True,
    on_error: str = "skip",
    loader_config: dict[str, dict[str, Any]] | None = None,
    **config: Any,
) -> None:
    super().__init__(metadata=metadata, recursive=recursive, on_error=on_error, **config)
    self.loader_config = dict(loader_config or {})
    self._cache: dict[str, Loader] = {}

can_handle classmethod

can_handle(source: SourceLike) -> bool

Return True — the auto loader accepts anything and routes it.

Source code in src\windlass\rag\loading.py

@classmethod
def can_handle(cls, source: SourceLike) -> bool:
    """Return True — the auto loader accepts anything and routes it."""
    return True

aload_source async

aload_source(source: SourceLike) -> list[Document]

Route one source to its loader and read it.

Parameters:

Name	Type	Description	Default
source	SourceLike	A single path, URL or byte payload.	required

Returns:

Type	Description
list[Document]	The extracted documents, or [] when no loader claims the source
list[Document]	and on_error='skip'.

Raises:

Type	Description
IngestionError	When no loader claims the source and on_error='raise'.

Source code in src\windlass\rag\loading.py

async def aload_source(self, source: SourceLike) -> list[Document]:
    """Route one source to its loader and read it.

    Args:
        source: A single path, URL or byte payload.

    Returns:
        The extracted documents, or ``[]`` when no loader claims the source
        and ``on_error='skip'``.

    Raises:
        IngestionError: When no loader claims the source and
            ``on_error='raise'``.
    """
    try:
        delegate = self._delegate(source)
    except IngestionError:
        if self.on_error == "raise":
            raise
        self._log.debug("No loader for %s; skipping.", source)
        return []
    return await delegate.aload_source(source)

RAGPipeline

RAGPipeline(
    *,
    llm: LLM,
    embedder: Embedder,
    vectorstore: VectorStore,
    retriever: Retriever | None = None,
    chunker: Chunker | None = None,
    loader: Loader | None = None,
    preprocessor: Preprocessor | None = None,
    guardrail: Guardrail | None = None,
    memory: Memory | None = None,
    tracer: Tracer | None = None,
    prompt: str = DEFAULT_PROMPT,
    top_k: int = 5,
    max_context_tokens: int = 6000,
    include_sources: bool = True,
    answer_without_context: bool = False,
    batch_size: int = 64
)

A fully wired retrieval-augmented generation pipeline.

Parameters:

Name	Type	Description	Default
llm	LLM	Model used to generate answers.	required
embedder	Embedder	Model used to embed chunks and queries.	required
vectorstore	VectorStore	Where chunk vectors are stored.	required
retriever	Retriever | None	Retrieval strategy. Defaults to dense search over vectorstore.	None
chunker	Chunker | None	Chunking strategy. Defaults to recursive.	None
loader	Loader | None	Loader used by :meth:aingest. Defaults to format auto-detection.	None
preprocessor	Preprocessor | None	Optional cleaning/enrichment stage.	None
guardrail	Guardrail | None	Optional input/output policy.	None
memory	Memory | None	Optional conversation memory, making ask multi-turn aware.	None
tracer	Tracer | None	Observability backend.	None
prompt	str	Prompt template. Must contain {context} and {question}.	DEFAULT_PROMPT
top_k	int	How many chunks to retrieve per question.	5
max_context_tokens	int	Ceiling on context placed in the prompt. Chunks are dropped from the tail until it fits, so the highest-ranked context always survives.	6000
include_sources	bool	Number each context block and ask the model to cite them.	True
answer_without_context	bool	When False, retrieval finding nothing returns :data:NO_CONTEXT_ANSWER instead of calling the model. This is the single most effective anti-hallucination setting in a RAG system.	False
batch_size	int	Chunks embedded and indexed per batch during ingestion.	64

Attributes:

Name	Type	Description
llm		The generation model.
retriever		The retrieval strategy.
vectorstore		The vector store.

Raises:

Type	Description
ConfigurationError	When a required component is missing or the prompt template lacks its placeholders.

Source code in src\windlass\rag\pipeline.py

def __init__(
    self,
    *,
    llm: LLM,
    embedder: Embedder,
    vectorstore: VectorStore,
    retriever: Retriever | None = None,
    chunker: Chunker | None = None,
    loader: Loader | None = None,
    preprocessor: Preprocessor | None = None,
    guardrail: Guardrail | None = None,
    memory: Memory | None = None,
    tracer: Tracer | None = None,
    prompt: str = DEFAULT_PROMPT,
    top_k: int = 5,
    max_context_tokens: int = 6000,
    include_sources: bool = True,
    answer_without_context: bool = False,
    batch_size: int = 64,
) -> None:
    if llm is None or embedder is None or vectorstore is None:
        raise ConfigurationError(
            "A RAG pipeline needs an llm, an embedder and a vector store.",
            hint="Build it with Windlass.rag(), which supplies working defaults.",
        )
    if "{context}" not in prompt or "{question}" not in prompt:
        raise ConfigurationError(
            "The RAG prompt template must contain {context} and {question}.",
            context={"prompt": prompt[:200]},
        )

    self.llm = llm
    self.embedder = embedder
    self.vectorstore = vectorstore
    self.chunker = chunker
    self.loader = loader
    self.preprocessor = preprocessor
    self.guardrail = guardrail
    self.memory = memory
    self.tracer = tracer or NullTracer()
    self.prompt = prompt
    self.top_k = top_k
    self.max_context_tokens = max_context_tokens
    self.include_sources = include_sources
    self.answer_without_context = answer_without_context
    self.batch_size = batch_size

    if retriever is None:
        from windlass.providers.retrievers.vector import VectorRetriever

        retriever = VectorRetriever(embedder=embedder, vectorstore=vectorstore, top_k=top_k)
    self.retriever = retriever

aingest async

aingest(
    source: Any, *, metadata: dict[str, Any] | None = None, show_progress: bool = False
) -> int

Load, preprocess, chunk, embed and index a source.

Parameters:

Name	Type	Description	Default
source	Any	A path, directory, URL, glob, byte payload, or a sequence of any of those. Directories are walked and each file routed to the right loader by extension.	required
metadata	dict[str, Any] | None	Extra metadata attached to every resulting chunk — a tenant id, a corpus name, an ingestion date.	None
show_progress	bool	Log a line per stage. Useful on long runs.	False

Returns:

Type	Description
int	How many chunks were indexed.

Raises:

Type	Description
IngestionError	When nothing could be loaded, or a stage fails.

Performance

Loading, embedding and indexing are all batched and concurrent. Re-ingesting unchanged content is cheap: chunk ids are content hashes, so the store upserts rather than duplicating, and a configured embedding cache skips the model entirely.

Example

import asyncio, tempfile, pathlib from windlass import Windlass folder = pathlib.Path(tempfile.mkdtemp()) _ = (folder / "a.txt").write_text("Retrieval augmented generation.") asyncio.run(Windlass.rag().aingest(folder)) >= 1 True

Source code in src\windlass\rag\pipeline.py

async def aingest(
    self,
    source: Any,
    *,
    metadata: dict[str, Any] | None = None,
    show_progress: bool = False,
) -> int:
    """Load, preprocess, chunk, embed and index a source.

    Args:
        source: A path, directory, URL, glob, byte payload, or a sequence of
            any of those. Directories are walked and each file routed to the
            right loader by extension.
        metadata: Extra metadata attached to every resulting chunk — a
            tenant id, a corpus name, an ingestion date.
        show_progress: Log a line per stage. Useful on long runs.

    Returns:
        How many chunks were indexed.

    Raises:
        IngestionError: When nothing could be loaded, or a stage fails.

    Performance:
        Loading, embedding and indexing are all batched and concurrent.
        Re-ingesting unchanged content is cheap: chunk ids are content
        hashes, so the store upserts rather than duplicating, and a
        configured embedding cache skips the model entirely.

    Example:
        >>> import asyncio, tempfile, pathlib
        >>> from windlass import Windlass
        >>> folder = pathlib.Path(tempfile.mkdtemp())
        >>> _ = (folder / "a.txt").write_text("Retrieval augmented generation.")
        >>> asyncio.run(Windlass.rag().aingest(folder)) >= 1
        True
    """
    started = time.perf_counter()
    with self.tracer.span("ingest", kind="ingestion", inputs=str(source)) as span:
        documents = await self._load(source, metadata)
        if show_progress:
            _log.info("Loaded %d documents", len(documents))

        documents = await self._preprocess(documents)
        if not documents:
            raise IngestionError(
                "Every document was dropped during preprocessing.",
                hint="Relax the preprocessor — a min_length or language filter "
                "is probably too strict.",
            )
        if show_progress:
            _log.info("Preprocessed to %d documents", len(documents))

        chunks = await self._chunk(documents)
        if show_progress:
            _log.info("Produced %d chunks", len(chunks))

        indexed = await self._index(chunks)
        span.set_output({"documents": len(documents), "chunks": indexed})
        span.set_metadata(elapsed_ms=round((time.perf_counter() - started) * 1000, 1))

    if show_progress:
        _log.info("Indexed %d chunks in %.1fs", indexed, time.perf_counter() - started)
    return indexed

ingest

ingest(
    source: Any, *, metadata: dict[str, Any] | None = None, show_progress: bool = False
) -> int

Blocking :meth:aingest.

Source code in src\windlass\rag\pipeline.py

def ingest(
    self, source: Any, *, metadata: dict[str, Any] | None = None, show_progress: bool = False
) -> int:
    """Blocking :meth:`aingest`."""
    return run_sync(self.aingest(source, metadata=metadata, show_progress=show_progress))

aingest_documents async

aingest_documents(documents: Sequence[Document]) -> int

Ingest already-loaded documents, skipping the loader.

Parameters:

Name	Type	Description	Default
documents	Sequence[Document]	Documents to preprocess, chunk, embed and index.	required

Returns:

Type	Description
int	How many chunks were indexed.

Source code in src\windlass\rag\pipeline.py

async def aingest_documents(self, documents: Sequence[Document]) -> int:
    """Ingest already-loaded documents, skipping the loader.

    Args:
        documents: Documents to preprocess, chunk, embed and index.

    Returns:
        How many chunks were indexed.
    """
    processed = await self._preprocess(list(documents))
    chunks = await self._chunk(processed)
    return await self._index(chunks)

ingest_documents

ingest_documents(documents: Sequence[Document]) -> int

Blocking :meth:aingest_documents.

Source code in src\windlass\rag\pipeline.py

def ingest_documents(self, documents: Sequence[Document]) -> int:
    """Blocking :meth:`aingest_documents`."""
    return run_sync(self.aingest_documents(documents))

aingest_text async

aingest_text(
    text: str, *, source: str | None = None, metadata: dict[str, Any] | None = None
) -> int

Ingest a raw string.

Parameters:

Name	Type	Description	Default
text	str	The text to index.	required
source	str | None	Label recorded as the document's source, used for citation.	None
metadata	dict[str, Any] | None	Extra metadata.	None

Returns:

Type	Description
int	How many chunks were indexed.

Example

import asyncio from windlass import Windlass asyncio.run(Windlass.rag().aingest_text("Some knowledge.")) >= 1 True

Source code in src\windlass\rag\pipeline.py

async def aingest_text(
    self, text: str, *, source: str | None = None, metadata: dict[str, Any] | None = None
) -> int:
    """Ingest a raw string.

    Args:
        text: The text to index.
        source: Label recorded as the document's source, used for citation.
        metadata: Extra metadata.

    Returns:
        How many chunks were indexed.

    Example:
        >>> import asyncio
        >>> from windlass import Windlass
        >>> asyncio.run(Windlass.rag().aingest_text("Some knowledge.")) >= 1
        True
    """
    document = Document(content=text, source=source, metadata=dict(metadata or {}))
    return await self.aingest_documents([document])

ingest_text

ingest_text(
    text: str, *, source: str | None = None, metadata: dict[str, Any] | None = None
) -> int

Blocking :meth:aingest_text.

Source code in src\windlass\rag\pipeline.py

def ingest_text(
    self, text: str, *, source: str | None = None, metadata: dict[str, Any] | None = None
) -> int:
    """Blocking :meth:`aingest_text`."""
    return run_sync(self.aingest_text(text, source=source, metadata=metadata))

asearch async

asearch(
    query: str, k: int | None = None, *, filters: MetadataFilter | None = None
) -> SearchResult

Retrieve without generating an answer.

Useful for building a search UI, for debugging why an answer was wrong, and for evaluating retrieval quality independently of generation.

Parameters:

Name	Type	Description	Default
query	str	The search query.	required
k	int | None	Override for :attr:top_k.	None
filters	MetadataFilter | None	Metadata constraints.	None

Returns:

Type	Description
SearchResult	The ranked results.

Source code in src\windlass\rag\pipeline.py

async def asearch(
    self, query: str, k: int | None = None, *, filters: MetadataFilter | None = None
) -> SearchResult:
    """Retrieve without generating an answer.

    Useful for building a search UI, for debugging why an answer was wrong,
    and for evaluating retrieval quality independently of generation.

    Args:
        query: The search query.
        k: Override for :attr:`top_k`.
        filters: Metadata constraints.

    Returns:
        The ranked results.
    """
    with self.tracer.span("retrieve", kind="retriever", inputs=query) as span:
        result = await self.retriever.aretrieve(query, k or self.top_k, filters=filters)
        span.set_output([h.chunk.content[:200] for h in result.hits])
        span.set_metadata(hits=len(result.hits))
    return result

search

search(
    query: str, k: int | None = None, *, filters: MetadataFilter | None = None
) -> SearchResult

Blocking :meth:asearch.

Source code in src\windlass\rag\pipeline.py

def search(
    self, query: str, k: int | None = None, *, filters: MetadataFilter | None = None
) -> SearchResult:
    """Blocking :meth:`asearch`."""
    return run_sync(self.asearch(query, k, filters=filters))

aask async

aask(
    question: str,
    *,
    k: int | None = None,
    filters: MetadataFilter | None = None,
    thread_id: str = "default",
    **llm_kwargs: Any
) -> RAGAnswer

Answer a question from the indexed corpus.

The full path: input guardrail, retrieval, context assembly, generation, output guardrail, memory update.

Parameters:

Name	Type	Description	Default
question	str	The user's question.	required
k	int | None	Override for :attr:top_k.	None
filters	MetadataFilter | None	Metadata constraints on retrieval.	None
thread_id	str	Conversation thread, when memory is configured.	'default'
**llm_kwargs	Any	Per-call model overrides (temperature, ...).	{}

Returns:

Type	Description
RAGAnswer	The answer, the contexts it was based on, token usage and latency.

Raises:

Type	Description
GuardrailViolation	When a guardrail blocks the question or answer.
PipelineError	When generation fails.

Performance

One retrieval (embedding + store query) plus one model call. Context is truncated to :attr:max_context_tokens from the tail, so a large top_k never blows the context window.

Example

import asyncio from windlass import Windlass rag = Windlass.rag() _ = rag.ingest_text("The Eiffel Tower is in Paris.") asyncio.run(rag.aask("Where is the Eiffel Tower?")).question 'Where is the Eiffel Tower?'

Source code in src\windlass\rag\pipeline.py

async def aask(
    self,
    question: str,
    *,
    k: int | None = None,
    filters: MetadataFilter | None = None,
    thread_id: str = "default",
    **llm_kwargs: Any,
) -> RAGAnswer:
    """Answer a question from the indexed corpus.

    The full path: input guardrail, retrieval, context assembly, generation,
    output guardrail, memory update.

    Args:
        question: The user's question.
        k: Override for :attr:`top_k`.
        filters: Metadata constraints on retrieval.
        thread_id: Conversation thread, when memory is configured.
        **llm_kwargs: Per-call model overrides (``temperature``, ...).

    Returns:
        The answer, the contexts it was based on, token usage and latency.

    Raises:
        GuardrailViolation: When a guardrail blocks the question or answer.
        PipelineError: When generation fails.

    Performance:
        One retrieval (embedding + store query) plus one model call. Context
        is truncated to :attr:`max_context_tokens` from the tail, so a large
        ``top_k`` never blows the context window.

    Example:
        >>> import asyncio
        >>> from windlass import Windlass
        >>> rag = Windlass.rag()
        >>> _ = rag.ingest_text("The Eiffel Tower is in Paris.")
        >>> asyncio.run(rag.aask("Where is the Eiffel Tower?")).question
        'Where is the Eiffel Tower?'
    """
    started = time.perf_counter()
    with self.tracer.span("rag.ask", kind="chain", inputs=question) as span:
        safe_question = await self._guard(question, "input")
        result = await self.asearch(safe_question, k, filters=filters)

        if not result.hits and not self.answer_without_context:
            answer = RAGAnswer(
                answer=NO_CONTEXT_ANSWER,
                question=question,
                latency_ms=(time.perf_counter() - started) * 1000,
                metadata={"no_context": True},
            )
            span.set_output(answer.answer)
            return answer

        context, used = self._build_context(result.hits)
        messages = await self._build_messages(safe_question, context, thread_id)

        with self.tracer.span("generate", kind="llm", inputs=safe_question) as llm_span:
            try:
                completion = await self.llm.acomplete(messages, **llm_kwargs)
            except Exception as exc:
                raise PipelineError(
                    f"Answer generation failed: {exc}",
                    context={"question": question[:200]},
                ) from exc
            llm_span.set_output(completion.content)
            llm_span.set_usage(completion.usage)
            llm_span.set_metadata(model=completion.model)

        safe_answer = await self._guard(completion.content, "output")
        await self._remember(question, safe_answer, thread_id)

        answer = RAGAnswer(
            answer=safe_answer,
            question=question,
            contexts=used,
            usage=completion.usage,
            latency_ms=(time.perf_counter() - started) * 1000,
            metadata={
                "model": completion.model,
                "retrieved": len(result.hits),
                "used_contexts": len(used),
                "retrieval_ms": round(result.latency_ms, 1),
            },
        )
        span.set_output(safe_answer)
        span.set_usage(completion.usage)
    return answer

ask

ask(
    question: str,
    *,
    k: int | None = None,
    filters: MetadataFilter | None = None,
    thread_id: str = "default",
    **llm_kwargs: Any
) -> RAGAnswer

Blocking :meth:aask.

Source code in src\windlass\rag\pipeline.py

def ask(
    self,
    question: str,
    *,
    k: int | None = None,
    filters: MetadataFilter | None = None,
    thread_id: str = "default",
    **llm_kwargs: Any,
) -> RAGAnswer:
    """Blocking :meth:`aask`."""
    return run_sync(
        self.aask(question, k=k, filters=filters, thread_id=thread_id, **llm_kwargs)
    )

astream_ask async

astream_ask(
    question: str,
    *,
    k: int | None = None,
    filters: MetadataFilter | None = None,
    thread_id: str = "default",
    **llm_kwargs: Any
) -> AsyncIterator[str]

Stream an answer token by token.

Retrieval completes first (it must — the context shapes the prompt), then generation streams. Guardrails run on the assembled answer after the stream finishes, so a blocking output policy cannot un-send text that has already reached the user; use redact semantics or non-streaming generation when output blocking must be authoritative.

Parameters:

Name	Type	Description	Default
question	str	The user's question.	required
k	int | None	Override for :attr:top_k.	None
filters	MetadataFilter | None	Metadata constraints.	None
thread_id	str	Conversation thread.	'default'
**llm_kwargs	Any	Per-call model overrides.	{}

Yields:

Type	Description
AsyncIterator[str]	Text fragments as the model produces them.

Source code in src\windlass\rag\pipeline.py

async def astream_ask(
    self,
    question: str,
    *,
    k: int | None = None,
    filters: MetadataFilter | None = None,
    thread_id: str = "default",
    **llm_kwargs: Any,
) -> AsyncIterator[str]:
    """Stream an answer token by token.

    Retrieval completes first (it must — the context shapes the prompt), then
    generation streams. Guardrails run on the assembled answer after the
    stream finishes, so a blocking output policy cannot un-send text that has
    already reached the user; use ``redact`` semantics or non-streaming
    generation when output blocking must be authoritative.

    Args:
        question: The user's question.
        k: Override for :attr:`top_k`.
        filters: Metadata constraints.
        thread_id: Conversation thread.
        **llm_kwargs: Per-call model overrides.

    Yields:
        Text fragments as the model produces them.
    """
    safe_question = await self._guard(question, "input")
    result = await self.asearch(safe_question, k, filters=filters)

    if not result.hits and not self.answer_without_context:
        yield NO_CONTEXT_ANSWER
        return

    context, _ = self._build_context(result.hits)
    messages = await self._build_messages(safe_question, context, thread_id)

    collected: list[str] = []
    async for event in self.llm.astream(messages, **llm_kwargs):
        if event.type == "text" and event.delta:
            collected.append(event.delta)
            yield event.delta

    answer = "".join(collected)
    if self.guardrail is not None:
        await self.guardrail.avalidate(answer, stage="output")
    await self._remember(question, answer, thread_id)

stream_ask

stream_ask(question: str, **kwargs: Any) -> Iterator[str]

Blocking :meth:astream_ask.

Parameters:

Name	Type	Description	Default
question	str	The user's question.	required
**kwargs	Any	Forwarded to :meth:astream_ask.	{}

Yields:

Type	Description
str	Text fragments.

Source code in src\windlass\rag\pipeline.py

def stream_ask(self, question: str, **kwargs: Any) -> Iterator[str]:
    """Blocking :meth:`astream_ask`.

    Args:
        question: The user's question.
        **kwargs: Forwarded to :meth:`astream_ask`.

    Yields:
        Text fragments.
    """
    return iter_sync(self.astream_ask(question, **kwargs))

aevaluate async

aevaluate(
    questions: Sequence[str] | Sequence[dict[str, Any]],
    *,
    evaluator: Any = None,
    metrics: Sequence[str] | None = None
) -> Any

Answer a question set and score the results.

Parameters:

Name	Type	Description	Default
questions	Sequence[str] | Sequence[dict[str, Any]]	Question strings, or dicts with question and an optional reference ground-truth answer.	required
evaluator	Any	An :class:~windlass.interfaces.evaluator.Evaluator. When omitted, the built-in evaluator is used with this pipeline's model as the judge.	None
metrics	Sequence[str] | None	Metric names, passed to the default evaluator.	None

Returns:

Name	Type	Description
An	Any	class:~windlass.core.types.EvaluationReport.

Example

import asyncio from windlass import Windlass rag = Windlass.rag() _ = rag.ingest_text("Paris is the capital of France.") report = asyncio.run(rag.aevaluate( ... [{"question": "Capital of France?", "reference": "Paris"}], ... metrics=["answer_relevancy_lexical"], ... )) report.samples 1

Source code in src\windlass\rag\pipeline.py

async def aevaluate(
    self,
    questions: Sequence[str] | Sequence[dict[str, Any]],
    *,
    evaluator: Any = None,
    metrics: Sequence[str] | None = None,
) -> Any:
    """Answer a question set and score the results.

    Args:
        questions: Question strings, or dicts with ``question`` and an
            optional ``reference`` ground-truth answer.
        evaluator: An :class:`~windlass.interfaces.evaluator.Evaluator`. When
            omitted, the built-in evaluator is used with this pipeline's
            model as the judge.
        metrics: Metric names, passed to the default evaluator.

    Returns:
        An :class:`~windlass.core.types.EvaluationReport`.

    Example:
        >>> import asyncio
        >>> from windlass import Windlass
        >>> rag = Windlass.rag()
        >>> _ = rag.ingest_text("Paris is the capital of France.")
        >>> report = asyncio.run(rag.aevaluate(
        ...     [{"question": "Capital of France?", "reference": "Paris"}],
        ...     metrics=["answer_relevancy_lexical"],
        ... ))
        >>> report.samples
        1
    """
    from windlass.interfaces.evaluator import EvalSample

    if evaluator is None:
        from windlass.providers.evaluation.builtin import BuiltinEvaluator

        evaluator = BuiltinEvaluator(metrics=metrics, llm=self.llm)

    samples: list[EvalSample] = []
    for item in questions:
        if isinstance(item, str):
            question, reference = item, ""
        else:
            question = str(item.get("question", ""))
            reference = str(item.get("reference", "") or item.get("ground_truth", ""))
        answer = await self.aask(question)
        samples.append(EvalSample.from_answer(answer, reference=reference))

    with self.tracer.span("evaluate", kind="evaluation") as span:
        report = await evaluator.aevaluate(samples)
        span.set_output(report.summary)
    return report

evaluate

evaluate(
    questions: Sequence[str] | Sequence[dict[str, Any]],
    *,
    evaluator: Any = None,
    metrics: Sequence[str] | None = None
) -> Any

Blocking :meth:aevaluate.

Source code in src\windlass\rag\pipeline.py

def evaluate(
    self,
    questions: Sequence[str] | Sequence[dict[str, Any]],
    *,
    evaluator: Any = None,
    metrics: Sequence[str] | None = None,
) -> Any:
    """Blocking :meth:`aevaluate`."""
    return run_sync(self.aevaluate(questions, evaluator=evaluator, metrics=metrics))

native_llm

native_llm() -> Any

Return the model provider's own client.

Source code in src\windlass\rag\pipeline.py

def native_llm(self) -> Any:
    """Return the model provider's own client."""
    return self.llm.native()

native_store

native_store() -> Any

Return the vector store's own handle (a FAISS index, a Chroma collection).

Source code in src\windlass\rag\pipeline.py

def native_store(self) -> Any:
    """Return the vector store's own handle (a FAISS index, a Chroma collection)."""
    return self.vectorstore.native()

native_retriever

native_retriever() -> Any

Return the retriever's underlying object.

Source code in src\windlass\rag\pipeline.py

def native_retriever(self) -> Any:
    """Return the retriever's underlying object."""
    return self.retriever.native()

describe

describe() -> dict[str, Any]

Return a JSON-safe description of every wired component.

Useful for logging what a deployment is actually running, and for diffing two environments that behave differently.

Returns:

Type	Description
dict[str, Any]	A dict keyed by component role.

Source code in src\windlass\rag\pipeline.py

def describe(self) -> dict[str, Any]:
    """Return a JSON-safe description of every wired component.

    Useful for logging what a deployment is actually running, and for
    diffing two environments that behave differently.

    Returns:
        A dict keyed by component role.
    """
    parts: dict[str, Any] = {
        "llm": self.llm.describe(),
        "embedder": self.embedder.describe(),
        "vectorstore": self.vectorstore.describe(),
        "retriever": self.retriever.describe(),
        "top_k": self.top_k,
        "max_context_tokens": self.max_context_tokens,
    }
    for label, component in (
        ("chunker", self.chunker),
        ("loader", self.loader),
        ("preprocessor", self.preprocessor),
        ("guardrail", self.guardrail),
        ("memory", self.memory),
        ("tracer", self.tracer),
    ):
        if component is not None:
            parts[label] = component.describe()
    return parts

acount async

acount() -> int

Return how many chunks are indexed.

Reads the vector store, falling back to the retriever's own index for purely lexical pipelines where no vectors are stored.

Source code in src\windlass\rag\pipeline.py

async def acount(self) -> int:
    """Return how many chunks are indexed.

    Reads the vector store, falling back to the retriever's own index for
    purely lexical pipelines where no vectors are stored.
    """
    count = await self.vectorstore.acount()
    if count:
        return count
    own = getattr(self.retriever, "chunks", None)
    return len(own) if own is not None else 0

count

count() -> int

Blocking :meth:acount.

Source code in src\windlass\rag\pipeline.py

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

aclose async

aclose() -> None

Release every component's resources.

Source code in src\windlass\rag\pipeline.py

async def aclose(self) -> None:
    """Release every component's resources."""
    for component in (
        self.llm,
        self.embedder,
        self.vectorstore,
        self.retriever,
        self.tracer,
    ):
        close = getattr(component, "aclose", None)
        if close is not None:
            try:
                await close()
            except Exception as exc:
                _log.debug("Closing %s failed: %s", component, exc)

close

close() -> None

Blocking :meth:aclose.

Source code in src\windlass\rag\pipeline.py

def close(self) -> None:
    """Blocking :meth:`aclose`."""
    run_sync(self.aclose())

save

save(directory: str | Path) -> Path

Persist the index to disk, when the store supports it.

Parameters:

Name	Type	Description	Default
directory	str | Path	Where to write. Created if absent.	required

Returns:

Type	Description
Path	The directory written to.

Raises:

Type	Description
PipelineError	When the configured store cannot persist.

Example

import tempfile from windlass import Windlass rag = Windlass.rag() _ = rag.ingest_text("something worth keeping") rag.save(tempfile.mkdtemp()).is_dir() True

Source code in src\windlass\rag\pipeline.py

def save(self, directory: str | Path) -> Path:
    """Persist the index to disk, when the store supports it.

    Args:
        directory: Where to write. Created if absent.

    Returns:
        The directory written to.

    Raises:
        PipelineError: When the configured store cannot persist.

    Example:
        >>> import tempfile
        >>> from windlass import Windlass
        >>> rag = Windlass.rag()
        >>> _ = rag.ingest_text("something worth keeping")
        >>> rag.save(tempfile.mkdtemp()).is_dir()
        True
    """
    target = Path(directory)
    target.mkdir(parents=True, exist_ok=True)
    saver = getattr(self.vectorstore, "save", None)
    if not callable(saver):
        raise PipelineError(
            f"{type(self.vectorstore).__name__} does not support saving to disk.",
            hint="Chroma and Pinecone persist automatically; use "
            "vectordb('memory', persist_path=...) or 'faiss' for explicit saves.",
        )
    # The in-memory store saves to a single JSON file; FAISS writes a
    # directory of index plus sidecar.
    if self.vectorstore.provider_name == "memory":
        saver(target / f"{self.vectorstore.collection}.json")
    else:
        saver(target)
    return target

load

load(directory: str | Path) -> RAGPipeline

Restore an index previously written by :meth:save.

Parameters:

Name	Type	Description	Default
directory	str | Path	Directory containing the saved index.	required

Returns:

Type	Description
RAGPipeline	self, so the call chains.

Raises:

Type	Description
PipelineError	When the configured store cannot load.

Source code in src\windlass\rag\pipeline.py

def load(self, directory: str | Path) -> RAGPipeline:
    """Restore an index previously written by :meth:`save`.

    Args:
        directory: Directory containing the saved index.

    Returns:
        ``self``, so the call chains.

    Raises:
        PipelineError: When the configured store cannot load.
    """
    source = Path(directory)
    loader = getattr(self.vectorstore, "load", None)
    if not callable(loader):
        raise PipelineError(
            f"{type(self.vectorstore).__name__} does not support loading from disk."
        )
    if self.vectorstore.provider_name == "memory":
        loader(source / f"{self.vectorstore.collection}.json")
    else:
        loader(source)

    # Rebuild any lexical index from the restored corpus.
    rebuild = getattr(self.vectorstore, "all_chunks", None)
    if callable(rebuild) and getattr(self.retriever, "requires_index", False):
        run_sync(self.retriever.aindex(rebuild()))
    return self

loader_for

loader_for(
    source: SourceLike | Sequence[SourceLike],
    *,
    metadata: dict[str, Any] | None = None,
    **config: Any
) -> Loader

Return a loader able to read source.

A heterogeneous sequence (or a directory containing several formats) yields an :class:AutoLoader, which dispatches per file.

Parameters:

Name	Type	Description	Default
source	SourceLike | Sequence[SourceLike]	A path, URL, directory, byte payload, or sequence of them.	required
metadata	dict[str, Any] | None	Metadata attached to every produced document.	None
**config	Any	Forwarded to the chosen loader's constructor.	{}

Returns:

Type	Description
Loader	A ready-to-use loader.

Raises:

Type	Description
IngestionError	When nothing can read the source.

Example

loader_for("report.pdf").provider_name 'pdf' loader_for(b"raw bytes").provider_name 'text'

Source code in src\windlass\rag\loading.py

def loader_for(
    source: SourceLike | Sequence[SourceLike],
    *,
    metadata: dict[str, Any] | None = None,
    **config: Any,
) -> Loader:
    """Return a loader able to read ``source``.

    A heterogeneous sequence (or a directory containing several formats) yields
    an :class:`AutoLoader`, which dispatches per file.

    Args:
        source: A path, URL, directory, byte payload, or sequence of them.
        metadata: Metadata attached to every produced document.
        **config: Forwarded to the chosen loader's constructor.

    Returns:
        A ready-to-use loader.

    Raises:
        IngestionError: When nothing can read the source.

    Example:
        >>> loader_for("report.pdf").provider_name
        'pdf'
        >>> loader_for(b"raw bytes").provider_name
        'text'
    """
    if isinstance(source, Sequence) and not isinstance(source, str | bytes | Path):
        return AutoLoader(metadata=metadata, **config)

    if isinstance(source, bytes):
        return REGISTRY.create("loader", "text", metadata=metadata, **config)

    if isinstance(source, str) and source.startswith(("http://", "https://")):
        name = "youtube" if _is_youtube(source) else "web"
        return REGISTRY.create("loader", name, metadata=metadata, **config)

    path = Path(str(source))
    if path.is_dir() or any(ch in str(path) for ch in "*?["):
        return AutoLoader(metadata=metadata, **config)

    suffix = path.suffix.lower()
    claimed = extension_map().get(suffix)
    if claimed:
        return REGISTRY.create("loader", claimed, metadata=metadata, **config)

    if suffix in _TEXT_FALLBACK:
        return REGISTRY.create("loader", "text", metadata=metadata, **config)

    supported = ", ".join(sorted(extension_map())) or "(no loaders registered)"
    raise IngestionError(
        f"No loader can read {suffix or 'that source'!r}.",
        hint=f"Supported extensions: {supported}\n"
        'Most formats need: pip install "windlass[loaders]"',
        context={"source": str(source), "extension": suffix},
    )