Guardrails¶

Composable input/output scanning middleware for any BaseAgent. Guardrails intercept the agent stream to validate, block, rewrite, or audit prompts and responses — without modifying agent code.

Two APIs are provided:

@guardrail decorator -- define rules as methods on your agent class (auto-discovered).
GuardrailMiddleware -- wrap any agent programmatically with scanner instances.

Both run pre-scanners (before generation) and post-scanners (after generation) concurrently, emit structured events to configurable sinks, and log metrics to MLflow when active.

Guardrails vs Scanners¶

A scanner is a reusable, stateless detection unit — a class with an async scan (or scan_messages) method that inspects text and returns a ScanResult. Scanners only detect; they don't decide policy, and the same scanner instance works across any agent or middleware. A guardrail is the policy and enforcement layer that runs scanners (or inline checks) against a specific agent: the @guardrail decorator and GuardrailMiddleware bind a check into an agent's stream() at a chosen timing (pre/post) with an outcome (active = block/retract, passive = log only). Think of it this way: scanners are the sensors; guardrails are the policy that runs them and decides what to do when one fires.

	Scanner	Guardrail
What it is	A detection unit (sensor)	A policy that runs scanners (enforcement)
Returns	`ScanResult` (passed/score/rewrites/blocked_response)	Block, retract, rewrite, or log — based on outcome
Reuse	Reusable across any agent/middleware	Bound to a specific agent + timing + outcome
API	`scan` / `scan_messages` method	`@guardrail` decorator or `GuardrailMiddleware`

A @guardrail factory method returns a scanner instance; an inline @guardrail is its own check. See Scanners for the detection units and @guardrail / GuardrailMiddleware below for the policy layer.

Installation¶

InvisibleTextScanner and TokenLimitScanner require no extras. LanguageScanner relies on langdetect, which ships in the guardrails extra. ML-based scanners and Hebrew locale scanners need their own extras:

uv add "latent[guardrails]"            # langdetect-backed scanners (LanguageScanner)
uv add "latent[guardrails-llmguard]"   # LLM Guard ML scanners
uv add "latent[guardrails-hebrew]"     # Hebrew locale scanners

`@guardrail` Method Decorator¶

Decorate methods on any BaseAgent subclass. The framework discovers them automatically at init time -- no registration required.

from latent.guardrails import guardrail
from latent.agents import ReActAgent

class MyAgent(ReActAgent):
    @guardrail(timing="pre", outcome="active", message="Blocked: profanity detected")
    async def no_profanity(self, prompt: str) -> bool:
        return "badword" in prompt.lower()  # True = violation

    @guardrail(timing="pre", outcome="active")
    def injection_scanner(self):
        from latent.guardrails.scanners.llmguard import PromptInjectionScanner
        return PromptInjectionScanner(threshold=0.95)

    @guardrail(timing="post", outcome="passive")
    async def pii_audit(self, prompt: str, output: str) -> float:
        return compute_pii_score(output)  # >= threshold (default 0.5) = flagged

Parameters¶

Parameter	Type	Default	Description
`timing`	`"pre"` / `"post"`	`"pre"`	When to run: before or after generation
`outcome`	`"active"` / `"passive"`	`"active"`	Active blocks/retracts; passive logs only
`threshold`	`float`	`0.5`	Score threshold for float-returning rules
`on_error`	`"ignore"` / `"raise"`	`"ignore"`	Error handling: swallow or propagate
`message`	`str`	`"Request blocked by guardrail."`	User-facing message when blocked
`every`	`int`	`0`	Post rules: 0 = end-of-stream only, N = every N streamed chunks (TextDelta events)
`on_block`	`Callable \\| None`	`None`	Invoked when an active rule blocks; receives a `GuardrailBlockContext` and its return value (sync or async) replaces the static `message` as the block response

Dynamic block messages¶

Pass on_block to compute the block response at runtime instead of using a static message. The callback receives a GuardrailBlockContext and may be sync or async; its return value becomes the blocked_response.

from latent.guardrails import guardrail
from latent.guardrails.base import GuardrailBlockContext
from latent.agents import ReActAgent

def handle_block(ctx: GuardrailBlockContext) -> str:
    return f"Blocked '{ctx.rule_name}' (score={ctx.score:.2f}). Please rephrase."

class MyAgent(ReActAgent):
    @guardrail(timing="pre", outcome="active", on_block=handle_block)
    async def no_profanity(self, prompt: str) -> bool:
        return "badword" in prompt.lower()

GuardrailBlockContext fields:

Field	Type	Description
`rule_name`	`str`	Which rule blocked
`score`	`float`	Violation score
`prompt`	`str`	The scanned prompt
`message`	`str`	The static block message
`metadata`	`dict`	Scanner-specific data
`agent`	`Any`	The agent instance

Signature Detection¶

The decorator infers the rule type from the method signature:

Signature	Type	Behavior
`(self)`	Factory	Called once at init; must return a scanner instance
`(self, prompt: str)`	Pre inline	Runs on each input
`(self, prompt: str, output: str)`	Post inline	Runs on each output

Return Types¶

Return type	Semantics
`-> bool`	`True` = violation, `False` = pass
`-> float`	Score compared against `threshold`; `>= threshold` = violation
`-> ScanResult`	Full control -- pass through directly

Async-only since 5.0.0

Inline methods ((self, prompt) / (self, prompt, output)) must be async def — the decorator raises TypeError otherwise. Factory methods ((self) returning a scanner instance) stay synchronous def.

`GuardrailMiddleware`¶

For programmatic composition when you want to attach scanners to an agent without subclassing.

from latent.guardrails import GuardrailMiddleware
from latent.guardrails.scanners.builtin import LanguageScanner, TokenLimitScanner
from latent.guardrails.scanners.llmguard import MaliciousURLsScanner

wrapped = GuardrailMiddleware(
    agent,
    pre_scanners=[
        LanguageScanner(allowed_languages=["en", "he"]),
        TokenLimitScanner(max_tokens=4000),
    ],
    post_scanners=[
        MaliciousURLsScanner(),
    ],
)

# Use wrapped.stream(messages) instead of agent.stream(messages)
async for event in wrapped.stream(messages):
    handle(event)

Constructor¶

GuardrailMiddleware(
    agent,
    *,
    pre_scanners: list | None = None,
    post_scanners: list | None = None,
    sinks: list[EventSink] | None = None,  # default: [StructlogSink()]
    enabled: bool = True,
    max_log_chars: int = 500,
)

Stream Behavior¶

Pre-scan -- all pre-scanners run concurrently. If an active scanner fails, the block message is first emitted as a leading TextDelta(text=message) (so downstream consumers persist it as assistant text), followed by the structured GuardrailViolation event, and then the stream stops. Passive violations are yielded as informational events.
Input rewriting -- if a scanner sets rewritten_input, the rewritten prompt is forwarded to the agent.
Agent stream -- TextDelta events are yielded as they arrive, buffered for post-scan.
Mid-stream post-scan -- scanners with every > 0 run every N streamed chunks (TextDelta events) during streaming. Active violations halt the stream.
End-of-stream post-scan -- scanners with every = 0 run concurrently after the full response is collected.
Metrics -- guardrail counters are logged to MLflow when active.

Metrics¶

Access counters with wrapped.get_metrics():

Key	Description
`guardrails.input.scanned`	Total inputs scanned
`guardrails.input.blocked`	Inputs blocked by active pre-scanners
`guardrails.output.scanned`	Outputs scanned
`guardrails.active.violations.<rule>`	Active violations per rule
`guardrails.passive.violations.<rule>`	Passive violations per rule

Scanners¶

Scanners are the reusable scanning units that pre/post-scanners and @guardrail factories run. Each implements an async scan (or scan_messages) method and returns a ScanResult. The built-in input/output scanners are:

LanguageScanner — blocks disallowed languages (needs the guardrails extra).
InvisibleTextScanner — detects zero-width / invisible Unicode characters.
TokenLimitScanner — blocks prompts over a token estimate.
LLMGuardrailScanner — LLM-backed custom rule via Judge[ViolationScore].

LLM Guard ML scanners (prompt injection, toxicity, PII, etc.) require guardrails-llmguard, and Hebrew locale scanners require guardrails-hebrew. Context-engineering scanners operate on the full message array via scan_messages.

See Scanners for the full catalog (constructor params, ML and Hebrew scanners, context scanners), the scan protocol, the ScanResult fields, and how to write a custom scanner.

Events and Observability¶

`GuardrailViolation`¶

Yielded from stream() when a guardrail fires. Inspect it to decide how to handle the violation in your application.

Field	Type	Description
`rule_name`	`str`	Which rule fired
`timing`	`"pre"` / `"post"`	When it fired
`outcome`	`"active"` / `"passive"`	Whether it blocks
`score`	`float`	Violation score
`input_text`	`str`	The scanned input prompt (truncated to `max_log_chars`)
`output_text`	`str` / `None`	The scanned output (post violations; rewritten output for redaction scanners)
`message`	`str`	User-facing message
`retraction`	`bool`	`True` if already-streamed content should be replaced
`scrub_history`	`bool`	`True` if the message should be removed from history

Active pre-violation: stream never starts. Use message as the response. scrub_history=True -- remove the bad input from conversation history.

Active post-violation: content already streamed. retraction=True -- the frontend should replace visible content with message. scrub_history=True -- remove the response from history.

Passive violation: informational only. Log it, alert on it, but don't block.

Event Sinks¶

All scan results (pass and fail) are emitted to configurable sinks. The canonical import is latent.observability; the old latent.guardrails.sinks is a deprecated shim that emits a DeprecationWarning.

Sink	Description
`StructlogSink`	Structured logging (default)
`WebhookSink`	POST events to an HTTP endpoint
`OpenTelemetrySink`	Emit spans to an OTel collector
`NoOpSink`	Discard all events (suppress default logging)

WebhookSink and OpenTelemetrySink both accept max_text_chars: int = 500 to cap the length of text attributes recorded on emitted events/spans.

from latent.observability import StructlogSink, WebhookSink, OpenTelemetrySink, NoOpSink

# Custom sinks
wrapped = GuardrailMiddleware(
    agent,
    pre_scanners=[...],
    sinks=[
        StructlogSink(),
        WebhookSink(url="https://hooks.example.com/guardrails", max_text_chars=500),
        OpenTelemetrySink(max_text_chars=500),
    ],
)

# Suppress all event logging
wrapped = GuardrailMiddleware(agent, pre_scanners=[...], sinks=[NoOpSink()])

Configuration¶

Guardrail settings can be loaded from TOML config files. The system searches in order:

[guardrails] section in latent.toml / config/latent.toml
Standalone guardrails.toml / config/guardrails.toml
Defaults

# latent.toml
[guardrails]
enabled = true
max_log_chars = 500

Note

Scanner and sink configuration via TOML is not yet supported. Configure scanners and sinks in code.

Integration Examples¶

Agent with Mixed Guardrails¶

Combining decorator-based rules with programmatic scanners:

from latent.guardrails import guardrail, GuardrailMiddleware
from latent.guardrails.scanners.builtin import LanguageScanner, InvisibleTextScanner
from latent.agents import ReActAgent

class SupportAgent(ReActAgent):
    """Agent with built-in guardrails via decorators."""

    @guardrail(timing="pre", outcome="active", message="Please use a supported language.")
    def language_check(self):
        return LanguageScanner(allowed_languages=["en", "he"])

    @guardrail(timing="pre", outcome="active")
    def invisible_chars(self):
        return InvisibleTextScanner()

    @guardrail(timing="post", outcome="passive", every=50)
    async def length_monitor(self, prompt: str, output: str) -> bool:
        return len(output) > 10000

agent = SupportAgent(model="gpt-4o", system_prompt="You are a helpful assistant.")

Consuming Violations in a Chat Loop¶

from latent.agents.events import TextDelta
from latent.guardrails.events import GuardrailViolation

async for event in agent.stream(messages):
    if isinstance(event, GuardrailViolation):
        if event.outcome == "active" and event.timing == "pre":
            # Input was blocked -- show the message, don't add to history
            print(f"[BLOCKED] {event.message}")
            break
        elif event.outcome == "active" and event.timing == "post":
            # Output retracted -- replace what was shown
            print(f"\n[RETRACTED] {event.message}")
        else:
            # Passive -- log for monitoring
            log_violation(event)
    elif isinstance(event, TextDelta):
        print(event.text, end="", flush=True)

Evaluation Metrics¶

Measure guardrail accuracy against labeled data using the built-in metrics:

from latent.guardrails.metrics import guardrail_adherence_metric, per_scanner_breakdown_metric

# Overall accuracy: did the system block when it should have?
score = guardrail_adherence_metric(
    {"expected_blocked": True},
    {"blocked": True},
)  # -> 1.0

# Per-rule accuracy
injection_metric = per_scanner_breakdown_metric("prompt_injection")
score = injection_metric(
    {"expected_blocked": True},
    {"violated_rules": ["prompt_injection", "toxicity"]},
)  # -> 1.0