windlass.core

core

Windlass core: the framework machinery that knows nothing about AI.

This package holds the parts every component depends on and that depend on nothing themselves:

• :mod:~windlass.core.types — the shared data model (Document, Chunk, Message, Completion, ...).
• :mod:~windlass.core.exceptions — one error hierarchy for the whole framework.
• :mod:~windlass.core.registry — name-to-implementation lookup and plugins.
• :mod:~windlass.core.container — dependency injection.
• :mod:~windlass.core.config — settings from env, file and code.
• :mod:~windlass.core.lazy — optional dependency loading with good errors.
• :mod:~windlass.core.concurrency — async/sync bridging and bounded parallelism.
• :mod:~windlass.core.retry, :mod:~windlass.core.cache — reliability and speed.
• :mod:~windlass.core.text, :mod:~windlass.core.vectors — shared primitives.

Importing this package is cheap and pulls in no optional dependency.

Cache

Bases: ABC

Key/value cache interface used across Windlass.

Implementations must be safe to call from multiple threads. Async methods default to running the sync ones, which is right for in-process backends; a network-backed cache should override them.

get abstractmethod

get(key: str) -> Any

Return the cached value, or None when absent or expired.

Source code in src\windlass\core\cache.py

@abstractmethod
def get(self, key: str) -> Any:
    """Return the cached value, or ``None`` when absent or expired."""

set abstractmethod

set(key: str, value: Any, ttl: float | None = None) -> None

Store value under key with an optional TTL in seconds.

Source code in src\windlass\core\cache.py

@abstractmethod
def set(self, key: str, value: Any, ttl: float | None = None) -> None:
    """Store ``value`` under ``key`` with an optional TTL in seconds."""

delete abstractmethod

delete(key: str) -> None

Remove key. No-op when it is not present.

Source code in src\windlass\core\cache.py

@abstractmethod
def delete(self, key: str) -> None:
    """Remove ``key``. No-op when it is not present."""

clear abstractmethod

clear() -> None

Drop every entry.

Source code in src\windlass\core\cache.py

@abstractmethod
def clear(self) -> None:
    """Drop every entry."""

aget async

aget(key: str) -> Any

Async :meth:get.

Source code in src\windlass\core\cache.py

async def aget(self, key: str) -> Any:
    """Async :meth:`get`."""
    return self.get(key)

aset async

aset(key: str, value: Any, ttl: float | None = None) -> None

Async :meth:set.

Source code in src\windlass\core\cache.py

async def aset(self, key: str, value: Any, ttl: float | None = None) -> None:
    """Async :meth:`set`."""
    self.set(key, value, ttl)

MemoryCache

MemoryCache(max_size: int = 1024, ttl: float | None = 3600.0)

Bases: Cache

Thread-safe in-process cache with TTL expiry and LRU eviction.

Parameters:

Name	Type	Description	Default
max_size	int	Maximum number of live entries. The least recently used entry is evicted when the cache is full.	1024
ttl	float | None	Default lifetime in seconds; None means entries never expire.	3600.0

Attributes:

Name	Type	Description
hits		Number of successful lookups since construction.
misses		Number of failed lookups since construction.

Example

cache = MemoryCache(max_size=2) cache.set("a", 1) cache.get("a") 1 cache.set("b", 2); cache.set("c", 3) # evicts "a" cache.get("a") is None True

Source code in src\windlass\core\cache.py

def __init__(self, max_size: int = 1024, ttl: float | None = 3600.0) -> None:
    self.max_size = max(1, max_size)
    self.ttl = ttl
    self.hits = 0
    self.misses = 0
    self._data: OrderedDict[str, tuple[Any, float | None]] = OrderedDict()
    self._lock = threading.RLock()

get

get(key: str) -> Any

Return the value for key, or None when missing or expired.

Source code in src\windlass\core\cache.py

def get(self, key: str) -> Any:
    """Return the value for ``key``, or ``None`` when missing or expired."""
    with self._lock:
        entry = self._data.get(key)
        if entry is None:
            self.misses += 1
            return None
        value, expires_at = entry
        if expires_at is not None and expires_at <= time.time():
            del self._data[key]
            self.misses += 1
            return None
        self._data.move_to_end(key)
        self.hits += 1
        return value

set

set(key: str, value: Any, ttl: float | None = None) -> None

Store value, evicting the least recently used entry if needed.

Source code in src\windlass\core\cache.py

def set(self, key: str, value: Any, ttl: float | None = None) -> None:
    """Store ``value``, evicting the least recently used entry if needed."""
    lifetime = self.ttl if ttl is None else ttl
    expires_at = time.time() + lifetime if lifetime else None
    with self._lock:
        self._data[key] = (value, expires_at)
        self._data.move_to_end(key)
        while len(self._data) > self.max_size:
            self._data.popitem(last=False)

delete

delete(key: str) -> None

Remove key if present.

Source code in src\windlass\core\cache.py

def delete(self, key: str) -> None:
    """Remove ``key`` if present."""
    with self._lock:
        self._data.pop(key, None)

clear

clear() -> None

Drop every entry and reset the counters.

Source code in src\windlass\core\cache.py

def clear(self) -> None:
    """Drop every entry and reset the counters."""
    with self._lock:
        self._data.clear()
        self.hits = self.misses = 0

stats

stats() -> dict[str, Any]

Return hit/miss counters and the current size.

Returns:

Type	Description
dict[str, Any]	A dict with hits, misses, size, max_size and
dict[str, Any]	hit_rate.

Source code in src\windlass\core\cache.py

def stats(self) -> dict[str, Any]:
    """Return hit/miss counters and the current size.

    Returns:
        A dict with ``hits``, ``misses``, ``size``, ``max_size`` and
        ``hit_rate``.
    """
    with self._lock:
        total = self.hits + self.misses
        return {
            "hits": self.hits,
            "misses": self.misses,
            "size": len(self._data),
            "max_size": self.max_size,
            "hit_rate": (self.hits / total) if total else 0.0,
        }

NullCache

Bases: Cache

A cache that never stores anything.

Used whenever caching is disabled, so call sites need no if cache: branches.

Example

c = NullCache() c.set("k", 1) c.get("k") is None True

get

get(key: str) -> Any

Always returns None.

Source code in src\windlass\core\cache.py

def get(self, key: str) -> Any:
    """Always returns ``None``."""
    return None

set

set(key: str, value: Any, ttl: float | None = None) -> None

Discards the value.

Source code in src\windlass\core\cache.py

def set(self, key: str, value: Any, ttl: float | None = None) -> None:
    """Discards the value."""

delete

delete(key: str) -> None

No-op.

Source code in src\windlass\core\cache.py

def delete(self, key: str) -> None:
    """No-op."""

clear

clear() -> None

No-op.

Source code in src\windlass\core\cache.py

def clear(self) -> None:
    """No-op."""

WindlassSettings

Bases: BaseSettings

Process-wide defaults and credentials.

Every field can be set with an environment variable. Provider keys accept both the Windlass-prefixed name and the vendor's conventional name, so an existing OPENAI_API_KEY in your shell just works.

Attributes:

Name	Type	Description
default_llm	str	Provider used when a builder does not name one.
default_model	str	Model id used when a builder does not name one.
default_embedding	str	Embedding provider used by RAG builders.
default_vectordb	str	Vector store used by RAG builders.
default_chunker	str	Chunking strategy used by RAG builders.
default_retriever	str	Retrieval strategy used by RAG builders.
temperature	float	Default sampling temperature.
max_tokens	int | None	Default completion ceiling; None defers to the provider.
request_timeout	float	Per-request timeout in seconds.
max_concurrency	int	Ceiling on in-flight provider calls for batch work.
batch_size	int	Default batch size for embedding and indexing.
openai_api_key	SecretStr | None	Credential for OpenAI-compatible endpoints.
openai_base_url	str | None	Override for proxies and Azure-style gateways.
anthropic_api_key	SecretStr | None	Credential for Anthropic.
google_api_key	SecretStr | None	Credential for Gemini.
groq_api_key	SecretStr | None	Credential for Groq.
huggingface_api_key	SecretStr | None	Token for the HuggingFace Inference API.
ollama_base_url	str	Where the local Ollama daemon listens.
pinecone_api_key	SecretStr | None	Credential for Pinecone.
langsmith_api_key	SecretStr | None	Credential for LangSmith tracing.
langfuse_public_key	SecretStr | None	Public key for Langfuse.
langfuse_secret_key	SecretStr | None	Secret key for Langfuse.
langfuse_host	str	Langfuse endpoint.
telemetry	bool	Whether Windlass may emit anonymous usage counters. Off by default; Windlass ships no telemetry backend, the flag exists so downstream distributions can honour it.
strict_plugins	bool	Fail loudly when a third-party plugin cannot load.
retry	RetryConfig	Retry policy.
cache	CacheConfig	Cache policy.

Example

s = WindlassSettings(default_llm="anthropic", temperature=0.0) s.default_llm, s.temperature ('anthropic', 0.0)

secret

secret(field: str) -> str | None

Return a credential's plaintext value, or None when unset.

Parameters:

Name	Type	Description	Default
field	str	Attribute name, e.g. "openai_api_key".	required

Returns:

Type	Description
str | None	The secret value, or None.

Raises:

Type	Description
ConfigurationError	If field is not a known setting.

Source code in src\windlass\core\config.py

def secret(self, field: str) -> str | None:
    """Return a credential's plaintext value, or ``None`` when unset.

    Args:
        field: Attribute name, e.g. ``"openai_api_key"``.

    Returns:
        The secret value, or ``None``.

    Raises:
        ConfigurationError: If ``field`` is not a known setting.
    """
    if not hasattr(self, field):
        raise ConfigurationError(f"Unknown setting {field!r}.")
    value = getattr(self, field)
    if value is None:
        return None
    return value.get_secret_value() if isinstance(value, SecretStr) else str(value)

require_secret

require_secret(field: str, *, provider: str, env_var: str) -> str

Return a credential or raise a clear authentication error.

Parameters:

Name	Type	Description	Default
field	str	Attribute name holding the credential.	required
provider	str	Provider name for the error message.	required
env_var	str	The environment variable a user would normally set.	required

Returns:

Type	Description
str	The secret value.

Raises:

Type	Description
AuthenticationError	When the credential is missing.

Source code in src\windlass\core\config.py

def require_secret(self, field: str, *, provider: str, env_var: str) -> str:
    """Return a credential or raise a clear authentication error.

    Args:
        field: Attribute name holding the credential.
        provider: Provider name for the error message.
        env_var: The environment variable a user would normally set.

    Returns:
        The secret value.

    Raises:
        AuthenticationError: When the credential is missing.
    """
    value = self.secret(field)
    if not value:
        from windlass.core.exceptions import AuthenticationError

        raise AuthenticationError(
            f"No API key configured for the {provider} provider.",
            provider=provider,
            hint=(
                f"Set {env_var} in your environment, add it to a .env file, or pass\n"
                f"    Windlass.agent().llm('{provider}', api_key='...')"
            ),
        )
    return value

masked

masked() -> dict[str, Any]

Return settings as a dict with every secret replaced by '***'.

Safe to log or print. Used by windlass config.

Source code in src\windlass\core\config.py

def masked(self) -> dict[str, Any]:
    """Return settings as a dict with every secret replaced by ``'***'``.

    Safe to log or print. Used by ``windlass config``.
    """
    data = self.model_dump(mode="json")
    for key, value in list(data.items()):
        if isinstance(getattr(self, key, None), SecretStr):
            data[key] = "***" if value else None
    return data

Container

Container(parent: Container | None = None, registry: Registry | None = None)

A hierarchical, thread-safe dependency container.

Parameters:

Name	Type	Description	Default
parent	Container | None	Optional parent whose bindings are inherited.	None
registry	Registry | None	Component registry used to resolve string names. Defaults to the global :data:~windlass.core.registry.REGISTRY.	None

Example

c = Container() _ = c.bind("greeting", lambda: "hello") c.resolve("greeting") 'hello' child = c.scope() _ = child.bind("greeting", lambda: "hi") c.resolve("greeting"), child.resolve("greeting") ('hello', 'hi')

Source code in src\windlass\core\container.py

def __init__(self, parent: Container | None = None, registry: Registry | None = None) -> None:
    self._parent = parent
    inherited: Registry = parent._registry if parent is not None else REGISTRY
    self._registry: Registry = registry or inherited
    self._bindings: dict[Any, Binding] = {}
    self._lock = threading.RLock()

bind

bind(
    key: Any,
    factory: Callable[..., Any],
    *,
    singleton: bool = True,
    override: bool = True
) -> Container

Bind key to a factory.

Parameters:

Name	Type	Description	Default
key	Any	A string, a type, or any hashable token.	required
factory	Callable[..., Any]	Zero-argument callable, or one accepting the container.	required
singleton	bool	Reuse the first produced value.	True
override	bool	Replace an existing binding. When False, an existing binding is kept and the call is a no-op.	True

Returns:

Type	Description
Container	self, so calls chain.

Raises:

Type	Description
ConfigurationError	If factory is not callable.

Source code in src\windlass\core\container.py

def bind(
    self,
    key: Any,
    factory: Callable[..., Any],
    *,
    singleton: bool = True,
    override: bool = True,
) -> Container:
    """Bind ``key`` to a factory.

    Args:
        key: A string, a type, or any hashable token.
        factory: Zero-argument callable, or one accepting the container.
        singleton: Reuse the first produced value.
        override: Replace an existing binding. When False, an existing
            binding is kept and the call is a no-op.

    Returns:
        ``self``, so calls chain.

    Raises:
        ConfigurationError: If ``factory`` is not callable.
    """
    if not callable(factory):
        raise ConfigurationError(
            f"Binding for {key!r} must be callable, got {type(factory).__name__}.",
            hint="Pass a factory (or use bind_instance for a ready-made object).",
        )
    with self._lock:
        if not override and key in self._bindings:
            return self
        self._bindings[key] = Binding(factory=factory, singleton=singleton)
    return self

bind_instance

bind_instance(key: Any, instance: Any, *, override: bool = True) -> Container

Bind key to an already-constructed object.

Parameters:

Name	Type	Description	Default
key	Any	Lookup key.	required
instance	Any	The object to return on resolution.	required
override	bool	Replace an existing binding.	True

Returns:

Type	Description
Container	self.

Source code in src\windlass\core\container.py

def bind_instance(self, key: Any, instance: Any, *, override: bool = True) -> Container:
    """Bind ``key`` to an already-constructed object.

    Args:
        key: Lookup key.
        instance: The object to return on resolution.
        override: Replace an existing binding.

    Returns:
        ``self``.
    """
    with self._lock:
        if not override and key in self._bindings:
            return self
        self._bindings[key] = Binding(
            factory=lambda: instance, singleton=True, value=instance, produced=True
        )
    return self

unbind

unbind(key: Any) -> None

Remove a local binding. Parent bindings become visible again.

Source code in src\windlass\core\container.py

def unbind(self, key: Any) -> None:
    """Remove a local binding. Parent bindings become visible again."""
    with self._lock:
        self._bindings.pop(key, None)

has

has(key: Any) -> bool

Return whether key resolves here or in any ancestor.

Source code in src\windlass\core\container.py

def has(self, key: Any) -> bool:
    """Return whether ``key`` resolves here or in any ancestor."""
    if key in self._bindings:
        return True
    return self._parent.has(key) if self._parent else False

resolve

resolve(key: type[_T]) -> _T

resolve(key: type[_T], default: _T) -> _T

resolve(key: Any, default: Any = ...) -> Any

resolve(key: Any, default: Any = _UNSET) -> Any

Produce the value bound to key.

Singletons are produced at most once, under a lock, so concurrent first resolutions do not both run the factory.

Parameters:

Name	Type	Description	Default
key	Any	Lookup key.	required
default	Any	Returned instead of raising when nothing is bound.	_UNSET

Returns:

Type	Description
Any	The resolved value.

Raises:

Type	Description
ConfigurationError	When nothing is bound and no default was given, or when the factory itself fails.

Source code in src\windlass\core\container.py

def resolve(self, key: Any, default: Any = _UNSET) -> Any:
    """Produce the value bound to ``key``.

    Singletons are produced at most once, under a lock, so concurrent first
    resolutions do not both run the factory.

    Args:
        key: Lookup key.
        default: Returned instead of raising when nothing is bound.

    Returns:
        The resolved value.

    Raises:
        ConfigurationError: When nothing is bound and no ``default`` was
            given, or when the factory itself fails.
    """
    binding = self._find(key)
    if binding is None:
        if default is not _UNSET:
            return default
        raise ConfigurationError(
            f"Nothing is bound for {_label(key)}.",
            hint="Bind it with container.bind(...) or pass it explicitly to the builder.",
            context={"key": _label(key), "bound": [_label(k) for k in self.keys()]},
        )

    if binding.singleton and binding.produced:
        return binding.value

    with self._lock:
        if binding.singleton and binding.produced:
            return binding.value
        try:
            value = _invoke(binding.factory, self)
        except ConfigurationError:
            raise
        except Exception as exc:
            raise ConfigurationError(
                f"Factory for {_label(key)} raised {type(exc).__name__}: {exc}"
            ) from exc
        if binding.singleton:
            binding.value = value
            binding.produced = True
        return value

keys

keys() -> list[Any]

Return every key visible from here, including inherited ones.

Source code in src\windlass\core\container.py

def keys(self) -> list[Any]:
    """Return every key visible from here, including inherited ones."""
    seen: dict[Any, None] = {}
    node: Container | None = self
    while node is not None:
        for key in node._bindings:
            seen.setdefault(key, None)
        node = node._parent
    return list(seen)

component

component(kind: str, spec: Any = None, /, **config: Any) -> Any

Resolve a component from a name, an instance, or a factory.

This is the single funnel through which every builder turns user input into a live component, and the reason .llm(...) accepts three different kinds of argument.

Parameters:

Name	Type	Description	Default
kind	str	Component kind, e.g. "chunker".	required
spec	Any	One of: None — resolve the binding for kind if present, otherwise fail with a clear message. str — a registry name; config is passed to the constructor. an instance — used as-is (config must be empty). a callable — invoked, optionally with the container.	None
**config	Any	Constructor keyword arguments for the string form.	{}

Returns:

Type	Description
Any	The live component.

Raises:

Type	Description
ComponentNotFoundError	For an unknown registry name.
ConfigurationError	For an unusable spec.

Example

c = Container() llm = c.component("llm", "fake", responses=["hi"]) llm.complete("x").content 'hi'

Source code in src\windlass\core\container.py

def component(
    self,
    kind: str,
    spec: Any = None,
    /,
    **config: Any,
) -> Any:
    """Resolve a component from a name, an instance, or a factory.

    This is the single funnel through which every builder turns user input
    into a live component, and the reason ``.llm(...)`` accepts three
    different kinds of argument.

    Args:
        kind: Component kind, e.g. ``"chunker"``.
        spec: One of:

            * ``None`` — resolve the binding for ``kind`` if present,
              otherwise fail with a clear message.
            * ``str`` — a registry name; ``config`` is passed to the
              constructor.
            * an instance — used as-is (``config`` must be empty).
            * a callable — invoked, optionally with the container.
        **config: Constructor keyword arguments for the string form.

    Returns:
        The live component.

    Raises:
        ComponentNotFoundError: For an unknown registry name.
        ConfigurationError: For an unusable ``spec``.

    Example:
        >>> c = Container()
        >>> llm = c.component("llm", "fake", responses=["hi"])
        >>> llm.complete("x").content
        'hi'
    """
    if spec is None:
        if self.has(kind):
            return self.resolve(kind)
        raise ConfigurationError(
            f"No {kind} configured.",
            hint=f"Pass one explicitly, e.g. .{kind}('name'), "
            f"or bind it with container.bind('{kind}', factory).",
        )

    if isinstance(spec, str):
        return self._registry.create(kind, spec, **config)

    if callable(spec) and (inspect.isfunction(spec) or inspect.ismethod(spec)):
        produced = _invoke(spec, self)
        return produced

    if isinstance(spec, type):
        return spec(**config)

    if config:
        raise ConfigurationError(
            f"Cannot apply configuration to an already-constructed {kind}.",
            hint=f"Either pass .{kind}('name', **config) or configure the instance yourself.",
            context={"kind": kind, "config": sorted(config)},
        )
    return spec

bind_component

bind_component(
    kind: str, spec: Any = None, /, *, singleton: bool = True, **config: Any
) -> Container

Bind kind to a lazily resolved component.

The component is not constructed until something resolves it, which keeps builder chains cheap and lets configuration keep changing right up until the first call.

Parameters:

Name	Type	Description	Default
kind	str	Component kind, used as the binding key.	required
spec	Any	Name, instance or factory — see :meth:component.	None
singleton	bool	Reuse the constructed component.	True
**config	Any	Constructor keyword arguments.	{}

Returns:

Type	Description
Container	self.

Source code in src\windlass\core\container.py

def bind_component(
    self, kind: str, spec: Any = None, /, *, singleton: bool = True, **config: Any
) -> Container:
    """Bind ``kind`` to a lazily resolved component.

    The component is not constructed until something resolves it, which
    keeps builder chains cheap and lets configuration keep changing right up
    until the first call.

    Args:
        kind: Component kind, used as the binding key.
        spec: Name, instance or factory — see :meth:`component`.
        singleton: Reuse the constructed component.
        **config: Constructor keyword arguments.

    Returns:
        ``self``.
    """
    return self.bind(kind, lambda: self.component(kind, spec, **config), singleton=singleton)

scope

scope() -> Container

Return a child container inheriting every binding from this one.

Source code in src\windlass\core\container.py

def scope(self) -> Container:
    """Return a child container inheriting every binding from this one."""
    return Container(parent=self, registry=self._registry)

AgentError

AgentError(
    message: str, *, hint: str | None = None, context: dict[str, Any] | None = None
)

Bases: WindlassError

The agent runtime could not complete the request.

Source code in src\windlass\core\exceptions.py

def __init__(
    self,
    message: str,
    *,
    hint: str | None = None,
    context: dict[str, Any] | None = None,
) -> None:
    self.message = message
    self.hint = hint
    self.context: dict[str, Any] = dict(context or {})
    super().__init__(self._render())

AuthenticationError

AuthenticationError(
    message: str,
    *,
    provider: str | None = None,
    status_code: int | None = None,
    hint: str | None = None,
    context: dict[str, Any] | None = None,
    original: BaseException | None = None
)

Bases: ProviderError

Credentials are missing, expired, or rejected by the provider.

Source code in src\windlass\core\exceptions.py

def __init__(
    self,
    message: str,
    *,
    provider: str | None = None,
    status_code: int | None = None,
    hint: str | None = None,
    context: dict[str, Any] | None = None,
    original: BaseException | None = None,
) -> None:
    self.provider = provider
    self.status_code = status_code
    self.original = original
    ctx = {"provider": provider, **(context or {})}
    if status_code is not None:
        ctx.setdefault("status_code", status_code)
    super().__init__(message, hint=hint, context=ctx)

ComponentNotFoundError

ComponentNotFoundError(kind: str, name: str, available: list[str] | None = None)

Bases: RegistryError

No component is registered under the requested name.

The message lists the available names so typos are obvious.

Source code in src\windlass\core\exceptions.py

def __init__(self, kind: str, name: str, available: list[str] | None = None) -> None:
    self.kind = kind
    self.name = name
    self.available = sorted(available or [])
    options = ", ".join(self.available) if self.available else "(none registered)"
    super().__init__(
        f"No {kind} named {name!r} is registered.",
        hint=(
            f"Available {kind}s: {options}\n"
            f"Register your own with @windlass.register.{kind}('{name}')."
        ),
        context={"kind": kind, "name": name, "available": self.available},
    )

ConfigurationError

ConfigurationError(
    message: str, *, hint: str | None = None, context: dict[str, Any] | None = None
)

Bases: WindlassError

A component was configured with invalid or incomplete settings.

Source code in src\windlass\core\exceptions.py

def __init__(
    self,
    message: str,
    *,
    hint: str | None = None,
    context: dict[str, Any] | None = None,
) -> None:
    self.message = message
    self.hint = hint
    self.context: dict[str, Any] = dict(context or {})
    super().__init__(self._render())

EvaluationError

EvaluationError(
    message: str, *, hint: str | None = None, context: dict[str, Any] | None = None
)

Bases: WindlassError

An evaluation run or metric computation failed.

Source code in src\windlass\core\exceptions.py

def __init__(
    self,
    message: str,
    *,
    hint: str | None = None,
    context: dict[str, Any] | None = None,
) -> None:
    self.message = message
    self.hint = hint
    self.context: dict[str, Any] = dict(context or {})
    super().__init__(self._render())

GuardrailViolation

GuardrailViolation(
    message: str,
    *,
    stage: str = "input",
    rule: str | None = None,
    detections: list[dict[str, Any]] | None = None
)

Bases: WindlassError

A guardrail blocked an input or an output.

Parameters:

Name	Type	Description	Default
message	str	What the guardrail objected to.	required
stage	str	"input" or "output".	'input'
rule	str | None	Identifier of the rule that fired.	None
detections	list[dict[str, Any]] | None	Structured detail about each detection.	None

Source code in src\windlass\core\exceptions.py

def __init__(
    self,
    message: str,
    *,
    stage: str = "input",
    rule: str | None = None,
    detections: list[dict[str, Any]] | None = None,
) -> None:
    self.stage = stage
    self.rule = rule
    self.detections = detections or []
    super().__init__(
        message,
        hint="Relax the policy with .guardrails(on_violation='redact') "
        "to sanitise instead of block.",
        context={"stage": stage, "rule": rule, "detections": self.detections},
    )

IngestionError

IngestionError(
    message: str, *, hint: str | None = None, context: dict[str, Any] | None = None
)

Bases: PipelineError

A document could not be loaded, preprocessed, chunked or indexed.

Source code in src\windlass\core\exceptions.py

def __init__(
    self,
    message: str,
    *,
    hint: str | None = None,
    context: dict[str, Any] | None = None,
) -> None:
    self.message = message
    self.hint = hint
    self.context: dict[str, Any] = dict(context or {})
    super().__init__(self._render())

MaxIterationsExceeded

MaxIterationsExceeded(limit: int)

Bases: AgentError

The agent hit its reasoning-step budget without producing an answer.

Source code in src\windlass\core\exceptions.py

def __init__(self, limit: int) -> None:
    super().__init__(
        f"Agent exceeded its budget of {limit} reasoning steps without finishing.",
        hint="Raise the ceiling with .max_iterations(n) or simplify the task.",
        context={"limit": limit},
    )

MCPError

MCPError(
    message: str, *, hint: str | None = None, context: dict[str, Any] | None = None
)

Bases: WindlassError

An MCP server could not be reached or returned an error.

Source code in src\windlass\core\exceptions.py

def __init__(
    self,
    message: str,
    *,
    hint: str | None = None,
    context: dict[str, Any] | None = None,
) -> None:
    self.message = message
    self.hint = hint
    self.context: dict[str, Any] = dict(context or {})
    super().__init__(self._render())

MissingDependencyError

MissingDependencyError(
    feature: str,
    *,
    extra: str,
    package: str | None = None,
    original: BaseException | None = None
)

Bases: WindlassError

An optional dependency required by the requested feature is not installed.

Windlass keeps its core install tiny; every integration lives behind an optional dependency group. When a feature is used without its group being installed this error explains exactly which command fixes it.

Parameters:

Name	Type	Description	Default
feature	str	Human readable feature name, e.g. "The Pinecone vector store".	required
extra	str	The extras group to install, e.g. "vectordb".	required
package	str | None	The distribution that failed to import, for the context dict.	None
original	BaseException | None	The underlying :class:ImportError, kept for debugging.	None

Example

raise MissingDependencyError("Evaluation", extra="evaluation") Traceback (most recent call last): windlass.core.exceptions.MissingDependencyError: Evaluation is not installed. Hint: Run: pip install "windlass[evaluation]"

Source code in src\windlass\core\exceptions.py

def __init__(
    self,
    feature: str,
    *,
    extra: str,
    package: str | None = None,
    original: BaseException | None = None,
) -> None:
    self.feature = feature
    self.extra = extra
    self.package = package
    self.original = original
    super().__init__(
        f"{feature} is not installed.",
        hint=f'Run:\n\n    pip install "windlass[{extra}]"',
        context={"feature": feature, "extra": extra, "package": package},
    )

PipelineError

PipelineError(
    message: str, *, hint: str | None = None, context: dict[str, Any] | None = None
)

Bases: WindlassError

A RAG or ingestion pipeline step failed.

Source code in src\windlass\core\exceptions.py

def __init__(
    self,
    message: str,
    *,
    hint: str | None = None,
    context: dict[str, Any] | None = None,
) -> None:
    self.message = message
    self.hint = hint
    self.context: dict[str, Any] = dict(context or {})
    super().__init__(self._render())

PluginError

PluginError(
    message: str, *, hint: str | None = None, context: dict[str, Any] | None = None
)

Bases: WindlassError

A third-party plugin failed to load or misbehaved during discovery.

Source code in src\windlass\core\exceptions.py

def __init__(
    self,
    message: str,
    *,
    hint: str | None = None,
    context: dict[str, Any] | None = None,
) -> None:
    self.message = message
    self.hint = hint
    self.context: dict[str, Any] = dict(context or {})
    super().__init__(self._render())

ProviderError

ProviderError(
    message: str,
    *,
    provider: str | None = None,
    status_code: int | None = None,
    hint: str | None = None,
    context: dict[str, Any] | None = None,
    original: BaseException | None = None
)

Bases: WindlassError

A backing provider (LLM, vector store, ...) returned an error.

Parameters:

Name	Type	Description	Default
message	str	What went wrong.	required
provider	str | None	Which provider raised it.	None
status_code	int | None	HTTP status, when the failure came from an HTTP call. Pass this whenever you have it. Retry classification reads it (see :func:windlass.core.retry.is_retryable), so an adapter that omits it turns a transient 429 or 503 — which one retry would have fixed — into a hard failure of the whole run.	None
hint	str | None	Actionable remediation.	None
context	dict[str, Any] | None	Structured detail.	None
original	BaseException | None	The underlying exception, kept for debugging.	None

Attributes:

Name	Type	Description
provider		The provider name.
status_code		The HTTP status, when known.
original		The wrapped exception, when there was one.

Example

err = ProviderError("upstream boom", provider="acme", status_code=503) from windlass.core.retry import is_retryable is_retryable(err) True

Source code in src\windlass\core\exceptions.py

def __init__(
    self,
    message: str,
    *,
    provider: str | None = None,
    status_code: int | None = None,
    hint: str | None = None,
    context: dict[str, Any] | None = None,
    original: BaseException | None = None,
) -> None:
    self.provider = provider
    self.status_code = status_code
    self.original = original
    ctx = {"provider": provider, **(context or {})}
    if status_code is not None:
        ctx.setdefault("status_code", status_code)
    super().__init__(message, hint=hint, context=ctx)

RateLimitError

RateLimitError(message: str, *, retry_after: float | None = None, **kwargs: Any)

Bases: ProviderError

The provider applied rate limiting or quota exhaustion.

Attributes:

Name	Type	Description
retry_after		Seconds the provider asked us to wait, when advertised.

Source code in src\windlass\core\exceptions.py

def __init__(self, message: str, *, retry_after: float | None = None, **kwargs: Any) -> None:
    self.retry_after = retry_after
    kwargs.setdefault("context", {})["retry_after"] = retry_after
    super().__init__(message, **kwargs)

RegistryError

RegistryError(
    message: str, *, hint: str | None = None, context: dict[str, Any] | None = None
)

Bases: WindlassError

Base class for component-registry problems.

Source code in src\windlass\core\exceptions.py

def __init__(
    self,
    message: str,
    *,
    hint: str | None = None,
    context: dict[str, Any] | None = None,
) -> None:
    self.message = message
    self.hint = hint
    self.context: dict[str, Any] = dict(context or {})
    super().__init__(self._render())

RetrievalError

RetrievalError(
    message: str, *, hint: str | None = None, context: dict[str, Any] | None = None
)

Bases: PipelineError

Retrieval failed or returned an unusable result.

Source code in src\windlass\core\exceptions.py

def __init__(
    self,
    message: str,
    *,
    hint: str | None = None,
    context: dict[str, Any] | None = None,
) -> None:
    self.message = message
    self.hint = hint
    self.context: dict[str, Any] = dict(context or {})
    super().__init__(self._render())

ToolError

ToolError(
    message: str, *, hint: str | None = None, context: dict[str, Any] | None = None
)

Bases: WindlassError

Base class for tool related failures.

Source code in src\windlass\core\exceptions.py

def __init__(
    self,
    message: str,
    *,
    hint: str | None = None,
    context: dict[str, Any] | None = None,
) -> None:
    self.message = message
    self.hint = hint
    self.context: dict[str, Any] = dict(context or {})
    super().__init__(self._render())

ToolExecutionError

ToolExecutionError(name: str, original: BaseException)

Bases: ToolError

A tool raised while executing.

The original exception is preserved on :attr:original and as the __cause__ of this error.

Source code in src\windlass\core\exceptions.py

def __init__(self, name: str, original: BaseException) -> None:
    self.tool = name
    self.original = original
    super().__init__(
        f"Tool {name!r} raised {type(original).__name__}: {original}",
        context={"tool": name, "error_type": type(original).__name__},
    )

ToolNotFoundError

ToolNotFoundError(name: str, available: list[str] | None = None)

Bases: ToolError

The model asked to call a tool that is not registered.

Source code in src\windlass\core\exceptions.py

def __init__(self, name: str, available: list[str] | None = None) -> None:
    options = ", ".join(sorted(available or [])) or "(no tools bound)"
    super().__init__(
        f"The model requested an unknown tool {name!r}.",
        hint=f"Tools available to this agent: {options}",
        context={"tool": name, "available": sorted(available or [])},
    )

ValidationError

ValidationError(
    message: str, *, hint: str | None = None, context: dict[str, Any] | None = None
)

Bases: WindlassError

User supplied data failed validation before reaching a provider.

Source code in src\windlass\core\exceptions.py

def __init__(
    self,
    message: str,
    *,
    hint: str | None = None,
    context: dict[str, Any] | None = None,
) -> None:
    self.message = message
    self.hint = hint
    self.context: dict[str, Any] = dict(context or {})
    super().__init__(self._render())

WindlassError

WindlassError(
    message: str, *, hint: str | None = None, context: dict[str, Any] | None = None
)

Bases: Exception

Base class for every error raised by Windlass.

Parameters:

Name	Type	Description	Default
message	str	Human readable description of what went wrong.	required
hint	str | None	Optional actionable remediation shown after the message.	None
context	dict[str, Any] | None	Arbitrary structured detail (provider name, model, ids, ...). Useful for logging and for tests that assert on specifics.	None

Attributes:

Name	Type	Description
message		The raw message without the hint appended.
hint		The remediation hint, if any.
context	dict[str, Any]	Structured detail about the failure.

Example

raise WindlassError("boom", hint="try again", context={"attempt": 2}) Traceback (most recent call last): windlass.core.exceptions.WindlassError: boom Hint: try again

Source code in src\windlass\core\exceptions.py

def __init__(
    self,
    message: str,
    *,
    hint: str | None = None,
    context: dict[str, Any] | None = None,
) -> None:
    self.message = message
    self.hint = hint
    self.context: dict[str, Any] = dict(context or {})
    super().__init__(self._render())

ComponentSpec dataclass

ComponentSpec(
    kind: str,
    name: str,
    target: Any = None,
    path: str | None = None,
    extra: str | None = None,
    aliases: tuple[str, ...] = (),
    description: str = "",
    defaults: dict[str, Any] = dict(),
    origin: str = "user",
)

A registry entry: how to obtain one implementation of a component kind.

Exactly one of target or path is set. path entries are resolved on first use, which is what makes registration free of import cost.

Attributes:

Name	Type	Description
kind	str	The component kind, e.g. "chunker".
name	str	The lookup name, e.g. "semantic".
target	Any	The resolved class or factory, once known.
path	str | None	Dotted "module:Attribute" path for deferred resolution.
extra	str | None	Extras group needed for this component, used in error messages.
aliases	tuple[str, ...]	Additional names that resolve to this spec.
description	str	One-line summary shown by windlass list.
defaults	dict[str, Any]	Keyword arguments merged under user config at construction.
origin	str	"builtin", "plugin" or "user". Purely informational.

resolve

resolve() -> Any

Import and cache the target if it was registered lazily.

Returns:

Type	Description
Any	The class or factory for this component.

Raises:

Type	Description
RegistryError	If the dotted path cannot be imported. Missing optional dependencies surface as :class:~windlass.core.exceptions.MissingDependencyError from inside the provider module instead.

Source code in src\windlass\core\registry.py

def resolve(self) -> Any:
    """Import and cache the target if it was registered lazily.

    Returns:
        The class or factory for this component.

    Raises:
        RegistryError: If the dotted path cannot be imported. Missing
            *optional dependencies* surface as
            :class:`~windlass.core.exceptions.MissingDependencyError` from
            inside the provider module instead.
    """
    if self.target is not None:
        return self.target
    if not self.path:  # pragma: no cover - guarded at registration
        raise RegistryError(f"Component {self.kind}:{self.name} has neither target nor path.")
    module_path, _, attr = self.path.partition(":")
    try:
        module = importlib.import_module(module_path)
    except Exception as exc:
        from windlass.core.exceptions import MissingDependencyError

        if isinstance(exc, MissingDependencyError):
            raise
        raise RegistryError(
            f"Failed to import {self.kind} {self.name!r} from {self.path!r}: {exc}",
            hint=(f'Try: pip install "windlass[{self.extra}]"' if self.extra else None),
            context={"kind": self.kind, "name": self.name, "path": self.path},
        ) from exc
    self.target = getattr(module, attr) if attr else module
    return self.target

Registry

Registry(*, discover: bool = False)

A thread-safe, kind-partitioned component registry.

You almost never construct this yourself — use the module-level :data:REGISTRY singleton, or the :data:register decorator namespace.

Parameters:

Name	Type	Description	Default
discover	bool	Whether the first lookup should scan installed distributions for entry-point plugins. The process-wide :data:REGISTRY does; a registry you construct yourself does not, so it stays an isolated container holding exactly what you put in it. Without that distinction, installing any third-party Windlass plugin would change the contents of every registry in the process, including the ones tests build to get a clean slate.	False

Example

reg = Registry() _ = reg.register("chunker", "shouty", lambda: None, description="demo") reg.has("chunker", "shouty") True reg.names("chunker") ['shouty']

Opt in explicitly when you do want the installed plugins:

discovered = Registry(discover=True) isinstance(discovered.load_plugins(), list) True

Source code in src\windlass\core\registry.py

def __init__(self, *, discover: bool = False) -> None:
    self._specs: dict[str, dict[str, ComponentSpec]] = {k: {} for k in COMPONENT_KINDS}
    self._aliases: dict[str, dict[str, str]] = {k: {} for k in COMPONENT_KINDS}
    self._lock = threading.RLock()
    self._discover = discover
    self._plugins_loaded = not discover

add_kind

add_kind(kind: str) -> None

Declare a brand new component kind at runtime.

Parameters:

Name	Type	Description	Default
kind	str	The new kind's name.	required

Source code in src\windlass\core\registry.py

def add_kind(self, kind: str) -> None:
    """Declare a brand new component kind at runtime.

    Args:
        kind: The new kind's name.
    """
    with self._lock:
        self._specs.setdefault(kind, {})
        self._aliases.setdefault(kind, {})

kinds

kinds() -> list[str]

Return every known component kind, sorted.

Source code in src\windlass\core\registry.py

def kinds(self) -> list[str]:
    """Return every known component kind, sorted."""
    return sorted(self._specs)

register

register(
    kind: str,
    name: str,
    target: Any,
    *,
    aliases: tuple[str, ...] | str = (),
    description: str = "",
    extra: str | None = None,
    defaults: dict[str, Any] | None = None,
    origin: str = "user",
    override: bool = False
) -> ComponentSpec

Register a concrete class or factory.

Parameters:

Name	Type	Description	Default
kind	str	Component kind.	required
name	str	Lookup name; case-insensitive.	required
target	Any	A class or a zero-or-more-argument factory callable.	required
aliases	tuple[str, ...] | str	Extra names resolving to the same component.	()
description	str	One-line summary for CLI listings and docs.	''
extra	str | None	Extras group required, used in error hints.	None
defaults	dict[str, Any] | None	Config defaults merged beneath user-supplied kwargs.	None
origin	str	builtin / plugin / user.	'user'
override	bool	Allow replacing an existing registration.	False

Returns:

Type	Description
ComponentSpec	The stored :class:ComponentSpec.

Raises:

Type	Description
DuplicateComponentError	If name is taken and override is False.

Source code in src\windlass\core\registry.py

def register(
    self,
    kind: str,
    name: str,
    target: Any,
    *,
    aliases: tuple[str, ...] | str = (),
    description: str = "",
    extra: str | None = None,
    defaults: dict[str, Any] | None = None,
    origin: str = "user",
    override: bool = False,
) -> ComponentSpec:
    """Register a concrete class or factory.

    Args:
        kind: Component kind.
        name: Lookup name; case-insensitive.
        target: A class or a zero-or-more-argument factory callable.
        aliases: Extra names resolving to the same component.
        description: One-line summary for CLI listings and docs.
        extra: Extras group required, used in error hints.
        defaults: Config defaults merged beneath user-supplied kwargs.
        origin: ``builtin`` / ``plugin`` / ``user``.
        override: Allow replacing an existing registration.

    Returns:
        The stored :class:`ComponentSpec`.

    Raises:
        DuplicateComponentError: If ``name`` is taken and ``override`` is False.
    """
    return self._store(
        ComponentSpec(
            kind=kind,
            name=name.lower(),
            target=target,
            aliases=(aliases,) if isinstance(aliases, str) else tuple(aliases),
            description=description,
            extra=extra,
            defaults=dict(defaults or {}),
            origin=origin,
        ),
        override=override,
    )

register_lazy

register_lazy(
    kind: str,
    name: str,
    path: str,
    *,
    aliases: tuple[str, ...] | str = (),
    description: str = "",
    extra: str | None = None,
    defaults: dict[str, Any] | None = None,
    origin: str = "builtin",
    override: bool = False
) -> ComponentSpec

Register a component by dotted path without importing it.

This is how every built-in provider is wired up: the module is only imported the first time somebody actually asks for that component.

Parameters:

Name	Type	Description	Default
kind	str	Component kind.	required
name	str	Lookup name.	required
path	str	"package.module:Attribute".	required
aliases	tuple[str, ...] | str	Extra names resolving here.	()
description	str	One-line summary.	''
extra	str | None	Extras group required by the target module.	None
defaults	dict[str, Any] | None	Config defaults.	None
origin	str	Provenance label.	'builtin'
override	bool	Allow replacing an existing registration.	False

Returns:

Type	Description
ComponentSpec	The stored :class:ComponentSpec.

Source code in src\windlass\core\registry.py

def register_lazy(
    self,
    kind: str,
    name: str,
    path: str,
    *,
    aliases: tuple[str, ...] | str = (),
    description: str = "",
    extra: str | None = None,
    defaults: dict[str, Any] | None = None,
    origin: str = "builtin",
    override: bool = False,
) -> ComponentSpec:
    """Register a component by dotted path without importing it.

    This is how every built-in provider is wired up: the module is only
    imported the first time somebody actually asks for that component.

    Args:
        kind: Component kind.
        name: Lookup name.
        path: ``"package.module:Attribute"``.
        aliases: Extra names resolving here.
        description: One-line summary.
        extra: Extras group required by the target module.
        defaults: Config defaults.
        origin: Provenance label.
        override: Allow replacing an existing registration.

    Returns:
        The stored :class:`ComponentSpec`.
    """
    return self._store(
        ComponentSpec(
            kind=kind,
            name=name.lower(),
            path=path,
            aliases=(aliases,) if isinstance(aliases, str) else tuple(aliases),
            description=description,
            extra=extra,
            defaults=dict(defaults or {}),
            origin=origin,
        ),
        override=override,
    )

unregister

unregister(kind: str, name: str) -> None

Remove a component and its aliases. No-op when absent.

Source code in src\windlass\core\registry.py

def unregister(self, kind: str, name: str) -> None:
    """Remove a component and its aliases. No-op when absent."""
    with self._lock:
        bucket = self._bucket(kind)
        resolved = self._aliases[kind].pop(name.lower(), name.lower())
        spec = bucket.pop(resolved, None)
        if spec is not None:
            for alias in spec.aliases:
                self._aliases[kind].pop(alias.lower(), None)

get

get(kind: str, name: str) -> ComponentSpec

Return the spec registered under name.

Parameters:

Name	Type	Description	Default
kind	str	Component kind.	required
name	str	Name or alias, case-insensitive.	required

Returns:

Type	Description
ComponentSpec	The matching spec.

Raises:

Type	Description
ComponentNotFoundError	If nothing matches. The error lists all available names for that kind.

Source code in src\windlass\core\registry.py

def get(self, kind: str, name: str) -> ComponentSpec:
    """Return the spec registered under ``name``.

    Args:
        kind: Component kind.
        name: Name or alias, case-insensitive.

    Returns:
        The matching spec.

    Raises:
        ComponentNotFoundError: If nothing matches. The error lists all
            available names for that kind.
    """
    self.ensure_plugins_loaded()
    with self._lock:
        bucket = self._bucket(kind)
        key = name.lower()
        key = self._aliases[kind].get(key, key)
        spec = bucket.get(key)
        if spec is None:
            raise ComponentNotFoundError(kind, name, list(bucket))
        return spec

has

has(kind: str, name: str) -> bool

Return whether name resolves for kind.

Source code in src\windlass\core\registry.py

def has(self, kind: str, name: str) -> bool:
    """Return whether ``name`` resolves for ``kind``."""
    try:
        self.get(kind, name)
    except (ComponentNotFoundError, RegistryError):
        return False
    return True

names

names(kind: str) -> list[str]

Return every registered name for kind, sorted.

Source code in src\windlass\core\registry.py

def names(self, kind: str) -> list[str]:
    """Return every registered name for ``kind``, sorted."""
    self.ensure_plugins_loaded()
    with self._lock:
        return sorted(self._bucket(kind))

specs

specs(kind: str) -> list[ComponentSpec]

Return every spec for kind, sorted by name.

Source code in src\windlass\core\registry.py

def specs(self, kind: str) -> list[ComponentSpec]:
    """Return every spec for ``kind``, sorted by name."""
    self.ensure_plugins_loaded()
    with self._lock:
        return [self._bucket(kind)[n] for n in sorted(self._bucket(kind))]

catalog

catalog() -> dict[str, list[ComponentSpec]]

Return every spec grouped by kind — powers windlass list.

Source code in src\windlass\core\registry.py

def catalog(self) -> dict[str, list[ComponentSpec]]:
    """Return every spec grouped by kind — powers ``windlass list``."""
    return {kind: self.specs(kind) for kind in self.kinds()}

create

create(kind: str, name: str, /, **config: Any) -> Any

Instantiate a registered component.

Registered defaults are merged underneath config, so caller arguments always win.

Parameters:

Name	Type	Description	Default
kind	str	Component kind.	required
name	str	Name or alias.	required
**config	Any	Constructor keyword arguments.	{}

Returns:

Type	Description
Any	The constructed component.

Raises:

Type	Description
ComponentNotFoundError	If the name is not registered.
ConfigurationError	If the constructor rejects the arguments.

Example

from windlass.core.registry import REGISTRY llm = REGISTRY.create("llm", "fake", responses=["hi"]) llm.complete("anything").content 'hi'

Source code in src\windlass\core\registry.py

def create(self, kind: str, name: str, /, **config: Any) -> Any:
    """Instantiate a registered component.

    Registered defaults are merged *underneath* ``config``, so caller
    arguments always win.

    Args:
        kind: Component kind.
        name: Name or alias.
        **config: Constructor keyword arguments.

    Returns:
        The constructed component.

    Raises:
        ComponentNotFoundError: If the name is not registered.
        ConfigurationError: If the constructor rejects the arguments.

    Example:
        >>> from windlass.core.registry import REGISTRY
        >>> llm = REGISTRY.create("llm", "fake", responses=["hi"])
        >>> llm.complete("anything").content
        'hi'
    """
    spec = self.get(kind, name)
    factory = spec.resolve()
    merged = {**spec.defaults, **config}
    try:
        return factory(**merged)
    except TypeError as exc:
        from windlass.core.exceptions import ConfigurationError

        raise ConfigurationError(
            f"Could not construct {kind} {name!r}: {exc}",
            hint=f"Check the accepted options for {getattr(factory, '__name__', factory)}.",
            context={"kind": kind, "name": name, "config": sorted(merged)},
        ) from exc

ensure_plugins_loaded

ensure_plugins_loaded() -> None

Discover entry-point plugins exactly once, on first lookup.

A no-op unless this registry was constructed with discover=True. :meth:load_plugins remains available on any registry for callers that want discovery on demand.

Source code in src\windlass\core\registry.py

def ensure_plugins_loaded(self) -> None:
    """Discover entry-point plugins exactly once, on first lookup.

    A no-op unless this registry was constructed with ``discover=True``.
    :meth:`load_plugins` remains available on any registry for callers that
    want discovery on demand.
    """
    if self._plugins_loaded:
        return
    with self._lock:
        if self._plugins_loaded:
            return
        self._plugins_loaded = True  # set first: discovery must not recurse
    self.load_plugins()

load_plugins

load_plugins(*, strict: bool = False) -> list[str]

Scan installed distributions for Windlass plugins.

Two entry-point conventions are supported:

• windlass.plugins — the value is a callable taking the registry; use this to register many components or to do custom setup.
• windlass.<kind> (e.g. windlass.chunker) — the value is the component class itself, registered under the entry-point name.

Parameters:

Name	Type	Description	Default
strict	bool	Raise on the first plugin that fails instead of warning.	False

Returns:

Type	Description
list[str]	Names of the plugins that loaded successfully.

Raises:

Type	Description
PluginError	Only when strict is True and a plugin fails.

Source code in src\windlass\core\registry.py

def load_plugins(self, *, strict: bool = False) -> list[str]:
    """Scan installed distributions for Windlass plugins.

    Two entry-point conventions are supported:

    * ``windlass.plugins`` — the value is a callable taking the registry;
      use this to register many components or to do custom setup.
    * ``windlass.<kind>`` (e.g. ``windlass.chunker``) — the value is the
      component class itself, registered under the entry-point name.

    Args:
        strict: Raise on the first plugin that fails instead of warning.

    Returns:
        Names of the plugins that loaded successfully.

    Raises:
        PluginError: Only when ``strict`` is True and a plugin fails.
    """
    from importlib.metadata import entry_points

    loaded: list[str] = []

    def _handle(err: Exception, label: str) -> None:
        if strict:
            raise PluginError(
                f"Plugin {label!r} failed to load: {err}", context={"plugin": label}
            ) from err
        _log.warning("Skipping Windlass plugin %r: %s", label, err)

    for ep in entry_points(group=PLUGIN_GROUP):
        try:
            hook = ep.load()
            hook(self)
            loaded.append(ep.name)
        except Exception as exc:
            _handle(exc, ep.name)

    for kind in self.kinds():
        for ep in entry_points(group=f"windlass.{kind}"):
            try:
                self.register_lazy(
                    kind,
                    ep.name,
                    ep.value.replace(":", ":", 1),
                    description=f"Plugin from {ep.value}",
                    origin="plugin",
                    override=True,
                )
                loaded.append(f"{kind}:{ep.name}")
            except Exception as exc:
                _handle(exc, f"{kind}:{ep.name}")

    return loaded

snapshot

snapshot() -> dict[str, dict[str, ComponentSpec]]

Return a shallow copy of the registry state (for tests).

Source code in src\windlass\core\registry.py

def snapshot(self) -> dict[str, dict[str, ComponentSpec]]:
    """Return a shallow copy of the registry state (for tests)."""
    with self._lock:
        return {kind: dict(bucket) for kind, bucket in self._specs.items()}

restore

restore(snapshot: dict[str, dict[str, ComponentSpec]]) -> None

Restore a previously captured :meth:snapshot (for tests).

Source code in src\windlass\core\registry.py

def restore(self, snapshot: dict[str, dict[str, ComponentSpec]]) -> None:
    """Restore a previously captured :meth:`snapshot` (for tests)."""
    with self._lock:
        self._specs = {kind: dict(bucket) for kind, bucket in snapshot.items()}

AgentResponse

Bases: WindlassModel

The final result of an agent run.

Attributes:

Name	Type	Description
output	str	The agent's answer.
messages	list[Message]	Full transcript including tool traffic.
steps	list[AgentStep]	Structured per-iteration trace.
usage	Usage	Aggregate token accounting across the whole run.
latency_ms	float	End-to-end wall-clock time.
thread_id	str | None	Checkpoint thread, when checkpointing is enabled.
interrupted	bool	True when the run paused for human approval.
metadata	dict[str, Any]	Runtime annotations.

tool_calls property

tool_calls: list[ToolCall]

Every tool call made during the run, in order.

AgentStep

Bases: WindlassModel

One iteration of an agent's reason/act loop.

Attributes:

Name	Type	Description
index	int	Zero-based step number.
thought	str	Assistant text emitted during this step, if any.
tool_calls	list[ToolCall]	Tools the model requested.
tool_results	list[ToolResult]	Results of executing them.
usage	Usage	Tokens spent on this step.
node	str	Name of the graph node that ran (multi-agent / LangGraph).

Chunk

Bases: WindlassModel

A retrievable slice of a :class:Document.

Attributes:

Name	Type	Description
id	str	Stable identifier, derived from the document id and index.
content	str	The chunk text as it will be embedded and shown to the model.
metadata	dict[str, Any]	Document metadata plus chunk-level annotations.
document_id	str	The document this came from.
index	int	Zero-based position within the document.
embedding	list[float] | None	Optional dense vector; populated during ingestion.
parent_id	str | None	Set by hierarchical chunkers; points at the larger parent chunk that should be expanded at retrieval time.
start_char	int	Character offset in the source document.
end_char	int	Exclusive end offset in the source document.

with_embedding

with_embedding(vector: list[float]) -> Chunk

Return a copy carrying vector as its embedding.

Source code in src\windlass\core\types.py

def with_embedding(self, vector: list[float]) -> Chunk:
    """Return a copy carrying ``vector`` as its embedding."""
    return self.model_copy(update={"embedding": vector})

Completion

Bases: WindlassModel

A non-streaming model response.

Attributes:

Name	Type	Description
content	str	The generated text.
tool_calls	list[ToolCall]	Tools the model wants invoked, if any.
finish_reason	FinishReason	Why generation stopped.
model	str	Model identifier reported by the provider.
usage	Usage	Token accounting.
raw	Any	The untouched provider response object (Level 3 escape hatch).
latency_ms	float	Wall-clock time for the call.

to_message

to_message() -> Message

Render this completion as an assistant :class:Message.

Source code in src\windlass\core\types.py

def to_message(self) -> Message:
    """Render this completion as an assistant :class:`Message`."""
    return Message(role=Role.ASSISTANT, content=self.content, tool_calls=self.tool_calls)

Document

Bases: WindlassModel

A source document before chunking.

Attributes:

Name	Type	Description
id	str	Stable identifier. Defaults to a content hash so re-ingestion is idempotent.
content	str	Extracted plain text.
metadata	dict[str, Any]	Anything worth filtering or citing on later (page, author, section, url, ...). Kept alongside every chunk derived from it.
source	str | None	Where the document came from (path, URL, ...).
mimetype	str | None	Detected content type, when known.
created_at	float	Unix timestamp of ingestion.

Embedding

Bases: WindlassModel

A dense vector plus the text it encodes.

Attributes:

Name	Type	Description
vector	list[float]	The embedding values.
text	str	The text that was embedded.
model	str	Model identifier that produced the vector.

EvaluationReport

Bases: WindlassModel

Aggregate results of an evaluation run.

Attributes:

Name	Type	Description
results	list[EvaluationResult]	Every individual metric/sample score.
summary	dict[str, float]	Mean score per metric.
pass_rate	float	Fraction of results that passed their threshold.
samples	int	Number of samples evaluated.

EvaluationResult

Bases: WindlassModel

Score for a single metric on a single sample.

Attributes:

Name	Type	Description
metric	str	Metric name, e.g. faithfulness.
score	float	Normalised score, conventionally in [0, 1].
passed	bool	Whether the score met the metric's threshold.
threshold	float | None	The threshold that was applied.
reason	str	Explanation, when the metric produces one.
sample_id	str	Identifier of the evaluated sample.

FinishReason

Bases: StrEnum

Why the model stopped generating.

GuardrailResult

Bases: WindlassModel

Verdict from a guardrail check.

Attributes:

Name	Type	Description
allowed	bool	False when the content must be blocked.
content	str	Possibly rewritten content (redaction returns a sanitised copy).
detections	list[dict[str, Any]]	One entry per rule that fired.
rule	str | None	The first rule that blocked, if any.
stage	str	input or output.

modified property

modified: bool

True when the guardrail rewrote the content.

Message

Bases: WindlassModel

One turn in a conversation.

Attributes:

Name	Type	Description
role	Role	Author of the message.
content	str	Text body. May be empty when the turn is purely tool calls.
tool_calls	list[ToolCall]	Tools the assistant wants to invoke.
tool_call_id	str | None	Set on tool messages to correlate with a call.
name	str | None	Optional speaker name (multi-agent transcripts use this).
metadata	dict[str, Any]	Free-form annotations that never reach the provider.

Example

Message.user("hello") Message(role=, content='hello', ...)

system classmethod

system(content: str, **kw: Any) -> Message

Build a system message.

Source code in src\windlass\core\types.py

@classmethod
def system(cls, content: str, **kw: Any) -> Message:
    """Build a system message."""
    return cls(role=Role.SYSTEM, content=content, **kw)

user classmethod

user(content: str, **kw: Any) -> Message

Build a user message.

Source code in src\windlass\core\types.py

@classmethod
def user(cls, content: str, **kw: Any) -> Message:
    """Build a user message."""
    return cls(role=Role.USER, content=content, **kw)

assistant classmethod

assistant(content: str = '', **kw: Any) -> Message

Build an assistant message.

Source code in src\windlass\core\types.py

@classmethod
def assistant(cls, content: str = "", **kw: Any) -> Message:
    """Build an assistant message."""
    return cls(role=Role.ASSISTANT, content=content, **kw)

tool classmethod

tool(result: ToolResult) -> Message

Build the tool message that answers a :class:ToolCall.

Source code in src\windlass\core\types.py

@classmethod
def tool(cls, result: ToolResult) -> Message:
    """Build the ``tool`` message that answers a :class:`ToolCall`."""
    return cls(
        role=Role.TOOL,
        content=result.content,
        tool_call_id=result.call_id,
        name=result.name,
        metadata={"is_error": result.is_error},
    )

RAGAnswer

Bases: WindlassModel

The answer produced by a RAG pipeline.

Attributes:

Name	Type	Description
answer	str	The generated answer text.
question	str	The question that was asked.
contexts	list[ScoredChunk]	Chunks that were placed in the prompt, in prompt order.
usage	Usage	Token accounting for the generation call.
latency_ms	float	End-to-end wall-clock time including retrieval.
metadata	dict[str, Any]	Pipeline annotations (guardrail decisions, rewrites, ...).

sources property

sources: list[str]

Unique, order-preserving list of the sources cited in the context.

Role

Bases: StrEnum

Who authored a :class:Message.

A :class:~enum.StrEnum, so members compare and format as their plain string value — Role.USER == "user" and f"{Role.USER}" both do what you would expect when the value reaches a provider's wire format.

ScoredChunk

Bases: WindlassModel

A chunk together with its retrieval score.

Attributes:

Name	Type	Description
chunk	Chunk	The retrieved chunk.
score	float	Relevance score. Higher is better; the scale depends on the retriever (cosine similarity, BM25, reciprocal rank fusion, ...).
rank	int	One-based position in the result list.
retriever	str	Which retriever produced this hit. Hybrid search keeps this so you can see which leg contributed each result.

content property

content: str

Shortcut to chunk.content.

metadata property

metadata: dict[str, Any]

Shortcut to chunk.metadata.

SearchResult

Bases: WindlassModel

The full result of a retrieval call.

Attributes:

Name	Type	Description
query	str	The query that was executed (post rewriting, if any).
hits	list[ScoredChunk]	Ranked chunks.
total	int	Number of candidates considered before truncation.
latency_ms	float	Wall-clock retrieval time.

texts

texts() -> list[str]

Return just the chunk texts, ranked.

Source code in src\windlass\core\types.py

def texts(self) -> list[str]:
    """Return just the chunk texts, ranked."""
    return [h.chunk.content for h in self.hits]

StreamEvent

Bases: WindlassModel

One incremental update from a streaming call.

Attributes:

Name	Type	Description
type	Literal['text', 'tool_call', 'usage', 'done']	text for content deltas, tool_call when a call is complete, usage for accounting, done for the terminal event.
delta	str	Incremental text for text events.
tool_call	ToolCall | None	The completed call for tool_call events.
finish_reason	FinishReason | None	Present on the done event.
usage	Usage | None	Present on usage and done events when the provider reports it.
raw	Any	Untouched provider chunk.

ToolCall

Bases: WindlassModel

A model's request to invoke a tool.

Attributes:

Name	Type	Description
id	str	Provider-assigned correlation id; echoed back in the tool result.
name	str	Registered tool name.
arguments	dict[str, Any]	Parsed keyword arguments for the tool.
raw_arguments	str | None	The original JSON string, kept when parsing failed.

ToolResult

Bases: WindlassModel

The outcome of executing a :class:ToolCall.

Attributes:

Name	Type	Description
call_id	str	The :attr:ToolCall.id this answers.
name	str	Tool that produced the result.
content	str	String rendering handed back to the model.
data	Any	The unserialised Python return value, for programmatic access.
is_error	bool	True when the tool raised and the error text was returned.
duration_ms	float	Wall-clock execution time.

Usage

Bases: WindlassModel

Token accounting for one or more model calls.

Attributes:

Name	Type	Description
prompt_tokens	int	Tokens consumed by the prompt.
completion_tokens	int	Tokens produced by the model.
total_tokens	int	Sum of the two; computed when a provider omits it.
cached_tokens	int	Prompt tokens served from the provider's prompt cache.
cost_usd	float | None	Estimated spend, when the adapter can compute it.
calls	int	How many provider round-trips this usage aggregates.

build_cache

build_cache(config: CacheConfig | None = None) -> Cache

Construct the cache described by config.

Falls back to :class:MemoryCache (with a warning) if the disk backend is requested but diskcache is missing — a broken cache should never take down the application.

Parameters:

Name	Type	Description	Default
config	CacheConfig | None	Cache policy. Defaults to the global configuration.	None

Returns:

Type	Description
Cache	A ready-to-use cache. :class:NullCache when caching is disabled.

Example

isinstance(build_cache(CacheConfig(enabled=False)), NullCache) True

Source code in src\windlass\core\cache.py

def build_cache(config: CacheConfig | None = None) -> Cache:
    """Construct the cache described by ``config``.

    Falls back to :class:`MemoryCache` (with a warning) if the disk backend is
    requested but ``diskcache`` is missing — a broken cache should never take
    down the application.

    Args:
        config: Cache policy. Defaults to the global configuration.

    Returns:
        A ready-to-use cache. :class:`NullCache` when caching is disabled.

    Example:
        >>> isinstance(build_cache(CacheConfig(enabled=False)), NullCache)
        True
    """
    cfg = config or settings().cache
    if not cfg.enabled:
        return NullCache()
    if cfg.backend == "disk":
        try:
            return DiskCache(cfg.directory, ttl=cfg.ttl)
        except Exception as exc:
            _log.warning("Disk cache unavailable (%s); falling back to memory cache.", exc)
    return MemoryCache(max_size=cfg.max_size, ttl=cfg.ttl)

make_key

make_key(*parts: Any, namespace: str = '') -> str

Build a deterministic cache key from arbitrary arguments.

Arguments are JSON-encoded with sorted keys, so dict ordering never changes the key. Values that are not JSON-serialisable fall back to repr.

Parameters:

Name	Type	Description	Default
*parts	Any	Anything identifying the call — provider, model, prompt, params.	()
namespace	str	Optional prefix to keep different caches from colliding.	''

Returns:

Type	Description
str	A namespace:hexdigest string.

Example

make_key("openai", {"b": 1, "a": 2}) == make_key("openai", {"a": 2, "b": 1}) True

Source code in src\windlass\core\cache.py

def make_key(*parts: Any, namespace: str = "") -> str:
    """Build a deterministic cache key from arbitrary arguments.

    Arguments are JSON-encoded with sorted keys, so dict ordering never changes
    the key. Values that are not JSON-serialisable fall back to ``repr``.

    Args:
        *parts: Anything identifying the call — provider, model, prompt, params.
        namespace: Optional prefix to keep different caches from colliding.

    Returns:
        A ``namespace:hexdigest`` string.

    Example:
        >>> make_key("openai", {"b": 1, "a": 2}) == make_key("openai", {"a": 2, "b": 1})
        True
    """
    payload = json.dumps(parts, sort_keys=True, default=repr, ensure_ascii=False)
    digest = hashlib.blake2b(payload.encode("utf-8"), digest_size=16).hexdigest()
    return f"{namespace}:{digest}" if namespace else digest

batched

batched(items: Iterable[_T], size: int) -> Iterator[list[_T]]

Split an iterable into lists of at most size elements.

Parameters:

Name	Type	Description	Default
items	Iterable[_T]	Any iterable; consumed lazily.	required
size	int	Maximum batch size. Must be positive.	required

Yields:

Type	Description
list[_T]	Successive batches. The final batch may be shorter.

Raises:

Type	Description
ValueError	If size is not positive.

Example

list(batched(range(5), 2)) [[0, 1], [2, 3], [4]]

Source code in src\windlass\core\concurrency.py

def batched(items: Iterable[_T], size: int) -> Iterator[list[_T]]:
    """Split an iterable into lists of at most ``size`` elements.

    Args:
        items: Any iterable; consumed lazily.
        size: Maximum batch size. Must be positive.

    Yields:
        Successive batches. The final batch may be shorter.

    Raises:
        ValueError: If ``size`` is not positive.

    Example:
        >>> list(batched(range(5), 2))
        [[0, 1], [2, 3], [4]]
    """
    if size <= 0:
        raise ValueError("batch size must be positive")
    batch: list[_T] = []
    for item in items:
        batch.append(item)
        if len(batch) >= size:
            yield batch
            batch = []
    if batch:
        yield batch

gather_bounded async

gather_bounded(
    awaitables: Sequence[Awaitable[_T]],
    *,
    limit: int = 8,
    return_exceptions: bool = False
) -> list[_T]

asyncio.gather with a concurrency ceiling.

Unbounded gather over a few thousand embedding calls will happily open a few thousand sockets and get you rate limited. This keeps at most limit in flight while preserving input order in the results.

Parameters:

Name	Type	Description	Default
awaitables	Sequence[Awaitable[_T]]	The coroutines to run.	required
limit	int	Maximum number in flight at once. <= 0 means unbounded.	8
return_exceptions	bool	When True, failures land in the result list instead of propagating.	False

Returns:

Type	Description
list[_T]	Results in the same order as awaitables.

Example

import asyncio async def double(x): return x * 2 asyncio.run(gather_bounded([double(i) for i in range(4)], limit=2)) [0, 2, 4, 6]

Source code in src\windlass\core\concurrency.py

async def gather_bounded(
    awaitables: Sequence[Awaitable[_T]],
    *,
    limit: int = 8,
    return_exceptions: bool = False,
) -> list[_T]:
    """``asyncio.gather`` with a concurrency ceiling.

    Unbounded ``gather`` over a few thousand embedding calls will happily open a
    few thousand sockets and get you rate limited. This keeps at most ``limit``
    in flight while preserving input order in the results.

    Args:
        awaitables: The coroutines to run.
        limit: Maximum number in flight at once. ``<= 0`` means unbounded.
        return_exceptions: When True, failures land in the result list instead
            of propagating.

    Returns:
        Results in the same order as ``awaitables``.

    Example:
        >>> import asyncio
        >>> async def double(x): return x * 2
        >>> asyncio.run(gather_bounded([double(i) for i in range(4)], limit=2))
        [0, 2, 4, 6]
    """
    if not awaitables:
        return []
    if limit <= 0 or limit >= len(awaitables):
        return await asyncio.gather(*awaitables, return_exceptions=return_exceptions)  # type: ignore[return-value]

    semaphore = asyncio.Semaphore(limit)

    async def _guarded(aw: Awaitable[_T]) -> _T:
        async with semaphore:
            return await aw

    return await asyncio.gather(  # type: ignore[return-value]
        *(_guarded(aw) for aw in awaitables), return_exceptions=return_exceptions
    )

iter_sync

iter_sync(agen: AsyncIterator[_T], *, timeout: float | None = None) -> Iterator[_T]

Consume an async iterator from synchronous code.

Pulls one item at a time, so streaming stays streaming — nothing is buffered ahead of the consumer.

Every __anext__ is dispatched to the same background event loop. That is not an optimisation, it is a correctness requirement: :func:asyncio.run calls loop.shutdown_asyncgens() on exit, which finalises any suspended async generator. Driving one with repeated asyncio.run calls therefore yields exactly one item and then silently stops.

Parameters:

Name	Type	Description	Default
agen	AsyncIterator[_T]	The async iterator to drain.	required
timeout	float | None	Optional per-item ceiling in seconds.	None

Yields:

Type	Description
_T	Each item produced by agen.

Raises:

Type	Description
TimeoutError	If timeout elapses while waiting for an item.
Exception	Anything the iterator raises, re-raised unchanged.

Example

async def gen(): ... for i in range(3): ... yield i list(iter_sync(gen())) [0, 1, 2]

Source code in src\windlass\core\concurrency.py

def iter_sync(agen: AsyncIterator[_T], *, timeout: float | None = None) -> Iterator[_T]:
    """Consume an async iterator from synchronous code.

    Pulls one item at a time, so streaming stays streaming — nothing is buffered
    ahead of the consumer.

    Every ``__anext__`` is dispatched to the *same* background event loop. That
    is not an optimisation, it is a correctness requirement: :func:`asyncio.run`
    calls ``loop.shutdown_asyncgens()`` on exit, which finalises any suspended
    async generator. Driving one with repeated ``asyncio.run`` calls therefore
    yields exactly one item and then silently stops.

    Args:
        agen: The async iterator to drain.
        timeout: Optional per-item ceiling in seconds.

    Yields:
        Each item produced by ``agen``.

    Raises:
        TimeoutError: If ``timeout`` elapses while waiting for an item.
        Exception: Anything the iterator raises, re-raised unchanged.

    Example:
        >>> async def gen():
        ...     for i in range(3):
        ...         yield i
        >>> list(iter_sync(gen()))
        [0, 1, 2]
    """
    sentinel = object()

    async def _next() -> Any:
        try:
            return await agen.__anext__()
        except StopAsyncIteration:
            return sentinel

    try:
        while True:
            item = _BACKGROUND.submit(_next()).result(timeout)
            if item is sentinel:
                return
            yield item
    finally:
        aclose = getattr(agen, "aclose", None)
        if aclose is not None:
            with contextlib.suppress(Exception):
                _BACKGROUND.submit(aclose()).result(timeout)

map_async async

map_async(
    fn: Callable[[_T], Awaitable[_R]],
    items: Sequence[_T],
    *,
    limit: int = 8,
    return_exceptions: bool = False
) -> list[_R]

Apply an async function across items with bounded concurrency.

Parameters:

Name	Type	Description	Default
fn	Callable[[_T], Awaitable[_R]]	Async callable applied to each item.	required
items	Sequence[_T]	Inputs.	required
limit	int	Maximum concurrent invocations.	8
return_exceptions	bool	Collect failures instead of raising.	False

Returns:

Type	Description
list[_R]	Results in input order.

Example

import asyncio async def square(x): return x * x asyncio.run(map_async(square, [1, 2, 3], limit=2)) [1, 4, 9]

Source code in src\windlass\core\concurrency.py

async def map_async(
    fn: Callable[[_T], Awaitable[_R]],
    items: Sequence[_T],
    *,
    limit: int = 8,
    return_exceptions: bool = False,
) -> list[_R]:
    """Apply an async function across ``items`` with bounded concurrency.

    Args:
        fn: Async callable applied to each item.
        items: Inputs.
        limit: Maximum concurrent invocations.
        return_exceptions: Collect failures instead of raising.

    Returns:
        Results in input order.

    Example:
        >>> import asyncio
        >>> async def square(x): return x * x
        >>> asyncio.run(map_async(square, [1, 2, 3], limit=2))
        [1, 4, 9]
    """
    return await gather_bounded(
        [fn(item) for item in items], limit=limit, return_exceptions=return_exceptions
    )

run_sync

run_sync(awaitable: Awaitable[_T], *, timeout: float | None = None) -> _T

Run an awaitable to completion from synchronous code.

Every call is dispatched to one long-lived background event loop, whether or not the caller is already inside a loop. That is a correctness requirement, not an optimisation — the same reasoning as :func:iter_sync.

:func:asyncio.run creates a loop and closes it again on every call. Any resource bound to that loop dies with it, and the most common such resource is a connection pool: every provider that keeps an httpx.AsyncClient alive between calls (Ollama, and every vendor SDK underneath OpenAI, Anthropic, Groq and Gemini) pools keep-alive sockets on the loop that opened them. With a fresh loop per call, the first blocking call succeeds and the second raises RuntimeError: Event loop is closed from deep inside httpx — a failure that mock transports never reproduce, because they open no sockets.

Sharing one loop also removes the per-call cost of standing a loop up and tearing it down, which dominates short calls.

Parameters:

Name	Type	Description	Default
awaitable	Awaitable[_T]	The coroutine or future to run.	required
timeout	float | None	Optional ceiling in seconds.	None

Returns:

Type	Description
_T	The awaited result.

Raises:

Type	Description
TimeoutError	If timeout elapses first.
RuntimeError	If called from inside the background loop itself, where blocking on it could never return.
Exception	Anything the awaitable raises, re-raised unchanged.

Note

Calling this from inside a coroutine works but blocks the calling thread. Prefer await on the a* method when you are already async.

Example

async def work() -> int: ... return 42 run_sync(work()) 42

Repeated calls share a loop, so a pooled client survives between them:

from windlass.providers.llm.fake import FakeLLM llm = FakeLLM(responses=["one", "two"]) llm.complete("a").content, llm.complete("b").content ('one', 'two')

Source code in src\windlass\core\concurrency.py

def run_sync(awaitable: Awaitable[_T], *, timeout: float | None = None) -> _T:
    """Run an awaitable to completion from synchronous code.

    Every call is dispatched to one long-lived background event loop, whether or
    not the caller is already inside a loop. That is a correctness requirement,
    not an optimisation — the same reasoning as :func:`iter_sync`.

    :func:`asyncio.run` creates a loop and closes it again on every call. Any
    resource bound to that loop dies with it, and the most common such resource
    is a connection pool: every provider that keeps an ``httpx.AsyncClient``
    alive between calls (Ollama, and every vendor SDK underneath OpenAI,
    Anthropic, Groq and Gemini) pools keep-alive sockets on the loop that opened
    them. With a fresh loop per call, the *first* blocking call succeeds and the
    second raises ``RuntimeError: Event loop is closed`` from deep inside httpx
    — a failure that mock transports never reproduce, because they open no
    sockets.

    Sharing one loop also removes the per-call cost of standing a loop up and
    tearing it down, which dominates short calls.

    Args:
        awaitable: The coroutine or future to run.
        timeout: Optional ceiling in seconds.

    Returns:
        The awaited result.

    Raises:
        TimeoutError: If ``timeout`` elapses first.
        RuntimeError: If called from inside the background loop itself, where
            blocking on it could never return.
        Exception: Anything the awaitable raises, re-raised unchanged.

    Note:
        Calling this from inside a coroutine works but blocks the calling
        thread. Prefer ``await`` on the ``a*`` method when you are already async.

    Example:
        >>> async def work() -> int:
        ...     return 42
        >>> run_sync(work())
        42

        Repeated calls share a loop, so a pooled client survives between them:

        >>> from windlass.providers.llm.fake import FakeLLM
        >>> llm = FakeLLM(responses=["one", "two"])
        >>> llm.complete("a").content, llm.complete("b").content
        ('one', 'two')
    """
    if _BACKGROUND.is_current_thread():
        raise RuntimeError(
            "run_sync() was called from inside the background event loop, where "
            "blocking on that same loop can never return. Await the a* coroutine "
            "directly instead."
        )
    return _BACKGROUND.submit(awaitable).result(timeout)

configure

configure(**overrides: Any) -> WindlassSettings

Update the process-wide settings.

Values are merged onto the current settings, so partial updates are fine.

Parameters:

Name	Type	Description	Default
**overrides	Any	Any field of :class:WindlassSettings. Nested retry and cache accept either a dict or the model instance.	{}

Returns:

Type	Description
WindlassSettings	The new settings object.

Raises:

Type	Description
ConfigurationError	If an override is unknown or fails validation.

Example

configure(temperature=0.0, retry={"attempts": 5}).retry.attempts 5 reset_settings()

Source code in src\windlass\core\config.py

def configure(**overrides: Any) -> WindlassSettings:
    """Update the process-wide settings.

    Values are merged onto the current settings, so partial updates are fine.

    Args:
        **overrides: Any field of :class:`WindlassSettings`. Nested ``retry`` and
            ``cache`` accept either a dict or the model instance.

    Returns:
        The new settings object.

    Raises:
        ConfigurationError: If an override is unknown or fails validation.

    Example:
        >>> configure(temperature=0.0, retry={"attempts": 5}).retry.attempts
        5
        >>> reset_settings()
    """
    global _SETTINGS
    with _LOCK:
        current = settings().model_dump()
        unknown = set(overrides) - set(WindlassSettings.model_fields)
        if unknown:
            raise ConfigurationError(
                f"Unknown setting(s): {', '.join(sorted(unknown))}.",
                hint=f"Valid settings: {', '.join(sorted(WindlassSettings.model_fields))}",
            )
        current.update(overrides)
        try:
            _SETTINGS = WindlassSettings(**current)
        except Exception as exc:
            raise ConfigurationError(f"Invalid configuration override: {exc}") from exc
        return _SETTINGS

reset_settings

reset_settings() -> None

Discard cached settings so the next :func:settings call re-reads the env.

Primarily a test helper.

Source code in src\windlass\core\config.py

def reset_settings() -> None:
    """Discard cached settings so the next :func:`settings` call re-reads the env.

    Primarily a test helper.
    """
    global _SETTINGS
    with _LOCK:
        _SETTINGS = None

settings

settings() -> WindlassSettings

Return the process-wide settings, constructing them on first use.

Returns:

Type	Description
WindlassSettings	The active :class:WindlassSettings.

Example

isinstance(settings().request_timeout, float) True

Source code in src\windlass\core\config.py

def settings() -> WindlassSettings:
    """Return the process-wide settings, constructing them on first use.

    Returns:
        The active :class:`WindlassSettings`.

    Example:
        >>> isinstance(settings().request_timeout, float)
        True
    """
    global _SETTINGS
    if _SETTINGS is None:
        with _LOCK:
            if _SETTINGS is None:
                _SETTINGS = _build_settings()
    return _SETTINGS

root_container

root_container() -> Container

Return the process-wide root container.

Bindings placed here are inherited by every builder created afterwards, which makes it the right place for application-wide wiring::

root_container().bind_instance("tracer", MyTracer())

Returns:

Type	Description
Container	The shared root :class:Container.

Source code in src\windlass\core\container.py

def root_container() -> Container:
    """Return the process-wide root container.

    Bindings placed here are inherited by every builder created afterwards,
    which makes it the right place for application-wide wiring::

        root_container().bind_instance("tracer", MyTracer())

    Returns:
        The shared root :class:`Container`.
    """
    return _ROOT

is_available

is_available(module: str) -> bool

Return whether module can be imported without importing it.

Uses :func:importlib.util.find_spec, so the module's top-level code does not execute. Safe to call on hot paths and at registration time.

Parameters:

Name	Type	Description	Default
module	str	Dotted module path, e.g. "faiss".	required

Returns:

Type	Description
bool	True when the module is installed and importable.

Example

is_available("json") True is_available("definitely_not_installed_xyz") False

Source code in src\windlass\core\lazy.py

def is_available(module: str) -> bool:
    """Return whether ``module`` can be imported without importing it.

    Uses :func:`importlib.util.find_spec`, so the module's top-level code does
    not execute. Safe to call on hot paths and at registration time.

    Args:
        module: Dotted module path, e.g. ``"faiss"``.

    Returns:
        True when the module is installed and importable.

    Example:
        >>> is_available("json")
        True
        >>> is_available("definitely_not_installed_xyz")
        False
    """
    if module in _CACHE:
        return True
    try:
        return importlib.util.find_spec(module) is not None
    except (ImportError, ValueError, ModuleNotFoundError):
        return False

require

require(module: str, *, extra: str, feature: str) -> ModuleType

Import an optional module or raise a helpful error.

Thread-safe and memoised: concurrent first-time callers import exactly once.

Parameters:

Name	Type	Description	Default
module	str	Dotted module path to import.	required
extra	str	The windlass extras group that provides it, used to build the pip install hint.	required
feature	str	Human readable feature name for the error message.	required

Returns:

Type	Description
ModuleType	The imported module.

Raises:

Type	Description
MissingDependencyError	When the module is not installed. The message contains the exact install command.

Example

require("json", extra="core", feature="JSON") # doctest: +ELLIPSIS

Source code in src\windlass\core\lazy.py

def require(module: str, *, extra: str, feature: str) -> ModuleType:
    """Import an optional module or raise a helpful error.

    Thread-safe and memoised: concurrent first-time callers import exactly once.

    Args:
        module: Dotted module path to import.
        extra: The ``windlass`` extras group that provides it, used to build
            the ``pip install`` hint.
        feature: Human readable feature name for the error message.

    Returns:
        The imported module.

    Raises:
        MissingDependencyError: When the module is not installed. The message
            contains the exact install command.

    Example:
        >>> require("json", extra="core", feature="JSON")  # doctest: +ELLIPSIS
        <module 'json' ...>
    """
    cached = _CACHE.get(module)
    if cached is not None:
        return cached
    with _LOCK:
        cached = _CACHE.get(module)
        if cached is not None:
            return cached
        try:
            mod = importlib.import_module(module)
        except ImportError as exc:
            # Distinguish "the package is missing" from "the package is broken".
            missing_root = module.split(".")[0]
            if _is_missing(exc, missing_root):
                raise MissingDependencyError(
                    feature, extra=extra, package=missing_root, original=exc
                ) from exc
            raise MissingDependencyError(
                f"{feature} failed to import ({exc})",
                extra=extra,
                package=missing_root,
                original=exc,
            ) from exc
        _CACHE[module] = mod
        return mod

configure_logging

configure_logging(
    level: int | str = "INFO",
    *,
    stream: Any = None,
    fmt: str | None = None,
    force: bool = False
) -> logging.Logger

Attach a human-readable stream handler to the windlass logger.

Intended for scripts, notebooks and the CLI. Applications with their own logging configuration should not call this.

Parameters:

Name	Type	Description	Default
level	int | str	Log level name or numeric value.	'INFO'
stream	Any	Target stream. Defaults to sys.stderr.	None
fmt	str | None	Custom :mod:logging format string.	None
force	bool	Replace handlers that a previous call installed.	False

Returns:

Type	Description
Logger	The configured windlass logger.

Example

logger = configure_logging("WARNING") logger.name 'windlass'

Source code in src\windlass\core\logging.py

def configure_logging(
    level: int | str = "INFO",
    *,
    stream: Any = None,
    fmt: str | None = None,
    force: bool = False,
) -> logging.Logger:
    """Attach a human-readable stream handler to the ``windlass`` logger.

    Intended for scripts, notebooks and the CLI. Applications with their own
    logging configuration should not call this.

    Args:
        level: Log level name or numeric value.
        stream: Target stream. Defaults to ``sys.stderr``.
        fmt: Custom :mod:`logging` format string.
        force: Replace handlers that a previous call installed.

    Returns:
        The configured ``windlass`` logger.

    Example:
        >>> logger = configure_logging("WARNING")
        >>> logger.name
        'windlass'
    """
    logger = logging.getLogger(ROOT_LOGGER_NAME)
    logger.setLevel(level if isinstance(level, int) else level.upper())

    existing = [h for h in logger.handlers if getattr(h, "_windlass_handler", False)]
    if existing and not force:
        return logger
    for handler in existing:
        logger.removeHandler(handler)

    handler = logging.StreamHandler(stream or sys.stderr)
    handler.setFormatter(
        logging.Formatter(fmt or "%(asctime)s %(levelname)-7s %(name)s: %(message)s", "%H:%M:%S")
    )
    handler.addFilter(_ContextFilter())
    handler._windlass_handler = True  # type: ignore[attr-defined]
    logger.addHandler(handler)
    logger.propagate = False
    return logger

get_logger

get_logger(name: str) -> logging.Logger

Return a Windlass-namespaced logger.

Parameters:

Name	Type	Description	Default
name	str	Usually __name__. A leading windlass. is not duplicated; anything else is nested under windlass..	required

Returns:

Type	Description
Logger	A configured :class:logging.Logger.

Example

get_logger("windlass.core.registry").name 'windlass.core.registry' get_logger("my_plugin").name 'windlass.my_plugin'

Source code in src\windlass\core\logging.py

def get_logger(name: str) -> logging.Logger:
    """Return a Windlass-namespaced logger.

    Args:
        name: Usually ``__name__``. A leading ``windlass.`` is not duplicated;
            anything else is nested under ``windlass.``.

    Returns:
        A configured :class:`logging.Logger`.

    Example:
        >>> get_logger("windlass.core.registry").name
        'windlass.core.registry'
        >>> get_logger("my_plugin").name
        'windlass.my_plugin'
    """
    if name == ROOT_LOGGER_NAME or name.startswith(f"{ROOT_LOGGER_NAME}."):
        logger = logging.getLogger(name)
    else:
        logger = logging.getLogger(f"{ROOT_LOGGER_NAME}.{name}")
    if not any(isinstance(f, _ContextFilter) for f in logger.filters):
        logger.addFilter(_ContextFilter())
    return logger

available

available(kind: str) -> list[str]

List registered names for kind in the global registry.

Source code in src\windlass\core\registry.py

def available(kind: str) -> list[str]:
    """List registered names for ``kind`` in the global registry."""
    return REGISTRY.names(kind)

create

create(kind: str, name: str, /, **config: Any) -> Any

Instantiate a component from the global registry.

Thin module-level shortcut for :meth:Registry.create.

Source code in src\windlass\core\registry.py

def create(kind: str, name: str, /, **config: Any) -> Any:
    """Instantiate a component from the global registry.

    Thin module-level shortcut for :meth:`Registry.create`.
    """
    return REGISTRY.create(kind, name, **config)

retry_async async

retry_async(
    fn: Callable[[], Awaitable[_T]],
    *,
    config: RetryConfig | None = None,
    retry_on: Iterable[type[BaseException]] = (),
    on_retry: Callable[[int, BaseException, float], None] | None = None,
    description: str = "provider call"
) -> _T

Await fn with retries on transient failures.

Parameters:

Name	Type	Description	Default
fn	Callable[[], Awaitable[_T]]	A zero-argument callable returning an awaitable. It is re-invoked on each attempt, so pass a lambda rather than a coroutine object — coroutines cannot be awaited twice.	required
config	RetryConfig | None	Retry policy. Defaults to the global configuration.	None
retry_on	Iterable[type[BaseException]]	Extra exception types to treat as transient.	()
on_retry	Callable[[int, BaseException, float], None] | None	Callback invoked as (attempt, exception, delay) before each sleep. Handy for metrics.	None
description	str	Label used in log messages.	'provider call'

Returns:

Type	Description
_T	Whatever fn eventually returns.

Raises:

Type	Description
Exception	The last failure, once attempts are exhausted or the error is classified as non-retryable.

Example

import asyncio calls = [] async def flaky(): ... calls.append(1) ... if len(calls) < 2: ... raise ConnectionError("reset") ... return "ok" asyncio.run(retry_async(flaky, config=RetryConfig(attempts=3, initial_delay=0))) 'ok'

Source code in src\windlass\core\retry.py

async def retry_async(
    fn: Callable[[], Awaitable[_T]],
    *,
    config: RetryConfig | None = None,
    retry_on: Iterable[type[BaseException]] = (),
    on_retry: Callable[[int, BaseException, float], None] | None = None,
    description: str = "provider call",
) -> _T:
    """Await ``fn`` with retries on transient failures.

    Args:
        fn: A zero-argument callable returning an awaitable. It is re-invoked on
            each attempt, so pass a lambda rather than a coroutine object —
            coroutines cannot be awaited twice.
        config: Retry policy. Defaults to the global configuration.
        retry_on: Extra exception types to treat as transient.
        on_retry: Callback invoked as ``(attempt, exception, delay)`` before each
            sleep. Handy for metrics.
        description: Label used in log messages.

    Returns:
        Whatever ``fn`` eventually returns.

    Raises:
        Exception: The last failure, once attempts are exhausted or the error is
            classified as non-retryable.

    Example:
        >>> import asyncio
        >>> calls = []
        >>> async def flaky():
        ...     calls.append(1)
        ...     if len(calls) < 2:
        ...         raise ConnectionError("reset")
        ...     return "ok"
        >>> asyncio.run(retry_async(flaky, config=RetryConfig(attempts=3, initial_delay=0)))
        'ok'
    """
    cfg = config or settings().retry
    last: BaseException | None = None

    for attempt in range(1, cfg.attempts + 1):
        try:
            return await fn()
        except BaseException as exc:
            last = exc
            if attempt >= cfg.attempts or not is_retryable(exc, retry_on):
                raise
            delay = _next_delay(attempt, cfg, exc)
            _log.warning(
                "%s failed (attempt %d/%d): %s — retrying in %.2fs",
                description,
                attempt,
                cfg.attempts,
                exc,
                delay,
            )
            if on_retry is not None:
                on_retry(attempt, exc, delay)
            await asyncio.sleep(delay)

    raise last if last is not None else WindlassError(f"{description} failed without an exception")

retry_sync

retry_sync(
    fn: Callable[[], _T],
    *,
    config: RetryConfig | None = None,
    retry_on: Iterable[type[BaseException]] = (),
    description: str = "provider call"
) -> _T

Blocking counterpart of :func:retry_async.

Parameters:

Name	Type	Description	Default
fn	Callable[[], _T]	Zero-argument callable, re-invoked per attempt.	required
config	RetryConfig | None	Retry policy. Defaults to the global configuration.	None
retry_on	Iterable[type[BaseException]]	Extra transient exception types.	()
description	str	Label used in log messages.	'provider call'

Returns:

Type	Description
_T	Whatever fn eventually returns.

Raises:

Type	Description
Exception	The last failure once attempts are exhausted.

Example

retry_sync(lambda: 7, config=RetryConfig(attempts=1)) 7

Source code in src\windlass\core\retry.py

def retry_sync(
    fn: Callable[[], _T],
    *,
    config: RetryConfig | None = None,
    retry_on: Iterable[type[BaseException]] = (),
    description: str = "provider call",
) -> _T:
    """Blocking counterpart of :func:`retry_async`.

    Args:
        fn: Zero-argument callable, re-invoked per attempt.
        config: Retry policy. Defaults to the global configuration.
        retry_on: Extra transient exception types.
        description: Label used in log messages.

    Returns:
        Whatever ``fn`` eventually returns.

    Raises:
        Exception: The last failure once attempts are exhausted.

    Example:
        >>> retry_sync(lambda: 7, config=RetryConfig(attempts=1))
        7
    """
    cfg = config or settings().retry
    last: BaseException | None = None

    for attempt in range(1, cfg.attempts + 1):
        try:
            return fn()
        except BaseException as exc:
            last = exc
            if attempt >= cfg.attempts or not is_retryable(exc, retry_on):
                raise
            delay = _next_delay(attempt, cfg, exc)
            _log.warning(
                "%s failed (attempt %d/%d): %s — retrying in %.2fs",
                description,
                attempt,
                cfg.attempts,
                exc,
                delay,
            )
            time.sleep(delay)

    raise last if last is not None else WindlassError(f"{description} failed without an exception")

with_retry

with_retry(
    *, config: RetryConfig | None = None, retry_on: Iterable[type[BaseException]] = ()
) -> Callable[[Callable[..., Any]], Callable[..., Any]]

Decorator applying the retry policy to a sync or async function.

Parameters:

Name	Type	Description	Default
config	RetryConfig | None	Retry policy. Defaults to the global configuration.	None
retry_on	Iterable[type[BaseException]]	Extra transient exception types.	()

Returns:

Type	Description
Callable[[Callable[..., Any]], Callable[..., Any]]	A decorator preserving the wrapped function's signature and sync/async
Callable[[Callable[..., Any]], Callable[..., Any]]	nature.

Example

@with_retry(config=RetryConfig(attempts=2, initial_delay=0)) ... def ping() -> str: ... return "pong" ping() 'pong'

Source code in src\windlass\core\retry.py

def with_retry(
    *,
    config: RetryConfig | None = None,
    retry_on: Iterable[type[BaseException]] = (),
) -> Callable[[Callable[..., Any]], Callable[..., Any]]:
    """Decorator applying the retry policy to a sync or async function.

    Args:
        config: Retry policy. Defaults to the global configuration.
        retry_on: Extra transient exception types.

    Returns:
        A decorator preserving the wrapped function's signature and sync/async
        nature.

    Example:
        >>> @with_retry(config=RetryConfig(attempts=2, initial_delay=0))
        ... def ping() -> str:
        ...     return "pong"
        >>> ping()
        'pong'
    """

    def decorate(fn: Callable[..., Any]) -> Callable[..., Any]:
        label = getattr(fn, "__qualname__", repr(fn))

        if asyncio.iscoroutinefunction(fn):

            @functools.wraps(fn)
            async def async_wrapper(*args: Any, **kwargs: Any) -> Any:
                return await retry_async(
                    lambda: fn(*args, **kwargs),
                    config=config,
                    retry_on=retry_on,
                    description=label,
                )

            return async_wrapper

        @functools.wraps(fn)
        def sync_wrapper(*args: Any, **kwargs: Any) -> Any:
            return retry_sync(
                lambda: fn(*args, **kwargs),
                config=config,
                retry_on=retry_on,
                description=label,
            )

        return sync_wrapper

    return decorate