Output Validation

Never trust LLM outputs blindly. Validation ensures agent responses are safe, accurate, and properly formatted before reaching users.

Validation Layers

LLM Response → Format Check → Content Safety → Factual Verify → Business Rules → User

Schema Validation

Ensure outputs match expected structure:

from pydantic import BaseModel, validator
from typing import List, Optional

class AgentResponse(BaseModel):
    answer: str
    confidence: float
    sources: List[str]
    tool_calls: Optional[List[dict]] = None

    @validator('confidence')
    def confidence_range(cls, v):
        if not 0 <= v <= 1:
            raise ValueError('Confidence must be between 0 and 1')
        return v

    @validator('answer')
    def answer_not_empty(cls, v):
        if not v.strip():
            raise ValueError('Answer cannot be empty')
        return v

def validate_response(raw_output):
    try:
        return AgentResponse.parse_raw(raw_output)
    except ValidationError as e:
        return {"error": "Invalid response format", "details": str(e)}

Content Safety Checks

class ContentValidator:
    def __init__(self):
        self.blocked_patterns = [
            r'\b(password|secret|api[_-]?key)\s*[:=]\s*\S+',
            r'\b\d{16}\b',  # Credit card patterns
            r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b'  # Emails
        ]

    def check_pii_leakage(self, text):
        """Check for accidentally exposed sensitive data"""
        for pattern in self.blocked_patterns:
            if re.search(pattern, text, re.IGNORECASE):
                return False, "Potential sensitive data detected"
        return True, None

    def check_harmful_content(self, text):
        """Use moderation API or classifier"""
        result = moderation_api.check(text)
        if result.flagged:
            return False, f"Content flagged: {result.categories}"
        return True, None

Factual Consistency

Verify claims against known facts:

class FactChecker:
    def __init__(self, knowledge_base):
        self.kb = knowledge_base

    def verify_claims(self, response, context):
        """Cross-reference response claims with source context"""
        claims = extract_claims(response)
        verified = []

        for claim in claims:
            # Check if claim is supported by context
            support = self.kb.find_support(claim, context)

            if support.score > 0.8:
                verified.append({"claim": claim, "status": "verified"})
            elif support.score > 0.5:
                verified.append({"claim": claim, "status": "partial"})
            else:
                verified.append({"claim": claim, "status": "unverified"})

        return verified

    def add_verification_note(self, response, verification):
        """Annotate response with verification status"""
        unverified = [v for v in verification if v["status"] == "unverified"]

        if unverified:
            note = f"\n\n⚠️ {len(unverified)} claim(s) could not be verified against sources."
            return response + note

        return response

Business Logic Validation

class BusinessValidator:
    def __init__(self, rules):
        self.rules = rules

    def validate(self, response, context):
        errors = []

        for rule in self.rules:
            if not rule.check(response, context):
                errors.append(rule.error_message)

        return len(errors) == 0, errors

# Example rules
rules = [
    Rule(
        name="price_sanity",
        check=lambda r, c: not contains_price(r) or price_in_range(r, 0, 10000),
        error_message="Price out of acceptable range"
    ),
    Rule(
        name="date_validity",
        check=lambda r, c: not contains_date(r) or date_is_valid(r),
        error_message="Invalid date in response"
    ),
    Rule(
        name="no_competitors",
        check=lambda r, c: not mentions_competitors(r),
        error_message="Response mentions competitor products"
    )
]

Validation Pipeline

Combine all validators:

class ValidationPipeline:
    def __init__(self):
        self.validators = [
            ("schema", SchemaValidator()),
            ("safety", ContentValidator()),
            ("facts", FactChecker(knowledge_base)),
            ("business", BusinessValidator(business_rules))
        ]

    def validate(self, response, context):
        results = {"passed": True, "checks": []}

        for name, validator in self.validators:
            is_valid, details = validator.validate(response, context)

            results["checks"].append({
                "name": name,
                "passed": is_valid,
                "details": details
            })

            if not is_valid:
                results["passed"] = False

        return results

    def process(self, response, context):
        validation = self.validate(response, context)

        if not validation["passed"]:
            # Option 1: Regenerate
            # Option 2: Return sanitized version
            # Option 3: Return with warnings
            return self.handle_failure(response, validation)

        return response

Best Practices

Validate	Why
Output format	Prevent parsing errors
Content safety	Protect users
Factual accuracy	Build trust
Business rules	Maintain compliance

Next: Learn debugging strategies for agent systems. :::