windlass.rag.builder

builder

The fluent RAG builder.

This is the API most people will actually use::

rag = (
    Windlass.rag()
    .loader("pdf")
    .preprocessor()
    .chunker("semantic")
    .embedding("huggingface")
    .vectordb("pinecone")
    .retriever("hybrid")
    .guardrails()
    .observe("langfuse")
)

rag.ingest("./documents")
print(rag.ask("Explain transformers"))

Three properties make it work:

Nothing is constructed until it is needed. Each call records intent; the pipeline is assembled on the first ingest or ask. So you can reconfigure freely, and a builder that names Pinecone does not import Pinecone until you use it.

Cross-component wiring is automatic. The semantic chunker needs the embedder, the vector retriever needs the embedder and the store, hybrid search needs a BM25 leg over the same corpus, FAISS needs the embedding dimensionality. :meth:RAGBuilder.build resolves that graph so you never wire it by hand.

Every argument accepts three forms. A registry name ("semantic"), a configured instance (MyChunker(...)), or a factory. That is the whole extensibility story: your component is indistinguishable from a built-in one.

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()