ant_ai.hooks.integrations.guardrails_ai

GuardrailsAIHook pydantic-model

Bases: AgentHook, BaseModel

Wraps a guardrails.Guard instance as an AgentHook.

Only overrides after_model — validates the LLM output text and returns PostModelRetry if validation fails.

Parameters:

Name	Type	Description	Default
guard		A configured guardrails.Guard instance.	required
num_reasks		How many times guardrails may internally call an LLM to fix invalid output before handing control back to ant-ai. The default (0) disables guardrails' own reask loop so ant-ai's retry mechanism stays in control. Set to a positive value only when the guard has an llm_api configured and you want guardrails to attempt self-correction before ant-ai retries.	required
api_key		API key forwarded to the LLM used by guardrails during internal reasks. Only relevant when num_reasks > 0.	required

.. note:: Guard is not thread-safe. Validation calls on a shared hook instance are serialized with an internal lock so concurrent agent invocations (e.g. inside an A2A server) do not race on the guard's internal history.

Example::

```python
from guardrails import Guard
from guardrails.hub import ValidJson

hook = GuardrailsAIHook(guard=Guard().use(ValidJson))
agent = Agent(..., hooks=[hook])
```

See examples/guardrails_agent.py for a full safety pipeline using ToxicLanguage and DetectPII validators.

Show JSON schema:

{
  "description": "Wraps a ``guardrails.Guard`` instance as an ``AgentHook``.\n\nOnly overrides ``after_model`` \u2014 validates the LLM output text and\nreturns ``PostModelRetry`` if validation fails.\n\nArgs:\n    guard: A configured ``guardrails.Guard`` instance.\n    num_reasks: How many times guardrails may internally call an LLM to\n        fix invalid output before handing control back to ant-ai.  The\n        default (``0``) disables guardrails' own reask loop so ant-ai's\n        retry mechanism stays in control.  Set to a positive value only\n        when the guard has an ``llm_api`` configured and you want\n        guardrails to attempt self-correction before ant-ai retries.\n    api_key: API key forwarded to the LLM used by guardrails during\n        internal reasks.  Only relevant when ``num_reasks > 0``.\n\n.. note::\n    ``Guard`` is not thread-safe. Validation calls on a shared hook\n    instance are serialized with an internal lock so concurrent agent\n    invocations (e.g. inside an A2A server) do not race on the guard's\n    internal history.\n\nExample::\n\n    ```python\n    from guardrails import Guard\n    from guardrails.hub import ValidJson\n\n    hook = GuardrailsAIHook(guard=Guard().use(ValidJson))\n    agent = Agent(..., hooks=[hook])\n    ```\n\nSee ``examples/guardrails_agent.py`` for a full safety pipeline using\n``ToxicLanguage`` and ``DetectPII`` validators.",
  "properties": {
    "guard": {
      "title": "Guard"
    },
    "num_reasks": {
      "default": 0,
      "minimum": 0,
      "title": "Num Reasks",
      "type": "integer"
    },
    "api_key": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Api Key"
    }
  },
  "required": [
    "guard"
  ],
  "title": "GuardrailsAIHook",
  "type": "object"
}

Config:

• arbitrary_types_allowed: True

Fields:

• guard (SkipValidation[Any])
• num_reasks (int)
• api_key (str | None)

Source code in src/ant_ai/hooks/integrations/guardrails_ai.py

class GuardrailsAIHook(AgentHook, BaseModel):
    """
    Wraps a ``guardrails.Guard`` instance as an ``AgentHook``.

    Only overrides ``after_model`` — validates the LLM output text and
    returns ``PostModelRetry`` if validation fails.

    Args:
        guard: A configured ``guardrails.Guard`` instance.
        num_reasks: How many times guardrails may internally call an LLM to
            fix invalid output before handing control back to ant-ai.  The
            default (``0``) disables guardrails' own reask loop so ant-ai's
            retry mechanism stays in control.  Set to a positive value only
            when the guard has an ``llm_api`` configured and you want
            guardrails to attempt self-correction before ant-ai retries.
        api_key: API key forwarded to the LLM used by guardrails during
            internal reasks.  Only relevant when ``num_reasks > 0``.

    .. note::
        ``Guard`` is not thread-safe. Validation calls on a shared hook
        instance are serialized with an internal lock so concurrent agent
        invocations (e.g. inside an A2A server) do not race on the guard's
        internal history.

    Example::

        ```python
        from guardrails import Guard
        from guardrails.hub import ValidJson

        hook = GuardrailsAIHook(guard=Guard().use(ValidJson))
        agent = Agent(..., hooks=[hook])
        ```

    See ``examples/guardrails_agent.py`` for a full safety pipeline using
    ``ToxicLanguage`` and ``DetectPII`` validators.
    """

    model_config = ConfigDict(arbitrary_types_allowed=True)

    name: ClassVar[str] = "guardrails_ai"
    guard: SkipValidation[Any]  # guardrails.Guard
    num_reasks: int = Field(default=0, ge=0)
    api_key: str | None = None
    _lock: threading.Lock = PrivateAttr(default_factory=threading.Lock)

    async def after_model(
        self,
        result: StepResult,
        ctx: InvocationContext | None,
    ) -> PostModelDecision:
        if not isinstance(result.output, LLMOutput):
            return PostModelPass(result=result)

        # Guard.validate is synchronous and mutates internal state (history,
        # reask counter). Serialize via _lock so concurrent callers sharing
        # this hook instance do not race on that state.
        raw = result.output.raw
        if not raw or not raw.strip():
            # No text content: check tool call arguments instead (the LLM may
            # have written content via tool calls, e.g. FilesystemTool).
            tool_calls = result.output.tool_calls
            if not tool_calls:
                return PostModelPass(result=result)
            raw = "\n".join(tc.function.arguments for tc in tool_calls)

        def _validate() -> Any:
            with self._lock:
                kwargs: dict[str, Any] = {}
                if self.api_key is not None:
                    kwargs["api_key"] = self.api_key
                return self.guard.validate(raw, num_reasks=self.num_reasks, **kwargs)

        try:
            outcome = await asyncio.to_thread(_validate)
        except Exception as exc:  # noqa: BLE001
            return PostModelRetry(reason=f"guardrails validation error: {exc}")

        # When on_fail="reask" and num_reasks=0, guardrails quirk: it sets
        # validation_passed=True even though validators failed (it expected to
        # reask but couldn't). Detect real failures via validator_status.
        failed = [
            s
            for s in (outcome.validation_summaries or [])
            if getattr(s, "validator_status", None) == "fail"
        ]
        if outcome.validation_passed and not failed:
            return PostModelPass(result=result)

        history = getattr(self.guard, "history", None)
        return PostModelRetry(
            reason=_failure_reason(
                outcome.validation_summaries,
                history.last if history else None,
            )
        )