ant_ai.hooks.adapters.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.

Example::

from guardrails import Guard
from guardrails.hub import ValidJson

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

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\nExample::\n\n    from guardrails import Guard\n    from guardrails.hub import ValidJson\n\n    hook = GuardrailsAIHook(guard=Guard().use(ValidJson))\n    agent = Agent(..., hooks=[hook])",
  "properties": {
    "name": {
      "default": "guardrails_ai",
      "title": "Name",
      "type": "string"
    },
    "guard": {
      "title": "Guard"
    }
  },
  "required": [
    "guard"
  ],
  "title": "GuardrailsAIHook",
  "type": "object"
}

Config:

• arbitrary_types_allowed: True

Fields:

• name (str)
• guard (SkipValidation[Any])

Source code in src/ant_ai/hooks/adapters/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.

    Example::

        from guardrails import Guard
        from guardrails.hub import ValidJson

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

    model_config = ConfigDict(arbitrary_types_allowed=True)

    name: str = "guardrails_ai"
    guard: SkipValidation[Any]  # guardrails.Guard

    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 — run in a thread to avoid blocking.
        outcome = await asyncio.to_thread(self.guard.validate, result.output.raw)

        if outcome.validation_passed:
            return PostModelPass(result=result)

        return PostModelRetry(reason=str(outcome.error))