windlass.interfaces.guardrail

guardrail

The guardrail interface.

Guardrails inspect what goes into a model and what comes out. Windlass runs them at two points — input before the prompt is sent, output before the answer is returned — and lets each guardrail decide between three verdicts:

• allow — nothing objectionable found.
• redact — rewrite the content (mask a credit card, strip a name) and continue. This is the default for PII.
• block — refuse. Raises :class:~windlass.core.exceptions.GuardrailViolation unless the pipeline was told to return a refusal message instead.

Implementers override one coroutine, :meth:Guardrail.acheck.

Example

from windlass.providers.guardrails.rules import RuleGuardrail g = RuleGuardrail(pii=True, on_violation="redact") g.check("mail me at a@b.com").content 'mail me at [EMAIL]'

Guardrail

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

Bases: Component

Abstract input/output safety check.

Parameters:

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

Attributes:

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

Raises:

Type	Description
ValueError	If on_violation is not a recognised action.

Example

Implementing a guardrail takes one method::

class NoShouting(Guardrail):
    provider_name = "no_shouting"

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

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

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

acheck abstractmethod async

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

Inspect content and return a verdict.

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

Parameters:

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

Returns:

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

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

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

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

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

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

avalidate async

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

Check content and apply the configured policy.

Parameters:

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

Returns:

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

Raises:

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

Example

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

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

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

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

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

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

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

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

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

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

validate

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

Blocking :meth:avalidate.

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

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

check

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

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

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

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

GuardrailChain

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

Bases: Guardrail

Runs several guardrails in order, threading redactions through.

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

Parameters:

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

Attributes:

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

Example

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

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

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

add

add(guard: Guardrail) -> GuardrailChain

Append a guardrail and return self.

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

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

acheck async

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

Aggregate every child's detections without enforcing them.

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

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

avalidate async

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

Run every guardrail's own policy in sequence.

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

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

describe

describe() -> dict[str, Any]

Return a summary including each guardrail.

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

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