latent.guardrails.middleware¶

Classes¶

GuardrailMiddleware¶

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

Stream-intercepting wrapper that runs guardrail scanners around any agent.

Methods¶

GuardrailMiddleware.get_metrics¶

get_metrics() -> dict[str, int]

GuardrailMiddleware.stream¶

stream(messages: list[Message], config: dict[str, Any] | None = None) -> AsyncIterator[AgentEvent | GuardrailEvent]

GuardrailMiddleware.warmup¶

warmup() -> None

Preload every pre/post scanner's model off the request path.

Best-effort — see latent.guardrails.warmup. Call once at boot so the first user request does not pay guardrail model cold-start. No-op when the middleware is disabled, since its scanners never run.