railtracks.guardrails

11class Guard(BaseModel):
12    """
13    Configuration for guardrails: input/output (and future tool) rails plus behavior flags.
14
15    ``input`` and ``output`` are lists of LLM guardrails (see :class:`BaseLLMGuardrail`).
16    The runner expects each rail to be callable with :class:`LLMGuardrailEvent` and
17    return a :class:`GuardrailDecision`. For output rails, the guarded assistant
18    message is ``event.output_message`` (see :class:`~railtracks.guardrails.core.event.LLMGuardrailEvent`).
19    """
20
21    model_config = ConfigDict(arbitrary_types_allowed=True)
22
23    input: list[InputGuard] = Field(
24        default_factory=list,
25        description="Guardrails run on LLM input (prompt / message history).",
26    )
27    output: list[OutputGuard] = Field(
28        default_factory=list,
29        description="Guardrails run on LLM output (model response).",
30    )
31    tool_call: list[Any] = Field(default_factory=list)
32    tool_response: list[Any] = Field(default_factory=list)
33    fail_open: bool = False
34    trace: bool = True
35
36    @field_validator("input", "output", "tool_call", "tool_response")
37    @classmethod
38    def _validate_callable_rails(
39        cls, value: list[InputGuard | OutputGuard]
40    ) -> list[InputGuard | OutputGuard]:
41        for rail in value:
42            if not callable(rail):
43                raise TypeError(
44                    f"Every guardrail must be callable, got {type(rail).__name__}."
45                )
46        return value

11class Guardrail(Protocol):
12    """
13    Base protocol for all guardrails: callable with a name.
14
15    Concrete ABC hierarchies (e.g. :class:`BaseLLMGuardrail`) narrow the event
16    type and add domain-specific attributes like ``phase``.
17    """
18
19    name: str
20
21    def __call__(self, event: Any) -> GuardrailDecision: ...

24class BaseGuardrail(ABC):
25    """Abstract base class for all guardrails."""
26
27    name: str
28
29    def __init__(self, name: str | None = None):
30        self.name = name or self.__class__.__name__
31
32    @abstractmethod
33    def __call__(self, event: Any) -> GuardrailDecision:
34        pass

12class GuardrailAction(str, Enum):
13    ALLOW = "allow"
14    TRANSFORM = "transform"
15    BLOCK = "block"

12class GuardrailBlockedError(NodeInvocationError):
13    """
14    Raised when guardrails deterministically block an operation (e.g. LLM input).
15
16    This error is intended to remain distinguishable from `LLMError` so callers/tests can
17    assert guardrail rejection explicitly.
18    """
19
20    def __init__(
21        self,
22        *,
23        rail_name: str | None = None,
24        reason: str,
25        user_facing_message: str | None = None,
26        traces: list["GuardrailTrace"] | None = None,
27        meta: dict[str, Any] | None = None,
28        notes: list[str] | None = None,
29        fatal: bool = False,
30    ):
31        self.rail_name = rail_name
32        self.reason = reason
33        self.user_facing_message = user_facing_message
34        self.traces = traces
35        self.meta = meta
36
37        base_message = "Blocked by guardrails"
38        if rail_name:
39            base_message += f" ({rail_name})"
40        base_message += f": {reason}"
41
42        derived_notes: list[str] = []
43        if user_facing_message:
44            derived_notes.append(f"user_message={user_facing_message!r}")
45        if meta:
46            derived_notes.append("meta attached (see exception.meta)")
47
48        super().__init__(
49            message=base_message,
50            notes=[*(notes or []), *derived_notes],
51            fatal=fatal,
52        )

18class GuardrailDecision(BaseModel):
19    model_config = ConfigDict(arbitrary_types_allowed=True)
20
21    action: GuardrailAction
22    reason: str
23    messages: MessageHistory | None = None
24    output_message: Message | None = None
25    user_facing_message: str | None = None
26    meta: dict[str, Any] | None = None
27
28    @classmethod
29    def allow(
30        cls, reason: str = "Allowed by guardrail.", meta: dict[str, Any] | None = None
31    ) -> "GuardrailDecision":
32        return cls(action=GuardrailAction.ALLOW, reason=reason, meta=meta)
33
34    @classmethod
35    def block(
36        cls,
37        reason: str,
38        user_facing_message: str | None = None,
39        meta: dict[str, Any] | None = None,
40    ) -> "GuardrailDecision":
41        return cls(
42            action=GuardrailAction.BLOCK,
43            reason=reason,
44            user_facing_message=user_facing_message,
45            meta=meta,
46        )
47
48    @classmethod
49    def transform_messages(
50        cls,
51        messages: MessageHistory,
52        reason: str,
53        meta: dict[str, Any] | None = None,
54    ) -> "GuardrailDecision":
55        return cls(
56            action=GuardrailAction.TRANSFORM,
57            reason=reason,
58            messages=messages,
59            meta=meta,
60        )
61
62    @classmethod
63    def transform_output(
64        cls,
65        output_message: Message,
66        reason: str,
67        meta: dict[str, Any] | None = None,
68    ) -> "GuardrailDecision":
69        return cls(
70            action=GuardrailAction.TRANSFORM,
71            reason=reason,
72            output_message=output_message,
73            meta=meta,
74        )

 9class GuardrailTrace(BaseModel):
10    rail_name: str
11    phase: str
12    action: str
13    reason: str
14    meta: dict[str, Any] | None = None

37class BaseLLMGuardrail(BaseGuardrail):
38    """Abstract base class for guardrails that run on LLM input or output."""
39
40    phase: LLMGuardrailPhase
41
42    @abstractmethod
43    def __call__(self, event: LLMGuardrailEvent) -> GuardrailDecision:
44        pass

47class InputGuard(BaseLLMGuardrail):
48    """Base for guardrails that run on LLM input (e.g. prompt / message history)."""
49
50    phase = LLMGuardrailPhase.INPUT

53class OutputGuard(BaseLLMGuardrail):
54    """
55    Base for guardrails that run on LLM output (e.g. model response).
56
57    Inspect ``event.output_message`` for the assistant message produced this turn.
58    ``event.messages`` is conversation context and may not yet include that reply.
59    """
60
61    phase = LLMGuardrailPhase.OUTPUT

19class LLMGuardrailEvent(BaseModel):
20    """
21    Event passed to LLM guardrails.
22
23    - ``messages``: conversation context (usually the history *before* the current
24      assistant reply is appended to the node).
25    - ``output_message``: for ``phase == OUTPUT``, the assistant ``Message`` being
26      guarded this turn; ``None`` for INPUT phase or when not applicable.
27    """
28
29    model_config = ConfigDict(arbitrary_types_allowed=True)
30
31    phase: LLMGuardrailPhase
32    messages: MessageHistory
33    output_message: Message | None = None
34
35    node_name: str | None = None
36    node_uuid: str | None = None
37    run_id: str | None = None
38    model_name: str | None = None
39    model_provider: str | None = None
40    tags: dict[str, str] | None = None

14class LLMGuardrailPhase(str, Enum):
15    INPUT = "llm_input"
16    OUTPUT = "llm_output"