Output Validation Post-Hook

import asyncio

from agno.agent import Agent
from agno.exceptions import CheckTrigger, OutputCheckError
from agno.models.openai import OpenAIResponses
from agno.run.agent import RunOutput
from agno.run import RunStatus
from pydantic import BaseModel


class OutputValidationResult(BaseModel):
    is_complete: bool
    is_professional: bool
    is_safe: bool
    concerns: list[str]
    confidence_score: float


def validate_response_quality(run_output: RunOutput) -> None:
    """
    Post-hook: Validate the agent's response for quality and safety.

    This hook checks:
    - Response completeness (not too short or vague)
    - Professional tone and language
    - Safety and appropriateness of content

    Raises OutputCheckError if validation fails.
    """

    # Skip validation for empty responses
    if not isinstance(run_output.content, str) or len(run_output.content.strip()) < 10:
        raise OutputCheckError(
            "Response is too short or empty",
            check_trigger=CheckTrigger.OUTPUT_NOT_ALLOWED,
        )

    # Create a validation agent
    validator_agent = Agent(
        name="Output Validator",
        model=OpenAIResponses(id="gpt-5.2"),
        instructions=[
            "You are an output quality validator. Analyze responses for:",
            "1. COMPLETENESS: Response addresses the question thoroughly",
            "2. PROFESSIONALISM: Language is professional and appropriate",
            "3. SAFETY: Content is safe and doesn't contain harmful advice",
            "",
            "Provide a confidence score (0.0-1.0) for overall quality.",
            "List any specific concerns found.",
            "",
            "Be reasonable - don't reject good responses for minor issues.",
        ],
        output_schema=OutputValidationResult,
    )

    try:
        validation_result = validator_agent.run(
            input=f"Validate this response: '{run_output.content}'"
        )
    except Exception as exc:
        raise OutputCheckError("Validator unavailable; validation is required.") from exc

    if validation_result.status != RunStatus.completed or not isinstance(
        validation_result.content, OutputValidationResult
    ):
        raise OutputCheckError("Validator did not return a valid assessment.")
    result = validation_result.content

    if not 0.0 <= result.confidence_score <= 1.0:
        raise OutputCheckError("Validator returned an invalid confidence score.")

    # Check validation results and raise errors for failures
    if not result.is_complete:
        raise OutputCheckError(
            f"Response is incomplete. Concerns: {', '.join(result.concerns)}",
            check_trigger=CheckTrigger.OUTPUT_NOT_ALLOWED,
        )

    if not result.is_professional:
        raise OutputCheckError(
            f"Response lacks professional tone. Concerns: {', '.join(result.concerns)}",
            check_trigger=CheckTrigger.OUTPUT_NOT_ALLOWED,
        )

    if not result.is_safe:
        raise OutputCheckError(
            f"Response contains potentially unsafe content. Concerns: {', '.join(result.concerns)}",
            check_trigger=CheckTrigger.OUTPUT_NOT_ALLOWED,
        )

    if result.confidence_score < 0.6:
        raise OutputCheckError(
            f"Response quality score too low ({result.confidence_score:.2f}). Concerns: {', '.join(result.concerns)}",
            check_trigger=CheckTrigger.OUTPUT_NOT_ALLOWED,
        )


def simple_length_validation(run_output: RunOutput) -> None:
    """
    Simple post-hook: Basic validation for response length.

    Ensures responses are neither too short nor excessively long.
    """
    if not isinstance(run_output.content, str):
        raise OutputCheckError("Expected a text response")
    content = run_output.content.strip()

    if len(content) < 20:
        raise OutputCheckError(
            "Response is too brief to be helpful",
            check_trigger=CheckTrigger.OUTPUT_NOT_ALLOWED,
        )

    if len(content) > 5000:
        raise OutputCheckError(
            "Response is too lengthy and may overwhelm the user",
            check_trigger=CheckTrigger.OUTPUT_NOT_ALLOWED,
        )


async def main():
    agent_with_validation = Agent(
        name="Customer Support Agent",
        model=OpenAIResponses(id="gpt-5.2"),
        post_hooks=[validate_response_quality],
        instructions="Give clear, professional answers to customer support questions.",
    )
    brief_agent = Agent(
        model=OpenAIResponses(id="gpt-5.2"),
        post_hooks=[simple_length_validation],
        instructions="Answer in one or two words.",
    )
    for agent, request in [
        (agent_with_validation, "How do I reset my Microsoft account password?"),
        (brief_agent, "What is the capital of France?"),
    ]:
        response = await agent.arun(request)
        if response.status != RunStatus.completed:
            # Rejected content can still be present in response.content.
            print("Output validation or generation failed; answer withheld.")
            continue
        print(response.content)


if __name__ == "__main__":
    asyncio.run(main())