railtracks.guardrails.llm.llm_guard

View Source

  1import inspect
  2from abc import abstractmethod
  3from typing import Any, Awaitable, Generic, TypeVar, cast
  4
  5from pydantic import BaseModel
  6from typing_extensions import Literal
  7
  8from railtracks.context.central import get_parent_id, get_run_id, is_context_present
  9from railtracks.guardrails.core.decision import GuardrailAction, GuardrailDecision
 10from railtracks.guardrails.core.errors import GuardrailBlockedError
 11from railtracks.guardrails.core.event import LLMGuardrailEvent, LLMGuardrailPhase
 12from railtracks.guardrails.core.interfaces import BaseGuardrail
 13from railtracks.guardrails.core.trace import GuardrailTrace
 14from railtracks.llm.history import MessageHistory
 15from railtracks.llm.message import Message, UserMessage
 16from railtracks.llm.response import Response
 17from railtracks.llm.tools.tool import Tool
 18from railtracks.utils.logging.create import get_rt_logger
 19from railtracks.utils.unpack import unpack_async_sync
 20
 21logger = get_rt_logger("guardrails")
 22
 23_TValue = TypeVar("_TValue", bound=MessageHistory | Message)
 24
 25
 26class BaseLLMGuardrail(
 27    BaseGuardrail[
 28        [MessageHistory, type[BaseModel] | None, list[Tool] | None], Response
 29    ],
 30    Generic[_TValue],
 31):
 32    """Abstract base class for guardrails that run on LLM input or output.
 33
 34    Attributes:
 35        phase: Whether this rail expects :class:`LLMGuardrailPhase` ``INPUT`` or
 36            ``OUTPUT`` events.
 37    """
 38
 39    phase: LLMGuardrailPhase
 40
 41    def __init__(self, name: str | None = None, fail_open: bool = False):
 42        """Initialize the guardrail.
 43
 44        Args:
 45            name: Rail name for traces and debugging; defaults to the class name.
 46            fail_open: Whether to allow the request to continue when this guard raises an unexpected exception.
 47        """
 48        super().__init__(name=name)
 49        self.fail_open = fail_open
 50
 51    @abstractmethod
 52    def __call__(
 53        self, event: LLMGuardrailEvent
 54    ) -> GuardrailDecision | Awaitable[GuardrailDecision]:
 55        """Evaluate the event and return a decision. Implemented by each concrete guard.
 56
 57        May be sync or async. An ``async def`` implementation is awaited before its
 58        decision is dispatched, so a guard is free to ``await rt.call(...)`` and
 59        delegate the judgement to another agent.
 60        """
 61        pass
 62
 63    @abstractmethod
 64    def convert(
 65        self, value: str | Message | MessageHistory | LLMGuardrailEvent, /
 66    ) -> LLMGuardrailEvent:
 67        """Build an event from a raw value. Implemented per phase by InputGuard and OutputGuard."""
 68        pass
 69
 70    def decide(
 71        self, value: str | Message | MessageHistory | LLMGuardrailEvent, /
 72    ) -> GuardrailDecision:
 73        """Convert value to an event and run this guard on it directly, outside a chain.
 74
 75        Synchronous guards only. An ``async def`` guard would return an un-awaited
 76        coroutine here, so use :meth:`adecide` for those.
 77        """
 78        if self._is_async():
 79            raise TypeError(
 80                f"Guardrail {self._rail_name()} is async; "
 81                f"use `await {type(self).__name__}.adecide(...)` instead of `.decide(...)`."
 82            )
 83        converted_event = self.convert(value)
 84
 85        return cast(GuardrailDecision, self(converted_event))
 86
 87    async def adecide(
 88        self, value: str | Message | MessageHistory | LLMGuardrailEvent, /
 89    ) -> GuardrailDecision:
 90        """Async counterpart to :meth:`decide`; accepts sync and async guards alike."""
 91        converted_event = self.convert(value)
 92
 93        return await unpack_async_sync(self(converted_event))
 94
 95    def _is_async(self) -> bool:
 96        """Whether this guard's ``__call__`` is a coroutine function."""
 97        return inspect.iscoroutinefunction(type(self).__call__)
 98
 99    @staticmethod
100    def _coerce_to_message_history(
101        input: str | Any | MessageHistory,
102    ) -> MessageHistory:
103        """Convert str, Message, or MessageHistory to a MessageHistory."""
104        if isinstance(input, MessageHistory):
105            return input
106        if isinstance(input, str):
107            return MessageHistory([UserMessage(input)])
108        if isinstance(input, Message):
109            return MessageHistory([input])
110        raise TypeError(
111            f"Expected str, Message, or MessageHistory got {type(input).__name__}"
112        )
113
114    @staticmethod
115    def _node_metadata() -> tuple[str | None, str | None]:
116        """``(node_uuid, run_id)`` for event observability.
117
118        Returns ``(None, None)`` when no run context is active — the gate logic never
119        depends on this metadata, so it must not raise just to populate it.
120        """
121        if not is_context_present():
122            return None, None
123        return get_parent_id(), get_run_id()
124
125    @staticmethod
126    def _raise_if_blocked(
127        decision: GuardrailDecision | None, traces: list[GuardrailTrace]
128    ) -> None:
129        """Raise :class:`GuardrailBlockedError` when a rail returned ``BLOCK``."""
130        if decision is not None and decision.action == GuardrailAction.BLOCK:
131            rail_name = traces[-1].rail_name if traces else None
132            raise GuardrailBlockedError(
133                rail_name=rail_name,
134                reason=decision.reason,
135                user_facing_message=decision.user_facing_message,
136                traces=traces,
137                meta=decision.meta,
138            )
139
140    def _handle_rail_exception(
141        self,
142        *,
143        exc: Exception,
144        traces: list[GuardrailTrace],
145        value: _TValue,
146        reason_prefix: str,
147    ) -> (
148        tuple[Literal["continue"], _TValue]
149        | tuple[Literal["stop"], _TValue, GuardrailDecision]
150    ):
151        """Record exc as a trace and return a stop outcome with a blocking decision."""
152        traces.append(self._trace_for_exception(exc=exc))
153
154        if self.fail_open:
155            return ("continue", value)
156
157        block = GuardrailDecision.block(
158            reason=f"{reason_prefix}: {self._rail_name()}",
159            user_facing_message="Request blocked by guardrails.",
160            meta={
161                "exception_type": exc.__class__.__name__,
162                "exception_message": str(exc),
163            },
164        )
165        return ("stop", value, block)
166
167    async def run(
168        self,
169        *,
170        event: LLMGuardrailEvent,
171        value: _TValue,
172    ) -> tuple[_TValue, list[GuardrailTrace], GuardrailDecision]:
173        """Run this guard once on event and value.
174
175        Returns the resulting value, the traces recorded, and the decision that
176        applies. A BLOCK (or a failed TRANSFORM) surfaces its own decision as-is.
177        A successful TRANSFORM also surfaces its own decision (its `reason`/`meta`
178        describe what changed, e.g. a PII redaction count -- callers like
179        `MiddlewareGuardInputResponseEvent` depend on that). A plain ALLOW, or an
180        exception swallowed via `fail_open=True`, has nothing rail-specific worth
181        keeping, so a generic `GuardrailDecision.allow()` is returned instead.
182        """
183        traces: list[GuardrailTrace] = []
184
185        step = await self._eval_one_rail(
186            event=event,
187            value=value,
188            traces=traces,
189        )
190
191        if step[0] == "stop":
192            return step[1], traces, step[2]
193
194        _, value, event, decision = step
195
196        return (
197            value,
198            traces,
199            decision if decision is not None else GuardrailDecision.allow(),
200        )
201
202    async def _eval_one_rail(
203        self,
204        event: LLMGuardrailEvent,
205        value: _TValue,
206        traces: list[GuardrailTrace],
207    ) -> (
208        tuple[Literal["continue"], _TValue, LLMGuardrailEvent, GuardrailDecision | None]
209        | tuple[Literal["stop"], _TValue, GuardrailDecision]
210    ):
211        """Call this guard, await it if async, validate the decision, and dispatch it."""
212        try:
213            decision = await unpack_async_sync(self(event))
214            if not isinstance(decision, GuardrailDecision):
215                raise TypeError(
216                    f"Guardrail {self._rail_name()} returned {type(decision).__name__}, expected GuardrailDecision."
217                )
218        except Exception as e:
219            outcome = self._handle_rail_exception(
220                exc=e,
221                traces=traces,
222                value=value,
223                reason_prefix="Guardrail raised exception",
224            )
225            if outcome[0] == "continue":
226                return ("continue", value, event, None)
227            return ("stop", outcome[1], outcome[2])
228
229        traces.append(self._trace_from_decision(decision=decision))
230
231        if decision.action == GuardrailAction.ALLOW:
232            return ("continue", value, event, None)
233
234        return self._dispatch_non_allow_decision(
235            event=event,
236            value=value,
237            decision=decision,
238            traces=traces,
239        )
240
241    def _dispatch_non_allow_decision(
242        self,
243        *,
244        event: LLMGuardrailEvent,
245        value: _TValue,
246        decision: GuardrailDecision,
247        traces: list[GuardrailTrace],
248    ) -> (
249        tuple[Literal["continue"], _TValue, LLMGuardrailEvent, GuardrailDecision | None]
250        | tuple[Literal["stop"], _TValue, GuardrailDecision]
251    ):
252        """Apply a TRANSFORM, BLOCK, or unknown-action decision returned by this guard."""
253        if decision.action == GuardrailAction.TRANSFORM:
254            try:
255                value = self._extract_transform_value(decision)
256                event = self._sync_event_after_transform(event, value)
257            except Exception as e:
258                outcome = self._handle_rail_exception(
259                    exc=e,
260                    traces=traces,
261                    value=value,
262                    reason_prefix="Guardrail transform failed",
263                )
264                if outcome[0] == "continue":
265                    return ("continue", value, event, None)
266                return ("stop", outcome[1], outcome[2])
267            # Success: unlike a plain ALLOW, the caller needs this decision's own
268            # reason/meta (e.g. a PII redaction count) -- preserve it as-is.
269            return ("continue", value, event, decision)
270
271        if decision.action == GuardrailAction.BLOCK:
272            return ("stop", value, decision)
273
274        traces.append(
275            GuardrailTrace(
276                rail_name=self._rail_name(),
277                phase=self.phase.value,
278                action="error",
279                reason="Unknown guardrail action",
280                meta={"action": str(decision.action)},
281            )
282        )
283
284        if self.fail_open:
285            return ("continue", value, event, None)
286
287        block = GuardrailDecision.block(
288            reason=f"Unknown guardrail action from {self._rail_name()}",
289            user_facing_message="Request blocked by guardrails.",
290        )
291        return ("stop", value, block)
292
293    @abstractmethod
294    def _sync_event_after_transform(
295        self,
296        event: LLMGuardrailEvent,
297        value: _TValue,
298    ) -> LLMGuardrailEvent:
299        """Return a copy of event updated with the transformed value. Implemented per phase."""
300        pass
301
302    @abstractmethod
303    def _extract_transform_value(self, decision: GuardrailDecision) -> _TValue:
304        """Extract the replacement value from a TRANSFORM decision. Implemented per phase."""
305        pass
306
307    def _rail_name(self) -> str:
308        """Return this guard's name, falling back to its class name if unset."""
309        name = self.name
310        if isinstance(name, str) and name.strip():
311            return name
312
313        return self.__class__.__name__
314
315    def _trace_from_decision(
316        self,
317        decision: GuardrailDecision,
318    ) -> GuardrailTrace:
319        """Build a trace recording this guard's decision."""
320        return GuardrailTrace(
321            rail_name=self._rail_name(),
322            phase=self.phase.value,
323            action=decision.action.value,
324            reason=decision.reason,
325            meta=decision.meta,
326        )
327
328    def _trace_for_exception(
329        self,
330        *,
331        exc: Exception,
332    ) -> GuardrailTrace:
333        """Build a trace recording an exception raised by this guard."""
334        return GuardrailTrace(
335            rail_name=self._rail_name(),
336            phase=self.phase.value,
337            action="error",
338            reason="Guardrail raised exception",
339            meta={
340                "exception_type": exc.__class__.__name__,
341                "exception_message": str(exc),
342            },
343        )

logger = <RTContextLoggingAdapter RT.guardrails (WARNING)>

class BaseLLMGuardrail(railtracks.guardrails.core.interfaces.BaseGuardrail[(<class 'railtracks.llm.history.MessageHistory'>, type[pydantic.main.BaseModel] | None, list[railtracks.llm.tools.tool.Tool] | None), railtracks.llm.response.Response], typing.Generic[~_TValue]): View Source

 27class BaseLLMGuardrail(
 28    BaseGuardrail[
 29        [MessageHistory, type[BaseModel] | None, list[Tool] | None], Response
 30    ],
 31    Generic[_TValue],
 32):
 33    """Abstract base class for guardrails that run on LLM input or output.
 34
 35    Attributes:
 36        phase: Whether this rail expects :class:`LLMGuardrailPhase` ``INPUT`` or
 37            ``OUTPUT`` events.
 38    """
 39
 40    phase: LLMGuardrailPhase
 41
 42    def __init__(self, name: str | None = None, fail_open: bool = False):
 43        """Initialize the guardrail.
 44
 45        Args:
 46            name: Rail name for traces and debugging; defaults to the class name.
 47            fail_open: Whether to allow the request to continue when this guard raises an unexpected exception.
 48        """
 49        super().__init__(name=name)
 50        self.fail_open = fail_open
 51
 52    @abstractmethod
 53    def __call__(
 54        self, event: LLMGuardrailEvent
 55    ) -> GuardrailDecision | Awaitable[GuardrailDecision]:
 56        """Evaluate the event and return a decision. Implemented by each concrete guard.
 57
 58        May be sync or async. An ``async def`` implementation is awaited before its
 59        decision is dispatched, so a guard is free to ``await rt.call(...)`` and
 60        delegate the judgement to another agent.
 61        """
 62        pass
 63
 64    @abstractmethod
 65    def convert(
 66        self, value: str | Message | MessageHistory | LLMGuardrailEvent, /
 67    ) -> LLMGuardrailEvent:
 68        """Build an event from a raw value. Implemented per phase by InputGuard and OutputGuard."""
 69        pass
 70
 71    def decide(
 72        self, value: str | Message | MessageHistory | LLMGuardrailEvent, /
 73    ) -> GuardrailDecision:
 74        """Convert value to an event and run this guard on it directly, outside a chain.
 75
 76        Synchronous guards only. An ``async def`` guard would return an un-awaited
 77        coroutine here, so use :meth:`adecide` for those.
 78        """
 79        if self._is_async():
 80            raise TypeError(
 81                f"Guardrail {self._rail_name()} is async; "
 82                f"use `await {type(self).__name__}.adecide(...)` instead of `.decide(...)`."
 83            )
 84        converted_event = self.convert(value)
 85
 86        return cast(GuardrailDecision, self(converted_event))
 87
 88    async def adecide(
 89        self, value: str | Message | MessageHistory | LLMGuardrailEvent, /
 90    ) -> GuardrailDecision:
 91        """Async counterpart to :meth:`decide`; accepts sync and async guards alike."""
 92        converted_event = self.convert(value)
 93
 94        return await unpack_async_sync(self(converted_event))
 95
 96    def _is_async(self) -> bool:
 97        """Whether this guard's ``__call__`` is a coroutine function."""
 98        return inspect.iscoroutinefunction(type(self).__call__)
 99
100    @staticmethod
101    def _coerce_to_message_history(
102        input: str | Any | MessageHistory,
103    ) -> MessageHistory:
104        """Convert str, Message, or MessageHistory to a MessageHistory."""
105        if isinstance(input, MessageHistory):
106            return input
107        if isinstance(input, str):
108            return MessageHistory([UserMessage(input)])
109        if isinstance(input, Message):
110            return MessageHistory([input])
111        raise TypeError(
112            f"Expected str, Message, or MessageHistory got {type(input).__name__}"
113        )
114
115    @staticmethod
116    def _node_metadata() -> tuple[str | None, str | None]:
117        """``(node_uuid, run_id)`` for event observability.
118
119        Returns ``(None, None)`` when no run context is active — the gate logic never
120        depends on this metadata, so it must not raise just to populate it.
121        """
122        if not is_context_present():
123            return None, None
124        return get_parent_id(), get_run_id()
125
126    @staticmethod
127    def _raise_if_blocked(
128        decision: GuardrailDecision | None, traces: list[GuardrailTrace]
129    ) -> None:
130        """Raise :class:`GuardrailBlockedError` when a rail returned ``BLOCK``."""
131        if decision is not None and decision.action == GuardrailAction.BLOCK:
132            rail_name = traces[-1].rail_name if traces else None
133            raise GuardrailBlockedError(
134                rail_name=rail_name,
135                reason=decision.reason,
136                user_facing_message=decision.user_facing_message,
137                traces=traces,
138                meta=decision.meta,
139            )
140
141    def _handle_rail_exception(
142        self,
143        *,
144        exc: Exception,
145        traces: list[GuardrailTrace],
146        value: _TValue,
147        reason_prefix: str,
148    ) -> (
149        tuple[Literal["continue"], _TValue]
150        | tuple[Literal["stop"], _TValue, GuardrailDecision]
151    ):
152        """Record exc as a trace and return a stop outcome with a blocking decision."""
153        traces.append(self._trace_for_exception(exc=exc))
154
155        if self.fail_open:
156            return ("continue", value)
157
158        block = GuardrailDecision.block(
159            reason=f"{reason_prefix}: {self._rail_name()}",
160            user_facing_message="Request blocked by guardrails.",
161            meta={
162                "exception_type": exc.__class__.__name__,
163                "exception_message": str(exc),
164            },
165        )
166        return ("stop", value, block)
167
168    async def run(
169        self,
170        *,
171        event: LLMGuardrailEvent,
172        value: _TValue,
173    ) -> tuple[_TValue, list[GuardrailTrace], GuardrailDecision]:
174        """Run this guard once on event and value.
175
176        Returns the resulting value, the traces recorded, and the decision that
177        applies. A BLOCK (or a failed TRANSFORM) surfaces its own decision as-is.
178        A successful TRANSFORM also surfaces its own decision (its `reason`/`meta`
179        describe what changed, e.g. a PII redaction count -- callers like
180        `MiddlewareGuardInputResponseEvent` depend on that). A plain ALLOW, or an
181        exception swallowed via `fail_open=True`, has nothing rail-specific worth
182        keeping, so a generic `GuardrailDecision.allow()` is returned instead.
183        """
184        traces: list[GuardrailTrace] = []
185
186        step = await self._eval_one_rail(
187            event=event,
188            value=value,
189            traces=traces,
190        )
191
192        if step[0] == "stop":
193            return step[1], traces, step[2]
194
195        _, value, event, decision = step
196
197        return (
198            value,
199            traces,
200            decision if decision is not None else GuardrailDecision.allow(),
201        )
202
203    async def _eval_one_rail(
204        self,
205        event: LLMGuardrailEvent,
206        value: _TValue,
207        traces: list[GuardrailTrace],
208    ) -> (
209        tuple[Literal["continue"], _TValue, LLMGuardrailEvent, GuardrailDecision | None]
210        | tuple[Literal["stop"], _TValue, GuardrailDecision]
211    ):
212        """Call this guard, await it if async, validate the decision, and dispatch it."""
213        try:
214            decision = await unpack_async_sync(self(event))
215            if not isinstance(decision, GuardrailDecision):
216                raise TypeError(
217                    f"Guardrail {self._rail_name()} returned {type(decision).__name__}, expected GuardrailDecision."
218                )
219        except Exception as e:
220            outcome = self._handle_rail_exception(
221                exc=e,
222                traces=traces,
223                value=value,
224                reason_prefix="Guardrail raised exception",
225            )
226            if outcome[0] == "continue":
227                return ("continue", value, event, None)
228            return ("stop", outcome[1], outcome[2])
229
230        traces.append(self._trace_from_decision(decision=decision))
231
232        if decision.action == GuardrailAction.ALLOW:
233            return ("continue", value, event, None)
234
235        return self._dispatch_non_allow_decision(
236            event=event,
237            value=value,
238            decision=decision,
239            traces=traces,
240        )
241
242    def _dispatch_non_allow_decision(
243        self,
244        *,
245        event: LLMGuardrailEvent,
246        value: _TValue,
247        decision: GuardrailDecision,
248        traces: list[GuardrailTrace],
249    ) -> (
250        tuple[Literal["continue"], _TValue, LLMGuardrailEvent, GuardrailDecision | None]
251        | tuple[Literal["stop"], _TValue, GuardrailDecision]
252    ):
253        """Apply a TRANSFORM, BLOCK, or unknown-action decision returned by this guard."""
254        if decision.action == GuardrailAction.TRANSFORM:
255            try:
256                value = self._extract_transform_value(decision)
257                event = self._sync_event_after_transform(event, value)
258            except Exception as e:
259                outcome = self._handle_rail_exception(
260                    exc=e,
261                    traces=traces,
262                    value=value,
263                    reason_prefix="Guardrail transform failed",
264                )
265                if outcome[0] == "continue":
266                    return ("continue", value, event, None)
267                return ("stop", outcome[1], outcome[2])
268            # Success: unlike a plain ALLOW, the caller needs this decision's own
269            # reason/meta (e.g. a PII redaction count) -- preserve it as-is.
270            return ("continue", value, event, decision)
271
272        if decision.action == GuardrailAction.BLOCK:
273            return ("stop", value, decision)
274
275        traces.append(
276            GuardrailTrace(
277                rail_name=self._rail_name(),
278                phase=self.phase.value,
279                action="error",
280                reason="Unknown guardrail action",
281                meta={"action": str(decision.action)},
282            )
283        )
284
285        if self.fail_open:
286            return ("continue", value, event, None)
287
288        block = GuardrailDecision.block(
289            reason=f"Unknown guardrail action from {self._rail_name()}",
290            user_facing_message="Request blocked by guardrails.",
291        )
292        return ("stop", value, block)
293
294    @abstractmethod
295    def _sync_event_after_transform(
296        self,
297        event: LLMGuardrailEvent,
298        value: _TValue,
299    ) -> LLMGuardrailEvent:
300        """Return a copy of event updated with the transformed value. Implemented per phase."""
301        pass
302
303    @abstractmethod
304    def _extract_transform_value(self, decision: GuardrailDecision) -> _TValue:
305        """Extract the replacement value from a TRANSFORM decision. Implemented per phase."""
306        pass
307
308    def _rail_name(self) -> str:
309        """Return this guard's name, falling back to its class name if unset."""
310        name = self.name
311        if isinstance(name, str) and name.strip():
312            return name
313
314        return self.__class__.__name__
315
316    def _trace_from_decision(
317        self,
318        decision: GuardrailDecision,
319    ) -> GuardrailTrace:
320        """Build a trace recording this guard's decision."""
321        return GuardrailTrace(
322            rail_name=self._rail_name(),
323            phase=self.phase.value,
324            action=decision.action.value,
325            reason=decision.reason,
326            meta=decision.meta,
327        )
328
329    def _trace_for_exception(
330        self,
331        *,
332        exc: Exception,
333    ) -> GuardrailTrace:
334        """Build a trace recording an exception raised by this guard."""
335        return GuardrailTrace(
336            rail_name=self._rail_name(),
337            phase=self.phase.value,
338            action="error",
339            reason="Guardrail raised exception",
340            meta={
341                "exception_type": exc.__class__.__name__,
342                "exception_message": str(exc),
343            },
344        )

Abstract base class for guardrails that run on LLM input or output.

Attributes:

phase: Whether this rail expects LLMGuardrailPhase INPUT or OUTPUT events.

BaseLLMGuardrail(name: str | None = None, fail_open: bool = False) View Source

42    def __init__(self, name: str | None = None, fail_open: bool = False):
43        """Initialize the guardrail.
44
45        Args:
46            name: Rail name for traces and debugging; defaults to the class name.
47            fail_open: Whether to allow the request to continue when this guard raises an unexpected exception.
48        """
49        super().__init__(name=name)
50        self.fail_open = fail_open

Initialize the guardrail.

Arguments:

name: Rail name for traces and debugging; defaults to the class name.
fail_open: Whether to allow the request to continue when this guard raises an unexpected exception.

phase: railtracks.guardrails.LLMGuardrailPhase

fail_open

@abstractmethod

def convert( self, value: str | railtracks.llm.Message | railtracks.llm.MessageHistory | railtracks.guardrails.LLMGuardrailEvent, /) -> railtracks.guardrails.LLMGuardrailEvent: View Source

64    @abstractmethod
65    def convert(
66        self, value: str | Message | MessageHistory | LLMGuardrailEvent, /
67    ) -> LLMGuardrailEvent:
68        """Build an event from a raw value. Implemented per phase by InputGuard and OutputGuard."""
69        pass

Build an event from a raw value. Implemented per phase by InputGuard and OutputGuard.

def decide( self, value: str | railtracks.llm.Message | railtracks.llm.MessageHistory | railtracks.guardrails.LLMGuardrailEvent, /) -> railtracks.guardrails.GuardrailDecision: View Source

71    def decide(
72        self, value: str | Message | MessageHistory | LLMGuardrailEvent, /
73    ) -> GuardrailDecision:
74        """Convert value to an event and run this guard on it directly, outside a chain.
75
76        Synchronous guards only. An ``async def`` guard would return an un-awaited
77        coroutine here, so use :meth:`adecide` for those.
78        """
79        if self._is_async():
80            raise TypeError(
81                f"Guardrail {self._rail_name()} is async; "
82                f"use `await {type(self).__name__}.adecide(...)` instead of `.decide(...)`."
83            )
84        converted_event = self.convert(value)
85
86        return cast(GuardrailDecision, self(converted_event))

Convert value to an event and run this guard on it directly, outside a chain.

Synchronous guards only. An async def guard would return an un-awaited coroutine here, so use adecide() for those.

async def adecide( self, value: str | railtracks.llm.Message | railtracks.llm.MessageHistory | railtracks.guardrails.LLMGuardrailEvent, /) -> railtracks.guardrails.GuardrailDecision: View Source

88    async def adecide(
89        self, value: str | Message | MessageHistory | LLMGuardrailEvent, /
90    ) -> GuardrailDecision:
91        """Async counterpart to :meth:`decide`; accepts sync and async guards alike."""
92        converted_event = self.convert(value)
93
94        return await unpack_async_sync(self(converted_event))

Async counterpart to decide(); accepts sync and async guards alike.

async def run( self, *, event: railtracks.guardrails.LLMGuardrailEvent, value: ~_TValue) -> tuple[~_TValue, list[railtracks.guardrails.GuardrailTrace], railtracks.guardrails.GuardrailDecision]: View Source

168    async def run(
169        self,
170        *,
171        event: LLMGuardrailEvent,
172        value: _TValue,
173    ) -> tuple[_TValue, list[GuardrailTrace], GuardrailDecision]:
174        """Run this guard once on event and value.
175
176        Returns the resulting value, the traces recorded, and the decision that
177        applies. A BLOCK (or a failed TRANSFORM) surfaces its own decision as-is.
178        A successful TRANSFORM also surfaces its own decision (its `reason`/`meta`
179        describe what changed, e.g. a PII redaction count -- callers like
180        `MiddlewareGuardInputResponseEvent` depend on that). A plain ALLOW, or an
181        exception swallowed via `fail_open=True`, has nothing rail-specific worth
182        keeping, so a generic `GuardrailDecision.allow()` is returned instead.
183        """
184        traces: list[GuardrailTrace] = []
185
186        step = await self._eval_one_rail(
187            event=event,
188            value=value,
189            traces=traces,
190        )
191
192        if step[0] == "stop":
193            return step[1], traces, step[2]
194
195        _, value, event, decision = step
196
197        return (
198            value,
199            traces,
200            decision if decision is not None else GuardrailDecision.allow(),
201        )

Run this guard once on event and value.

Returns the resulting value, the traces recorded, and the decision that applies. A BLOCK (or a failed TRANSFORM) surfaces its own decision as-is. A successful TRANSFORM also surfaces its own decision (its reason/meta describe what changed, e.g. a PII redaction count -- callers like MiddlewareGuardInputResponseEvent depend on that). A plain ALLOW, or an exception swallowed via fail_open=True, has nothing rail-specific worth keeping, so a generic GuardrailDecision.allow() is returned instead.