windlass.interfaces

interfaces

Windlass interfaces — the contracts every component implements.

This package is the architectural heart of the framework. Everything else depends on these abstractions and nothing depends on a concrete provider, which is what the dependency-inversion principle buys us in practice:

• Swapping OpenAI for Ollama is a string change, not a refactor.
• A custom retriever written in a user's repo is indistinguishable from a built-in one.
• The whole framework can be exercised in tests with fakes that implement these same interfaces.

There is exactly one interface per component kind:

============ ========================================================= Kind Interface ============ ========================================================= llm :class:~windlass.interfaces.llm.LLM embedding :class:~windlass.interfaces.embedding.Embedder loader :class:~windlass.interfaces.loader.Loader preprocessor :class:~windlass.interfaces.preprocessor.Preprocessor chunker :class:~windlass.interfaces.chunker.Chunker retriever :class:~windlass.interfaces.retriever.Retriever vectordb :class:~windlass.interfaces.vectordb.VectorStore memory :class:~windlass.interfaces.memory.Memory guardrail :class:~windlass.interfaces.guardrail.Guardrail evaluator :class:~windlass.interfaces.evaluator.Evaluator tracer :class:~windlass.interfaces.tracer.Tracer tool :class:~windlass.interfaces.tool.Tool mcp :class:~windlass.interfaces.mcp.MCPClient ============ =========================================================

Every one of them shares :class:~windlass.interfaces.base.Component, which supplies naming, configuration, lifecycle and the native() escape hatch.

Component

Component(*, name: str | None = None, **config: Any)

Bases: SupportsNative

Abstract base for every registrable Windlass component.

Parameters:

Name	Type	Description	Default
name	str | None	Identifier used in traces and error messages. Defaults to the class's registered name, or its class name.	None
**config	Any	Arbitrary component configuration, kept on :attr:config.	{}

Attributes:

Name	Type	Description
name		The component's identifier.
config	dict[str, Any]	The configuration it was constructed with.

Example

class Upper(Component): ... def apply(self, text: str) -> str: ... return text.upper() Upper(name="upper").name 'upper'

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

def __init__(self, *, name: str | None = None, **config: Any) -> None:
    self.name = name or getattr(type(self), "provider_name", None) or type(self).__name__
    self.config: dict[str, Any] = dict(config)
    self._log = get_logger(f"{self.kind}.{self.name}")

option

option(key: str, default: Any = None) -> Any

Read a configuration value.

Parameters:

Name	Type	Description	Default
key	str	Configuration key.	required
default	Any	Returned when the key is absent.	None

Returns:

Type	Description
Any	The configured value or default.

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

def option(self, key: str, default: Any = None) -> Any:
    """Read a configuration value.

    Args:
        key: Configuration key.
        default: Returned when the key is absent.

    Returns:
        The configured value or ``default``.
    """
    return self.config.get(key, default)

require_option

require_option(key: str) -> Any

Read a configuration value that must be present.

Parameters:

Name	Type	Description	Default
key	str	Configuration key.	required

Returns:

Type	Description
Any	The configured value.

Raises:

Type	Description
ConfigurationError	When the key is missing or None.

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

def require_option(self, key: str) -> Any:
    """Read a configuration value that must be present.

    Args:
        key: Configuration key.

    Returns:
        The configured value.

    Raises:
        ConfigurationError: When the key is missing or ``None``.
    """
    value = self.config.get(key)
    if value is None:
        raise ConfigurationError(
            f"{type(self).__name__} requires the {key!r} option.",
            hint=f"Pass {key}=... when constructing the {self.kind}.",
            context={"component": self.name, "kind": self.kind, "option": key},
        )
    return value

with_config

with_config(**overrides: Any) -> Component

Return a new component of the same type with merged configuration.

Components are treated as immutable once built; reconfiguring produces a copy so a shared component cannot be mutated out from under another pipeline.

Parameters:

Name	Type	Description	Default
**overrides	Any	Configuration to merge over the current values.	{}

Returns:

Type	Description
Component	A new instance of the same class.

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

def with_config(self, **overrides: Any) -> Component:
    """Return a new component of the same type with merged configuration.

    Components are treated as immutable once built; reconfiguring produces a
    copy so a shared component cannot be mutated out from under another
    pipeline.

    Args:
        **overrides: Configuration to merge over the current values.

    Returns:
        A new instance of the same class.
    """
    merged = {**self.config, **overrides}
    return type(self)(name=self.name, **merged)

aclose async

aclose() -> None

Release resources held by the component.

The default is a no-op. Adapters holding sockets, file handles or thread pools should override it. Safe to call more than once.

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

async def aclose(self) -> None:
    """Release resources held by the component.

    The default is a no-op. Adapters holding sockets, file handles or thread
    pools should override it. Safe to call more than once.
    """

close

close() -> None

Blocking counterpart of :meth:aclose.

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

def close(self) -> None:
    """Blocking counterpart of :meth:`aclose`."""
    from windlass.core.concurrency import run_sync

    run_sync(self.aclose())

native

native() -> Any

Return the wrapped provider object.

The default returns self, which is correct for pure-Python components that wrap nothing.

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

def native(self) -> Any:
    """Return the wrapped provider object.

    The default returns ``self``, which is correct for pure-Python
    components that wrap nothing.
    """
    return self

describe

describe() -> dict[str, Any]

Return a JSON-safe summary of this component.

Used by tracing, by Pipeline.describe() and by the CLI.

Returns:

Type	Description
dict[str, Any]	A dict with kind, name, class and config.

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

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

    Used by tracing, by ``Pipeline.describe()`` and by the CLI.

    Returns:
        A dict with ``kind``, ``name``, ``class`` and ``config``.
    """
    return {
        "kind": self.kind,
        "name": self.name,
        "class": f"{type(self).__module__}.{type(self).__qualname__}",
        "config": {k: _safe(k, v) for k, v in self.config.items()},
    }

SupportsNative

Bases: ABC

Mixin declaring the Level 3 escape hatch.

Any adapter wrapping a third-party object should return it from :meth:native so advanced users can reach past Windlass without forking it.

native abstractmethod

native() -> Any

Return the underlying provider object.

Returns:

Type	Description
Any	The wrapped SDK client, index, graph or handle. Adapters with no
Any	third-party object behind them return self.

Example

from windlass.providers.llm.fake import FakeLLM FakeLLM().native() is not None True

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

@abc.abstractmethod
def native(self) -> Any:
    """Return the underlying provider object.

    Returns:
        The wrapped SDK client, index, graph or handle. Adapters with no
        third-party object behind them return ``self``.

    Example:
        >>> from windlass.providers.llm.fake import FakeLLM
        >>> FakeLLM().native() is not None
        True
    """

Chunker

Chunker(
    *,
    chunk_size: int = 1000,
    overlap: int | None = None,
    min_chunk_size: int = 50,
    keep_separator: bool = True,
    name: str | None = None,
    **config: Any
)

Bases: Component

Abstract text splitter.

Parameters:

Name	Type	Description	Default
chunk_size	int	Target size of each chunk, measured in :attr:unit.	1000
overlap	int | None	How much each chunk repeats from the previous one. Overlap keeps a fact that straddles a boundary retrievable from either side. None scales it to :data:DEFAULT_OVERLAP_RATIO of chunk_size, so shrinking chunk_size alone always produces a working chunker.	None
min_chunk_size	int	Chunks shorter than this are merged into their neighbour rather than emitted — stray one-line fragments pollute retrieval.	50
keep_separator	bool	Whether the split separator stays in the output.	True
name	str | None	Component name for traces.	None
**config	Any	Strategy-specific options.	{}

Attributes:

Name	Type	Description
unit	str	"char" or "token" — how chunk_size is measured.

Raises:

Type	Description
ValueError	If an explicit overlap is negative or not smaller than chunk_size. A derived default is never invalid.

Example

Implementing a chunker takes one method::

class LineChunker(Chunker):
    provider_name = "line"

    def split_text(self, text: str) -> list[str]:
        return [ln for ln in text.splitlines() if ln.strip()]

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

def __init__(
    self,
    *,
    chunk_size: int = 1000,
    overlap: int | None = None,
    min_chunk_size: int = 50,
    keep_separator: bool = True,
    name: str | None = None,
    **config: Any,
) -> None:
    if chunk_size <= 0:
        raise ValueError("chunk_size must be positive")

    if overlap is None:
        # Scale with chunk_size. A fixed default would make the perfectly
        # reasonable `chunker(chunk_size=20)` raise, which is a poor trade:
        # the user changed one obvious knob and got an error about another
        # they never set.
        overlap = int(chunk_size * DEFAULT_OVERLAP_RATIO)
    elif overlap < 0:
        raise ValueError("overlap must not be negative")
    elif overlap >= chunk_size:
        raise ValueError(
            f"overlap ({overlap}) must be smaller than chunk_size ({chunk_size}); "
            "otherwise chunking cannot make progress."
        )
    super().__init__(
        name=name or self.provider_name,
        chunk_size=chunk_size,
        overlap=overlap,
        min_chunk_size=min_chunk_size,
        keep_separator=keep_separator,
        **config,
    )
    self.chunk_size = chunk_size
    self.overlap = overlap
    self.min_chunk_size = min_chunk_size
    self.keep_separator = keep_separator

split_text abstractmethod

split_text(text: str) -> list[str]

Split raw text into chunk strings.

The only method a strategy must implement. Return the pieces in reading order; the base class turns them into :class:Chunk objects with ids, offsets and inherited metadata.

Parameters:

Name	Type	Description	Default
text	str	The text to split.	required

Returns:

Type	Description
list[str]	Chunk strings in document order.

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

@abc.abstractmethod
def split_text(self, text: str) -> list[str]:
    """Split raw text into chunk strings.

    The only method a strategy must implement. Return the pieces in reading
    order; the base class turns them into :class:`Chunk` objects with ids,
    offsets and inherited metadata.

    Args:
        text: The text to split.

    Returns:
        Chunk strings in document order.
    """

asplit_text async

asplit_text(text: str) -> list[str]

Async :meth:split_text.

Override this instead when your strategy needs to await something — the semantic chunker calls an embedding model here.

Parameters:

Name	Type	Description	Default
text	str	The text to split.	required

Returns:

Type	Description
list[str]	Chunk strings in document order.

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

async def asplit_text(self, text: str) -> list[str]:
    """Async :meth:`split_text`.

    Override this instead when your strategy needs to await something — the
    semantic chunker calls an embedding model here.

    Args:
        text: The text to split.

    Returns:
        Chunk strings in document order.
    """
    return self.split_text(text)

achunk_document async

achunk_document(document: Document) -> list[Chunk]

Split one document into chunks.

Parameters:

Name	Type	Description	Default
document	Document	The document to split.	required

Returns:

Type	Description
list[Chunk]	Chunks carrying the document's metadata plus chunk_index,
list[Chunk]	chunk_total and character offsets.

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

async def achunk_document(self, document: Document) -> list[Chunk]:
    """Split one document into chunks.

    Args:
        document: The document to split.

    Returns:
        Chunks carrying the document's metadata plus ``chunk_index``,
        ``chunk_total`` and character offsets.
    """
    pieces = await self.asplit_text(document.content)
    pieces = self._merge_undersized(pieces)
    chunks: list[Chunk] = []
    cursor = 0
    for index, piece in enumerate(pieces):
        start = document.content.find(piece, cursor)
        if start < 0:  # strategy rewrote the text (e.g. added a heading path)
            start = cursor
        else:
            cursor = start + len(piece)
        chunks.append(
            Chunk(
                content=piece,
                document_id=document.id,
                index=index,
                start_char=start,
                end_char=start + len(piece),
                metadata={
                    **document.metadata,
                    "document_id": document.id,
                    "chunk_index": index,
                    "chunk_total": len(pieces),
                    "chunker": self.name,
                    **({"source": document.source} if document.source else {}),
                },
            )
        )
    return chunks

achunk async

achunk(
    documents: Sequence[Document] | Document, *, concurrency: int | None = None
) -> list[Chunk]

Split many documents concurrently.

Parameters:

Name	Type	Description	Default
documents	Sequence[Document] | Document	One document or a sequence of them.	required
concurrency	int | None	Maximum simultaneous splits. Defaults to the global max_concurrency setting.	None

Returns:

Type	Description
list[Chunk]	All chunks, grouped by source document in input order.

Performance

CPU-bound strategies see little benefit from concurrency; strategies that await a model (semantic, contextual) see a lot.

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

async def achunk(
    self, documents: Sequence[Document] | Document, *, concurrency: int | None = None
) -> list[Chunk]:
    """Split many documents concurrently.

    Args:
        documents: One document or a sequence of them.
        concurrency: Maximum simultaneous splits. Defaults to the global
            ``max_concurrency`` setting.

    Returns:
        All chunks, grouped by source document in input order.

    Performance:
        CPU-bound strategies see little benefit from concurrency; strategies
        that await a model (``semantic``, ``contextual``) see a lot.
    """
    items = [documents] if isinstance(documents, Document) else list(documents)
    if not items:
        return []
    limit = concurrency or settings().max_concurrency
    grouped = await gather_bounded([self.achunk_document(doc) for doc in items], limit=limit)
    return [chunk for group in grouped for chunk in group]

chunk

chunk(documents: Sequence[Document] | Document) -> list[Chunk]

Blocking :meth:achunk.

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

def chunk(self, documents: Sequence[Document] | Document) -> list[Chunk]:
    """Blocking :meth:`achunk`."""
    return run_sync(self.achunk(documents))

chunk_text

chunk_text(text: str, *, metadata: dict[str, Any] | None = None) -> list[Chunk]

Split a bare string, without constructing a Document first.

Parameters:

Name	Type	Description	Default
text	str	The text to split.	required
metadata	dict[str, Any] | None	Metadata copied onto every chunk.	None

Returns:

Type	Description
list[Chunk]	The resulting chunks.

Example

from windlass.providers.chunkers.recursive import RecursiveChunker RecursiveChunker(chunk_size=100, overlap=10).chunk_text("hello")[0].content 'hello'

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

def chunk_text(self, text: str, *, metadata: dict[str, Any] | None = None) -> list[Chunk]:
    """Split a bare string, without constructing a Document first.

    Args:
        text: The text to split.
        metadata: Metadata copied onto every chunk.

    Returns:
        The resulting chunks.

    Example:
        >>> from windlass.providers.chunkers.recursive import RecursiveChunker
        >>> RecursiveChunker(chunk_size=100, overlap=10).chunk_text("hello")[0].content
        'hello'
    """
    return run_sync(self.achunk(Document(content=text, metadata=dict(metadata or {}))))

Embedder

Embedder(
    model: str = "",
    *,
    dimensions: int | None = None,
    batch_size: int | None = None,
    normalize: bool = True,
    cache: Cache | None = None,
    name: str | None = None,
    **config: Any
)

Bases: Component

Abstract text-embedding model.

Parameters:

Name	Type	Description	Default
model	str	Model identifier passed to the provider.	''
dimensions	int | None	Output dimensionality. Providers that expose a native value should override :meth:dimension.	None
batch_size	int | None	How many texts to send per provider request.	None
normalize	bool	L2-normalise every vector. Recommended: it makes cosine similarity a plain dot product and matches what FAISS expects.	True
cache	Cache | None	Optional cache for embeddings. Embeddings are deterministic, so caching them is always safe.	None
name	str | None	Component name for traces.	None
**config	Any	Provider-specific options.	{}

Attributes:

Name	Type	Description
model		Configured model identifier.
batch_size	int	Configured request batch size.
normalize	bool	Whether vectors are normalised on the way out.

Example

Implementing a provider takes one method::

class MyEmbedder(Embedder):
    provider_name = "mine"

    async def aembed_texts(self, texts, *, kind="document"):
        return [await my_sdk.embed(t) for t in texts]

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

def __init__(
    self,
    model: str = "",
    *,
    dimensions: int | None = None,
    batch_size: int | None = None,
    normalize: bool = True,
    cache: Cache | None = None,
    name: str | None = None,
    **config: Any,
) -> None:
    super().__init__(
        name=name or self.provider_name,
        model=model,
        dimensions=dimensions,
        batch_size=batch_size or settings().batch_size,
        normalize=normalize,
        **config,
    )
    self.model = model or self.default_model()
    self.batch_size: int = self.config["batch_size"]
    self.normalize: bool = normalize
    self._dimensions = dimensions
    # `cache or NullCache()` looks equivalent and is not: every Cache
    # implements __len__, so a freshly constructed (empty) cache is falsy
    # and would be silently discarded — caching would appear to be
    # configured and never happen. Test for None explicitly.
    self._cache: Cache = NullCache() if cache is None else cache

default_model classmethod

default_model() -> str

Return the model used when the caller does not name one.

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

@classmethod
def default_model(cls) -> str:
    """Return the model used when the caller does not name one."""
    return ""

aembed_texts abstractmethod async

aembed_texts(texts: list[str], *, kind: str = 'document') -> list[list[float]]

Embed a batch of texts.

This is the only method a provider must implement. It receives at most :attr:batch_size texts, already prefixed for kind.

Parameters:

Name	Type	Description	Default
texts	list[str]	The texts to embed. Never empty.	required
kind	str	"document" or "query".	'document'

Returns:

Type	Description
list[list[float]]	One vector per input, in the same order.

Raises:

Type	Description
ProviderError	For any provider-side failure.

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

@abc.abstractmethod
async def aembed_texts(self, texts: list[str], *, kind: str = "document") -> list[list[float]]:
    """Embed a batch of texts.

    This is the only method a provider must implement. It receives at most
    :attr:`batch_size` texts, already prefixed for ``kind``.

    Args:
        texts: The texts to embed. Never empty.
        kind: ``"document"`` or ``"query"``.

    Returns:
        One vector per input, in the same order.

    Raises:
        ProviderError: For any provider-side failure.
    """

dimension

dimension() -> int

Return the vector dimensionality.

Determined from configuration when possible, otherwise by embedding a one-character probe and measuring the result. The probe runs at most once per instance.

Returns:

Type	Description
int	The number of components in each vector.

Example

from windlass.providers.embeddings.hash import HashEmbedder HashEmbedder(dimensions=64).dimension() 64

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

def dimension(self) -> int:
    """Return the vector dimensionality.

    Determined from configuration when possible, otherwise by embedding a
    one-character probe and measuring the result. The probe runs at most
    once per instance.

    Returns:
        The number of components in each vector.

    Example:
        >>> from windlass.providers.embeddings.hash import HashEmbedder
        >>> HashEmbedder(dimensions=64).dimension()
        64
    """
    if self._dimensions is None:
        self._dimensions = len(self.embed_one("dimension probe"))
    return self._dimensions

aembed async

aembed(
    texts: Sequence[str], *, kind: str = "document", concurrency: int | None = None
) -> list[list[float]]

Embed many texts with batching, bounded concurrency and caching.

Parameters:

Name	Type	Description	Default
texts	Sequence[str]	The texts to embed.	required
kind	str	"document" for corpus text, "query" for search input.	'document'
concurrency	int | None	Maximum simultaneous provider requests. Defaults to the global max_concurrency setting.	None

Returns:

Type	Description
list[list[float]]	One vector per input, in input order.

Raises:

Type	Description
ProviderError	For any provider-side failure.
ValueError	If kind is not document or query.

Performance

Texts are grouped into :attr:batch_size batches and the batches run concurrently up to concurrency. Cached texts never reach the provider, so re-ingesting a mostly-unchanged corpus is cheap.

Example

import asyncio from windlass.providers.embeddings.hash import HashEmbedder len(asyncio.run(HashEmbedder(dimensions=8).aembed(["a", "b"]))) 2

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

async def aembed(
    self,
    texts: Sequence[str],
    *,
    kind: str = "document",
    concurrency: int | None = None,
) -> list[list[float]]:
    """Embed many texts with batching, bounded concurrency and caching.

    Args:
        texts: The texts to embed.
        kind: ``"document"`` for corpus text, ``"query"`` for search input.
        concurrency: Maximum simultaneous provider requests. Defaults to the
            global ``max_concurrency`` setting.

    Returns:
        One vector per input, in input order.

    Raises:
        ProviderError: For any provider-side failure.
        ValueError: If ``kind`` is not ``document`` or ``query``.

    Performance:
        Texts are grouped into :attr:`batch_size` batches and the batches
        run concurrently up to ``concurrency``. Cached texts never reach the
        provider, so re-ingesting a mostly-unchanged corpus is cheap.

    Example:
        >>> import asyncio
        >>> from windlass.providers.embeddings.hash import HashEmbedder
        >>> len(asyncio.run(HashEmbedder(dimensions=8).aembed(["a", "b"])))
        2
    """
    if kind not in {"document", "query"}:
        raise ValueError(f"kind must be 'document' or 'query', got {kind!r}")
    if not texts:
        return []

    prefix = self.query_prefix if kind == "query" else self.document_prefix
    prepared = [f"{prefix}{t}" if prefix else t for t in texts]

    # Split into cache hits and the misses we actually have to compute.
    results: list[list[float] | None] = [None] * len(prepared)
    pending: list[tuple[int, str]] = []
    for i, text in enumerate(prepared):
        hit = await self._cache.aget(self._key(text, kind))
        if hit is not None:
            results[i] = hit
        else:
            pending.append((i, text))

    if pending:
        batches = list(batched(pending, self.batch_size))
        limit = concurrency or settings().max_concurrency
        computed = await gather_bounded(
            [self._embed_batch([t for _, t in batch], kind) for batch in batches],
            limit=limit,
        )
        for batch, vectors in zip(batches, computed, strict=True):
            for (index, text), vector in zip(batch, vectors, strict=True):
                final = normalize(vector) if self.normalize else list(vector)
                results[index] = final
                await self._cache.aset(self._key(text, kind), final)

    return [r for r in results if r is not None]

embed

embed(texts: Sequence[str], *, kind: str = 'document') -> list[list[float]]

Blocking :meth:aembed.

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

def embed(self, texts: Sequence[str], *, kind: str = "document") -> list[list[float]]:
    """Blocking :meth:`aembed`."""
    return run_sync(self.aembed(texts, kind=kind))

aembed_one async

aembed_one(text: str, *, kind: str = 'document') -> list[float]

Embed a single text.

Parameters:

Name	Type	Description	Default
text	str	The text to embed.	required
kind	str	"document" or "query".	'document'

Returns:

Type	Description
list[float]	One vector.

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

async def aembed_one(self, text: str, *, kind: str = "document") -> list[float]:
    """Embed a single text.

    Args:
        text: The text to embed.
        kind: ``"document"`` or ``"query"``.

    Returns:
        One vector.
    """
    vectors = await self.aembed([text], kind=kind)
    return vectors[0]

embed_one

embed_one(text: str, *, kind: str = 'document') -> list[float]

Blocking :meth:aembed_one.

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

def embed_one(self, text: str, *, kind: str = "document") -> list[float]:
    """Blocking :meth:`aembed_one`."""
    return run_sync(self.aembed_one(text, kind=kind))

aembed_query async

aembed_query(text: str) -> list[float]

Embed a search query, applying any query instruction prefix.

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

async def aembed_query(self, text: str) -> list[float]:
    """Embed a search query, applying any query instruction prefix."""
    return await self.aembed_one(text, kind="query")

embed_query

embed_query(text: str) -> list[float]

Blocking :meth:aembed_query.

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

def embed_query(self, text: str) -> list[float]:
    """Blocking :meth:`aembed_query`."""
    return run_sync(self.aembed_query(text))

set_cache

set_cache(cache: Cache) -> None

Attach a cache to this embedder.

Parameters:

Name	Type	Description	Default
cache	Cache	Any :class:~windlass.core.cache.Cache implementation.	required

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

def set_cache(self, cache: Cache) -> None:
    """Attach a cache to this embedder.

    Args:
        cache: Any :class:`~windlass.core.cache.Cache` implementation.
    """
    self._cache = cache

EvalSample

Bases: WindlassModel

One evaluated interaction.

Attributes:

Name	Type	Description
id	str	Sample identifier, used to correlate results.
question	str	The user's input.
answer	str	What the system produced.
contexts	list[str]	Retrieved texts that were placed in the prompt. Required by faithfulness and context-precision metrics.
reference	str	The ground-truth answer, when you have one.
reference_contexts	list[str]	The ideal contexts, for retrieval recall metrics.
metadata	dict[str, Any]	Anything else worth slicing results by (tenant, model, ...).

Example

EvalSample(question="q", answer="a").id != "" True

from_answer classmethod

from_answer(answer: RAGAnswer, *, reference: str = '') -> EvalSample

Build a sample straight from a RAG answer.

This is the ergonomic path: run your pipeline, feed the answers to the evaluator, done.

Parameters:

Name	Type	Description	Default
answer	RAGAnswer	What :meth:~windlass.rag.pipeline.RAGPipeline.ask returned.	required
reference	str	Ground truth, when available.	''

Returns:

Type	Description
EvalSample	A populated sample.

Example

from windlass.core.types import RAGAnswer EvalSample.from_answer(RAGAnswer(answer="a", question="q")).question 'q'

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

@classmethod
def from_answer(cls, answer: RAGAnswer, *, reference: str = "") -> EvalSample:
    """Build a sample straight from a RAG answer.

    This is the ergonomic path: run your pipeline, feed the answers to the
    evaluator, done.

    Args:
        answer: What :meth:`~windlass.rag.pipeline.RAGPipeline.ask` returned.
        reference: Ground truth, when available.

    Returns:
        A populated sample.

    Example:
        >>> from windlass.core.types import RAGAnswer
        >>> EvalSample.from_answer(RAGAnswer(answer="a", question="q")).question
        'q'
    """
    return cls(
        question=answer.question,
        answer=answer.answer,
        contexts=[hit.chunk.content for hit in answer.contexts],
        reference=reference,
        metadata=dict(answer.metadata),
    )

Evaluator

Evaluator(
    *,
    metrics: Sequence[str] | None = None,
    threshold: float = 0.5,
    llm: Any = None,
    concurrency: int | None = None,
    name: str | None = None,
    **config: Any
)

Bases: Component

Abstract evaluation backend.

Parameters:

Name	Type	Description	Default
metrics	Sequence[str] | None	Metric names to compute. Which names are valid depends on the backend; ask it with :meth:available_metrics.	None
threshold	float	Score at or above which a result counts as passing.	0.5
llm	Any	Judge model for metrics that need one. Many quality metrics are themselves LLM calls.	None
concurrency	int | None	Maximum simultaneous sample evaluations.	None
name	str | None	Component name for traces.	None
**config	Any	Backend-specific options.	{}

Attributes:

Name	Type	Description
metrics	list[str]	The configured metric names.
threshold		The configured pass threshold.

Example

Implementing an evaluator takes one method::

class LengthEvaluator(Evaluator):
    provider_name = "length"

    async def aevaluate_sample(self, sample):
        score = min(1.0, len(sample.answer) / 100)
        return [EvaluationResult(metric="length", score=score)]

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

def __init__(
    self,
    *,
    metrics: Sequence[str] | None = None,
    threshold: float = 0.5,
    llm: Any = None,
    concurrency: int | None = None,
    name: str | None = None,
    **config: Any,
) -> None:
    super().__init__(
        name=name or self.provider_name,
        metrics=list(metrics or self.default_metrics()),
        threshold=threshold,
        **config,
    )
    self.metrics: list[str] = list(metrics or self.default_metrics())
    self.threshold = threshold
    self.llm = llm
    self.concurrency = concurrency or settings().max_concurrency

default_metrics classmethod

default_metrics() -> tuple[str, ...]

Return the metrics used when the caller does not choose any.

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

@classmethod
def default_metrics(cls) -> tuple[str, ...]:
    """Return the metrics used when the caller does not choose any."""
    return ()

available_metrics classmethod

available_metrics() -> tuple[str, ...]

Return every metric name this backend understands.

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

@classmethod
def available_metrics(cls) -> tuple[str, ...]:
    """Return every metric name this backend understands."""
    return cls.default_metrics()

aevaluate_sample abstractmethod async

aevaluate_sample(sample: EvalSample) -> list[EvaluationResult]

Score one sample against every configured metric.

Parameters:

Name	Type	Description	Default
sample	EvalSample	The interaction to score.	required

Returns:

Name	Type	Description
One	list[EvaluationResult]	class:~windlass.core.types.EvaluationResult per metric.

Raises:

Type	Description
EvaluationError	When a metric cannot be computed.

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

@abc.abstractmethod
async def aevaluate_sample(self, sample: EvalSample) -> list[EvaluationResult]:
    """Score one sample against every configured metric.

    Args:
        sample: The interaction to score.

    Returns:
        One :class:`~windlass.core.types.EvaluationResult` per metric.

    Raises:
        EvaluationError: When a metric cannot be computed.
    """

aevaluate async

aevaluate(
    samples: Sequence[EvalSample | RAGAnswer | dict[str, Any]],
) -> EvaluationReport

Evaluate a dataset and aggregate the results.

Parameters:

Name	Type	Description	Default
samples	Sequence[EvalSample | RAGAnswer | dict[str, Any]]	Samples, RAG answers, or dicts that coerce into samples.	required

Returns:

Type	Description
EvaluationReport	An aggregated :class:~windlass.core.types.EvaluationReport.

Raises:

Type	Description
EvaluationError	When every sample fails to evaluate.

Performance

Samples run concurrently up to concurrency. LLM-judged metrics dominate the cost, so keep an eye on that budget when evaluating thousands of rows.

Example

import asyncio from windlass.providers.evaluation.builtin import BuiltinEvaluator ev = BuiltinEvaluator(metrics=["answer_relevancy_lexical"]) report = asyncio.run(ev.aevaluate([ ... EvalSample(question="what is rag", answer="rag is retrieval") ... ])) report.samples 1

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

async def aevaluate(
    self, samples: Sequence[EvalSample | RAGAnswer | dict[str, Any]]
) -> EvaluationReport:
    """Evaluate a dataset and aggregate the results.

    Args:
        samples: Samples, RAG answers, or dicts that coerce into samples.

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

    Raises:
        EvaluationError: When every sample fails to evaluate.

    Performance:
        Samples run concurrently up to ``concurrency``. LLM-judged metrics
        dominate the cost, so keep an eye on that budget when evaluating
        thousands of rows.

    Example:
        >>> import asyncio
        >>> from windlass.providers.evaluation.builtin import BuiltinEvaluator
        >>> ev = BuiltinEvaluator(metrics=["answer_relevancy_lexical"])
        >>> report = asyncio.run(ev.aevaluate([
        ...     EvalSample(question="what is rag", answer="rag is retrieval")
        ... ]))
        >>> report.samples
        1
    """
    prepared = [self._coerce(s) for s in samples]
    if not prepared:
        return EvaluationReport(samples=0)

    outcomes = await gather_bounded(
        [self.aevaluate_sample(s) for s in prepared],
        limit=self.concurrency,
        return_exceptions=True,
    )

    results: list[EvaluationResult] = []
    failures = 0
    for sample, outcome in zip(prepared, outcomes, strict=True):
        if isinstance(outcome, BaseException):
            failures += 1
            self._log.warning("Evaluation failed for sample %s: %s", sample.id, outcome)
            continue
        for result in outcome:
            if not result.sample_id:
                result.sample_id = sample.id
            if result.threshold is None:
                result.threshold = self.threshold
            result.passed = result.score >= result.threshold
            results.append(result)

    if failures and not results:
        from windlass.core.exceptions import EvaluationError

        raise EvaluationError(
            f"All {failures} samples failed to evaluate.",
            hint="Check that the judge LLM is configured and that samples "
            "carry the fields your metrics need (contexts, reference).",
        )

    return EvaluationReport(results=results, samples=len(prepared))

evaluate

evaluate(
    samples: Sequence[EvalSample | RAGAnswer | dict[str, Any]],
) -> EvaluationReport

Blocking :meth:aevaluate.

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

def evaluate(
    self, samples: Sequence[EvalSample | RAGAnswer | dict[str, Any]]
) -> EvaluationReport:
    """Blocking :meth:`aevaluate`."""
    return run_sync(self.aevaluate(samples))

Guardrail

Guardrail(
    *,
    on_violation: str = "block",
    stages: tuple[str, ...] = ("input", "output"),
    name: str | None = None,
    **config: Any
)

Bases: Component

Abstract input/output safety check.

Parameters:

Name	Type	Description	Default
on_violation	str	One of block, redact, warn or allow. warn logs and lets the content through unchanged, which is the right setting while you are calibrating a new policy in production.	'block'
stages	tuple[str, ...]	Which stages this guardrail runs at — any of input and output.	('input', 'output')
name	str | None	Component name for traces.	None
**config	Any	Policy-specific options.	{}

Attributes:

Name	Type	Description
on_violation		The configured action.
stages		The stages this guardrail participates in.

Raises:

Type	Description
ValueError	If on_violation is not a recognised action.

Example

Implementing a guardrail takes one method::

class NoShouting(Guardrail):
    provider_name = "no_shouting"

    async def acheck(self, content, *, stage="input", context=None):
        if content.isupper():
            return GuardrailResult(
                allowed=False, content=content,
                rule="shouting", stage=stage,
            )
        return GuardrailResult(allowed=True, content=content, stage=stage)

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

def __init__(
    self,
    *,
    on_violation: str = "block",
    stages: tuple[str, ...] = ("input", "output"),
    name: str | None = None,
    **config: Any,
) -> None:
    if on_violation not in VIOLATION_ACTIONS:
        raise ValueError(
            f"on_violation must be one of {VIOLATION_ACTIONS}, got {on_violation!r}"
        )
    super().__init__(
        name=name or self.provider_name,
        on_violation=on_violation,
        stages=stages,
        **config,
    )
    self.on_violation = on_violation
    self.stages = tuple(stages)

acheck abstractmethod async

acheck(
    content: str, *, stage: str = "input", context: dict[str, Any] | None = None
) -> GuardrailResult

Inspect content and return a verdict.

Implementations should report what they found and leave the policy decision to :attr:on_violation — that keeps one detector usable in both blocking and redacting configurations.

Parameters:

Name	Type	Description	Default
content	str	The text to inspect.	required
stage	str	"input" or "output".	'input'
context	dict[str, Any] | None	Extra signals — retrieved chunks, the user id, the tool about to be called.	None

Returns:

Type	Description
GuardrailResult	The verdict, with content set to the (possibly rewritten) text.

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

@abc.abstractmethod
async def acheck(
    self,
    content: str,
    *,
    stage: str = "input",
    context: dict[str, Any] | None = None,
) -> GuardrailResult:
    """Inspect content and return a verdict.

    Implementations should report *what they found* and leave the policy
    decision to :attr:`on_violation` — that keeps one detector usable in
    both blocking and redacting configurations.

    Args:
        content: The text to inspect.
        stage: ``"input"`` or ``"output"``.
        context: Extra signals — retrieved chunks, the user id, the tool
            about to be called.

    Returns:
        The verdict, with ``content`` set to the (possibly rewritten) text.
    """

avalidate async

avalidate(
    content: str, *, stage: str = "input", context: dict[str, Any] | None = None
) -> str

Check content and apply the configured policy.

Parameters:

Name	Type	Description	Default
content	str	The text to check.	required
stage	str	"input" or "output".	'input'
context	dict[str, Any] | None	Extra signals for the check.	None

Returns:

Type	Description
str	The content to use downstream — unchanged, or redacted.

Raises:

Type	Description
GuardrailViolation	When a rule fires and on_violation='block'.

Example

import asyncio from windlass.providers.guardrails.rules import RuleGuardrail g = RuleGuardrail(banned_words=["secret"], on_violation="redact") asyncio.run(g.avalidate("the secret plan")) 'the [REDACTED] plan'

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

async def avalidate(
    self,
    content: str,
    *,
    stage: str = "input",
    context: dict[str, Any] | None = None,
) -> str:
    """Check content and apply the configured policy.

    Args:
        content: The text to check.
        stage: ``"input"`` or ``"output"``.
        context: Extra signals for the check.

    Returns:
        The content to use downstream — unchanged, or redacted.

    Raises:
        GuardrailViolation: When a rule fires and ``on_violation='block'``.

    Example:
        >>> import asyncio
        >>> from windlass.providers.guardrails.rules import RuleGuardrail
        >>> g = RuleGuardrail(banned_words=["secret"], on_violation="redact")
        >>> asyncio.run(g.avalidate("the secret plan"))
        'the [REDACTED] plan'
    """
    if stage not in self.stages:
        return content

    result = await self.acheck(content, stage=stage, context=context)
    if result.allowed and not result.detections:
        return content

    if self.on_violation == "allow":
        return content
    if self.on_violation == "warn":
        self._log.warning("Guardrail %s flagged %s content: %s", self.name, stage, result.rule)
        return content
    if self.on_violation == "redact":
        return result.content or content

    raise GuardrailViolation(
        f"Guardrail {self.name!r} blocked the {stage}: {result.rule or 'policy violation'}",
        stage=stage,
        rule=result.rule,
        detections=result.detections,
    )

validate

validate(
    content: str, *, stage: str = "input", context: dict[str, Any] | None = None
) -> str

Blocking :meth:avalidate.

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

def validate(
    self,
    content: str,
    *,
    stage: str = "input",
    context: dict[str, Any] | None = None,
) -> str:
    """Blocking :meth:`avalidate`."""
    return run_sync(self.avalidate(content, stage=stage, context=context))

check

check(
    content: str, *, stage: str = "input", context: dict[str, Any] | None = None
) -> GuardrailResult

Blocking :meth:acheck — returns the verdict without enforcing it.

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

def check(
    self,
    content: str,
    *,
    stage: str = "input",
    context: dict[str, Any] | None = None,
) -> GuardrailResult:
    """Blocking :meth:`acheck` — returns the verdict without enforcing it."""
    return run_sync(self.acheck(content, stage=stage, context=context))

GuardrailChain

GuardrailChain(guards: list[Guardrail] | None = None, *, name: str | None = None)

Bases: Guardrail

Runs several guardrails in order, threading redactions through.

Each guardrail sees whatever the previous one produced, so a PII redactor followed by an injection detector inspects the already-masked text.

Parameters:

Name	Type	Description	Default
guards	list[Guardrail] | None	The guardrails to run.	None
name	str | None	Component name for traces.	None

Attributes:

Name	Type	Description
guards	list[Guardrail]	The configured guardrails.

Example

from windlass.providers.guardrails.rules import RuleGuardrail chain = RuleGuardrail(pii=True, on_violation="redact") & RuleGuardrail( ... banned_words=["nope"], on_violation="redact" ... ) chain.validate("a@b.com says nope") '[EMAIL] says [REDACTED]'

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

def __init__(self, guards: list[Guardrail] | None = None, *, name: str | None = None) -> None:
    super().__init__(name=name or "chain", on_violation="block")
    self.guards: list[Guardrail] = list(guards or [])
    self.stages = ("input", "output")

add

add(guard: Guardrail) -> GuardrailChain

Append a guardrail and return self.

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

def add(self, guard: Guardrail) -> GuardrailChain:
    """Append a guardrail and return ``self``."""
    self.guards.append(guard)
    return self

acheck async

acheck(
    content: str, *, stage: str = "input", context: dict[str, Any] | None = None
) -> GuardrailResult

Aggregate every child's detections without enforcing them.

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

async def acheck(
    self,
    content: str,
    *,
    stage: str = "input",
    context: dict[str, Any] | None = None,
) -> GuardrailResult:
    """Aggregate every child's detections without enforcing them."""
    current = content
    detections: list[dict[str, Any]] = []
    rule: str | None = None
    allowed = True
    for guard in self.guards:
        if stage not in guard.stages:
            continue
        result = await guard.acheck(current, stage=stage, context=context)
        detections.extend(result.detections)
        if result.detections and guard.on_violation == "redact":
            current = result.content or current
        if not result.allowed and guard.on_violation == "block":
            allowed = False
            rule = rule or result.rule
    return GuardrailResult(
        allowed=allowed, content=current, detections=detections, rule=rule, stage=stage
    )

avalidate async

avalidate(
    content: str, *, stage: str = "input", context: dict[str, Any] | None = None
) -> str

Run every guardrail's own policy in sequence.

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

async def avalidate(
    self,
    content: str,
    *,
    stage: str = "input",
    context: dict[str, Any] | None = None,
) -> str:
    """Run every guardrail's own policy in sequence."""
    current = content
    for guard in self.guards:
        current = await guard.avalidate(current, stage=stage, context=context)
    return current

describe

describe() -> dict[str, Any]

Return a summary including each guardrail.

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

def describe(self) -> dict[str, Any]:
    """Return a summary including each guardrail."""
    return {**super().describe(), "guards": [g.describe() for g in self.guards]}

LLM

LLM(
    model: str = "",
    *,
    temperature: float | None = None,
    max_tokens: int | None = None,
    timeout: float | None = None,
    system_prompt: str | None = None,
    name: str | None = None,
    **config: Any
)

Bases: Component

Abstract chat-completion model.

Parameters:

Name	Type	Description	Default
model	str	Model identifier passed to the provider.	''
temperature	float | None	Sampling temperature. Falls back to the global setting.	None
max_tokens	int | None	Completion ceiling. None defers to the provider.	None
timeout	float | None	Per-request timeout in seconds.	None
system_prompt	str | None	Prepended to every call that does not already start with a system message.	None
name	str | None	Component name for traces. Defaults to the provider name.	None
**config	Any	Provider-specific options forwarded verbatim to the SDK.	{}

Attributes:

Name	Type	Description
model		The configured model identifier.
temperature	float	The configured sampling temperature.
max_tokens	int | None	The configured completion ceiling.
supports_tools	bool	Whether this provider can do native tool calling.
supports_streaming	bool	Whether this provider can stream.

Example

Implementing a provider takes one method::

class MyLLM(LLM):
    provider_name = "mine"

    async def agenerate(self, messages, *, tools=None, **kw):
        text = await my_sdk.chat(messages[-1].content)
        return Completion(content=text, model=self.model)

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

def __init__(
    self,
    model: str = "",
    *,
    temperature: float | None = None,
    max_tokens: int | None = None,
    timeout: float | None = None,
    system_prompt: str | None = None,
    name: str | None = None,
    **config: Any,
) -> None:
    cfg = settings()
    super().__init__(
        name=name or self.provider_name,
        model=model,
        temperature=cfg.temperature if temperature is None else temperature,
        max_tokens=cfg.max_tokens if max_tokens is None else max_tokens,
        timeout=cfg.request_timeout if timeout is None else timeout,
        system_prompt=system_prompt,
        **config,
    )
    self.model = model or self.default_model()
    self.temperature: float = self.config["temperature"]
    self.max_tokens: int | None = self.config["max_tokens"]
    self.timeout: float = self.config["timeout"]
    self.system_prompt: str | None = system_prompt

default_model classmethod

default_model() -> str

Return the model used when the caller does not name one.

Returns:

Type	Description
str	A provider-appropriate default model identifier.

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

@classmethod
def default_model(cls) -> str:
    """Return the model used when the caller does not name one.

    Returns:
        A provider-appropriate default model identifier.
    """
    return ""

agenerate abstractmethod async

agenerate(
    messages: list[Message], *, tools: list[dict[str, Any]] | None = None, **kwargs: Any
) -> Completion

Produce one completion.

This is the only method a provider must implement.

Parameters:

Name	Type	Description	Default
messages	list[Message]	The conversation, already normalised and with the system prompt applied.	required
tools	list[dict[str, Any]] | None	JSON-schema tool definitions in OpenAI function-calling format. Adapters translate these into their own shape.	None
**kwargs	Any	Per-call overrides (temperature, max_tokens, response_format, ...).	{}

Returns:

Type	Description
Completion	The completion, with raw set to the provider's response.

Raises:

Type	Description
ProviderError	For any provider-side failure.

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

@abc.abstractmethod
async def agenerate(
    self,
    messages: list[Message],
    *,
    tools: list[dict[str, Any]] | None = None,
    **kwargs: Any,
) -> Completion:
    """Produce one completion.

    This is the only method a provider must implement.

    Args:
        messages: The conversation, already normalised and with the system
            prompt applied.
        tools: JSON-schema tool definitions in OpenAI function-calling
            format. Adapters translate these into their own shape.
        **kwargs: Per-call overrides (``temperature``, ``max_tokens``,
            ``response_format``, ...).

    Returns:
        The completion, with ``raw`` set to the provider's response.

    Raises:
        ProviderError: For any provider-side failure.
    """

astream_generate async

astream_generate(
    messages: list[Message], *, tools: list[dict[str, Any]] | None = None, **kwargs: Any
) -> AsyncIterator[StreamEvent]

Produce incremental completion events.

The default implementation calls :meth:agenerate and emits the result as a single text event followed by done — correct, but not actually incremental. Providers that support streaming should override this.

Parameters:

Name	Type	Description	Default
messages	list[Message]	The normalised conversation.	required
tools	list[dict[str, Any]] | None	Tool definitions.	None
**kwargs	Any	Per-call overrides.	{}

Yields:

Type	Description
AsyncIterator[StreamEvent]	class:~windlass.core.types.StreamEvent values.

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

async def astream_generate(
    self,
    messages: list[Message],
    *,
    tools: list[dict[str, Any]] | None = None,
    **kwargs: Any,
) -> AsyncIterator[StreamEvent]:
    """Produce incremental completion events.

    The default implementation calls :meth:`agenerate` and emits the result
    as a single ``text`` event followed by ``done`` — correct, but not
    actually incremental. Providers that support streaming should override
    this.

    Args:
        messages: The normalised conversation.
        tools: Tool definitions.
        **kwargs: Per-call overrides.

    Yields:
        :class:`~windlass.core.types.StreamEvent` values.
    """
    completion = await self.agenerate(messages, tools=tools, **kwargs)
    if completion.content:
        yield StreamEvent(type="text", delta=completion.content)
    for call in completion.tool_calls:
        yield StreamEvent(type="tool_call", tool_call=call)
    yield StreamEvent(
        type="done", finish_reason=completion.finish_reason, usage=completion.usage
    )

acomplete async

acomplete(
    prompt: PromptLike,
    *,
    tools: list[dict[str, Any]] | None = None,
    retry: bool = True,
    **kwargs: Any
) -> Completion

Generate a completion, with retries and timing.

Parameters:

Name	Type	Description	Default
prompt	PromptLike	A string, a message, or a full transcript.	required
tools	list[dict[str, Any]] | None	Tool definitions the model may call.	None
retry	bool	Apply the configured retry policy to transient failures.	True
**kwargs	Any	Per-call overrides forwarded to the provider.	{}

Returns:

Type	Description
Completion	The completion. latency_ms is always populated.

Raises:

Type	Description
ProviderError	When the provider fails and retries are exhausted.
AuthenticationError	When credentials are missing or rejected.

Performance

One network round trip. Set retry=False on latency-critical paths where you would rather fail fast than wait out a backoff.

Example

import asyncio from windlass.providers.llm.fake import FakeLLM asyncio.run(FakeLLM(responses=["hi"]).acomplete("hello")).content 'hi'

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

async def acomplete(
    self,
    prompt: PromptLike,
    *,
    tools: list[dict[str, Any]] | None = None,
    retry: bool = True,
    **kwargs: Any,
) -> Completion:
    """Generate a completion, with retries and timing.

    Args:
        prompt: A string, a message, or a full transcript.
        tools: Tool definitions the model may call.
        retry: Apply the configured retry policy to transient failures.
        **kwargs: Per-call overrides forwarded to the provider.

    Returns:
        The completion. ``latency_ms`` is always populated.

    Raises:
        ProviderError: When the provider fails and retries are exhausted.
        AuthenticationError: When credentials are missing or rejected.

    Performance:
        One network round trip. Set ``retry=False`` on latency-critical
        paths where you would rather fail fast than wait out a backoff.

    Example:
        >>> import asyncio
        >>> from windlass.providers.llm.fake import FakeLLM
        >>> asyncio.run(FakeLLM(responses=["hi"]).acomplete("hello")).content
        'hi'
    """
    messages = self._prepare(prompt)
    started = time.perf_counter()

    async def _call() -> Completion:
        return await self.agenerate(messages, tools=tools, **kwargs)

    completion = (
        await retry_async(_call, description=f"{self.name}.generate")
        if retry
        else await _call()
    )
    if not completion.latency_ms:
        completion.latency_ms = (time.perf_counter() - started) * 1000
    if not completion.model:
        completion.model = self.model
    return completion

complete

complete(
    prompt: PromptLike,
    *,
    tools: list[dict[str, Any]] | None = None,
    retry: bool = True,
    **kwargs: Any
) -> Completion

Blocking :meth:acomplete.

Safe to call from a notebook or from inside a running event loop; the call is dispatched to a background loop when necessary.

Parameters:

Name	Type	Description	Default
prompt	PromptLike	A string, a message, or a full transcript.	required
tools	list[dict[str, Any]] | None	Tool definitions the model may call.	None
retry	bool	Apply the configured retry policy.	True
**kwargs	Any	Per-call overrides.	{}

Returns:

Type	Description
Completion	The completion.

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

def complete(
    self,
    prompt: PromptLike,
    *,
    tools: list[dict[str, Any]] | None = None,
    retry: bool = True,
    **kwargs: Any,
) -> Completion:
    """Blocking :meth:`acomplete`.

    Safe to call from a notebook or from inside a running event loop; the
    call is dispatched to a background loop when necessary.

    Args:
        prompt: A string, a message, or a full transcript.
        tools: Tool definitions the model may call.
        retry: Apply the configured retry policy.
        **kwargs: Per-call overrides.

    Returns:
        The completion.
    """
    return run_sync(self.acomplete(prompt, tools=tools, retry=retry, **kwargs))

astream async

astream(
    prompt: PromptLike, *, tools: list[dict[str, Any]] | None = None, **kwargs: Any
) -> AsyncIterator[StreamEvent]

Stream a completion as it is generated.

Parameters:

Name	Type	Description	Default
prompt	PromptLike	A string, a message, or a full transcript.	required
tools	list[dict[str, Any]] | None	Tool definitions.	None
**kwargs	Any	Per-call overrides.	{}

Yields:

Type	Description
AsyncIterator[StreamEvent]	Stream events. Text deltas arrive as they are produced; the final
AsyncIterator[StreamEvent]	event is always type="done".

Example

import asyncio from windlass.providers.llm.fake import FakeLLM async def main(): ... llm = FakeLLM(responses=["hey"]) ... return "".join([e.delta async for e in llm.astream("hi")]) asyncio.run(main()) 'hey'

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

async def astream(
    self,
    prompt: PromptLike,
    *,
    tools: list[dict[str, Any]] | None = None,
    **kwargs: Any,
) -> AsyncIterator[StreamEvent]:
    """Stream a completion as it is generated.

    Args:
        prompt: A string, a message, or a full transcript.
        tools: Tool definitions.
        **kwargs: Per-call overrides.

    Yields:
        Stream events. Text deltas arrive as they are produced; the final
        event is always ``type="done"``.

    Example:
        >>> import asyncio
        >>> from windlass.providers.llm.fake import FakeLLM
        >>> async def main():
        ...     llm = FakeLLM(responses=["hey"])
        ...     return "".join([e.delta async for e in llm.astream("hi")])
        >>> asyncio.run(main())
        'hey'
    """
    messages = self._prepare(prompt)
    async for event in self.astream_generate(messages, tools=tools, **kwargs):
        yield event

stream

stream(
    prompt: PromptLike, *, tools: list[dict[str, Any]] | None = None, **kwargs: Any
) -> Iterator[StreamEvent]

Blocking :meth:astream.

Parameters:

Name	Type	Description	Default
prompt	PromptLike	A string, a message, or a full transcript.	required
tools	list[dict[str, Any]] | None	Tool definitions.	None
**kwargs	Any	Per-call overrides.	{}

Yields:

Type	Description
StreamEvent	Stream events, pulled one at a time so nothing is buffered ahead of
StreamEvent	the consumer.

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

def stream(
    self,
    prompt: PromptLike,
    *,
    tools: list[dict[str, Any]] | None = None,
    **kwargs: Any,
) -> Iterator[StreamEvent]:
    """Blocking :meth:`astream`.

    Args:
        prompt: A string, a message, or a full transcript.
        tools: Tool definitions.
        **kwargs: Per-call overrides.

    Yields:
        Stream events, pulled one at a time so nothing is buffered ahead of
        the consumer.
    """
    return iter_sync(self.astream(prompt, tools=tools, **kwargs))

astream_text async

astream_text(prompt: PromptLike, **kwargs: Any) -> AsyncIterator[str]

Stream only the text deltas — the common case for chat UIs.

Parameters:

Name	Type	Description	Default
prompt	PromptLike	A string, a message, or a full transcript.	required
**kwargs	Any	Per-call overrides.	{}

Yields:

Type	Description
AsyncIterator[str]	Non-empty text fragments.

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

async def astream_text(self, prompt: PromptLike, **kwargs: Any) -> AsyncIterator[str]:
    """Stream only the text deltas — the common case for chat UIs.

    Args:
        prompt: A string, a message, or a full transcript.
        **kwargs: Per-call overrides.

    Yields:
        Non-empty text fragments.
    """
    async for event in self.astream(prompt, **kwargs):
        if event.type == "text" and event.delta:
            yield event.delta

abatch async

abatch(
    prompts: Sequence[PromptLike], *, concurrency: int | None = None, **kwargs: Any
) -> list[Completion]

Complete many prompts with bounded concurrency.

Parameters:

Name	Type	Description	Default
prompts	Sequence[PromptLike]	The prompts to run.	required
concurrency	int | None	Maximum simultaneous requests. Defaults to the global max_concurrency setting.	None
**kwargs	Any	Per-call overrides applied to every prompt.	{}

Returns:

Type	Description
list[Completion]	Completions in the same order as prompts.

Performance

Bounded on purpose — an unbounded fan-out over a few hundred prompts is the fastest way to get rate limited.

Example

import asyncio from windlass.providers.llm.fake import FakeLLM llm = FakeLLM(responses=["a", "b"]) [c.content for c in asyncio.run(llm.abatch(["1", "2"]))]['a', 'b']

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

async def abatch(
    self,
    prompts: Sequence[PromptLike],
    *,
    concurrency: int | None = None,
    **kwargs: Any,
) -> list[Completion]:
    """Complete many prompts with bounded concurrency.

    Args:
        prompts: The prompts to run.
        concurrency: Maximum simultaneous requests. Defaults to the global
            ``max_concurrency`` setting.
        **kwargs: Per-call overrides applied to every prompt.

    Returns:
        Completions in the same order as ``prompts``.

    Performance:
        Bounded on purpose — an unbounded fan-out over a few hundred prompts
        is the fastest way to get rate limited.

    Example:
        >>> import asyncio
        >>> from windlass.providers.llm.fake import FakeLLM
        >>> llm = FakeLLM(responses=["a", "b"])
        >>> [c.content for c in asyncio.run(llm.abatch(["1", "2"]))]
        ['a', 'b']
    """
    from windlass.core.concurrency import gather_bounded

    limit = concurrency or settings().max_concurrency
    return await gather_bounded([self.acomplete(p, **kwargs) for p in prompts], limit=limit)

batch

batch(
    prompts: Sequence[PromptLike], *, concurrency: int | None = None, **kwargs: Any
) -> list[Completion]

Blocking :meth:abatch.

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

def batch(
    self, prompts: Sequence[PromptLike], *, concurrency: int | None = None, **kwargs: Any
) -> list[Completion]:
    """Blocking :meth:`abatch`."""
    return run_sync(self.abatch(prompts, concurrency=concurrency, **kwargs))

Loader

Loader(
    *,
    encoding: str = "utf-8",
    metadata: dict[str, Any] | None = None,
    recursive: bool = True,
    on_error: str = "skip",
    name: str | None = None,
    **config: Any
)

Bases: Component

Abstract document loader.

Parameters:

Name	Type	Description	Default
encoding	str	Text encoding used for character-based formats.	'utf-8'
metadata	dict[str, Any] | None	Extra metadata merged into every produced document — useful for tagging a whole corpus with a tenant or collection id.	None
recursive	bool	Whether directory sources are walked recursively.	True
on_error	str	"raise" aborts the batch, "skip" logs and continues. "skip" is the right default for a 10,000-file corpus where one corrupt PDF should not lose the run.	'skip'
name	str | None	Component name for traces.	None
**config	Any	Loader-specific options.	{}

Attributes:

Name	Type	Description
extensions	tuple[str, ...]	File suffixes this loader claims, used for auto-detection.
mimetypes	tuple[str, ...]	MIME types this loader claims.

Example

Implementing a loader takes one method::

class MyLoader(Loader):
    provider_name = "mine"
    extensions = (".mine",)

    async def aload_source(self, source):
        return [Document(content=read(source), source=str(source))]

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

def __init__(
    self,
    *,
    encoding: str = "utf-8",
    metadata: dict[str, Any] | None = None,
    recursive: bool = True,
    on_error: str = "skip",
    name: str | None = None,
    **config: Any,
) -> None:
    if on_error not in {"raise", "skip"}:
        raise ValueError("on_error must be 'raise' or 'skip'")
    super().__init__(
        name=name or self.provider_name,
        encoding=encoding,
        metadata=metadata or {},
        recursive=recursive,
        on_error=on_error,
        **config,
    )
    self.encoding = encoding
    self.extra_metadata: dict[str, Any] = dict(metadata or {})
    self.recursive = recursive
    self.on_error = on_error

aload_source abstractmethod async

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

Load one concrete source.

Called once per file. Directory expansion and error handling are done for you by :meth:aload.

Parameters:

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

Returns:

Type	Description
list[Document]	The documents extracted from it. A PDF may return one document per
list[Document]	page or one for the whole file — that is the loader's choice.

Raises:

Type	Description
IngestionError	When the source cannot be parsed.

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

@abc.abstractmethod
async def aload_source(self, source: SourceLike) -> list[Document]:
    """Load one concrete source.

    Called once per file. Directory expansion and error handling are done
    for you by :meth:`aload`.

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

    Returns:
        The documents extracted from it. A PDF may return one document per
        page or one for the whole file — that is the loader's choice.

    Raises:
        IngestionError: When the source cannot be parsed.
    """

can_handle classmethod

can_handle(source: SourceLike) -> bool

Return whether this loader claims source.

Parameters:

Name	Type	Description	Default
source	SourceLike	Path, URL or payload to test.	required

Returns:

Type	Description
bool	True when the extension or scheme matches this loader.

Example

from windlass.providers.loaders.text import TextLoader TextLoader.can_handle("notes.txt") True TextLoader.can_handle("scan.pdf") False

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

@classmethod
def can_handle(cls, source: SourceLike) -> bool:
    """Return whether this loader claims ``source``.

    Args:
        source: Path, URL or payload to test.

    Returns:
        True when the extension or scheme matches this loader.

    Example:
        >>> from windlass.providers.loaders.text import TextLoader
        >>> TextLoader.can_handle("notes.txt")
        True
        >>> TextLoader.can_handle("scan.pdf")
        False
    """
    if isinstance(source, str | Path):
        text = str(source)
        if text.startswith(("http://", "https://")):
            return cls.handles_urls
        return Path(text).suffix.lower() in cls.extensions
    return False

aload async

aload(
    source: SourceLike | Sequence[SourceLike], *, concurrency: int | None = None
) -> list[Document]

Load one or many sources.

Directories are expanded (respecting :attr:recursive and this loader's :attr:extensions), and files are read concurrently.

Parameters:

Name	Type	Description	Default
source	SourceLike | Sequence[SourceLike]	A path, URL, byte payload, or a sequence of them. A directory path expands to its matching files.	required
concurrency	int | None	Maximum simultaneous reads. Defaults to the global max_concurrency setting.	None

Returns:

Type	Description
list[Document]	Every extracted document, with :attr:extra_metadata merged in.

Raises:

Type	Description
IngestionError	When on_error='raise' and a source fails, or when no source could be read at all.

Performance

Blocking parsers run on the thread pool, so a folder of 500 PDFs is parsed in parallel rather than serially.

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

async def aload(
    self, source: SourceLike | Sequence[SourceLike], *, concurrency: int | None = None
) -> list[Document]:
    """Load one or many sources.

    Directories are expanded (respecting :attr:`recursive` and this loader's
    :attr:`extensions`), and files are read concurrently.

    Args:
        source: A path, URL, byte payload, or a sequence of them. A
            directory path expands to its matching files.
        concurrency: Maximum simultaneous reads. Defaults to the global
            ``max_concurrency`` setting.

    Returns:
        Every extracted document, with :attr:`extra_metadata` merged in.

    Raises:
        IngestionError: When ``on_error='raise'`` and a source fails, or
            when no source could be read at all.

    Performance:
        Blocking parsers run on the thread pool, so a folder of 500 PDFs is
        parsed in parallel rather than serially.
    """
    sources = self._expand(source)
    if not sources:
        return []

    limit = concurrency or settings().max_concurrency
    results = await gather_bounded(
        [self._load_guarded(s) for s in sources], limit=limit, return_exceptions=True
    )

    documents: list[Document] = []
    failures: list[tuple[Any, BaseException]] = []
    for src, result in zip(sources, results, strict=True):
        if isinstance(result, BaseException):
            failures.append((src, result))
            continue
        documents.extend(result)

    if failures:
        if self.on_error == "raise":
            src, exc = failures[0]
            raise IngestionError(
                f"Failed to load {src}: {exc}",
                context={"source": str(src), "failures": len(failures)},
            ) from exc
        for src, exc in failures:
            self._log.warning("Skipping %s: %s", src, exc)
        if not documents:
            src, exc = failures[0]
            raise IngestionError(
                f"No documents could be loaded; the first failure was {src}: {exc}",
                hint="Check the paths and that the right extra is installed.",
                context={"failures": len(failures)},
            ) from exc

    for doc in documents:
        if self.extra_metadata:
            doc.metadata = {**self.extra_metadata, **doc.metadata}
        doc.metadata.setdefault("loader", self.name)
    return documents

load

load(source: SourceLike | Sequence[SourceLike]) -> list[Document]

Blocking :meth:aload.

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

def load(self, source: SourceLike | Sequence[SourceLike]) -> list[Document]:
    """Blocking :meth:`aload`."""
    return run_sync(self.aload(source))

astream_load async

astream_load(source: SourceLike | Sequence[SourceLike]) -> AsyncIterator[Document]

Yield documents one at a time instead of building a list.

Use this for corpora too large to hold in memory: combined with Pipeline.aingest_stream it keeps peak memory flat regardless of corpus size.

Parameters:

Name	Type	Description	Default
source	SourceLike | Sequence[SourceLike]	A path, URL, payload, or sequence of them.	required

Yields:

Type	Description
AsyncIterator[Document]	Documents as each source finishes parsing.

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

async def astream_load(
    self, source: SourceLike | Sequence[SourceLike]
) -> AsyncIterator[Document]:
    """Yield documents one at a time instead of building a list.

    Use this for corpora too large to hold in memory: combined with
    ``Pipeline.aingest_stream`` it keeps peak memory flat regardless of
    corpus size.

    Args:
        source: A path, URL, payload, or sequence of them.

    Yields:
        Documents as each source finishes parsing.
    """
    for item in self._expand(source):
        try:
            docs = await self.aload_source(item)
        except Exception as exc:
            if self.on_error == "raise":
                raise IngestionError(f"Failed to load {_label(item)}: {exc}") from exc
            self._log.warning("Skipping %s: %s", item, exc)
            continue
        for doc in docs:
            if self.extra_metadata:
                doc.metadata = {**self.extra_metadata, **doc.metadata}
            doc.metadata.setdefault("loader", self.name)
            yield doc

MCPClient

MCPClient(
    *,
    server: str = "",
    namespace: bool = False,
    timeout: float = 30.0,
    name: str | None = None,
    **config: Any
)

Bases: Component

Abstract MCP client.

Parameters:

Name	Type	Description	Default
server	str	Server label used in logs and to namespace tools.	''
namespace	bool	When True, remote tools are exposed as server_toolname so two servers offering search do not collide.	False
timeout	float	Per-call timeout in seconds.	30.0
name	str | None	Component name.	None
**config	Any	Transport-specific options (command, args, url, env, ...).	{}

Attributes:

Name	Type	Description
connected		Whether the transport is currently open.

Example

Implementing a client means three methods::

class MyClient(MCPClient):
    provider_name = "mine"

    async def aconnect(self): ...
    async def alist_tools(self): ...
    async def acall_tool(self, name, arguments): ...

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

def __init__(
    self,
    *,
    server: str = "",
    namespace: bool = False,
    timeout: float = 30.0,
    name: str | None = None,
    **config: Any,
) -> None:
    super().__init__(
        name=name or server or self.provider_name,
        server=server,
        namespace=namespace,
        timeout=timeout,
        **config,
    )
    self.server = server or self.provider_name
    self.namespace = namespace
    self.timeout = timeout
    self.connected = False

aconnect abstractmethod async

aconnect() -> None

Open the transport and perform the MCP handshake.

Must be idempotent — calling it twice should not open two sessions.

Raises:

Type	Description
MCPError	When the server cannot be reached or the handshake fails.

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

@abc.abstractmethod
async def aconnect(self) -> None:
    """Open the transport and perform the MCP handshake.

    Must be idempotent — calling it twice should not open two sessions.

    Raises:
        MCPError: When the server cannot be reached or the handshake fails.
    """

alist_tools abstractmethod async

alist_tools() -> list[Tool]

Discover the server's tools.

Returns:

Type	Description
list[Tool]	class:MCPToolProxy instances ready to bind to an agent.

Raises:

Type	Description
MCPError	When discovery fails.

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

@abc.abstractmethod
async def alist_tools(self) -> list[Tool]:
    """Discover the server's tools.

    Returns:
        :class:`MCPToolProxy` instances ready to bind to an agent.

    Raises:
        MCPError: When discovery fails.
    """

acall_tool abstractmethod async

acall_tool(name: str, arguments: dict[str, Any]) -> Any

Invoke a remote tool.

Parameters:

Name	Type	Description	Default
name	str	Tool name as advertised by the server (without any namespace prefix Windlass may have added).	required
arguments	dict[str, Any]	Arguments matching the remote schema.	required

Returns:

Type	Description
Any	The server's result, unwrapped from the MCP content envelope.

Raises:

Type	Description
MCPError	When the call fails.

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

@abc.abstractmethod
async def acall_tool(self, name: str, arguments: dict[str, Any]) -> Any:
    """Invoke a remote tool.

    Args:
        name: Tool name as advertised by the server (without any namespace
            prefix Windlass may have added).
        arguments: Arguments matching the remote schema.

    Returns:
        The server's result, unwrapped from the MCP content envelope.

    Raises:
        MCPError: When the call fails.
    """

alist_resources async

alist_resources() -> list[MCPResource]

Discover readable resources. Returns [] when unsupported.

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

async def alist_resources(self) -> list[MCPResource]:
    """Discover readable resources. Returns ``[]`` when unsupported."""
    return []

aread_resource async

aread_resource(uri: str) -> str

Read a resource's contents.

Parameters:

Name	Type	Description	Default
uri	str	The resource identifier.	required

Returns:

Type	Description
str	The resource body as text.

Raises:

Type	Description
MCPError	When the resource cannot be read.

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

async def aread_resource(self, uri: str) -> str:
    """Read a resource's contents.

    Args:
        uri: The resource identifier.

    Returns:
        The resource body as text.

    Raises:
        MCPError: When the resource cannot be read.
    """
    raise MCPError(
        f"{type(self).__name__} does not support reading resources.",
        context={"uri": uri},
    )

alist_prompts async

alist_prompts() -> list[MCPPrompt]

Discover prompt templates. Returns [] when unsupported.

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

async def alist_prompts(self) -> list[MCPPrompt]:
    """Discover prompt templates. Returns ``[]`` when unsupported."""
    return []

aget_prompt async

aget_prompt(name: str, arguments: dict[str, Any] | None = None) -> str

Instantiate a prompt template.

Parameters:

Name	Type	Description	Default
name	str	Prompt name.	required
arguments	dict[str, Any] | None	Template arguments.	None

Returns:

Type	Description
str	The rendered prompt text.

Raises:

Type	Description
MCPError	When the prompt cannot be rendered.

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

async def aget_prompt(self, name: str, arguments: dict[str, Any] | None = None) -> str:
    """Instantiate a prompt template.

    Args:
        name: Prompt name.
        arguments: Template arguments.

    Returns:
        The rendered prompt text.

    Raises:
        MCPError: When the prompt cannot be rendered.
    """
    raise MCPError(f"{type(self).__name__} does not support prompts.", context={"prompt": name})

adisconnect async

adisconnect() -> None

Close the transport. Safe to call when already closed.

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

async def adisconnect(self) -> None:
    """Close the transport. Safe to call when already closed."""
    self.connected = False

connect

connect() -> None

Blocking :meth:aconnect.

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

def connect(self) -> None:
    """Blocking :meth:`aconnect`."""
    run_sync(self.aconnect())

list_tools

list_tools() -> list[Tool]

Blocking :meth:alist_tools.

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

def list_tools(self) -> list[Tool]:
    """Blocking :meth:`alist_tools`."""
    return run_sync(self.alist_tools())

call_tool

call_tool(name: str, arguments: dict[str, Any] | None = None) -> Any

Blocking :meth:acall_tool.

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

def call_tool(self, name: str, arguments: dict[str, Any] | None = None) -> Any:
    """Blocking :meth:`acall_tool`."""
    return run_sync(self.acall_tool(name, arguments or {}))

list_resources

list_resources() -> list[MCPResource]

Blocking :meth:alist_resources.

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

def list_resources(self) -> list[MCPResource]:
    """Blocking :meth:`alist_resources`."""
    return run_sync(self.alist_resources())

read_resource

read_resource(uri: str) -> str

Blocking :meth:aread_resource.

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

def read_resource(self, uri: str) -> str:
    """Blocking :meth:`aread_resource`."""
    return run_sync(self.aread_resource(uri))

list_prompts

list_prompts() -> list[MCPPrompt]

Blocking :meth:alist_prompts.

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

def list_prompts(self) -> list[MCPPrompt]:
    """Blocking :meth:`alist_prompts`."""
    return run_sync(self.alist_prompts())

get_prompt

get_prompt(name: str, arguments: dict[str, Any] | None = None) -> str

Blocking :meth:aget_prompt.

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

def get_prompt(self, name: str, arguments: dict[str, Any] | None = None) -> str:
    """Blocking :meth:`aget_prompt`."""
    return run_sync(self.aget_prompt(name, arguments))

disconnect

disconnect() -> None

Blocking :meth:adisconnect.

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

def disconnect(self) -> None:
    """Blocking :meth:`adisconnect`."""
    run_sync(self.adisconnect())

aclose async

aclose() -> None

Alias for :meth:adisconnect, so containers can close uniformly.

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

async def aclose(self) -> None:
    """Alias for :meth:`adisconnect`, so containers can close uniformly."""
    await self.adisconnect()

MCPPrompt

Bases: WindlassModel

A parameterised prompt template advertised by an MCP server.

Attributes:

Name	Type	Description
name	str	Prompt identifier.
description	str	What the prompt is for.
arguments	list[dict[str, Any]]	Declared arguments, each a dict with name, description and required.
server	str	Which server advertised it.

MCPResource

Bases: WindlassModel

A readable resource advertised by an MCP server.

Attributes:

Name	Type	Description
uri	str	Resource identifier, e.g. file:///docs/readme.md.
name	str	Human readable name.
description	str	What the resource contains.
mimetype	str | None	Content type, when the server declares one.
server	str	Which server advertised it.

MCPToolProxy

MCPToolProxy(
    client: MCPClient,
    *,
    name: str,
    description: str = "",
    parameters: dict[str, Any] | None = None,
    server: str = "",
    **config: Any
)

Bases: Tool

A local :class:~windlass.interfaces.tool.Tool backed by a remote MCP tool.

Created by :meth:MCPClient.alist_tools. Calling it forwards to the server and returns whatever comes back, so agents bind it exactly like a local function.

Parameters:

Name	Type	Description	Default
client	MCPClient	The MCP client that owns the connection.	required
name	str	Remote tool name.	required
description	str	Remote tool description.	''
parameters	dict[str, Any] | None	Remote JSON Schema for the arguments.	None
server	str	Server label, kept in metadata and used to disambiguate same-named tools across servers.	''
**config	Any	Forwarded to :class:~windlass.interfaces.tool.Tool.	{}

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

def __init__(
    self,
    client: MCPClient,
    *,
    name: str,
    description: str = "",
    parameters: dict[str, Any] | None = None,
    server: str = "",
    **config: Any,
) -> None:
    super().__init__(
        name=name,
        description=description or f"MCP tool {name!r} from {server or 'server'}.",
        parameters=parameters,
        **config,
    )
    self.client = client
    self.server = server

acall async

acall(**kwargs: Any) -> Any

Forward the call to the MCP server.

Parameters:

Name	Type	Description	Default
**kwargs	Any	Arguments matching the remote schema.	{}

Returns:

Type	Description
Any	The server's result.

Raises:

Type	Description
MCPError	When the call fails or the server is unreachable.

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

async def acall(self, **kwargs: Any) -> Any:
    """Forward the call to the MCP server.

    Args:
        **kwargs: Arguments matching the remote schema.

    Returns:
        The server's result.

    Raises:
        MCPError: When the call fails or the server is unreachable.
    """
    return await self.client.acall_tool(self.name, kwargs)

native

native() -> Any

Return the underlying MCP client session.

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

def native(self) -> Any:
    """Return the underlying MCP client session."""
    return self.client.native()

Memory

Memory(
    *,
    max_messages: int | None = None,
    return_system: bool = False,
    name: str | None = None,
    **config: Any
)

Bases: Component

Abstract conversation / long-term memory.

Memory is keyed by thread_id so one instance can serve many concurrent users — an agent handling a hundred chat sessions needs one memory object, not a hundred.

Parameters:

Name	Type	Description	Default
max_messages	int | None	Ceiling on how many messages :meth:aget returns. None means unlimited.	None
return_system	bool	Whether system messages are included in recall. Usually False: the system prompt is supplied by the agent, not the history.	False
name	str | None	Component name for traces.	None
**config	Any	Strategy-specific options.	{}

Example

Implementing a memory takes two methods::

class NullMemory(Memory):
    provider_name = "null"

    async def aadd(self, messages, *, thread_id="default"): ...
    async def aget(self, *, thread_id="default", query=None):
        return []

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

def __init__(
    self,
    *,
    max_messages: int | None = None,
    return_system: bool = False,
    name: str | None = None,
    **config: Any,
) -> None:
    super().__init__(
        name=name or self.provider_name,
        max_messages=max_messages,
        return_system=return_system,
        **config,
    )
    self.max_messages = max_messages
    self.return_system = return_system

aadd abstractmethod async

aadd(messages: Message | Sequence[Message], *, thread_id: str = DEFAULT_THREAD) -> None

Record one or more messages.

Parameters:

Name	Type	Description	Default
messages	Message | Sequence[Message]	A message or a sequence of them.	required
thread_id	str	Conversation this belongs to.	DEFAULT_THREAD

Raises:

Type	Description
MemoryError_	When the backend cannot persist the messages.

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

@abc.abstractmethod
async def aadd(
    self, messages: Message | Sequence[Message], *, thread_id: str = DEFAULT_THREAD
) -> None:
    """Record one or more messages.

    Args:
        messages: A message or a sequence of them.
        thread_id: Conversation this belongs to.

    Raises:
        WindlassMemoryError: When the backend cannot persist the messages.
    """

aget abstractmethod async

aget(
    *,
    thread_id: str = DEFAULT_THREAD,
    query: str | None = None,
    limit: int | None = None
) -> list[Message]

Recall messages for a thread.

Parameters:

Name	Type	Description	Default
thread_id	str	Conversation to recall.	DEFAULT_THREAD
query	str | None	Current user input. Semantic memories use it to rank recall; recency-based memories ignore it.	None
limit	int | None	Override for :attr:max_messages.	None

Returns:

Type	Description
list[Message]	Messages in chronological order, oldest first.

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

@abc.abstractmethod
async def aget(
    self,
    *,
    thread_id: str = DEFAULT_THREAD,
    query: str | None = None,
    limit: int | None = None,
) -> list[Message]:
    """Recall messages for a thread.

    Args:
        thread_id: Conversation to recall.
        query: Current user input. Semantic memories use it to rank recall;
            recency-based memories ignore it.
        limit: Override for :attr:`max_messages`.

    Returns:
        Messages in chronological order, oldest first.
    """

aclear async

aclear(*, thread_id: str | None = None) -> None

Forget a thread, or everything when thread_id is None.

Parameters:

Name	Type	Description	Default
thread_id	str | None	Thread to clear. None clears every thread.	None

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

async def aclear(self, *, thread_id: str | None = None) -> None:
    """Forget a thread, or everything when ``thread_id`` is ``None``.

    Args:
        thread_id: Thread to clear. ``None`` clears every thread.
    """

athreads async

athreads() -> list[str]

Return the thread ids this memory knows about.

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

async def athreads(self) -> list[str]:
    """Return the thread ids this memory knows about."""
    return []

add

add(messages: Message | Sequence[Message], *, thread_id: str = DEFAULT_THREAD) -> None

Blocking :meth:aadd.

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

def add(
    self, messages: Message | Sequence[Message], *, thread_id: str = DEFAULT_THREAD
) -> None:
    """Blocking :meth:`aadd`."""
    run_sync(self.aadd(messages, thread_id=thread_id))

get

get(
    *,
    thread_id: str = DEFAULT_THREAD,
    query: str | None = None,
    limit: int | None = None
) -> list[Message]

Blocking :meth:aget.

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

def get(
    self,
    *,
    thread_id: str = DEFAULT_THREAD,
    query: str | None = None,
    limit: int | None = None,
) -> list[Message]:
    """Blocking :meth:`aget`."""
    return run_sync(self.aget(thread_id=thread_id, query=query, limit=limit))

clear

clear(*, thread_id: str | None = None) -> None

Blocking :meth:aclear.

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

def clear(self, *, thread_id: str | None = None) -> None:
    """Blocking :meth:`aclear`."""
    run_sync(self.aclear(thread_id=thread_id))

threads

threads() -> list[str]

Blocking :meth:athreads.

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

def threads(self) -> list[str]:
    """Blocking :meth:`athreads`."""
    return run_sync(self.athreads())

Preprocessor

Preprocessor(*, name: str | None = None, **config: Any)

Bases: Component

Abstract document preprocessor.

Parameters:

Name	Type	Description	Default
name	str | None	Component name for traces.	None
**config	Any	Preprocessor-specific options.	{}

Example

Implementing a preprocessor takes one method::

class Shout(Preprocessor):
    provider_name = "shout"

    async def aprocess_one(self, document):
        return [document.model_copy(
            update={"content": document.content.upper()}
        )]

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

def __init__(self, *, name: str | None = None, **config: Any) -> None:
    self.name = name or getattr(type(self), "provider_name", None) or type(self).__name__
    self.config: dict[str, Any] = dict(config)
    self._log = get_logger(f"{self.kind}.{self.name}")

aprocess_one abstractmethod async

aprocess_one(document: Document) -> list[Document]

Transform a single document.

Parameters:

Name	Type	Description	Default
document	Document	The document to process.	required

Returns:

Type	Description
list[Document]	Zero, one or many documents. Return [] to drop the input.

Raises:

Type	Description
IngestionError	When the document cannot be processed and dropping it silently would hide a real problem.

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

@abc.abstractmethod
async def aprocess_one(self, document: Document) -> list[Document]:
    """Transform a single document.

    Args:
        document: The document to process.

    Returns:
        Zero, one or many documents. Return ``[]`` to drop the input.

    Raises:
        IngestionError: When the document cannot be processed and dropping
            it silently would hide a real problem.
    """

aprocess async

aprocess(
    documents: Sequence[Document], *, concurrency: int | None = None
) -> list[Document]

Process a batch of documents concurrently.

Parameters:

Name	Type	Description	Default
documents	Sequence[Document]	The documents to process.	required
concurrency	int | None	Maximum simultaneous invocations. Defaults to the global max_concurrency setting.	None

Returns:

Type	Description
list[Document]	The flattened output, preserving input order.

Performance

Order is preserved even though execution is concurrent, so a preprocessor that calls an LLM (metadata extraction, contextual enrichment) still yields a deterministic corpus.

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

async def aprocess(
    self, documents: Sequence[Document], *, concurrency: int | None = None
) -> list[Document]:
    """Process a batch of documents concurrently.

    Args:
        documents: The documents to process.
        concurrency: Maximum simultaneous invocations. Defaults to the
            global ``max_concurrency`` setting.

    Returns:
        The flattened output, preserving input order.

    Performance:
        Order is preserved even though execution is concurrent, so a
        preprocessor that calls an LLM (metadata extraction, contextual
        enrichment) still yields a deterministic corpus.
    """
    if not documents:
        return []
    limit = concurrency or settings().max_concurrency
    batches = await gather_bounded([self.aprocess_one(doc) for doc in documents], limit=limit)
    return [doc for batch in batches for doc in batch]

process

process(documents: Sequence[Document] | Document) -> list[Document]

Blocking :meth:aprocess.

Parameters:

Name	Type	Description	Default
documents	Sequence[Document] | Document	One document or a sequence of them.	required

Returns:

Type	Description
list[Document]	The processed documents.

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

def process(self, documents: Sequence[Document] | Document) -> list[Document]:
    """Blocking :meth:`aprocess`.

    Args:
        documents: One document or a sequence of them.

    Returns:
        The processed documents.
    """
    items = [documents] if isinstance(documents, Document) else list(documents)
    return run_sync(self.aprocess(items))

PreprocessorChain

PreprocessorChain(steps: Sequence[Preprocessor] = (), *, name: str | None = None)

Bases: Preprocessor

Runs several preprocessors in sequence.

Each step sees the full output of the previous one, which matters for corpus-level steps like deduplication that need to compare documents against each other rather than in isolation.

Parameters:

Name	Type	Description	Default
steps	Sequence[Preprocessor]	The preprocessors to run, in order.	()
name	str | None	Component name for traces.	None

Attributes:

Name	Type	Description
steps	list[Preprocessor]	The configured steps.

Example

from windlass.providers.preprocessors.clean import CleanPreprocessor chain = PreprocessorChain([CleanPreprocessor(min_length=0)]) from windlass.core.types import Document chain.process([Document(content=" hi ")])[0].content 'hi'

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

def __init__(self, steps: Sequence[Preprocessor] = (), *, name: str | None = None) -> None:
    super().__init__(name=name or "chain")
    self.steps: list[Preprocessor] = list(steps)

add

add(step: Preprocessor) -> PreprocessorChain

Append a step and return self for chaining.

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

def add(self, step: Preprocessor) -> PreprocessorChain:
    """Append a step and return ``self`` for chaining."""
    self.steps.append(step)
    return self

aprocess_one async

aprocess_one(document: Document) -> list[Document]

Run every step against a single document.

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

async def aprocess_one(self, document: Document) -> list[Document]:
    """Run every step against a single document."""
    return await self.aprocess([document])

aprocess async

aprocess(
    documents: Sequence[Document], *, concurrency: int | None = None
) -> list[Document]

Run every step in order over the whole batch.

Parameters:

Name	Type	Description	Default
documents	Sequence[Document]	The documents to process.	required
concurrency	int | None	Forwarded to each step.	None

Returns:

Type	Description
list[Document]	The output of the final step. Short-circuits to [] as soon as a
list[Document]	step drops everything.

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

async def aprocess(
    self, documents: Sequence[Document], *, concurrency: int | None = None
) -> list[Document]:
    """Run every step in order over the whole batch.

    Args:
        documents: The documents to process.
        concurrency: Forwarded to each step.

    Returns:
        The output of the final step. Short-circuits to ``[]`` as soon as a
        step drops everything.
    """
    current = list(documents)
    for step in self.steps:
        if not current:
            break
        current = await step.aprocess(current, concurrency=concurrency)
    return current

describe

describe() -> dict[str, Any]

Return a summary including each step.

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

def describe(self) -> dict[str, Any]:
    """Return a summary including each step."""
    return {**super().describe(), "steps": [s.describe() for s in self.steps]}

Retriever

Retriever(
    *,
    top_k: int = 5,
    score_threshold: float | None = None,
    rerank: Any = None,
    fetch_k: int | None = None,
    name: str | None = None,
    **config: Any
)

Bases: Component

Abstract retrieval strategy.

Parameters:

Name	Type	Description	Default
top_k	int	Default number of chunks to return.	5
score_threshold	float | None	Drop hits scoring below this. None keeps everything. Useful for "answer only if we actually found something" flows.	None
rerank	Any	Optional reranker applied to the candidate set before truncation. Any object with an arerank(query, hits, k) coroutine.	None
fetch_k	int | None	How many candidates to pull before reranking/filtering. Defaults to top_k when no reranker is configured, 4 * top_k when one is.	None
name	str | None	Component name for traces.	None
**config	Any	Strategy-specific options.	{}

Attributes:

Name	Type	Description
top_k		The configured result count.
requires_index	bool	Whether :meth:aindex must be called before searching.

Example

Implementing a retriever takes one method::

class RandomRetriever(Retriever):
    provider_name = "random"

    async def aretrieve_chunks(self, query, k, *, filters=None, **kw):
        picks = random.sample(self.corpus, k)
        return [ScoredChunk(chunk=c, score=1.0) for c in picks]

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

def __init__(
    self,
    *,
    top_k: int = 5,
    score_threshold: float | None = None,
    rerank: Any = None,
    fetch_k: int | None = None,
    name: str | None = None,
    **config: Any,
) -> None:
    if top_k <= 0:
        raise ValueError("top_k must be positive")
    super().__init__(
        name=name or self.provider_name,
        top_k=top_k,
        score_threshold=score_threshold,
        fetch_k=fetch_k,
        **config,
    )
    self.top_k = top_k
    self.score_threshold = score_threshold
    self.reranker = rerank
    self.fetch_k = fetch_k or (top_k * 4 if rerank is not None else top_k)

aretrieve_chunks abstractmethod async

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

Return candidate chunks for query.

The only method a strategy must implement. Thresholding, reranking, truncation and timing are applied by :meth:aretrieve.

Parameters:

Name	Type	Description	Default
query	str	The search query.	required
k	int	How many candidates to produce. This is fetch_k, not top_k, when a reranker is configured.	required
filters	MetadataFilter | None	Metadata constraints.	None
**kwargs	Any	Strategy-specific options.	{}

Returns:

Type	Description
list[ScoredChunk]	Scored chunks, ideally already sorted by descending score.

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

@abc.abstractmethod
async def aretrieve_chunks(
    self,
    query: str,
    k: int,
    *,
    filters: MetadataFilter | None = None,
    **kwargs: Any,
) -> list[ScoredChunk]:
    """Return candidate chunks for ``query``.

    The only method a strategy must implement. Thresholding, reranking,
    truncation and timing are applied by :meth:`aretrieve`.

    Args:
        query: The search query.
        k: How many candidates to produce. This is ``fetch_k``, not
            ``top_k``, when a reranker is configured.
        filters: Metadata constraints.
        **kwargs: Strategy-specific options.

    Returns:
        Scored chunks, ideally already sorted by descending score.
    """

aindex async

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

Add chunks to whatever index this retriever maintains.

The default is a no-op, which is right for retrievers that read from a shared vector store. Lexical retrievers override it.

Parameters:

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

Returns:

Type	Description
int	How many chunks were indexed.

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

async def aindex(self, chunks: Sequence[Chunk]) -> int:
    """Add chunks to whatever index this retriever maintains.

    The default is a no-op, which is right for retrievers that read from a
    shared vector store. Lexical retrievers override it.

    Args:
        chunks: Chunks to index.

    Returns:
        How many chunks were indexed.
    """
    return 0

index

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

Blocking :meth:aindex.

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

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

aretrieve async

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

Retrieve chunks for a query.

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
**kwargs	Any	Strategy-specific options.	{}

Returns:

Name	Type	Description
A	SearchResult	class:~windlass.core.types.SearchResult with ranked hits,
	SearchResult	candidate count and latency.

Raises:

Type	Description
RetrievalError	When the underlying strategy fails.

Performance

With a reranker configured, fetch_k candidates are retrieved and then narrowed to k. Raising fetch_k improves recall at the cost of one larger rerank call.

Example

import asyncio from windlass.providers.retrievers.bm25 import BM25Retriever from windlass.core.types import Chunk r = BM25Retriever() _ = r.index([Chunk(content="vector search rocks")]) asyncio.run(r.aretrieve("vector")).hits[0].score > 0 True

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

async def aretrieve(
    self,
    query: str,
    k: int | None = None,
    *,
    filters: MetadataFilter | None = None,
    **kwargs: Any,
) -> SearchResult:
    """Retrieve chunks for a query.

    Args:
        query: The search query.
        k: Override for :attr:`top_k`.
        filters: Metadata constraints.
        **kwargs: Strategy-specific options.

    Returns:
        A :class:`~windlass.core.types.SearchResult` with ranked hits,
        candidate count and latency.

    Raises:
        RetrievalError: When the underlying strategy fails.

    Performance:
        With a reranker configured, ``fetch_k`` candidates are retrieved and
        then narrowed to ``k``. Raising ``fetch_k`` improves recall at the
        cost of one larger rerank call.

    Example:
        >>> import asyncio
        >>> from windlass.providers.retrievers.bm25 import BM25Retriever
        >>> from windlass.core.types import Chunk
        >>> r = BM25Retriever()
        >>> _ = r.index([Chunk(content="vector search rocks")])
        >>> asyncio.run(r.aretrieve("vector")).hits[0].score > 0
        True
    """
    limit = k or self.top_k
    candidates = max(limit, self.fetch_k) if self.reranker is not None else limit
    started = time.perf_counter()

    hits = await self.aretrieve_chunks(query, candidates, filters=filters, **kwargs)

    if self.reranker is not None and hits:
        hits = await self.reranker.arerank(query, hits, limit)

    if self.score_threshold is not None:
        hits = [h for h in hits if h.score >= self.score_threshold]

    hits = sorted(hits, key=lambda h: h.score, reverse=True)[:limit]
    for position, hit in enumerate(hits, start=1):
        hit.rank = position
        if not hit.retriever:
            hit.retriever = self.name

    return SearchResult(
        query=query,
        hits=hits,
        total=len(hits),
        latency_ms=(time.perf_counter() - started) * 1000,
    )

retrieve

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

Blocking :meth:aretrieve.

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

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

abatch_retrieve async

abatch_retrieve(
    queries: Sequence[str],
    k: int | None = None,
    *,
    filters: MetadataFilter | None = None,
    concurrency: int | None = None
) -> list[SearchResult]

Retrieve for many queries concurrently.

Parameters:

Name	Type	Description	Default
queries	Sequence[str]	The queries to run.	required
k	int | None	Override for :attr:top_k.	None
filters	MetadataFilter | None	Metadata constraints applied to every query.	None
concurrency	int | None	Maximum simultaneous retrievals.	None

Returns:

Type	Description
list[SearchResult]	One result per query, in input order.

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

async def abatch_retrieve(
    self,
    queries: Sequence[str],
    k: int | None = None,
    *,
    filters: MetadataFilter | None = None,
    concurrency: int | None = None,
) -> list[SearchResult]:
    """Retrieve for many queries concurrently.

    Args:
        queries: The queries to run.
        k: Override for :attr:`top_k`.
        filters: Metadata constraints applied to every query.
        concurrency: Maximum simultaneous retrievals.

    Returns:
        One result per query, in input order.
    """
    from windlass.core.config import settings

    limit = concurrency or settings().max_concurrency
    return await gather_bounded(
        [self.aretrieve(q, k, filters=filters) for q in queries], limit=limit
    )

batch_retrieve

batch_retrieve(queries: Sequence[str], k: int | None = None) -> list[SearchResult]

Blocking :meth:abatch_retrieve.

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

def batch_retrieve(self, queries: Sequence[str], k: int | None = None) -> list[SearchResult]:
    """Blocking :meth:`abatch_retrieve`."""
    return run_sync(self.abatch_retrieve(queries, k))

Tool

Tool(
    *,
    name: str,
    description: str = "",
    parameters: dict[str, Any] | None = None,
    timeout: float | None = None,
    requires_approval: bool = False,
    **config: Any
)

Bases: Component

Abstract agent-callable capability.

Parameters:

Name	Type	Description	Default
name	str	Tool name the model sees. Must match ^[a-zA-Z0-9_-]{1,64}$, which is what the major providers accept.	required
description	str	What the tool does. This is prompt text — the model chooses tools based on it, so vague descriptions cause bad calls.	''
parameters	dict[str, Any] | None	JSON Schema for the arguments. Derived automatically by the decorator; supply it manually when subclassing.	None
timeout	float | None	Seconds before the call is abandoned.	None
requires_approval	bool	When True the agent pauses for human approval before invoking it. Use this for anything that spends money or mutates state.	False
**config	Any	Tool-specific options.	{}

Attributes:

Name	Type	Description
description		The model-facing description.
parameters	dict[str, Any]	The JSON Schema for arguments.
requires_approval		Whether human-in-the-loop approval is required.

Example

Implementing a stateful tool::

class SearchTool(Tool):
    def __init__(self, client):
        super().__init__(
            name="search",
            description="Search the product catalogue.",
            parameters={
                "type": "object",
                "properties": {"q": {"type": "string"}},
                "required": ["q"],
            },
        )
        self.client = client

    async def acall(self, **kwargs):
        return await self.client.search(kwargs["q"])

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

def __init__(
    self,
    *,
    name: str,
    description: str = "",
    parameters: dict[str, Any] | None = None,
    timeout: float | None = None,
    requires_approval: bool = False,
    **config: Any,
) -> None:
    _validate_name(name)
    super().__init__(
        name=name,
        description=description,
        parameters=parameters,
        timeout=timeout,
        requires_approval=requires_approval,
        **config,
    )
    self.description = description or (self.__doc__ or "").strip().split("\n")[0]
    self.parameters: dict[str, Any] = parameters or {
        "type": "object",
        "properties": {},
    }
    self.timeout = timeout if timeout is not None else DEFAULT_TOOL_TIMEOUT
    self.requires_approval = requires_approval

acall abstractmethod async

acall(**kwargs: Any) -> Any

Execute the tool.

Parameters:

Name	Type	Description	Default
**kwargs	Any	Arguments matching :attr:parameters.	{}

Returns:

Type	Description
Any	Any JSON-serialisable value. It is rendered to a string for the
Any	model and kept intact on :attr:~windlass.core.types.ToolResult.data.

Raises:

Type	Description
Exception	Anything. The agent catches it and reports the failure back to the model rather than crashing the run.

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

@abc.abstractmethod
async def acall(self, **kwargs: Any) -> Any:
    """Execute the tool.

    Args:
        **kwargs: Arguments matching :attr:`parameters`.

    Returns:
        Any JSON-serialisable value. It is rendered to a string for the
        model and kept intact on :attr:`~windlass.core.types.ToolResult.data`.

    Raises:
        Exception: Anything. The agent catches it and reports the failure
            back to the model rather than crashing the run.
    """

schema

schema(*, style: str = 'openai') -> dict[str, Any]

Return the tool definition in a provider's format.

Parameters:

Name	Type	Description	Default
style	str	"openai" (the default, also used by Groq and Ollama), "anthropic", or "gemini".	'openai'

Returns:

Type	Description
dict[str, Any]	A dict ready to drop into that provider's request.

Raises:

Type	Description
ValueError	For an unknown style.

Example

from windlass.tools import tool @tool ... def ping() -> str: ... '''Ping.''' ... return "pong" ping.schema(style="anthropic")["name"] 'ping'

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

def schema(self, *, style: str = "openai") -> dict[str, Any]:
    """Return the tool definition in a provider's format.

    Args:
        style: ``"openai"`` (the default, also used by Groq and Ollama),
            ``"anthropic"``, or ``"gemini"``.

    Returns:
        A dict ready to drop into that provider's request.

    Raises:
        ValueError: For an unknown style.

    Example:
        >>> from windlass.tools import tool
        >>> @tool
        ... def ping() -> str:
        ...     '''Ping.'''
        ...     return "pong"
        >>> ping.schema(style="anthropic")["name"]
        'ping'
    """
    match style:
        case "openai":
            return {
                "type": "function",
                "function": {
                    "name": self.name,
                    "description": self.description,
                    "parameters": self.parameters,
                },
            }
        case "anthropic":
            return {
                "name": self.name,
                "description": self.description,
                "input_schema": self.parameters,
            }
        case "gemini":
            return {
                "name": self.name,
                "description": self.description,
                "parameters": _strip_unsupported(self.parameters),
            }
        case _:
            raise ValueError(
                f"Unknown tool schema style {style!r}; use openai, anthropic or gemini."
            )

ainvoke async

ainvoke(call: ToolCall) -> ToolResult

Execute a model-issued tool call and wrap the outcome.

Failures are captured rather than raised: the agent needs to hand the error text back to the model so it can recover, which is impossible if the exception unwinds the loop.

Parameters:

Name	Type	Description	Default
call	ToolCall	The call requested by the model.	required

Returns:

Name	Type	Description
A	ToolResult	class:~windlass.core.types.ToolResult, with is_error set
	ToolResult	when the tool raised or timed out.

Performance

Enforces :attr:timeout. A hung tool fails the call, not the run.

Example

import asyncio from windlass.core.types import ToolCall from windlass.tools import tool @tool ... def echo(text: str) -> str: ... '''Echo text.''' ... return text asyncio.run(echo.ainvoke(ToolCall(name="echo", arguments={"text": "hi"}))).content 'hi'

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

async def ainvoke(self, call: ToolCall) -> ToolResult:
    """Execute a model-issued tool call and wrap the outcome.

    Failures are captured rather than raised: the agent needs to hand the
    error text back to the model so it can recover, which is impossible if
    the exception unwinds the loop.

    Args:
        call: The call requested by the model.

    Returns:
        A :class:`~windlass.core.types.ToolResult`, with ``is_error`` set
        when the tool raised or timed out.

    Performance:
        Enforces :attr:`timeout`. A hung tool fails the call, not the run.

    Example:
        >>> import asyncio
        >>> from windlass.core.types import ToolCall
        >>> from windlass.tools import tool
        >>> @tool
        ... def echo(text: str) -> str:
        ...     '''Echo text.'''
        ...     return text
        >>> asyncio.run(echo.ainvoke(ToolCall(name="echo", arguments={"text": "hi"}))).content
        'hi'
    """
    import asyncio

    started = time.perf_counter()
    try:
        value = await asyncio.wait_for(self.acall(**call.arguments), timeout=self.timeout)
    except TimeoutError:
        return ToolResult(
            call_id=call.id,
            name=self.name,
            content=f"Tool {self.name!r} timed out after {self.timeout}s.",
            is_error=True,
            duration_ms=(time.perf_counter() - started) * 1000,
        )
    except Exception as exc:
        self._log.warning("Tool %s failed: %s", self.name, exc)
        return ToolResult(
            call_id=call.id,
            name=self.name,
            content=f"{type(exc).__name__}: {exc}",
            data=None,
            is_error=True,
            duration_ms=(time.perf_counter() - started) * 1000,
        )

    return ToolResult(
        call_id=call.id,
        name=self.name,
        content=render_result(value),
        data=value,
        duration_ms=(time.perf_counter() - started) * 1000,
    )

invoke

invoke(call: ToolCall) -> ToolResult

Blocking :meth:ainvoke.

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

def invoke(self, call: ToolCall) -> ToolResult:
    """Blocking :meth:`ainvoke`."""
    return run_sync(self.ainvoke(call))

arun async

arun(**kwargs: Any) -> ToolResult

Execute the tool directly, outside an agent.

Parameters:

Name	Type	Description	Default
**kwargs	Any	Tool arguments.	{}

Returns:

Type	Description
ToolResult	The result.

Raises:

Type	Description
ToolExecutionError	When the tool fails. Unlike :meth:ainvoke, direct invocation raises — there is no model to recover.

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

async def arun(self, **kwargs: Any) -> ToolResult:
    """Execute the tool directly, outside an agent.

    Args:
        **kwargs: Tool arguments.

    Returns:
        The result.

    Raises:
        ToolExecutionError: When the tool fails. Unlike :meth:`ainvoke`,
            direct invocation raises — there is no model to recover.
    """
    result = await self.ainvoke(ToolCall(name=self.name, arguments=kwargs))
    if result.is_error:
        raise ToolExecutionError(self.name, RuntimeError(result.content))
    return result

run

run(**kwargs: Any) -> ToolResult

Blocking :meth:arun.

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

def run(self, **kwargs: Any) -> ToolResult:
    """Blocking :meth:`arun`."""
    return run_sync(self.arun(**kwargs))

describe

describe() -> dict[str, Any]

Return a JSON-safe summary including the schema.

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

def describe(self) -> dict[str, Any]:
    """Return a JSON-safe summary including the schema."""
    return {
        **super().describe(),
        "description": self.description,
        "parameters": self.parameters,
        "requires_approval": self.requires_approval,
    }

NullTracer

NullTracer(**config: Any)

Bases: Tracer

A tracer that discards everything.

Used whenever observability is not configured, so instrumented code paths never need a if tracer is not None: guard.

Example

with NullTracer().span("x") as span: ... span.set_output(1)

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

def __init__(self, **config: Any) -> None:
    super().__init__(enabled=False, **config)

start_span

start_span(span: Span) -> None

Discards the span.

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

def start_span(self, span: Span) -> None:
    """Discards the span."""

end_span

end_span(span: Span) -> None

Discards the span.

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

def end_span(self, span: Span) -> None:
    """Discards the span."""

Span

Span(
    name: str,
    *,
    kind: str = "chain",
    parent_id: str | None = None,
    metadata: dict[str, Any] | None = None,
    trace_id: str | None = None
)

One timed, attributed unit of work.

Spans are created by :meth:Tracer.span and finished automatically when the context manager exits. Exceptions are recorded before propagating, so a failed run is still fully traced.

Parameters:

Name	Type	Description	Default
name	str	Human readable operation name.	required
kind	str	One of :data:SPAN_KINDS.	'chain'
parent_id	str | None	Enclosing span's id, when nested.	None
metadata	dict[str, Any] | None	Static attributes attached at creation.	None
trace_id	str | None	Groups spans belonging to one top-level request.	None

Attributes:

Name	Type	Description
id		Unique span id.
started_at		Monotonic start time.
ended_at	float | None	Monotonic end time, set on :meth:end.
error	str | None	Error message, when the span failed.
usage	Usage | None	Token accounting, for model spans.

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

def __init__(
    self,
    name: str,
    *,
    kind: str = "chain",
    parent_id: str | None = None,
    metadata: dict[str, Any] | None = None,
    trace_id: str | None = None,
) -> None:
    self.id = uuid.uuid4().hex[:16]
    self.trace_id = trace_id or self.id
    self.parent_id = parent_id
    self.name = name
    self.kind = kind
    self.metadata: dict[str, Any] = dict(metadata or {})
    self.inputs: Any = None
    self.outputs: Any = None
    self.usage: Usage | None = None
    self.error: str | None = None
    self.started_at = time.perf_counter()
    self.ended_at: float | None = None
    self._native: Any = None

duration_ms property

duration_ms: float

Elapsed time in milliseconds — live until the span ends.

set_input

set_input(value: Any) -> Span

Record the operation's input. Returns self for chaining.

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

def set_input(self, value: Any) -> Span:
    """Record the operation's input. Returns ``self`` for chaining."""
    self.inputs = value
    return self

set_output

set_output(value: Any) -> Span

Record the operation's output. Returns self for chaining.

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

def set_output(self, value: Any) -> Span:
    """Record the operation's output. Returns ``self`` for chaining."""
    self.outputs = value
    return self

set_usage

set_usage(usage: Usage) -> Span

Record token accounting. Returns self for chaining.

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

def set_usage(self, usage: Usage) -> Span:
    """Record token accounting. Returns ``self`` for chaining."""
    self.usage = usage
    return self

set_metadata

set_metadata(**fields: Any) -> Span

Merge extra attributes. Returns self for chaining.

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

def set_metadata(self, **fields: Any) -> Span:
    """Merge extra attributes. Returns ``self`` for chaining."""
    self.metadata.update(fields)
    return self

set_error

set_error(error: BaseException | str) -> Span

Mark the span as failed. Returns self for chaining.

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

def set_error(self, error: BaseException | str) -> Span:
    """Mark the span as failed. Returns ``self`` for chaining."""
    self.error = str(error)
    return self

end

end() -> Span

Stop the clock. Idempotent.

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

def end(self) -> Span:
    """Stop the clock. Idempotent."""
    if self.ended_at is None:
        self.ended_at = time.perf_counter()
    return self

attach_native

attach_native(obj: Any) -> Span

Associate the backend's own span object, for Level 3 access.

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

def attach_native(self, obj: Any) -> Span:
    """Associate the backend's own span object, for Level 3 access."""
    self._native = obj
    return self

native

native() -> Any

Return the backend's span object, if the tracer created one.

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

def native(self) -> Any:
    """Return the backend's span object, if the tracer created one."""
    return self._native

to_dict

to_dict() -> dict[str, Any]

Return a JSON-safe representation of the span.

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

def to_dict(self) -> dict[str, Any]:
    """Return a JSON-safe representation of the span."""
    return {
        "id": self.id,
        "trace_id": self.trace_id,
        "parent_id": self.parent_id,
        "name": self.name,
        "kind": self.kind,
        "duration_ms": round(self.duration_ms, 2),
        "metadata": self.metadata,
        "inputs": _clip(self.inputs),
        "outputs": _clip(self.outputs),
        "usage": self.usage.model_dump() if self.usage else None,
        "error": self.error,
    }

Tracer

Tracer(
    *,
    enabled: bool = True,
    project: str | None = None,
    tags: tuple[str, ...] = (),
    name: str | None = None,
    **config: Any
)

Bases: Component

Abstract tracing backend.

Parameters:

Name	Type	Description	Default
enabled	bool	Master switch. Disabled tracers still create spans (so code paths are identical) but never export them.	True
project	str | None	Project / session name shown in the backend's UI.	None
tags	tuple[str, ...]	Tags applied to every span.	()
name	str | None	Component name.	None
**config	Any	Backend-specific options.	{}

Example

Implementing a tracer takes one method::

class PrintTracer(Tracer):
    provider_name = "print"

    def start_span(self, span): ...
    def end_span(self, span):
        print(span.to_dict())

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

def __init__(
    self,
    *,
    enabled: bool = True,
    project: str | None = None,
    tags: tuple[str, ...] = (),
    name: str | None = None,
    **config: Any,
) -> None:
    from windlass.core.config import settings

    super().__init__(
        name=name or self.provider_name,
        enabled=enabled,
        project=project or settings().project,
        tags=tags,
        **config,
    )
    self.enabled = enabled
    self.project = project or settings().project
    self.tags = tuple(tags)
    self._stack: list[Span] = []

start_span abstractmethod

start_span(span: Span) -> None

Called when a span begins.

Parameters:

Name	Type	Description	Default
span	Span	The span that just started. Attach a backend object with :meth:Span.attach_native if the backend has one.	required

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

@abc.abstractmethod
def start_span(self, span: Span) -> None:
    """Called when a span begins.

    Args:
        span: The span that just started. Attach a backend object with
            :meth:`Span.attach_native` if the backend has one.
    """

end_span

end_span(span: Span) -> None

Called when a span ends, successfully or not.

Parameters:

Name	Type	Description	Default
span	Span	The finished span.	required

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

def end_span(self, span: Span) -> None:
    """Called when a span ends, successfully or not.

    Args:
        span: The finished span.
    """

flush

flush() -> None

Force any buffered spans to be exported.

Backends that batch should override this; a process that exits without flushing loses its last traces.

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

def flush(self) -> None:
    """Force any buffered spans to be exported.

    Backends that batch should override this; a process that exits without
    flushing loses its last traces.
    """

span

span(
    name: str, *, kind: str = "chain", inputs: Any = None, **metadata: Any
) -> Iterator[Span]

Open a span around a block of work.

Spans nest automatically: a span opened inside another becomes its child, which is what produces a readable trace tree from plain code.

Parameters:

Name	Type	Description	Default
name	str	Operation name.	required
kind	str	One of :data:SPAN_KINDS.	'chain'
inputs	Any	The operation's input, recorded immediately.	None
**metadata	Any	Extra attributes.	{}

Yields:

Type	Description
Span	The live :class:Span.

Raises:

Type	Description
Exception	Anything the block raises, after recording it.

Example

from windlass.providers.observability.console import ConsoleTracer with ConsoleTracer(enabled=False).span("work", kind="tool") as s: ... s.set_output(42)

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

@contextmanager
def span(
    self,
    name: str,
    *,
    kind: str = "chain",
    inputs: Any = None,
    **metadata: Any,
) -> Iterator[Span]:
    """Open a span around a block of work.

    Spans nest automatically: a span opened inside another becomes its
    child, which is what produces a readable trace tree from plain code.

    Args:
        name: Operation name.
        kind: One of :data:`SPAN_KINDS`.
        inputs: The operation's input, recorded immediately.
        **metadata: Extra attributes.

    Yields:
        The live :class:`Span`.

    Raises:
        Exception: Anything the block raises, after recording it.

    Example:
        >>> from windlass.providers.observability.console import ConsoleTracer
        >>> with ConsoleTracer(enabled=False).span("work", kind="tool") as s:
        ...     s.set_output(42)
        <Span tool:work ...>
    """
    parent = self._stack[-1] if self._stack else None
    current = Span(
        name,
        kind=kind,
        parent_id=parent.id if parent else None,
        trace_id=parent.trace_id if parent else None,
        metadata={**dict.fromkeys(self.tags, True), **metadata},
    )
    if inputs is not None:
        current.set_input(inputs)

    self._stack.append(current)
    if self.enabled:
        try:
            self.start_span(current)
        except Exception as exc:
            self._log.debug("Tracer %s failed to start a span: %s", self.name, exc)
    try:
        yield current
    except BaseException as exc:
        current.set_error(exc)
        raise
    finally:
        current.end()
        self._stack.pop()
        if self.enabled:
            try:
                self.end_span(current)
            except Exception as exc:
                self._log.debug("Tracer %s failed to end a span: %s", self.name, exc)

current_span

current_span() -> Span | None

Return the innermost open span, if any.

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

def current_span(self) -> Span | None:
    """Return the innermost open span, if any."""
    return self._stack[-1] if self._stack else None

event

event(name: str, **fields: Any) -> None

Record a zero-duration event on the current span.

Parameters:

Name	Type	Description	Default
name	str	Event name.	required
**fields	Any	Event attributes.	{}

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

def event(self, name: str, **fields: Any) -> None:
    """Record a zero-duration event on the current span.

    Args:
        name: Event name.
        **fields: Event attributes.
    """
    span = self.current_span()
    if span is not None:
        span.metadata.setdefault("events", []).append({"name": name, **fields})

VectorStore

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

Bases: Component

Abstract vector database.

Parameters:

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

Attributes:

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

Example

Implementing a store means four methods::

class MyStore(VectorStore):
    provider_name = "mine"

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

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

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

aadd abstractmethod async

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

Insert or update chunks.

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

Parameters:

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

Returns:

Type	Description
int	How many chunks were written.

Raises:

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

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

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

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

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

    Returns:
        How many chunks were written.

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

asearch abstractmethod async

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

Find the k nearest chunks to vector.

Parameters:

Name	Type	Description	Default
vector	Sequence[float]	The query embedding.	required
k	int	How many results to return.	5
filters	MetadataFilter | None	Metadata constraints. Stores without native filtering should call :meth:match_filters client-side.	None
**kwargs	Any	Store-specific search options.	{}

Returns:

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

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

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

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

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

adelete abstractmethod async

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

Delete chunks by id or by metadata filter.

Parameters:

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

Returns:

Type	Description
int	How many chunks were deleted.

Raises:

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

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

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

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

    Returns:
        How many chunks were deleted.

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

acount abstractmethod async

acount() -> int

Return how many chunks the collection holds.

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

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

aget async

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

Fetch chunks by id.

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

Parameters:

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

Returns:

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

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

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

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

    Args:
        ids: Chunk ids to fetch.

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

aclear async

aclear() -> None

Remove every chunk from the collection.

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

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

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

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

apersist async

apersist() -> None

Flush pending writes to durable storage.

A no-op for stores that write through.

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

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

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

add

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

Blocking :meth:aadd.

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

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

search

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

Blocking :meth:asearch.

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

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

delete

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

Blocking :meth:adelete.

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

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

count

count() -> int

Blocking :meth:acount.

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

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

get

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

Blocking :meth:aget.

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

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

clear

clear() -> None

Blocking :meth:aclear.

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

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

persist

persist() -> None

Blocking :meth:apersist.

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

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

validate_embeddings staticmethod

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

Assert that every chunk carries an embedding.

Parameters:

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

Raises:

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

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

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

    Args:
        chunks: Chunks about to be written.

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

match_filters staticmethod

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

Evaluate a metadata filter client-side.

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

Parameters:

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

Returns:

Type	Description
bool	True when the metadata satisfies every constraint.

Example

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

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

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

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

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

    Returns:
        True when the metadata satisfies every constraint.

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

rank staticmethod

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

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

Parameters:

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

Returns:

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

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

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

    Args:
        hits: Unordered hits.

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