railtracks.guardrails.llm.input

14class BlockTextInputGuard(InputGuard):
15    """Blocks LLM input when any user or system message matches a regex pattern."""
16
17    def __init__(
18        self,
19        pattern: str,
20        *,
21        name: str | None = None,
22        user_facing_message: str | None = None,
23    ) -> None:
24        """Initialize the input block-text guard.
25
26        Args:
27            pattern: Regex pattern; if it matches any scannable message content
28                the guard returns ``BLOCK``.
29            name: Optional rail name for traces (see :class:`InputGuard`).
30            user_facing_message: Optional message surfaced to UIs and
31                visualizers when the guard blocks.
32
33        Raises:
34            re.error: If *pattern* is not a valid regular expression.
35        """
36        super().__init__(name=name)
37        self._pattern = re.compile(pattern)
38        self._user_facing_message = user_facing_message
39
40    def __call__(self, event: LLMGuardrailEvent) -> GuardrailDecision:
41        """Block if any user/system string message matches the pattern.
42
43        Returns:
44            ``BLOCK`` when the pattern is found, ``ALLOW`` otherwise.
45        """
46        for msg in event.messages:
47            if msg.role not in _SCANNABLE_ROLES or not isinstance(msg.content, str):
48                continue
49            if self._pattern.search(msg.content):
50                return GuardrailDecision.block(
51                    reason=("Input blocked: prohibited content detected."),
52                    user_facing_message=self._user_facing_message,
53                )
54        return GuardrailDecision.allow(reason="No blocked patterns detected in input.")

18class PIIRedactInputGuard(InputGuard):
19    """Redacts PII from user and system string messages before they reach the LLM."""
20
21    def __init__(
22        self,
23        config: PIIRedactConfig | None = None,
24        *,
25        name: str | None = None,
26    ) -> None:
27        """Initialize the input PII redactor.
28
29        Args:
30            config: Redaction settings; defaults to all built-in entity kinds and no
31                custom patterns (see :class:`~railtracks.guardrails.llm.PIIRedactConfig`).
32            name: Optional rail name for traces (see :class:`InputGuard`).
33        """
34        super().__init__(name=name)
35        self._config = config or PIIRedactConfig()
36        self._engine = PIIEngine(self._config)
37
38    def __call__(self, event: LLMGuardrailEvent) -> GuardrailDecision:
39        """Scan user/system string content and redact matches.
40
41        Returns:
42            ``ALLOW`` when no PII is found, or ``TRANSFORM`` with rewritten messages on
43            :attr:`~railtracks.guardrails.core.decision.GuardrailDecision.messages`
44            and redaction metadata in ``meta``.
45        """
46        all_records: list[RedactionRecord] = []
47        new_messages: list[Message] = []
48        messages_affected = 0
49
50        for msg in event.messages:
51            if msg.role not in _SCANNABLE_ROLES or not isinstance(msg.content, str):
52                new_messages.append(msg)
53                continue
54
55            redacted_text, records = self._engine.redact(msg.content)
56            if records:
57                all_records.extend(records)
58                messages_affected += 1
59                clone = deepcopy(msg)
60                clone._content = redacted_text
61                new_messages.append(clone)
62            else:
63                new_messages.append(msg)
64
65        if not all_records:
66            return GuardrailDecision.allow(reason="No PII detected in input.")
67
68        return GuardrailDecision.transform_messages(
69            messages=MessageHistory(new_messages),
70            reason=f"Redacted {len(all_records)} PII span(s) from input messages.",
71            meta=build_redaction_meta(all_records, messages_affected=messages_affected),
72        )