Guardrails

A guardrail always returns a GuardrailResult; it never raises for a failed check. Whether a BLOCK halts a pipeline is the caller's policy. Import openbtk.guardrails to register them all.

General

openbtk.guardrails.phi_leakage.PHILeakageGuardrail

PHILeakageGuardrail(*, recognizers: tuple[str, ...] = ('rule',))

Bases: BaseGuardrail

Flags PHI-shaped spans in a payload.

Parameters:

Name	Type	Description	Default
recognizers	tuple[str, ...]	Forwarded to the wrapped DeidEngine -- defaults to the same zero-extras-safe default DeidEngine itself uses (rule-based only; pass ("rule", "ner") for the opt-in NER ensemble if the text extra is installed).	('rule',)

Example

import contextlib, io with contextlib.redirect_stdout(io.StringIO()): # registry logging ... guardrail = PHILeakageGuardrail() ... result = guardrail.check("Patient SSN: 123-45-6789.") ... clean = guardrail.check("No identifiers here.") result.passed False result.severity.value 'block' clean.passed True

Source code in src/openbtk/guardrails/phi_leakage.py

def __init__(self, *, recognizers: tuple[str, ...] = ("rule",)) -> None:
    self._engine = DeidEngine(mode=DeidMode.TAG, recognizers=list(recognizers))

openbtk.guardrails.terminology_validity.TerminologyValidityGuardrail

TerminologyValidityGuardrail(*, terminology: BaseTerminologyService | None = None)

Bases: BaseGuardrail

Checks that every code found in a payload validates against its declared terminology system.

Parameters:

Name	Type	Description	Default
terminology	BaseTerminologyService | None	The service to validate codes against. Defaults to :class:~openbtk.terminology.bundled.BundledMinimalBackend -- a real, working, zero-config default so this guardrail is usable and contract-testable with no required constructor args, at the cost of only covering ICD-10-CM out of the box. Inject a more complete backend (UMLSRestBackend, LocalVocabBackend, or CachedTerminologyService wrapping either) for real coverage of other systems.	None

A code the backend cannot confirm (TerminologyError -- an unlicensed or unavailable backend) is reported as a WARNING, not silently treated as valid or as a BLOCK-worthy invalid code: docs/03_ARCHITECTURE.md section 8.2 is explicit that a terminology backend "does not silently return valid," and conflating "could not check" with "is invalid" would be its own kind of silent wrong answer.

Example

from openbtk.core.schemas import CodeSystem guardrail = TerminologyValidityGuardrail() guardrail.check(("E11.9", CodeSystem.ICD10CM)).passed True guardrail.check(("Z99.999", CodeSystem.ICD10CM)).passed False

Source code in src/openbtk/guardrails/terminology_validity.py

def __init__(self, *, terminology: BaseTerminologyService | None = None) -> None:
    self._terminology = (
        terminology if terminology is not None else BundledMinimalBackend()
    )

openbtk.guardrails.groundedness.GroundednessGuardrail

GroundednessGuardrail(*, support_threshold: float = 0.5, decompose_claims: Callable[[str], list[str]] = _default_decompose_claims, is_supported: Callable[[str, list[str]], bool] | None = None)

Bases: BaseGuardrail

Checks whether every claim in a generated answer is supported by its retrieved context, returning unsupported spans.

Parameters:

Name	Type	Description	Default
support_threshold	float	Fraction of a claim's significant words that must appear in the context for the default is_supported to count it as grounded. Ignored if is_supported is overridden.	0.5
decompose_claims	Callable[[str], list[str]]	text -> list[claim]. Defaults to a sentence splitter.	_default_decompose_claims
is_supported	Callable[[str, list[str]], bool] | None	(claim, context) -> bool. Defaults to a significant-word-overlap heuristic against support_threshold.	None

check() only understands a :class:GroundednessCheckInput payload (it needs both the answer AND its context together, which no other schema in this project bundles) -- any other payload shape passes with an INFO result, the same "not my concern, not a failure" treatment TerminologyValidityGuardrail gives an unrelated payload.

Example

guardrail = GroundednessGuardrail() grounded = GroundednessCheckInput( ... answer="The patient has type 2 diabetes.", ... context=["Assessment: type 2 diabetes mellitus, stable."], ... ) guardrail.check(grounded).passed True ungrounded = GroundednessCheckInput( ... answer="The patient has a fractured femur.", ... context=["Assessment: type 2 diabetes mellitus, stable."], ... ) result = guardrail.check(ungrounded) result.passed False len(result.spans) 1

Source code in src/openbtk/guardrails/groundedness.py

def __init__(
    self,
    *,
    support_threshold: float = 0.5,
    decompose_claims: Callable[[str], list[str]] = _default_decompose_claims,
    is_supported: Callable[[str, list[str]], bool] | None = None,
) -> None:
    self._decompose_claims = decompose_claims
    self._is_supported = is_supported or (
        lambda claim, context: _default_is_supported(
            claim, context, support_threshold
        )
    )

openbtk.guardrails.groundedness.GroundednessCheckInput

Bases: BaseModel

The payload this guardrail's check() understands: a generated answer plus the source text it was supposed to be grounded in.

Example

GroundednessCheckInput(answer="The sky is blue.").context []

openbtk.guardrails.dose.DosePlausibilityGuardrail

DosePlausibilityGuardrail(*, reference: str | None = None, limits: Sequence[DoseLimit | Mapping[str, object]] | None = None)

Bases: BaseGuardrail

Flags a dose, daily total or route that breaks a limit in a reference table.

Parameters:

Name	Type	Description	Default
reference	str | None	Path to a JSON file holding a list of :class:DoseLimit objects. Read on first use, not at construction.	None
limits	Sequence[DoseLimit | Mapping[str, object]] | None	The same limits given directly, instead of (or as well as) a file.	None

Example

limits = [DoseLimit( ... drug="examplamine", unit="mg", max_single=100, max_daily=300, ... source="illustrative values for documentation only", ... )] guardrail = DosePlausibilityGuardrail(limits=limits) guardrail.check("Give examplamine 500 mg PO BID.").passed False guardrail.check("Give examplamine 50 mg PO BID.").passed True

Source code in src/openbtk/guardrails/dose.py

def __init__(
    self,
    *,
    reference: str | None = None,
    limits: Sequence[DoseLimit | Mapping[str, object]] | None = None,
) -> None:
    self._reference = reference
    self._inline = list(limits) if limits else []
    self._loaded: list[DoseLimit] | None = None
    self._matcher: re.Pattern[str] | None = None
    self._by_name: dict[str, DoseLimit] = {}

provenance

provenance() -> ComponentProvenance

Records the number of limits and a hash of the table, not its content.

Source code in src/openbtk/guardrails/dose.py

def provenance(self) -> ComponentProvenance:
    """Records the number of limits and a hash of the table, not its content."""
    base = super().provenance()
    try:
        limits = self._load()
    except (ConfigError, ValueError):
        return base.model_copy(update={"config": {"reference": "unreadable"}})
    digest = hashlib.sha256(
        json.dumps(
            sorted((lim.model_dump() for lim in limits), key=lambda d: d["drug"]),
            sort_keys=True,
        ).encode()
    ).hexdigest()
    return base.model_copy(
        update={"config": {"limits": len(limits), "reference_sha256": digest}}
    )

openbtk.guardrails.dose.DoseLimit

Bases: BaseModel

One drug's reference limits, with the source they came from.

Example

limit = DoseLimit( ... drug="examplamine", unit="mg", max_single=100, ... source="illustrative values for documentation only", ... ) limit.max_daily is None True

EHR

openbtk.guardrails.ehr.EHRCodeValidityGuardrail

EHRCodeValidityGuardrail(*, terminology: BaseTerminologyService | None = None)

Bases: TerminologyValidityGuardrail

Every code in a PatientRecord (conditions/medications/ procedures/observations) exists in its declared system.

Genuinely the identical check :class:TerminologyValidityGuardrail already performs -- PatientRecord was already one of its supported payload shapes. Registered under this separate, modality- scoped key because docs/05_DATA_MODALITY_SPEC.md section 2.5 names one explicitly, and a real subclass (not TERMINOLOGY_REGISTRY.register_alias, whose own docstring scopes it to a deprecation transition for the same class under a renamed key, not two permanent, independent semantic roles) keeps registry_key/provenance honest about which key actually constructed a given instance.

Source code in src/openbtk/guardrails/terminology_validity.py

def __init__(self, *, terminology: BaseTerminologyService | None = None) -> None:
    self._terminology = (
        terminology if terminology is not None else BundledMinimalBackend()
    )

openbtk.guardrails.ehr.ReferentialIntegrityGuardrail

Bases: BaseGuardrail

Every event's encounter_ref resolves to a real encounter, and its timestamp (when known) falls within that encounter's window.

Example

import contextlib, io from datetime import datetime, UTC from openbtk.core.schemas import CodeSystem with contextlib.redirect_stdout(io.StringIO()): ... from openbtk.data.ehr.schemas import ( ... CodedEvent, Demographics, Encounter, PatientRecord, ... ) record = PatientRecord( ... patient_id="pt-1", ... demographics=Demographics(), ... encounters=[Encounter( ... encounter_id="enc-1", ... start=datetime(2024, 3, 14, tzinfo=UTC), ... end=datetime(2024, 3, 15, tzinfo=UTC), ... )], ... conditions=[CodedEvent( ... code="x", system=CodeSystem.SNOMED, encounter_ref="enc-1", ... timestamp=datetime(2024, 3, 20, tzinfo=UTC), ... )], ... source_system="fhir-r4", ... ) ReferentialIntegrityGuardrail().check(record).passed False

openbtk.guardrails.ehr.UnitPlausibilityGuardrail

Bases: BaseGuardrail

Every observation's unit is a recognised UCUM string, and its value falls within a wide, illustrative physiological range for known LOINC codes.

A data-quality check, not a clinical decision support tool -- see this module's own _PHYSIOLOGICAL_RANGES docstring comment for the real, disclosed scope.

Example

from openbtk.data.ehr.schemas import ( ... Demographics, Measurement, PatientRecord, ... ) record = PatientRecord( ... patient_id="pt-1", ... demographics=Demographics(), ... observations=[Measurement(code="6690-2", value=400000, unit="10*3/uL")], ... source_system="fhir-r4", ... ) UnitPlausibilityGuardrail().check(record).passed False

openbtk.guardrails.ehr.CohortKAnonymityGuardrail

CohortKAnonymityGuardrail(*, k: int = 5, quasi_identifiers: Sequence[str] = ('birth_year', 'gender', 'race', 'ethnicity'), as_of: date | str | None = None)

Bases: BaseGuardrail

A cohort is k-anonymous over its quasi-identifiers, or the check warns.

Give it the cohort (any iterable of PatientRecord, consumed once) and it counts how many patients share each combination of the chosen attributes. If the smallest group is under k the result is a WARNING (per docs/06_SECURITY_COMPLIANCE.md section 3.10) saying how many patients are in a too-small group and how many are unique. The result carries counts only, never a value.

Parameters:

Name	Type	Description	Default
k	int	The smallest acceptable group. 5 is a common floor, not a rule; the right value is a policy decision for your data.	5
quasi_identifiers	Sequence[str]	Attributes to combine; see openbtk.data.ehr.cohort.QUASI_IDENTIFIER_FIELDS.	('birth_year', 'gender', 'race', 'ethnicity')
as_of	date | str | None	The date age is computed on (required only when age is used).	None

It reaches "satisfied" only over the attributes you name, and says nothing about what a group has in common. See openbtk.deid.kanonymity for what k-anonymity does and does not give you, and anonymise_to_k to generalise a table until it passes.

Example

from openbtk.data.ehr.schemas import Demographics, PatientRecord cohort = [ ... PatientRecord( ... patient_id=f"pt-{i}", ... demographics=Demographics(gender="female"), ... source_system="fhir-r4", ... ) ... for i in range(6) ... ] guardrail = CohortKAnonymityGuardrail(k=5, quasi_identifiers=["gender"]) guardrail.check(cohort).passed True guardrail.check(cohort[:2]).passed False

Source code in src/openbtk/guardrails/ehr.py

def __init__(
    self,
    *,
    k: int = 5,
    quasi_identifiers: Sequence[str] = (
        "birth_year",
        "gender",
        "race",
        "ethnicity",
    ),
    as_of: date | str | None = None,
) -> None:
    self._k = k
    self._fields = list(quasi_identifiers)
    self._as_of = as_of

provenance

provenance() -> ComponentProvenance

Records the threshold and the attributes used.

Source code in src/openbtk/guardrails/ehr.py

def provenance(self) -> ComponentProvenance:
    """Records the threshold and the attributes used."""
    return (
        super()
        .provenance()
        .model_copy(
            update={
                "config": {
                    "k": self._k,
                    "quasi_identifiers": list(self._fields),
                    "as_of": str(self._as_of) if self._as_of else None,
                }
            }
        )
    )

Composing guardrails

openbtk.guardrails.pipeline.GuardrailPipeline

GuardrailPipeline(guardrails: Sequence[BaseGuardrail | str], *, policy: PolicyConfig | None = None, short_circuit: bool = True)

Run an ordered list of guardrails over one payload.

Parameters:

Name	Type	Description	Default
guardrails	Sequence[BaseGuardrail | str]	Each entry is either a real BaseGuardrail instance or a registry key string (constructed via GUARDRAIL_REGISTRY.create).	required
policy	PolicyConfig | None	Forwarded to GUARDRAIL_REGISTRY.create for any string entries -- irrelevant for guardrails that don't send data offsite (true of every guardrail this project ships).	None
short_circuit	bool	When True (the default), stop running further guardrails once one returns a failed BLOCK-severity result. When False, every guardrail always runs, for a complete diagnostic report rather than a fail-fast check.	True

Example

import contextlib, io with contextlib.redirect_stdout(io.StringIO()): ... from openbtk.guardrails.phi_leakage import PHILeakageGuardrail ... pipeline = GuardrailPipeline([PHILeakageGuardrail()]) ... result = pipeline.run("Patient SSN: 123-45-6789.") result.passed False len(result.results) 1

Source code in src/openbtk/guardrails/pipeline.py

def __init__(
    self,
    guardrails: Sequence[BaseGuardrail | str],
    *,
    policy: PolicyConfig | None = None,
    short_circuit: bool = True,
) -> None:
    self._guardrails = [_resolve(g, policy) for g in guardrails]
    self._short_circuit = short_circuit

openbtk.guardrails.pipeline.GuardrailPipelineResult

Bases: BaseModel

The outcome of one :meth:GuardrailPipeline.run call.

Example

result = GuardrailPipelineResult( ... passed=True, ... results=[GuardrailResult( ... passed=True, severity=GuardrailSeverity.INFO, ... guardrail_key="guardrail.general.phi_leakage", ... message="No PHI detected.", ... )], ... ) result.short_circuited False

to_guardrail_outcomes

to_guardrail_outcomes(at: str) -> list[GuardrailOutcome]

Reshape each result into a GuardrailOutcome (checked_count=1 each, since this pipeline checks one payload, not a whole streamed run) for attaching to a RunManifest.

Parameters:

Name	Type	Description	Default
at	str	An attachment-point-shaped label the caller chooses -- this pipeline has no step id of its own, unlike the executor's own guardrail attachment.	required

Source code in src/openbtk/guardrails/pipeline.py

def to_guardrail_outcomes(self, at: str) -> list[GuardrailOutcome]:
    """Reshape each result into a ``GuardrailOutcome`` (``checked_count=1``
    each, since this pipeline checks one payload, not a whole
    streamed run) for attaching to a ``RunManifest``.

    Args:
        at: An attachment-point-shaped label the caller chooses --
            this pipeline has no step id of its own, unlike the
            executor's own guardrail attachment.
    """
    return [
        GuardrailOutcome(
            guardrail_key=r.guardrail_key,
            at=at,
            checked_count=1,
            blocked_count=int(
                not r.passed and r.severity is GuardrailSeverity.BLOCK
            ),
            warned_count=int(
                not r.passed and r.severity is GuardrailSeverity.WARNING
            ),
            sample_messages=[] if r.passed else [r.message],
        )
        for r in self.results
    ]