Core framework

The registry, base classes, configuration, and provenance primitives every other part of Openbtk builds on. import openbtk.core (or its submodules) never requires an optional dependency.

Everything on this page is under the stability policy: it is what you write custom components and plugins against.

Registry

openbtk.core.registry.Registry

Registry(category: str, base_type: type[T])

Bases: Generic[T]

A namespaced, type-checked registry of component classes.

Each registry is bound to exactly one category (e.g. "loader") and one base type (e.g. BaseLoader). Registering a class that is not a subclass of that base type, or whose key does not belong to this category, raises immediately -- at import time, not when the component is eventually used.

Source code in src/openbtk/core/registry.py

def __init__(self, category: str, base_type: type[T]) -> None:
    if category not in _VALID_CATEGORIES:
        raise RegistryError(
            f"Unknown registry category {category!r}. "
            f"Available: {sorted(_VALID_CATEGORIES)}",
            context={"category": category},
        )
    self._category = category
    self._base_type = base_type
    self._items: dict[str, type[T]] = {}
    self._deprecated_aliases: dict[str, tuple[str, str, str]] = {}

create

create(key: str, *, policy: PolicyConfig | None = None, **kwargs: Any) -> T

Instantiate the registered class for key with kwargs.

Enforces sends_data_offsite (docs/06_SECURITY_COMPLIANCE.md section 3.3, FR-V-07) before construction, i.e. before any data can possibly be read: a class with sends_data_offsite = True refuses to construct unless policy.allow_offsite_providers is True. policy=None (the default -- no caller has opted in to anything) is treated the same as the safe PolicyConfig() default, not as "no policy, no check": a component that sends data offsite must never be constructible by accident.

Parameters:

Name	Type	Description	Default
key	str	The registry key to instantiate.	required
policy	PolicyConfig | None	The active pipeline policy. Only allow_offsite_providers is consulted here. Keyword-only so callers can never confuse it with a component's own constructor kwarg.	None
**kwargs	Any	Forwarded to the class's constructor.	{}

Raises:

Type	Description
RegistryError	If key is not registered.
PolicyError	If the class sends data offsite and the policy (explicit or default) does not allow that.
Exception	Whatever the class's constructor raises.

Source code in src/openbtk/core/registry.py

def create(
    self, key: str, *, policy: PolicyConfig | None = None, **kwargs: Any
) -> T:
    """Instantiate the registered class for ``key`` with ``kwargs``.

    Enforces ``sends_data_offsite`` (docs/06_SECURITY_COMPLIANCE.md
    section 3.3, FR-V-07) before construction, i.e. before any data can
    possibly be read: a class with ``sends_data_offsite = True`` refuses
    to construct unless ``policy.allow_offsite_providers`` is ``True``.
    ``policy=None`` (the default -- no caller has opted in to anything)
    is treated the same as the safe ``PolicyConfig()`` default, not as
    "no policy, no check": a component that sends data offsite must
    never be constructible by accident.

    Args:
        key: The registry key to instantiate.
        policy: The active pipeline policy. Only ``allow_offsite_providers``
            is consulted here. Keyword-only so callers can never confuse
            it with a component's own constructor kwarg.
        **kwargs: Forwarded to the class's constructor.

    Raises:
        RegistryError: If ``key`` is not registered.
        PolicyError: If the class sends data offsite and the policy
            (explicit or default) does not allow that.
        Exception: Whatever the class's constructor raises.
    """
    cls = self.get(key)
    if getattr(cls, "sends_data_offsite", False):
        allow_offsite = policy is not None and policy.allow_offsite_providers
        if not allow_offsite:
            raise PolicyError(
                f"{cls.__qualname__} ({key}) sends data offsite "
                "(sends_data_offsite=True), but the active policy does "
                "not allow offsite providers. Pass policy=PolicyConfig("
                "allow_offsite_providers=True) to permit this explicitly.",
                context={
                    "registry": self._category,
                    "key": key,
                    "class_name": cls.__qualname__,
                },
            )
    log.debug("registry.create", registry=self._category, key=key)
    return cls(**kwargs)

create_from_config

create_from_config(config: dict[str, Any], *, policy: PolicyConfig | None = None) -> T

Instantiate from a {"type": ..., "params": {...}} dict.

Parameters:

Name	Type	Description	Default
config	dict[str, Any]	Must contain a "type" key (the registry key); an optional "params" dict supplies constructor kwargs.	required
policy	PolicyConfig | None	Forwarded to :meth:create -- see its docstring for the sends_data_offsite enforcement this triggers.	None

Raises:

Type	Description
RegistryError	If config has no "type" key, or the type is not registered.
PolicyError	See :meth:create.

Example

from openbtk.core.registry import GUARDRAIL_REGISTRY GUARDRAIL_REGISTRY.create_from_config( ... {"not_type": "nope"} ... ) # doctest: +IGNORE_EXCEPTION_DETAIL Traceback (most recent call last): ... openbtk.core.errors.RegistryError: ...must have a 'type' key

Source code in src/openbtk/core/registry.py

def create_from_config(
    self, config: dict[str, Any], *, policy: PolicyConfig | None = None
) -> T:
    """Instantiate from a ``{"type": ..., "params": {...}}`` dict.

    Args:
        config: Must contain a ``"type"`` key (the registry key); an
            optional ``"params"`` dict supplies constructor kwargs.
        policy: Forwarded to :meth:`create` -- see its docstring for the
            ``sends_data_offsite`` enforcement this triggers.

    Raises:
        RegistryError: If ``config`` has no ``"type"`` key, or the type
            is not registered.
        PolicyError: See :meth:`create`.

    Example:
        >>> from openbtk.core.registry import GUARDRAIL_REGISTRY
        >>> GUARDRAIL_REGISTRY.create_from_config(
        ...     {"not_type": "nope"}
        ... )  # doctest: +IGNORE_EXCEPTION_DETAIL
        Traceback (most recent call last):
            ...
        openbtk.core.errors.RegistryError: ...must have a 'type' key
    """
    if "type" not in config:
        raise RegistryError(
            f"Registry {self._category!r}: config dict must have a 'type' key.",
            # Only the keys, never the values -- params may legitimately
            # hold operational config; there is no reason for an error
            # about a MISSING key to echo the rest of it back.
            context={"registry": self._category, "config_keys": sorted(config)},
        )
    key = config["type"]
    params = config.get("params", {})
    return self.create(key, policy=policy, **params)

describe

describe(key: str) -> ComponentInfo

Return an introspective summary of the class registered at key.

Raises:

Type	Description
RegistryError	If key is not registered.

Source code in src/openbtk/core/registry.py

def describe(self, key: str) -> ComponentInfo:
    """Return an introspective summary of the class registered at ``key``.

    Raises:
        RegistryError: If ``key`` is not registered.
    """
    cls = self.get(key)
    return ComponentInfo(
        registry_key=key,
        class_name=cls.__name__,
        module=cls.__module__,
        docstring=inspect.getdoc(cls),
        signature=str(inspect.signature(cls)),
    )

get

get(key: str) -> type[T]

Return the registered class for key.

Triggers plugin discovery on first call (docs/04_API_DESIGN.md section 10) -- lookup methods are where "first registry access" actually happens; register() is what plugins themselves call, so triggering discovery there would be re-entrant.

Raises:

Type	Description
RegistryError	If key is not registered.

Source code in src/openbtk/core/registry.py

def get(self, key: str) -> type[T]:
    """Return the registered class for ``key``.

    Triggers plugin discovery on first call (docs/04_API_DESIGN.md
    section 10) -- lookup methods are where "first registry access"
    actually happens; register() is what plugins themselves call, so
    triggering discovery there would be re-entrant.

    Raises:
        RegistryError: If ``key`` is not registered.
    """
    load_plugins()
    if key not in self._items:
        raise RegistryError(
            f"Registry {self._category!r}: key {key!r} not found. "
            f"Available: {self.list_keys()}",
            context={"registry": self._category, "key": key},
        )
    deprecated = self._deprecated_aliases.get(key)
    if deprecated is not None:
        new_key, since, removal = deprecated
        warn_deprecated(
            f"registry key {key!r}",
            since=since,
            removal=removal,
            use=repr(new_key),
            stacklevel=3,
        )
    return self._items[key]

is_registered

is_registered(key: str) -> bool

Return whether key (or an alias of it) is registered.

Triggers plugin discovery on first call -- see :meth:get.

Source code in src/openbtk/core/registry.py

def is_registered(self, key: str) -> bool:
    """Return whether ``key`` (or an alias of it) is registered.

    Triggers plugin discovery on first call -- see :meth:`get`.
    """
    load_plugins()
    return key in self._items

list_keys

list_keys() -> list[str]

Return every registered key and alias, sorted.

Triggers plugin discovery on first call -- see :meth:get.

Source code in src/openbtk/core/registry.py

def list_keys(self) -> list[str]:
    """Return every registered key and alias, sorted.

    Triggers plugin discovery on first call -- see :meth:`get`.
    """
    load_plugins()
    return sorted(self._items)

register

register(key: str) -> Callable[[type[T]], type[T]]

Return a decorator that registers a class under key.

Parameters:

Name	Type	Description	Default
key	str	A key following <category>.<scope>.<name>, matching this registry's own category. Keys are permanent once released -- treat as public API.	required

Raises:

Type	Description
RegistryError	If the key is malformed, already registered, or the decorated class does not subclass this registry's base type.

Source code in src/openbtk/core/registry.py

def register(self, key: str) -> Callable[[type[T]], type[T]]:
    """Return a decorator that registers a class under ``key``.

    Args:
        key: A key following ``<category>.<scope>.<name>``, matching
            this registry's own category. Keys are permanent once
            released -- treat as public API.

    Raises:
        RegistryError: If the key is malformed, already registered, or
            the decorated class does not subclass this registry's base
            type.
    """
    _validate_key(key, expected_category=self._category)

    def decorator(cls: type[T]) -> type[T]:
        if not (isinstance(cls, type) and issubclass(cls, self._base_type)):
            raise RegistryError(
                f"Registry {self._category!r}: {cls!r} does not "
                f"subclass {self._base_type.__name__}.",
                context={"registry": self._category, "key": key},
            )
        if key in self._items:
            raise RegistryError(
                f"Registry {self._category!r}: key {key!r} already "
                f"registered by {self._items[key].__qualname__}.",
                context={"registry": self._category, "key": key},
            )
        self._items[key] = cls
        cls.registry_key = key
        log.debug(
            "registry.register",
            registry=self._category,
            key=key,
            cls=cls.__qualname__,
        )
        return cls

    return decorator

register_alias

register_alias(alias: str, existing_key: str, *, since: str | None = None, removal: str | None = None) -> None

Register alias as a second name for an already-registered key.

The primary use is a deprecation transition (mkdocs/stability.md): a key is renamed by registering the new key normally, then aliasing the old key to it, so existing configs keep working. Pass since and removal (release numbers, e.g. "1.2.0" / "2.0.0") to make every lookup through the old key emit an :class:~openbtk.core.deprecation.OpenBTKDeprecationWarning. A key is never removed within a major version, only deprecated.

Raises:

Type	Description
RegistryError	If alias is malformed, or existing_key is not already registered.
ValueError	If only one of since / removal is given.

Source code in src/openbtk/core/registry.py

def register_alias(
    self,
    alias: str,
    existing_key: str,
    *,
    since: str | None = None,
    removal: str | None = None,
) -> None:
    """Register ``alias`` as a second name for an already-registered key.

    The primary use is a deprecation transition (``mkdocs/stability.md``): a
    key is renamed by registering the new key normally, then aliasing the
    old key to it, so existing configs keep working. Pass ``since`` and
    ``removal`` (release numbers, e.g. ``"1.2.0"`` / ``"2.0.0"``) to make
    every lookup through the old key emit an
    :class:`~openbtk.core.deprecation.OpenBTKDeprecationWarning`. A key is
    never removed within a major version, only deprecated.

    Raises:
        RegistryError: If ``alias`` is malformed, or ``existing_key`` is
            not already registered.
        ValueError: If only one of ``since`` / ``removal`` is given.
    """
    if (since is None) != (removal is None):
        raise ValueError("give both since= and removal=, or neither")
    _validate_key(alias, expected_category=self._category)
    if existing_key not in self._items:
        raise RegistryError(
            f"Registry {self._category!r}: cannot alias {alias!r} to "
            f"{existing_key!r} -- {existing_key!r} is not registered.",
            context={"registry": self._category, "existing_key": existing_key},
        )
    self._items[alias] = self._items[existing_key]
    if since is not None and removal is not None:
        self._deprecated_aliases[alias] = (existing_key, since, removal)
    log.debug(
        "registry.alias",
        registry=self._category,
        alias=alias,
        existing_key=existing_key,
    )

openbtk.core.registry.get_registry

get_registry(category: str) -> Registry[Any]

Return the global registry for category.

Raises:

Type	Description
RegistryError	If category is not a known registry -- including "finetuner", which is reserved in the key grammar but has no backing base class yet.

Source code in src/openbtk/core/registry.py

def get_registry(category: str) -> Registry[Any]:
    """Return the global registry for ``category``.

    Raises:
        RegistryError: If ``category`` is not a known registry -- including
            ``"finetuner"``, which is reserved in the key grammar but has no
            backing base class yet.
    """
    if category not in _ALL_REGISTRIES:
        raise RegistryError(
            f"Unknown registry category {category!r}. "
            f"Available: {sorted(_ALL_REGISTRIES)}",
            context={"category": category},
        )
    return _ALL_REGISTRIES[category]

openbtk.core.registry.list_categories

list_categories() -> list[str]

Every registry category that exists, sorted.

Example

"loader" in list_categories() True

Source code in src/openbtk/core/registry.py

def list_categories() -> list[str]:
    """Every registry category that exists, sorted.

    Example:
        >>> "loader" in list_categories()
        True
    """
    return sorted(_ALL_REGISTRIES)

openbtk.core.registry.ComponentInfo

Bases: BaseModel

Introspective summary of one registered component.

Deliberately does not attempt a full JSON Schema of constructor parameters: Openbtk components take plain keyword arguments, not a single Pydantic config object, so a fabricated "schema" for an arbitrary __init__ signature would often be misleading (a default value that is not itself a JSON type, for instance). signature gives the same information in an honest, always-obtainable form.

Base classes

Every component role has one. Subclass it, register it under a key, and it is usable from a config file, the CLI and a pipeline.

openbtk.core.base.Component

Bases: ABC

Root of every Openbtk component. Supplies identity and provenance.

Subclasses are never instantiated directly through this class -- always through one of the role base classes below, via a registry.

registry_key class-attribute

registry_key: str = ''

Set by the registry's register() decorator at class-definition time. Empty string is the "not yet registered" sentinel -- a class used only in tests without going through the registry will report this rather than raising, since not every use of a component needs a key.

provenance

provenance() -> ComponentProvenance

Identity and configuration of this instance, for a run manifest.

The default here is deliberately conservative: an empty config dict, no model_identity. It does NOT attempt to introspect self.__dict__ for constructor arguments -- a generic introspection default risks capturing something that is not JSON-serialisable (a loaded model object, an open file handle) or, worse, something that should never reach a manifest at all. PHI safety is the project's central claim; a clever default that occasionally leaks is a worse failure mode than a plain default that never does.

Override this in any component whose configuration or wrapped model is worth recording -- which is the expected, normal case for anything beyond the simplest components.

Returns:

Type	Description
ComponentProvenance	A ComponentProvenance with this component's registry key,
ComponentProvenance	class name and package version populated, empty config, and no
ComponentProvenance	model identity.

Source code in src/openbtk/core/base.py

def provenance(self) -> ComponentProvenance:
    """Identity and configuration of this instance, for a run manifest.

    The default here is deliberately conservative: an empty ``config``
    dict, no ``model_identity``. It does NOT attempt to introspect
    ``self.__dict__`` for constructor arguments -- a generic
    introspection default risks capturing something that is not
    JSON-serialisable (a loaded model object, an open file handle) or,
    worse, something that should never reach a manifest at all. PHI
    safety is the project's central claim; a clever default that
    occasionally leaks is a worse failure mode than a plain default that
    never does.

    Override this in any component whose configuration or wrapped model
    is worth recording -- which is the expected, normal case for
    anything beyond the simplest components.

    Returns:
        A ``ComponentProvenance`` with this component's registry key,
        class name and package version populated, empty config, and no
        model identity.
    """
    return ComponentProvenance(
        registry_key=self.registry_key,
        class_name=type(self).__name__,
        package_version=_openbtk_version(),
        config={},
        model_identity=None,
    )

openbtk.core.base.BaseLoader

Bases: Component, Generic[InputT, RecordT]

Load raw data from a source into typed domain records.

Thin wrappers around format-specific libraries (pydicom, wfdb, fhir.resources, ...). Loaders stream: they yield records one at a time so a corpus larger than memory can still be processed.

load abstractmethod

load(source: InputT) -> Iterator[RecordT]

Yield records lazily from source.

MUST yield the first record without first consuming the whole source -- this is what makes the memory guarantee real rather than aspirational, and the contract suite checks it behaviourally (first record arrives fast even from a source too large to read fully).

Parameters:

Name	Type	Description	Default
source	InputT	Source identifier -- file path, directory, URL, or connection string, depending on the implementation.	required

Yields:

Type	Description
RecordT	Typed domain records, in a deterministic order.

Raises:

Type	Description
LoaderError	On any I/O or format-parsing failure.

Source code in src/openbtk/core/base.py

@abstractmethod
def load(self, source: InputT) -> Iterator[RecordT]:
    """Yield records lazily from ``source``.

    MUST yield the first record without first consuming the whole
    source -- this is what makes the memory guarantee real rather than
    aspirational, and the contract suite checks it behaviourally (first
    record arrives fast even from a source too large to read fully).

    Args:
        source: Source identifier -- file path, directory, URL, or
            connection string, depending on the implementation.

    Yields:
        Typed domain records, in a deterministic order.

    Raises:
        LoaderError: On any I/O or format-parsing failure.
    """

load_all

load_all(source: InputT) -> list[RecordT]

Materialise every record from source into a list.

Warning

Memory hazard -- O(corpus), not O(batch). Never used in Openbtk's own code, docs, or examples (ADR-0004; enforced by a grep gate in CI). Provided only for interactive/notebook use on small sources.

Source code in src/openbtk/core/base.py

def load_all(self, source: InputT) -> list[RecordT]:
    """Materialise every record from ``source`` into a list.

    Warning:
        Memory hazard -- O(corpus), not O(batch). Never used in
        Openbtk's own code, docs, or examples (ADR-0004; enforced by a
        grep gate in CI). Provided only for interactive/notebook use on
        small sources.
    """
    return list(self.load(source))

openbtk.core.base.BasePreprocessor

Bases: Component, Generic[RecordT]

Transform a loaded record into a cleaner or more enriched one.

Preprocessors handle de-identification, normalisation, enrichment and quality checks. They return the same record type and must be stateless per call -- the same input always produces the same output.

process abstractmethod

process(record: RecordT) -> RecordT

Return a processed record of the same type.

Parameters:

Name	Type	Description	Default
record	RecordT	A loaded (or previously processed) domain record.	required

Returns:

Type	Description
RecordT	A record of the same type, cleaned or enriched.

Raises:

Type	Description
ProcessingError	On any preprocessing failure.

Source code in src/openbtk/core/base.py

@abstractmethod
def process(self, record: RecordT) -> RecordT:
    """Return a processed record of the same type.

    Args:
        record: A loaded (or previously processed) domain record.

    Returns:
        A record of the same type, cleaned or enriched.

    Raises:
        ProcessingError: On any preprocessing failure.
    """

process_stream

process_stream(records: Iterable[RecordT]) -> Iterator[RecordT]

Lazily process a stream of records.

Default implementation applies :meth:process one record at a time. Override for batched efficiency (e.g. a model that processes in batches of N more cheaply than N separate calls).

Source code in src/openbtk/core/base.py

def process_stream(self, records: Iterable[RecordT]) -> Iterator[RecordT]:
    """Lazily process a stream of records.

    Default implementation applies :meth:`process` one record at a time.
    Override for batched efficiency (e.g. a model that processes in
    batches of N more cheaply than N separate calls).
    """
    return (self.process(r) for r in records)

openbtk.core.base.BaseChunker

Bases: Component, Generic[RecordT, ChunkT]

Split a record into smaller, retrievable chunks.

Returns an Iterator, not a list (ADR-0004) -- a single large record can produce thousands of chunks, and the executor must be able to stream them through embedding and indexing without holding them all in memory at once.

chunk abstractmethod

chunk(record: RecordT) -> Iterator[ChunkT]

Yield ordered chunks lazily.

Parameters:

Name	Type	Description	Default
record	RecordT	A preprocessed domain record.	required

Yields:

Type	Description
ChunkT	Chunks in document order. Yielding nothing is valid (e.g. a
ChunkT	record below the minimum chunkable length).

Raises:

Type	Description
ProcessingError	On chunking failure.

Source code in src/openbtk/core/base.py

@abstractmethod
def chunk(self, record: RecordT) -> Iterator[ChunkT]:
    """Yield ordered chunks lazily.

    Args:
        record: A preprocessed domain record.

    Yields:
        Chunks in document order. Yielding nothing is valid (e.g. a
        record below the minimum chunkable length).

    Raises:
        ProcessingError: On chunking failure.
    """

openbtk.core.base.BaseSegmenter

Bases: Component, Generic[RecordT, ChunkT]

Non-text equivalent of :class:BaseChunker: patches, windows, clips.

Semantically identical contract to BaseChunker -- a separate name is used for imaging/signal/video modalities where "segment" reads more naturally than "chunk."

segment abstractmethod

segment(record: RecordT) -> Iterator[ChunkT]

Yield ordered segments lazily. See :meth:BaseChunker.chunk.

Source code in src/openbtk/core/base.py

@abstractmethod
def segment(self, record: RecordT) -> Iterator[ChunkT]:
    """Yield ordered segments lazily. See :meth:`BaseChunker.chunk`."""

openbtk.core.base.BaseFeatureExtractor

Bases: Component, Generic[ChunkT]

Extract named scalar features from a chunk (biosignals, audio).

extract abstractmethod

extract(chunk: ChunkT) -> dict[str, float]

Extract scalar features from a chunk.

Parameters:

Name	Type	Description	Default
chunk	ChunkT	A chunk or segment of a domain record.	required

Returns:

Type	Description
dict[str, float]	Feature name to value.

Raises:

Type	Description
ProcessingError	On extraction failure.

Source code in src/openbtk/core/base.py

@abstractmethod
def extract(self, chunk: ChunkT) -> dict[str, float]:
    """Extract scalar features from a chunk.

    Args:
        chunk: A chunk or segment of a domain record.

    Returns:
        Feature name to value.

    Raises:
        ProcessingError: On extraction failure.
    """

openbtk.core.base.BaseEmbeddingProvider

Bases: Component

Provide text (or modality-specific) embeddings.

Implementations wrap PubMedBERT, BioBERT, OpenAI, BiomedCLIP and similar. All return float32 arrays so downstream vector stores have one dtype to plan around.

batch_size property

batch_size: int

Default batch size for streaming embedding. Override to tune.

dimension abstractmethod property

dimension: int

Dimensionality of the vectors this provider produces.

sends_data_offsite class-attribute

sends_data_offsite: bool

Whether embedding with this provider sends data to a third party. Enforced by the pipeline layer under a local-only policy (FR-V-07).

embed abstractmethod

embed(texts: list[str]) -> NDArray[np.float32]

Embed a list of texts.

Parameters:

Name	Type	Description	Default
texts	list[str]	Strings to embed. Must be non-empty.	required

Returns:

Type	Description
NDArray[float32]	Array of shape (len(texts), self.dimension), dtype float32.

Raises:

Type	Description
ProviderError	On model inference or API failure.

Source code in src/openbtk/core/base.py

@abstractmethod
def embed(self, texts: list[str]) -> NDArray[np.float32]:
    """Embed a list of texts.

    Args:
        texts: Strings to embed. Must be non-empty.

    Returns:
        Array of shape ``(len(texts), self.dimension)``, dtype float32.

    Raises:
        ProviderError: On model inference or API failure.
    """

embed_one

embed_one(text: str) -> NDArray[np.float32]

Embed a single text. Returns a 1D array of shape (dimension,).

Source code in src/openbtk/core/base.py

def embed_one(self, text: str) -> NDArray[np.float32]:
    """Embed a single text. Returns a 1D array of shape (dimension,)."""
    # numpy's __getitem__ stubs return Any for a general index -- too
    # polymorphic to narrow statically. embed()'s own contract guarantees
    # shape (1, dimension) here, so row 0 genuinely is NDArray[np.float32];
    # this cast documents that assumption rather than suppressing the
    # check silently.
    return cast("NDArray[np.float32]", self.embed([text])[0])

model_identity

model_identity() -> ModelIdentity

The specific model this provider wraps.

Not abstract, so a lightweight test double can be embedded in the contract suite without declaring one -- but calling this without an override is a programming error for any real provider, hence the explicit failure rather than a fabricated placeholder identity.

Raises:

Type	Description
NotImplementedError	Unless overridden.

Source code in src/openbtk/core/base.py

def model_identity(self) -> ModelIdentity:
    """The specific model this provider wraps.

    Not abstract, so a lightweight test double can be embedded in the
    contract suite without declaring one -- but calling this without an
    override is a programming error for any real provider, hence the
    explicit failure rather than a fabricated placeholder identity.

    Raises:
        NotImplementedError: Unless overridden.
    """
    raise NotImplementedError(
        f"{type(self).__name__} must override model_identity() to report "
        "the specific model it wraps."
    )

openbtk.core.base.BaseLLMProvider

Bases: Component

Generate text from a language model.

Implementations wrap OpenAI, Anthropic, local HuggingFace/vLLM, and biomedical-tuned model presets (Meditron, MedGemma, OpenBioLLM).

sends_data_offsite class-attribute

sends_data_offsite: bool

Whether calling this provider sends data to a third party. Enforced by the pipeline layer under a local-only policy (FR-V-07).

chat abstractmethod

chat(messages: list[Message], **kwargs: Any) -> LLMResponse

Generate a response to a multi-turn conversation.

Parameters:

Name	Type	Description	Default
messages	list[Message]	The conversation so far, oldest first.	required
**kwargs	Any	Provider-specific generation parameters.	{}

Raises:

Type	Description
ProviderError	On API or model failure.

Source code in src/openbtk/core/base.py

@abstractmethod
def chat(self, messages: list[Message], **kwargs: Any) -> LLMResponse:
    """Generate a response to a multi-turn conversation.

    Args:
        messages: The conversation so far, oldest first.
        **kwargs: Provider-specific generation parameters.

    Raises:
        ProviderError: On API or model failure.
    """

generate abstractmethod

generate(prompt: str, **kwargs: Any) -> LLMResponse

Generate a completion for a single prompt string.

Parameters:

Name	Type	Description	Default
prompt	str	The input prompt.	required
**kwargs	Any	Provider-specific generation parameters (temperature, max_tokens, stop sequences, ...).	{}

Raises:

Type	Description
ProviderError	On API or model failure.

Source code in src/openbtk/core/base.py

@abstractmethod
def generate(self, prompt: str, **kwargs: Any) -> LLMResponse:
    """Generate a completion for a single prompt string.

    Args:
        prompt: The input prompt.
        **kwargs: Provider-specific generation parameters (temperature,
            max_tokens, stop sequences, ...).

    Raises:
        ProviderError: On API or model failure.
    """

model_identity

model_identity() -> ModelIdentity

The specific model this provider wraps.

See :meth:BaseEmbeddingProvider.model_identity for why this is not abstract but still requires an override to use meaningfully.

Raises:

Type	Description
NotImplementedError	Unless overridden.

Source code in src/openbtk/core/base.py

def model_identity(self) -> ModelIdentity:
    """The specific model this provider wraps.

    See :meth:`BaseEmbeddingProvider.model_identity` for why this is not
    abstract but still requires an override to use meaningfully.

    Raises:
        NotImplementedError: Unless overridden.
    """
    raise NotImplementedError(
        f"{type(self).__name__} must override model_identity() to report "
        "the specific model it wraps."
    )

stream abstractmethod

stream(prompt: str, **kwargs: Any) -> Iterator[str]

Stream a completion token-by-token (or chunk-by-chunk).

Raises:

Type	Description
ProviderError	On API or model failure.

Source code in src/openbtk/core/base.py

@abstractmethod
def stream(self, prompt: str, **kwargs: Any) -> Iterator[str]:
    """Stream a completion token-by-token (or chunk-by-chunk).

    Raises:
        ProviderError: On API or model failure.
    """

structured

structured(messages: list[Message], schema: type[BaseModel], **kwargs: Any) -> BaseModel

Generate output validated against a Pydantic schema.

Default implementation asks the underlying model for JSON via :meth:chat and validates the response text against schema. This is a naive fallback, not the intended steady state: providers with native structured-output support (OpenAI's response_format, Anthropic's tool-use-based extraction) should override this for reliability, since asking a model to return JSON in a plain prompt is the least reliable way to get it.

Raises:

Type	Description
ProviderError	On API or model failure.
ValidationError	If the response text does not validate against schema.

Source code in src/openbtk/core/base.py

def structured(
    self, messages: list[Message], schema: type[BaseModel], **kwargs: Any
) -> BaseModel:
    """Generate output validated against a Pydantic schema.

    Default implementation asks the underlying model for JSON via
    :meth:`chat` and validates the response text against ``schema``.
    This is a naive fallback, not the intended steady state: providers
    with native structured-output support (OpenAI's ``response_format``,
    Anthropic's tool-use-based extraction) should override this for
    reliability, since asking a model to return JSON in a plain prompt
    is the least reliable way to get it.

    Raises:
        ProviderError: On API or model failure.
        pydantic.ValidationError: If the response text does not validate
            against ``schema``.
    """
    response = self.chat(messages, **kwargs)
    return schema.model_validate_json(response.text)

openbtk.core.base.BaseVectorStore

Bases: Component

Store and retrieve embedding vectors.

Implementations wrap FAISS, Chroma, Qdrant and similar.

delete abstractmethod

delete(ids: list[str]) -> None

Delete vectors by id.

Raises:

Type	Description
RetrievalError	On deletion failure.

Source code in src/openbtk/core/base.py

@abstractmethod
def delete(self, ids: list[str]) -> None:
    """Delete vectors by id.

    Raises:
        RetrievalError: On deletion failure.
    """

load classmethod

load(path: str) -> BaseVectorStore

Load a previously persisted store from disk.

Raises:

Type	Description
NotImplementedError	Unless overridden.

Source code in src/openbtk/core/base.py

@classmethod
def load(cls, path: str) -> BaseVectorStore:
    """Load a previously persisted store from disk.

    Raises:
        NotImplementedError: Unless overridden.
    """
    raise NotImplementedError(f"{cls.__name__} does not support load().")

persist

persist(path: str) -> None

Persist this store to disk.

Not every store supports persistence (an in-memory test double, for instance) -- the default raises rather than silently no-op'ing, so a caller who expects persistence to have happened finds out immediately.

Raises:

Type	Description
NotImplementedError	Unless overridden.

Source code in src/openbtk/core/base.py

def persist(self, path: str) -> None:
    """Persist this store to disk.

    Not every store supports persistence (an in-memory test double,
    for instance) -- the default raises rather than silently no-op'ing,
    so a caller who expects persistence to have happened finds out
    immediately.

    Raises:
        NotImplementedError: Unless overridden.
    """
    raise NotImplementedError(f"{type(self).__name__} does not support persist().")

query abstractmethod

query(vector: NDArray[float32], top_k: int = 5, filter: dict[str, Any] | None = None) -> list[SearchResult]

Query for nearest neighbours.

Parameters:

Name	Type	Description	Default
vector	NDArray[float32]	Query vector of shape (dimension,).	required
top_k	int	Number of results to return.	5
filter	dict[str, Any] | None	Optional metadata filter, store-specific in semantics.	None

Returns:

Type	Description
list[SearchResult]	Up to top_k results, ordered by decreasing relevance.

Raises:

Type	Description
RetrievalError	On query failure.

Source code in src/openbtk/core/base.py

@abstractmethod
def query(
    self,
    vector: NDArray[np.float32],
    top_k: int = 5,
    filter: dict[str, Any] | None = None,
) -> list[SearchResult]:
    """Query for nearest neighbours.

    Args:
        vector: Query vector of shape ``(dimension,)``.
        top_k: Number of results to return.
        filter: Optional metadata filter, store-specific in semantics.

    Returns:
        Up to ``top_k`` results, ordered by decreasing relevance.

    Raises:
        RetrievalError: On query failure.
    """

upsert abstractmethod

upsert(ids: list[str], vectors: NDArray[float32], metadata: list[dict[str, Any]]) -> None

Insert or update vectors with associated metadata.

Parameters:

Name	Type	Description	Default
ids	list[str]	Unique string identifiers, one per vector.	required
vectors	NDArray[float32]	Array of shape (len(ids), dimension).	required
metadata	list[dict[str, Any]]	One metadata dict per vector.	required

Raises:

Type	Description
RetrievalError	On storage failure.

Source code in src/openbtk/core/base.py

@abstractmethod
def upsert(
    self,
    ids: list[str],
    vectors: NDArray[np.float32],
    metadata: list[dict[str, Any]],
) -> None:
    """Insert or update vectors with associated metadata.

    Args:
        ids: Unique string identifiers, one per vector.
        vectors: Array of shape ``(len(ids), dimension)``.
        metadata: One metadata dict per vector.

    Raises:
        RetrievalError: On storage failure.
    """

openbtk.core.base.BaseReranker

Bases: Component

Reorder search results by a criterion other than the original score.

Implementations include concept-overlap reranking (shared UMLS CUIs between query and result) and cross-encoder reranking.

rerank abstractmethod

rerank(query: str, results: list[SearchResult], top_k: int | None = None) -> list[SearchResult]

Reorder (and optionally truncate) a list of search results.

Parameters:

Name	Type	Description	Default
query	str	The original query string.	required
results	list[SearchResult]	Results to reorder, in their original order.	required
top_k	int | None	If given, return at most this many results.	None

Returns:

Type	Description
list[SearchResult]	Results reordered by this reranker's criterion.

Source code in src/openbtk/core/base.py

@abstractmethod
def rerank(
    self, query: str, results: list[SearchResult], top_k: int | None = None
) -> list[SearchResult]:
    """Reorder (and optionally truncate) a list of search results.

    Args:
        query: The original query string.
        results: Results to reorder, in their original order.
        top_k: If given, return at most this many results.

    Returns:
        Results reordered by this reranker's criterion.
    """

openbtk.core.base.BaseGuardrail

Bases: Component

Check a payload for a policy violation.

Used for PHI detection, clinical hallucination checks, terminology validity, and signal-quality thresholds.

check abstractmethod

check(payload: Any) -> GuardrailResult

Check a payload and return a structured result.

Never raises for a failed check -- always returns a result with passed=False. The pipeline layer, not this method, decides whether a BLOCK severity should raise GuardrailViolation.

Parameters:

Name	Type	Description	Default
payload	Any	The item to check -- raw text, generated text, a record, or any domain object, depending on the guardrail.	required

Returns:

Type	Description
GuardrailResult	A GuardrailResult describing the outcome.

Source code in src/openbtk/core/base.py

@abstractmethod
def check(self, payload: Any) -> GuardrailResult:
    """Check a payload and return a structured result.

    Never raises for a failed check -- always returns a result with
    ``passed=False``. The pipeline layer, not this method, decides
    whether a ``BLOCK`` severity should raise ``GuardrailViolation``.

    Args:
        payload: The item to check -- raw text, generated text, a
            record, or any domain object, depending on the guardrail.

    Returns:
        A ``GuardrailResult`` describing the outcome.
    """

openbtk.core.base.BaseDatasetAdapter

Bases: Component

Load a known biomedical dataset by name.

Handles credentialed access, local caching, and synthetic data generation (Synthea and similar). Never auto-downloads restricted data.

license abstractmethod property

license: str

SPDX identifier, or an access-request URL for restricted data.

name abstractmethod property

name: str

Human-readable dataset name.

requires_credentials property

requires_credentials: bool

Whether this dataset requires credentialed access.

load abstractmethod

load(**kwargs: Any) -> Any

Load the dataset.

Parameters:

Name	Type	Description	Default
**kwargs	Any	Dataset-specific parameters (split, subset, path, credentials, filters, ...).	{}

Returns:

Type	Description
Any	Dataset-appropriate return type -- typically a list of records,
Any	a DataFrame, or a HuggingFace Dataset.

Raises:

Type	Description
DatasetError	On access failure (missing credentials, bad path, network error).

Source code in src/openbtk/core/base.py

@abstractmethod
def load(self, **kwargs: Any) -> Any:
    """Load the dataset.

    Args:
        **kwargs: Dataset-specific parameters (split, subset, path,
            credentials, filters, ...).

    Returns:
        Dataset-appropriate return type -- typically a list of records,
        a DataFrame, or a HuggingFace ``Dataset``.

    Raises:
        DatasetError: On access failure (missing credentials, bad path,
            network error).
    """

openbtk.core.base.BaseTerminologyService

Bases: Component

Resolve, validate and map codes across clinical terminology systems.

Implementations back onto UMLS, a locally supplied vocabulary release, or a small bundled permissively-licensed subset -- never a bundled restricted vocabulary.

is_authoritative

is_authoritative(system: CodeSystem) -> bool

Whether a False from :meth:validate for system means the code definitely does not exist, as opposed to "this backend cannot confirm it".

A complete vocabulary (the full UMLS, a whole vocabulary release the caller supplied) is authoritative: absence is a real answer. A partial one -- the small bundled ICD-10-CM subset, or a supplied file that simply has no rows for system -- is not: absence proves nothing, so a guardrail must report "unverifiable", not "invalid". Defaults to True; a backend that knows it is incomplete overrides it.

Source code in src/openbtk/core/base.py

def is_authoritative(self, system: CodeSystem) -> bool:  # noqa: ARG002
    """Whether a ``False`` from :meth:`validate` for ``system`` means the
    code definitely does not exist, as opposed to "this backend cannot
    confirm it".

    A complete vocabulary (the full UMLS, a whole vocabulary release the
    caller supplied) is authoritative: absence is a real answer. A partial
    one -- the small bundled ICD-10-CM subset, or a supplied file that
    simply has no rows for ``system`` -- is not: absence proves nothing, so
    a guardrail must report "unverifiable", not "invalid". Defaults to
    ``True``; a backend that knows it is incomplete overrides it.
    """
    return True

map abstractmethod

map(code: str, from_system: CodeSystem, to_system: CodeSystem) -> list[Concept]

Map a code from one terminology system to another, if a crosswalk exists.

Returns:

Type	Description
list[Concept]	Zero or more equivalent concepts in to_system.

Raises:

Type	Description
TerminologyError	If the backend is unavailable or unlicensed.

Source code in src/openbtk/core/base.py

@abstractmethod
def map(
    self, code: str, from_system: CodeSystem, to_system: CodeSystem
) -> list[Concept]:
    """Map a code from one terminology system to another, if a
    crosswalk exists.

    Returns:
        Zero or more equivalent concepts in ``to_system``.

    Raises:
        TerminologyError: If the backend is unavailable or unlicensed.
    """

resolve abstractmethod

resolve(code: str, system: CodeSystem) -> Concept | None

Resolve a code to its concept, if known.

Raises:

Type	Description
TerminologyError	If the backend is unavailable or unlicensed.

Source code in src/openbtk/core/base.py

@abstractmethod
def resolve(self, code: str, system: CodeSystem) -> Concept | None:
    """Resolve a code to its concept, if known.

    Raises:
        TerminologyError: If the backend is unavailable or unlicensed.
    """

validate abstractmethod

validate(code: str, system: CodeSystem) -> bool

Return whether code exists in system.

Raises:

Type	Description
TerminologyError	If the backend is unavailable or unlicensed.

Source code in src/openbtk/core/base.py

@abstractmethod
def validate(self, code: str, system: CodeSystem) -> bool:
    """Return whether ``code`` exists in ``system``.

    Raises:
        TerminologyError: If the backend is unavailable or unlicensed.
    """

Shared schemas

openbtk.core.schemas.TextSpan

Bases: BaseModel

A character range within some text, with a label and confidence.

Used for de-identification detections, linked-entity mentions, and guardrail-flagged regions alike -- one span type serves all of them so downstream code has a single shape to reason about.

Example

span = TextSpan(start=10, end=14, label="DATE", confidence=0.97) span.end - span.start 4

openbtk.core.schemas.LinkedEntity

Bases: BaseModel

A clinical entity mention linked to one or more terminology codes.

negated / uncertain / historical / subject carry the ConText-style assertions (docs/11_GLOSSARY.md Part 4) that determine whether a mention should be treated as a present, first-person finding -- getting these wrong inverts clinical meaning, so they are first-class fields rather than free-text metadata.

Example

entity = LinkedEntity( ... text="diabetes", ... span=TextSpan(start=0, end=8, label="CONDITION", confidence=0.9), ... cui="C0011849", ... codes={"SNOMED": "73211009"}, ... ) entity.subject 'patient'

openbtk.core.schemas.SourceRef

Bases: BaseModel

A pointer back from a retrieved item to its exact source location.

This is what makes verifiable citations possible: a retrieved chunk carries a SourceRef so a GroundednessGuardrail -- or a human reviewer -- can check a generated claim against the precise offsets in the original record, not just "some document somewhere."

Example

ref = SourceRef(record_id="rec-1", chunk_id="chunk-3") ref.char_start is None True

openbtk.core.schemas.SearchResult

Bases: BaseModel

One hit returned from a vector store or reranker query.

Example

hit = SearchResult(id="chunk-3", score=0.82, metadata={"section": "Plan"}) hit.source is None True

openbtk.core.schemas.GuardrailResult

Bases: BaseModel

The outcome of one BaseGuardrail.check() call.

A guardrail never raises on a failed check (docs/04_API_DESIGN.md section 3) -- it always returns one of these. Whether a BLOCK severity halts the pipeline is a policy decision made by the caller, not by the guardrail itself.

Example

result = GuardrailResult( ... passed=False, ... severity=GuardrailSeverity.BLOCK, ... guardrail_key="guardrail.general.phi_leakage", ... message="Detected a phone number in generated output.", ... ) result.passed False

openbtk.core.schemas.GuardrailSeverity

Bases: StrEnum

How seriously a guardrail result should be treated.

openbtk.core.schemas.Message

Bases: BaseModel

One turn in a conversation passed to an LLM provider.

Example

Message(role="user", content="Summarise this note.").role 'user'

openbtk.core.schemas.LLMResponse

Bases: BaseModel

The result of one LLM generation call.

Example

LLMResponse(text="The patient has type 2 diabetes.").text 'The patient has type 2 diabetes.' resp = LLMResponse(text="...", usage=TokenUsage(total_tokens=42)) resp.usage.total_tokens 42

openbtk.core.schemas.TokenUsage

Bases: BaseModel

Token accounting for one or more LLM calls.

Defined here, not in core.provenance (where RunManifest.token_usage lives) -- core.provenance already imports from this module (JsonValue), so defining TokenUsage there and referencing it from LLMResponse here would be a real import cycle, not a hypothetical one. core.provenance re-exports this exact class for backward compatibility with the name it originally shipped under.

Example

TokenUsage().total_tokens 0 a = TokenUsage(prompt_tokens=10, completion_tokens=5, total_tokens=15) b = TokenUsage(prompt_tokens=3, completion_tokens=2, total_tokens=5) (a + b).total_tokens 20

__add__

__add__(other: TokenUsage) -> TokenUsage

Combine two calls' usage into a running total -- real accounting for a pipeline that makes more than one LLM call in a run, not just a single-call struct with nothing to aggregate it.

Source code in src/openbtk/core/schemas.py

def __add__(self, other: TokenUsage) -> TokenUsage:
    """Combine two calls' usage into a running total -- real
    accounting for a pipeline that makes more than one LLM call in a
    run, not just a single-call struct with nothing to aggregate it."""
    if not isinstance(other, TokenUsage):
        return NotImplemented
    return TokenUsage(
        prompt_tokens=self.prompt_tokens + other.prompt_tokens,
        completion_tokens=self.completion_tokens + other.completion_tokens,
        total_tokens=self.total_tokens + other.total_tokens,
    )

openbtk.core.schemas.RAGAnswer

Bases: BaseModel

The result of a full retrieve-(rerank)-generate RAG call (:class:~openbtk.pipelines.rag.RAGPipeline, task 5.8).

sources is the whole point of carrying this as its own type rather than returning a bare LLMResponse: every generated answer is traceable back to the exact retrieved chunks it was grounded in, in the order they were given to the model -- what makes a GroundednessGuardrail (or a human reviewer) able to check a claim against a precise source rather than "some document somewhere" (:class:~openbtk.core.schemas.SourceRef's own docstring).

Example

answer = RAGAnswer( ... text="The patient has type 2 diabetes.", ... sources=[SourceRef(record_id="note-1", chunk_id="chunk-3")], ... ) answer.sources[0].record_id 'note-1'

openbtk.core.schemas.CodeSystem

Bases: StrEnum

A clinical terminology or unit-of-measure system.

Named exactly per docs/11_GLOSSARY.md's "Naming Conventions Derived From This Glossary" section, which is authoritative for this enum. Note: docs/05_DATA_MODALITY_SPEC.md's CodedEvent.system comment lists a different set (SNOMED, ICD10, ICD10CM, RXNORM, LOINC, CPT -- both ICD10 and ICD10CM, no UCUM). That is a documentation inconsistency to reconcile when the EHR module is built (M6), not a second source of truth -- the glossary's dedicated naming section wins here.

openbtk.core.schemas.Concept

Bases: BaseModel

A single resolved terminology concept.

Example

c = Concept( ... code="73211009", system=CodeSystem.SNOMED, display="Diabetes mellitus" ... ) c.system.value 'SNOMED'

Configuration

openbtk.core.config.PipelineConfig

Bases: BaseModel

A pipeline, described declaratively.

Example

cfg = PipelineConfig( ... name="probe", ... steps=[StepConfig(id="load", type="loader.general.probe")], ... ) cfg.version 1 issues = cfg.validate_registry() issues[0].message "Step 'load': type 'loader.general.probe' not found in registry 'loader'."

from_yaml classmethod

from_yaml(path: str | Path) -> PipelineConfig

Load and validate a pipeline config from a YAML file.

Environment variable references (${VAR_NAME}) anywhere in the YAML are resolved against os.environ before validation -- see :func:_interpolate_env_vars for what that means for secret handling downstream.

Raises:

Type	Description
ConfigError	If the file cannot be read, is not valid YAML, an interpolated environment variable is unset, or the content does not satisfy this schema.

Source code in src/openbtk/core/config.py

@classmethod
def from_yaml(cls, path: str | Path) -> PipelineConfig:
    """Load and validate a pipeline config from a YAML file.

    Environment variable references (``${VAR_NAME}``) anywhere in the
    YAML are resolved against ``os.environ`` before validation -- see
    :func:`_interpolate_env_vars` for what that means for secret
    handling downstream.

    Raises:
        ConfigError: If the file cannot be read, is not valid YAML, an
            interpolated environment variable is unset, or the content
            does not satisfy this schema.
    """
    path = Path(path)
    try:
        raw_text = path.read_text(encoding="utf-8")
    except OSError as e:
        raise ConfigError(
            f"Could not read pipeline config: {path}",
            context={"path": str(path)},
        ) from e

    try:
        raw = yaml.safe_load(raw_text)
    except yaml.YAMLError as e:
        raise ConfigError(
            f"Invalid YAML in pipeline config: {path}",
            context={"path": str(path)},
        ) from e

    interpolated = _interpolate_env_vars(raw)

    try:
        return cls.model_validate(interpolated)
    except Exception as e:
        raise ConfigError(
            f"Pipeline config failed validation: {path}\n{e}",
            context={"path": str(path)},
        ) from e

json_schema_export classmethod

json_schema_export() -> dict[str, Any]

Return this config's JSON Schema, for editor validation / docs.

Source code in src/openbtk/core/config.py

@classmethod
def json_schema_export(cls) -> dict[str, Any]:
    """Return this config's JSON Schema, for editor validation / docs."""
    return cls.model_json_schema()

validate_registry

validate_registry() -> list[ValidationIssue]

Check this config against the live component registries.

Does not instantiate anything or touch data (docs/03_ARCHITECTURE.md section 7.3) -- every check here is a lookup or a graph property. See this module's docstring for which checks from that section are NOT yet implemented, and why.

Returns:

Type	Description
list[ValidationIssue]	One ValidationIssue per problem found. An empty list means
list[ValidationIssue]	every check that IS implemented passed -- not that the config
list[ValidationIssue]	is guaranteed runnable.

Source code in src/openbtk/core/config.py

def validate_registry(self) -> list[ValidationIssue]:
    """Check this config against the live component registries.

    Does not instantiate anything or touch data (docs/03_ARCHITECTURE.md
    section 7.3) -- every check here is a lookup or a graph property.
    See this module's docstring for which checks from that section are
    NOT yet implemented, and why.

    Returns:
        One ``ValidationIssue`` per problem found. An empty list means
        every check that IS implemented passed -- not that the config
        is guaranteed runnable.
    """
    issues: list[ValidationIssue] = []
    step_ids = {step.id for step in self.steps}

    for step in self.steps:
        category = step.type.split(".", 1)[0]
        try:
            registry = get_registry(category)
        except Exception:
            issues.append(
                ValidationIssue(
                    severity="error",
                    message=(
                        f"Step {step.id!r}: type {step.type!r} has an "
                        f"unknown category {category!r}."
                    ),
                    step_id=step.id,
                )
            )
            continue
        if not registry.is_registered(step.type):
            issues.append(
                ValidationIssue(
                    severity="error",
                    message=(
                        f"Step {step.id!r}: type {step.type!r} not found "
                        f"in registry {category!r}."
                    ),
                    step_id=step.id,
                )
            )
        else:
            issues.extend(_component_issues(step, registry, self.policy))
        for dep in step.after:
            if dep not in step_ids:
                issues.append(
                    ValidationIssue(
                        severity="error",
                        message=(
                            f"Step {step.id!r} depends on {dep!r}, which "
                            "is not a step in this pipeline."
                        ),
                        step_id=step.id,
                    )
                )

    issues.extend(self._find_cycles(step_ids))

    for guardrail in self.guardrails:
        if not GUARDRAIL_REGISTRY.is_registered(guardrail.type):
            issues.append(
                ValidationIssue(
                    severity="error",
                    message=(
                        f"Guardrail type {guardrail.type!r} not found in "
                        "the guardrail registry."
                    ),
                )
            )

    return issues

openbtk.core.config.StepConfig

Bases: BaseModel

One step in a pipeline: a registered component plus its parameters.

Example

StepConfig(id="load", type="loader.clinical_text.plain_text").after []

openbtk.core.config.PolicyConfig

Bases: BaseModel

Pipeline-wide safety policy.

Example

PolicyConfig().allow_offsite_providers False

openbtk.core.config.ProvenanceConfig

Bases: BaseModel

Where and how this pipeline's run manifests are written.

Minimal by design: the only field with any specification anywhere is manifest_dir, from docs/03_ARCHITECTURE.md section 7.3's worked example. Extend this when core/provenance.py's RunManifest (deferred -- see that module's docstring) actually lands and more is known about what a pipeline run needs to configure.

Example

ProvenanceConfig().manifest_dir './runs'

openbtk.core.config.GuardrailConfig

Bases: BaseModel

A guardrail attached to specific points in a pipeline.

Example

gc = GuardrailConfig( ... type="guardrail.general.phi_leakage", ... at=["after:deid", "after:chunk"], ... ) gc.on_violation 'block'

openbtk.core.config.ValidationIssue

Bases: BaseModel

One problem found by PipelineConfig.validate_registry().

Not specified anywhere beyond a bare return-type reference (docs/04_API_DESIGN.md section 6) -- designed from scratch, matching the shape of every other result schema in this codebase (severity, message, plus enough context to locate the problem).

Example

issue = ValidationIssue( ... severity="error", message="bad step", step_id="load" ... ) issue.step_id 'load'

Provenance

openbtk.core.provenance.RunManifest

Bases: BaseModel

The audit record of one pipeline execution (ADR-0005). Emitted by every run -- success, failure or partial -- there is no manifest-off switch.

Example

from datetime import UTC, datetime manifest = RunManifest( ... run_id="a1b2c3", ... status="success", ... config={"name": "probe", "steps": []}, ... started_at=datetime(2026, 1, 1, tzinfo=UTC), ... ) manifest.manifest_version '1.0.0'

openbtk.core.provenance.StepProvenance

Bases: BaseModel

Identity, throughput and outcome of one executed pipeline step.

Example

sp = StepProvenance( ... step_id="load", ... component=ComponentProvenance( ... registry_key="loader.clinical_text.plain_text", ... class_name="PlainTextLoader", ... package_version="0.1.0", ... ), ... records_in=0, ... records_out=3, ... status="success", ... ) sp.records_out 3

openbtk.core.provenance.DataDigest

Bases: BaseModel

A content digest for one input source, for a run manifest.

Example

DataDigest(uri="./notes", sha256=None, record_count=3).record_count 3

openbtk.core.provenance.GuardrailOutcome

Bases: BaseModel

Aggregated outcome of one guardrail at one attachment point over an entire run. See this module's docstring for why this is aggregated rather than one entry per record.

Example

outcome = GuardrailOutcome( ... guardrail_key="guardrail.general.phi_leakage", ... at="after:deid", ... checked_count=100, ... blocked_count=0, ... ) outcome.warned_count 0

openbtk.core.provenance.ComponentProvenance

Bases: BaseModel

Identity and configuration of one component instance, for a run manifest.

Every Component exposes a default provenance() that builds one of these from class name, registry key and package version alone. A component whose behaviour depends on a specific model overrides provenance() to populate model_identity too.

Example

cp = ComponentProvenance( ... registry_key="loader.clinical_text.plain_text", ... class_name="PlainTextLoader", ... package_version="0.1.0", ... config={"encoding": "utf-8"}, ... ) cp.model_identity is None True ComponentProvenance( ... registry_key="", class_name="Unregistered", package_version="0.1.0" ... ).registry_key ''

openbtk.core.provenance.ModelIdentity

Bases: BaseModel

The identity of a specific model, pinned to an immutable revision.

Example

ModelIdentity( ... name="NeuML/pubmedbert-base-embeddings", ... revision="a1b2c3d4e5f6", # pragma: allowlist secret ... source="huggingface", ... ).source 'huggingface' ModelIdentity( # doctest: +IGNORE_EXCEPTION_DETAIL ... name="gpt-4o", revision="latest", source="api" ... ) Traceback (most recent call last): ... pydantic_core._pydantic_core.ValidationError: ...revision is a floating tag

Logging

openbtk.core.logging.get_logger

get_logger(name: str) -> Any

Return a PHI-safe structured logger for name.

Use __name__ as name, matching every other logging convention in this codebase. The returned logger's processor chain is fixed at creation and independent of any global structlog.configure() call -- by a host application or by Openbtk itself, since Openbtk never makes one.

Example

log = get_logger(name) log.info("probe.event", safe_field=1) # doctest: +SKIP

Source code in src/openbtk/core/logging.py

def get_logger(name: str) -> Any:
    """Return a PHI-safe structured logger for ``name``.

    Use ``__name__`` as ``name``, matching every other logging convention in
    this codebase. The returned logger's processor chain is fixed at
    creation and independent of any global ``structlog.configure()`` call --
    by a host application or by Openbtk itself, since Openbtk never makes
    one.

    Example:
        >>> log = get_logger(__name__)
        >>> log.info("probe.event", safe_field=1)  # doctest: +SKIP
    """
    return structlog.wrap_logger(
        structlog.PrintLogger(),
        processors=[
            structlog.contextvars.merge_contextvars,
            structlog.processors.add_log_level,
            _level_filter,
            structlog.processors.TimeStamper(fmt="iso", utc=True),
            _redact_processor,
            structlog.processors.JSONRenderer(),
        ],
        logger_name=name,
    )

openbtk.core.logging.configure_logging

configure_logging(*, allow_phi: bool = False) -> None

Enable or disable PHI redaction process-wide.

Parameters:

Name	Type	Description	Default
allow_phi	bool	If True, disables the redaction processor for every logger obtained via :func:get_logger, past and future -- checked fresh on each log call, not baked in at logger-creation time. Logs a warning immediately, because making the unsafe path loud is the point (docs/06_SECURITY_COMPLIANCE.md section 3.1): nobody should be able to end up here by accident.	False

Example

configure_logging(allow_phi=True) # doctest: +SKIP configure_logging(allow_phi=False) # restore the safe default

Source code in src/openbtk/core/logging.py

def configure_logging(*, allow_phi: bool = False) -> None:
    """Enable or disable PHI redaction process-wide.

    Args:
        allow_phi: If True, disables the redaction processor for every
            logger obtained via :func:`get_logger`, past and future --
            checked fresh on each log call, not baked in at logger-creation
            time. Logs a warning immediately, because making the unsafe
            path loud is the point (docs/06_SECURITY_COMPLIANCE.md section
            3.1): nobody should be able to end up here by accident.

    Example:
        >>> configure_logging(allow_phi=True)  # doctest: +SKIP
        >>> configure_logging(allow_phi=False)  # restore the safe default
    """
    global _allow_phi
    _allow_phi = allow_phi
    if allow_phi:
        get_logger(__name__).warning(
            "logging.phi_redaction_disabled",
            message=(
                "PHI redaction has been explicitly disabled via "
                "configure_logging(allow_phi=True). Log output may now "
                "contain PHI."
            ),
        )

openbtk.core.logging.set_log_level

set_log_level(level: str) -> None

Set the minimum level of Openbtk's own log lines, process-wide.

Parameters:

Name	Type	Description	Default
level	str	"debug" (the default: everything), "info", "warning", "error" or "critical". The same setting is available before import as the OPENBTK_LOG_LEVEL environment variable.	required

Raises:

Type	Description
ValueError	For any other value.

Example

set_log_level("warning") # quiet the routine debug/info lines set_log_level("debug") # restore the default

Source code in src/openbtk/core/logging.py

def set_log_level(level: str) -> None:
    """Set the minimum level of Openbtk's own log lines, process-wide.

    Args:
        level: ``"debug"`` (the default: everything), ``"info"``, ``"warning"``,
            ``"error"`` or ``"critical"``. The same setting is available before
            import as the ``OPENBTK_LOG_LEVEL`` environment variable.

    Raises:
        ValueError: For any other value.

    Example:
        >>> set_log_level("warning")  # quiet the routine debug/info lines
        >>> set_log_level("debug")  # restore the default
    """
    global _min_level
    key = level.strip().lower()
    if key not in _LEVELS:
        raise ValueError(f"level must be one of {sorted(_LEVELS)}, got {level!r}")
    _min_level = _LEVELS[key]

Retrying

openbtk.core.retry.retry_with_backoff

retry_with_backoff(func: Callable[[], T], *, max_attempts: int = 5, base_delay: float = 1.0, max_delay: float = 60.0, jitter: float = 0.1, sleep: Callable[[float], None] = time.sleep) -> T

Call func, retrying on :class:RateLimitError with exponential backoff.

Parameters:

Name	Type	Description	Default
func	Callable[[], T]	A zero-argument callable to invoke, e.g. lambda: client.chat.completions.create(...).	required
max_attempts	int	Total attempts before giving up. Must be >= 1; the final attempt's :class:RateLimitError propagates unretried.	5
base_delay	float	Delay in seconds before the first retry.	1.0
max_delay	float	Upper bound on the computed delay, before jitter is applied.	60.0
jitter	float	Fraction of the computed delay to randomise by (+/-), so many callers backing off together do not retry in lockstep -- a real thundering-herd risk against a shared rate limit, not a hypothetical one.	0.1
sleep	Callable[[float], None]	Injectable in place of time.sleep -- tests pass a no-op so a multi-attempt backoff runs in milliseconds, not tens of seconds of real wall-clock delay.	sleep

Returns:

Type	Description
T	Whatever func() returns, from whichever attempt first succeeds.

Raises:

Type	Description
RateLimitError	If every attempt is exhausted still rate-limited.
ValueError	If max_attempts is less than 1.

Example

retry_with_backoff(lambda: "ok") 'ok'

The retry-on-RateLimitError path is not shown here since a successful retry logs a provider.rate_limited_retrying warning (docs/09_CODING_STANDARDS.md's structlog rule) whose timestamp would make this doctest non-deterministic; see tests/unit/core/test_retry.py for that behaviour, exercised directly.

Source code in src/openbtk/core/retry.py

def retry_with_backoff(
    func: Callable[[], T],
    *,
    max_attempts: int = 5,
    base_delay: float = 1.0,
    max_delay: float = 60.0,
    jitter: float = 0.1,
    sleep: Callable[[float], None] = time.sleep,
) -> T:
    """Call ``func``, retrying on :class:`RateLimitError` with exponential backoff.

    Args:
        func: A zero-argument callable to invoke, e.g.
            ``lambda: client.chat.completions.create(...)``.
        max_attempts: Total attempts before giving up. Must be >= 1; the
            final attempt's :class:`RateLimitError` propagates unretried.
        base_delay: Delay in seconds before the first retry.
        max_delay: Upper bound on the computed delay, before jitter is applied.
        jitter: Fraction of the computed delay to randomise by (+/-), so
            many callers backing off together do not retry in lockstep --
            a real thundering-herd risk against a shared rate limit, not a
            hypothetical one.
        sleep: Injectable in place of ``time.sleep`` -- tests pass a no-op
            so a multi-attempt backoff runs in milliseconds, not tens of
            seconds of real wall-clock delay.

    Returns:
        Whatever ``func()`` returns, from whichever attempt first succeeds.

    Raises:
        RateLimitError: If every attempt is exhausted still rate-limited.
        ValueError: If ``max_attempts`` is less than 1.

    Example:
        >>> retry_with_backoff(lambda: "ok")
        'ok'

    The retry-on-``RateLimitError`` path is not shown here since a
    successful retry logs a ``provider.rate_limited_retrying`` warning
    (docs/09_CODING_STANDARDS.md's structlog rule) whose timestamp would
    make this doctest non-deterministic; see
    ``tests/unit/core/test_retry.py`` for that behaviour, exercised directly.
    """
    if max_attempts < 1:
        raise ValueError(f"max_attempts must be >= 1, got {max_attempts}.")

    for attempt in range(max_attempts):
        try:
            return func()
        except RateLimitError:
            if attempt == max_attempts - 1:
                raise
            delay = min(base_delay * (2**attempt), max_delay)
            # Retry jitter, not a security value: a predictable PRNG is what we want.
            delay = max(
                delay + random.uniform(-jitter * delay, jitter * delay),  # nosec B311
                0.0,
            )
            log.warning(
                "provider.rate_limited_retrying",
                attempt=attempt + 1,
                max_attempts=max_attempts,
                delay_seconds=round(delay, 3),
            )
            sleep(delay)

    # Unreachable: max_attempts >= 1 (checked above) guarantees at least one
    # loop iteration, and every iteration either returns or raises -- the
    # last one unconditionally. Satisfies mypy's "missing return" check
    # without asserting something false; see executor.py's own
    # defensive-backstop pragma for the same pattern.
    raise RuntimeError("unreachable")  # pragma: no cover

Deprecation

openbtk.core.deprecation.deprecated

deprecated(*, since: str, removal: str, use: str | None = None) -> Callable[[_F], _F]

Mark a function, method or class as deprecated.

Calling it (or instantiating the class) still works and emits an :class:OpenBTKDeprecationWarning naming the release that removes it and what to use instead. A "Deprecated" note is appended to its docstring so the API reference shows it, and __deprecated__ holds the message.

Example

import warnings @deprecated(since="1.2.0", removal="2.0.0", use="new()") ... def old() -> int: ... return 1 with warnings.catch_warnings(record=True) as caught: ... warnings.simplefilter("always") ... result = old() result, len(caught) (1, 1)

Source code in src/openbtk/core/deprecation.py

def deprecated(
    *, since: str, removal: str, use: str | None = None
) -> Callable[[_F], _F]:
    """Mark a function, method or class as deprecated.

    Calling it (or instantiating the class) still works and emits an
    :class:`OpenBTKDeprecationWarning` naming the release that removes it and what
    to use instead. A "Deprecated" note is appended to its docstring so the API
    reference shows it, and ``__deprecated__`` holds the message.

    Example:
        >>> import warnings
        >>> @deprecated(since="1.2.0", removal="2.0.0", use="new()")
        ... def old() -> int:
        ...     return 1
        >>> with warnings.catch_warnings(record=True) as caught:
        ...     warnings.simplefilter("always")
        ...     result = old()
        >>> result, len(caught)
        (1, 1)
    """
    _check_version("since", since)
    _check_version("removal", removal)

    def decorate(obj: Any) -> Any:
        name = (
            f"{obj.__qualname__}()" if not isinstance(obj, type) else obj.__qualname__
        )
        message = _message(name, since, removal, use)
        note = f"\n\nDeprecated since {since}: {message}"
        obj.__doc__ = (obj.__doc__ or "") + note
        obj.__deprecated__ = message

        if isinstance(obj, type):
            target: Any = obj
            original_init = target.__init__

            @functools.wraps(original_init)
            def init(self: Any, *args: Any, **kwargs: Any) -> None:
                warnings.warn(message, OpenBTKDeprecationWarning, stacklevel=2)
                original_init(self, *args, **kwargs)

            target.__init__ = init
            return obj

        @functools.wraps(obj)
        def wrapper(*args: Any, **kwargs: Any) -> Any:
            warnings.warn(message, OpenBTKDeprecationWarning, stacklevel=2)
            return obj(*args, **kwargs)

        wrapper.__deprecated__ = message  # type: ignore[attr-defined]
        wrapper.__doc__ = obj.__doc__
        return wrapper

    return decorate

openbtk.core.deprecation.warn_deprecated

warn_deprecated(name: str, *, since: str, removal: str, use: str | None = None, stacklevel: int = 2) -> None

Emit an :class:OpenBTKDeprecationWarning for name.

Parameters:

Name	Type	Description	Default
name	str	What is deprecated, as the caller knows it ("registry key 'x'").	required
since	str	The release that deprecated it, e.g. "1.2.0".	required
removal	str	The release that will remove it, e.g. "2.0.0".	required
use	str | None	What to use instead, if anything.	None
stacklevel	int	As for :func:warnings.warn; the default points at the caller.	2

Example

import warnings with warnings.catch_warnings(record=True) as caught: ... warnings.simplefilter("always") ... warn_deprecated("old()", since="1.2.0", removal="2.0.0", use="new()") message = str(caught[0].message) "1.2.0" in message and "2.0.0" in message and "new()" in message True

Source code in src/openbtk/core/deprecation.py

def warn_deprecated(
    name: str, *, since: str, removal: str, use: str | None = None, stacklevel: int = 2
) -> None:
    """Emit an :class:`OpenBTKDeprecationWarning` for ``name``.

    Args:
        name: What is deprecated, as the caller knows it (``"registry key 'x'"``).
        since: The release that deprecated it, e.g. ``"1.2.0"``.
        removal: The release that will remove it, e.g. ``"2.0.0"``.
        use: What to use instead, if anything.
        stacklevel: As for :func:`warnings.warn`; the default points at the caller.

    Example:
        >>> import warnings
        >>> with warnings.catch_warnings(record=True) as caught:
        ...     warnings.simplefilter("always")
        ...     warn_deprecated("old()", since="1.2.0", removal="2.0.0", use="new()")
        >>> message = str(caught[0].message)
        >>> "1.2.0" in message and "2.0.0" in message and "new()" in message
        True
    """
    _check_version("since", since)
    _check_version("removal", removal)
    warnings.warn(
        _message(name, since, removal, use),
        OpenBTKDeprecationWarning,
        stacklevel=stacklevel,
    )

openbtk.core.deprecation.OpenBTKDeprecationWarning

Bases: DeprecationWarning

Raised (as a warning) when a deprecated Openbtk API is used.

Errors

Every Openbtk error derives from OpenBTKError, carries a context dict of identifiers only (never content), and is safe to log.

openbtk.core.errors.OpenBTKError

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

Bases: Exception

Root of every exception raised by Openbtk.

Parameters:

Name	Type	Description	Default
message	str	Human-readable description of what failed, why, and what to do about it. Never includes PHI or raw document content.	required
context	dict[str, Any] | None	Structured, PHI-free metadata about where the failure occurred -- e.g. {"modality": "clinical_text", "stage": "load", "source": "/data/notes.csv"}. Identifiers only, never content.	None

Attributes:

Name	Type	Description
context	dict[str, Any]	The context dict passed at construction, always a dict (never None) so callers can update it uniformly.

Example

try: ... raise OpenBTKError("boom", context={"stage": "load"}) ... except OpenBTKError as e: ... e.context["stage"] 'load'

Source code in src/openbtk/core/errors.py

def __init__(self, message: str, *, context: dict[str, Any] | None = None) -> None:
    super().__init__(message)
    self.context: dict[str, Any] = dict(context) if context is not None else {}

openbtk.core.errors.ConfigError

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

Bases: OpenBTKError

A pipeline or component configuration is malformed or invalid.

Source code in src/openbtk/core/errors.py

def __init__(self, message: str, *, context: dict[str, Any] | None = None) -> None:
    super().__init__(message)
    self.context: dict[str, Any] = dict(context) if context is not None else {}

openbtk.core.errors.RegistryError

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

Bases: OpenBTKError

A registry operation failed: unknown key, duplicate key, or a class that does not subclass the registry's declared base type.

Source code in src/openbtk/core/errors.py

def __init__(self, message: str, *, context: dict[str, Any] | None = None) -> None:
    super().__init__(message)
    self.context: dict[str, Any] = dict(context) if context is not None else {}

openbtk.core.errors.MissingDependencyError

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

Bases: OpenBTKError

An optional dependency is required but not installed.

Always names the extra to install (pip install 'openbtk[<extra>]') rather than letting a bare ImportError propagate. Raised by openbtk.core._lazy.require().

Source code in src/openbtk/core/errors.py

def __init__(self, message: str, *, context: dict[str, Any] | None = None) -> None:
    super().__init__(message)
    self.context: dict[str, Any] = dict(context) if context is not None else {}

openbtk.core.errors.PolicyError

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

Bases: OpenBTKError

An operation is disallowed by the active pipeline policy.

The primary case: constructing or invoking a provider whose sends_data_offsite is True while policy.allow_offsite_providers is False.

Source code in src/openbtk/core/errors.py

def __init__(self, message: str, *, context: dict[str, Any] | None = None) -> None:
    super().__init__(message)
    self.context: dict[str, Any] = dict(context) if context is not None else {}

openbtk.core.errors.PluginError

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

Bases: OpenBTKError

Entry-point plugin discovery or registration failed.

Raised internally during plugin loading; load_plugins() catches these per-plugin and logs a warning rather than letting one broken third-party plugin prevent import openbtk from succeeding.

Source code in src/openbtk/core/errors.py

def __init__(self, message: str, *, context: dict[str, Any] | None = None) -> None:
    super().__init__(message)
    self.context: dict[str, Any] = dict(context) if context is not None else {}

openbtk.core.errors.LoaderError

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

Bases: OpenBTKError

A loader failed to read or parse its source.

Source code in src/openbtk/core/errors.py

def __init__(self, message: str, *, context: dict[str, Any] | None = None) -> None:
    super().__init__(message)
    self.context: dict[str, Any] = dict(context) if context is not None else {}

openbtk.core.errors.ProcessingError

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

Bases: OpenBTKError

A preprocessor, chunker, segmenter, or feature extractor failed.

Source code in src/openbtk/core/errors.py

def __init__(self, message: str, *, context: dict[str, Any] | None = None) -> None:
    super().__init__(message)
    self.context: dict[str, Any] = dict(context) if context is not None else {}

openbtk.core.errors.DatasetError

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

Bases: OpenBTKError

A dataset adapter failed: missing credentials, bad path, or a network error while accessing a named dataset.

Source code in src/openbtk/core/errors.py

def __init__(self, message: str, *, context: dict[str, Any] | None = None) -> None:
    super().__init__(message)
    self.context: dict[str, Any] = dict(context) if context is not None else {}

openbtk.core.errors.DeidError

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

Bases: OpenBTKError

A de-identification component failed: a malformed consistency key, an unresolvable transform, or a recognizer failure.

Never carries the text being de-identified or any detected value -- only identifiers (document id, category, recognizer key). The whole point of openbtk.deid is to keep PHI from spreading; an exception about it must not become a new place PHI spread to.

Source code in src/openbtk/core/errors.py

def __init__(self, message: str, *, context: dict[str, Any] | None = None) -> None:
    super().__init__(message)
    self.context: dict[str, Any] = dict(context) if context is not None else {}

openbtk.core.errors.ProviderError

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

Bases: OpenBTKError

An LLM or embedding provider failed to fulfil a request.

Source code in src/openbtk/core/errors.py

def __init__(self, message: str, *, context: dict[str, Any] | None = None) -> None:
    super().__init__(message)
    self.context: dict[str, Any] = dict(context) if context is not None else {}

openbtk.core.errors.RateLimitError

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

Bases: ProviderError

A provider rejected a request due to rate limiting.

Distinguished from a generic ProviderError so callers can apply backoff-and-retry logic specifically to this case.

Source code in src/openbtk/core/errors.py

def __init__(self, message: str, *, context: dict[str, Any] | None = None) -> None:
    super().__init__(message)
    self.context: dict[str, Any] = dict(context) if context is not None else {}

openbtk.core.errors.AuthenticationError

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

Bases: ProviderError

A provider rejected a request due to invalid or missing credentials.

Never includes the credential value itself in the message or context -- only that authentication failed and which provider rejected it.

Source code in src/openbtk/core/errors.py

def __init__(self, message: str, *, context: dict[str, Any] | None = None) -> None:
    super().__init__(message)
    self.context: dict[str, Any] = dict(context) if context is not None else {}

openbtk.core.errors.RetrievalError

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

Bases: OpenBTKError

A vector store operation (upsert, query, delete) failed.

Source code in src/openbtk/core/errors.py

def __init__(self, message: str, *, context: dict[str, Any] | None = None) -> None:
    super().__init__(message)
    self.context: dict[str, Any] = dict(context) if context is not None else {}

openbtk.core.errors.TerminologyError

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

Bases: OpenBTKError

A terminology lookup failed: unavailable backend, unlicensed vocabulary, or an unresolvable code.

Source code in src/openbtk/core/errors.py

def __init__(self, message: str, *, context: dict[str, Any] | None = None) -> None:
    super().__init__(message)
    self.context: dict[str, Any] = dict(context) if context is not None else {}

openbtk.core.errors.GuardrailViolation

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

Bases: OpenBTKError

A guardrail result was severity BLOCK and the pipeline is configured to raise rather than continue.

This is not a bug in the guardrail -- BaseGuardrail.check() never raises. This exception is raised by the pipeline layer after inspecting a GuardrailResult, so it is meant to be caught at that level, not inside component code.

Source code in src/openbtk/core/errors.py

def __init__(self, message: str, *, context: dict[str, Any] | None = None) -> None:
    super().__init__(message)
    self.context: dict[str, Any] = dict(context) if context is not None else {}