railtracks.guardrails

12class GuardrailAction(str, Enum):
13    """What the runner should do after a guardrail returns.
14
15    Members:
16        ALLOW: Keep the current input or output unchanged.
17        TRANSFORM: Replace input messages or the output message from the decision.
18        BLOCK: Stop and treat the interaction as blocked.
19    """
20
21    ALLOW = "allow"
22    TRANSFORM = "transform"
23    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        """Create a block error with optional trace and user-facing context.
32
33        Args:
34            rail_name: Name of the rail that blocked, if known.
35            reason: Machine-oriented explanation (also embedded in the base message).
36            user_facing_message: Optional short text for clients or UIs.
37            traces: Optional list of :class:`~railtracks.guardrails.core.trace.GuardrailTrace`
38                from the failed run.
39            meta: Optional structured details copied from the blocking decision.
40            notes: Extra debug lines forwarded to :class:`NodeInvocationError`.
41            fatal: Passed through to :class:`NodeInvocationError` (whether the run is
42                considered unrecoverable).
43        """
44        self.rail_name = rail_name
45        self.reason = reason
46        self.user_facing_message = user_facing_message
47        self.traces = traces
48        self.meta = meta
49
50        base_message = "Blocked by guardrails"
51        if rail_name:
52            base_message += f" ({rail_name})"
53        base_message += f": {reason}"
54
55        derived_notes: list[str] = []
56        if user_facing_message:
57            derived_notes.append(f"user_message={user_facing_message!r}")
58        if meta:
59            derived_notes.append("meta attached (see exception.meta)")
60
61        super().__init__(
62            message=base_message,
63            notes=[*(notes or []), *derived_notes],
64            fatal=fatal,
65        )

 26class GuardrailDecision(BaseModel):
 27    """Result of one guardrail invocation.
 28
 29    Which fields are set depends on :attr:`action`:
 30
 31    * ``ALLOW``: only :attr:`reason` and optional :attr:`meta` are typically used.
 32    * ``TRANSFORM``: for input phase, :attr:`messages` holds the new history; for
 33      output phase, :attr:`output_message` holds the new assistant message.
 34    * ``BLOCK``: :attr:`user_facing_message` and :attr:`meta` may carry details for
 35      callers or UIs.
 36    """
 37
 38    model_config = ConfigDict(arbitrary_types_allowed=True)
 39
 40    action: GuardrailAction
 41    reason: str
 42    messages: MessageHistory | None = None
 43    output_message: Message | None = None
 44    user_facing_message: str | None = None
 45    meta: dict[str, Any] | None = None
 46
 47    @classmethod
 48    def allow(
 49        cls, reason: str = "Allowed by guardrail.", meta: dict[str, Any] | None = None
 50    ) -> "GuardrailDecision":
 51        """Build an ``ALLOW`` decision with no content changes.
 52
 53        Args:
 54            reason: Explanation for traces and debugging.
 55            meta: Optional extra fields for observability.
 56
 57        Returns:
 58            A decision with :attr:`action` ``ALLOW``.
 59        """
 60        return cls(action=GuardrailAction.ALLOW, reason=reason, meta=meta)
 61
 62    @classmethod
 63    def block(
 64        cls,
 65        reason: str,
 66        user_facing_message: str | None = None,
 67        meta: dict[str, Any] | None = None,
 68    ) -> "GuardrailDecision":
 69        """Build a ``BLOCK`` decision.
 70
 71        Args:
 72            reason: Explanation for logs, traces, and raised errors.
 73            user_facing_message: Optional message safe to show to end users.
 74            meta: Optional extra fields (e.g. exception details).
 75
 76        Returns:
 77            A decision with :attr:`action` ``BLOCK``.
 78        """
 79        return cls(
 80            action=GuardrailAction.BLOCK,
 81            reason=reason,
 82            user_facing_message=user_facing_message,
 83            meta=meta,
 84        )
 85
 86    @classmethod
 87    def transform_messages(
 88        cls,
 89        messages: MessageHistory,
 90        reason: str,
 91        meta: dict[str, Any] | None = None,
 92    ) -> "GuardrailDecision":
 93        """Build a ``TRANSFORM`` decision for LLM input (message history).
 94
 95        Args:
 96            messages: Replacement conversation history for the model call.
 97            reason: Explanation for traces and debugging.
 98            meta: Optional extra fields (e.g. redaction counts).
 99
100        Returns:
101            A decision with :attr:`action` ``TRANSFORM`` and :attr:`messages` set.
102        """
103        return cls(
104            action=GuardrailAction.TRANSFORM,
105            reason=reason,
106            messages=messages,
107            meta=meta,
108        )
109
110    @classmethod
111    def transform_output(
112        cls,
113        output_message: Message,
114        reason: str,
115        meta: dict[str, Any] | None = None,
116    ) -> "GuardrailDecision":
117        """Build a ``TRANSFORM`` decision for LLM output (assistant message).
118
119        Args:
120            output_message: Replacement assistant message to return.
121            reason: Explanation for traces and debugging.
122            meta: Optional extra fields (e.g. redaction counts).
123
124        Returns:
125            A decision with :attr:`action` ``TRANSFORM`` and :attr:`output_message`
126            set.
127        """
128        return cls(
129            action=GuardrailAction.TRANSFORM,
130            reason=reason,
131            output_message=output_message,
132            meta=meta,
133        )

 9class GuardrailTrace(BaseModel):
10    """One guardrail step recorded during a run (for logging or debugging).
11
12    Attributes:
13        rail_name: The guard's :attr:`~railtracks.guardrails.core.interfaces.BaseGuardrail.name`
14            or class name if unset.
15        phase: ``LLMGuardrailPhase`` value string (e.g. ``llm_input``).
16        action: ``allow``, ``transform``, ``block``, or ``error`` when the runner
17            caught an exception, invalid return type, or unknown action.
18        reason: Short explanation, or a fixed message for error traces.
19        meta: Optional details (e.g. :attr:`GuardrailDecision.meta` or exception info).
20    """
21
22    rail_name: str
23    phase: str
24    action: str
25    reason: str
26    meta: dict[str, Any] | None = None

 76class InputGuard(BaseLLMGuardrail):
 77    """Base for guardrails that run on LLM input (e.g. prompt / message history).
 78
 79    Attributes:
 80        phase: Always :attr:`LLMGuardrailPhase.INPUT`.
 81    """
 82
 83    phase = LLMGuardrailPhase.INPUT
 84
 85    def decide(
 86        self,
 87        input: str | Any | MessageHistory | LLMGuardrailEvent,
 88    ) -> GuardrailDecision:
 89        """Run this guard without building an :class:`LLMGuardrailEvent` by hand.
 90
 91        Args:
 92            input: A :class:`LLMGuardrailEvent` (passed through), a ``str`` (treated
 93                as a single user message), a :class:`~railtracks.llm.message.Message`,
 94                or a :class:`~railtracks.llm.history.MessageHistory`.
 95
 96        Returns:
 97            The :class:`GuardrailDecision` from :meth:`__call__`.
 98
 99        Raises:
100            TypeError: If ``input`` is not a ``str``, ``Message``, ``MessageHistory``,
101                or :class:`LLMGuardrailEvent`.
102        """
103        if isinstance(input, LLMGuardrailEvent):
104            return self(input)
105
106        messages = _coerce_to_message_history(input)
107        event = LLMGuardrailEvent(
108            phase=LLMGuardrailPhase.INPUT,
109            messages=messages,
110        )
111        return self(event)

114class OutputGuard(BaseLLMGuardrail):
115    """Base for guardrails that run on LLM output (e.g. model response).
116
117    Inspect ``event.output_message`` for the assistant message produced this turn.
118    ``event.messages`` is conversation context and may not yet include that reply.
119
120    Attributes:
121        phase: Always :attr:`LLMGuardrailPhase.OUTPUT`.
122    """
123
124    phase = LLMGuardrailPhase.OUTPUT
125
126    def decide(
127        self,
128        output: str | Any | MessageHistory | LLMGuardrailEvent,
129    ) -> GuardrailDecision:
130        """Run this guard without building an :class:`LLMGuardrailEvent` by hand.
131
132        Args:
133            output: A :class:`LLMGuardrailEvent` (passed through), a ``str`` (becomes
134                the assistant message with empty prior history), a
135                :class:`~railtracks.llm.message.Message`, or a non-empty
136                :class:`~railtracks.llm.history.MessageHistory` (last message is the
137                output under test; earlier entries become ``event.messages``).
138
139        Returns:
140            The :class:`GuardrailDecision` from :meth:`__call__`.
141
142        Raises:
143            ValueError: If ``output`` is an empty :class:`~railtracks.llm.history.MessageHistory`.
144            TypeError: If ``output`` is not a ``str``, ``Message``, ``MessageHistory``,
145                or :class:`LLMGuardrailEvent`.
146        """
147        if isinstance(output, LLMGuardrailEvent):
148            return self(output)
149
150        if isinstance(output, str):
151            output_message = AssistantMessage(output)
152            messages = MessageHistory()
153        elif isinstance(output, Message):
154            output_message = output
155            messages = MessageHistory()
156        elif isinstance(output, MessageHistory):
157            if not output:
158                raise ValueError("Cannot decide with an empty MessageHistory.")
159            output_message = output[-1]
160            messages = MessageHistory(output[:-1])
161        else:
162            raise TypeError(
163                f"Expected str, Message, MessageHistory, or LLMGuardrailEvent, "
164                f"got {type(output).__name__}"
165            )
166
167        event = LLMGuardrailEvent(
168            phase=LLMGuardrailPhase.OUTPUT,
169            messages=messages,
170            output_message=output_message,
171        )
172        return self(event)

26class LLMGuardrailEvent(BaseModel):
27    """Payload for LLM input and output guardrails.
28
29    Attributes:
30        phase: Whether this is an input-phase or output-phase check.
31        messages: Conversation context, usually the history before the current
32            assistant reply is appended by the node.
33        output_message: For ``OUTPUT`` phase, the assistant :class:`~railtracks.llm.message.Message`
34            under inspection; ``None`` for input phase or when not applicable.
35        node_name: Optional node label for observability.
36        node_uuid: Optional stable node id for observability.
37        run_id: Optional run correlation id.
38        model_name: Optional resolved model name for observability.
39        model_provider: Optional provider string for observability.
40        tags: Optional key/value metadata (e.g. ``agent_kind`` from the mixin).
41    """
42
43    model_config = ConfigDict(arbitrary_types_allowed=True)
44
45    phase: LLMGuardrailPhase
46    messages: MessageHistory
47    output_message: Message | None = None
48
49    node_name: str | None = None
50    node_uuid: str | None = None
51    run_id: str | None = None
52    model_name: str | None = None
53    model_provider: str | None = None
54    tags: dict[str, str] | None = None

14class LLMGuardrailPhase(str, Enum):
15    """Which side of the LLM call a guardrail observes.
16
17    Members:
18        INPUT: Before the model (prompt / history), value ``llm_input``.
19        OUTPUT: After the model (assistant message), value ``llm_output``.
20    """
21
22    INPUT = "llm_input"
23    OUTPUT = "llm_output"