TL;DR:
- Agent output validation is a separate engineering concern from input guardrails — outputs need JSON schema enforcement, PII detection, and safety filtering before reaching users or downstream systems
- Three validation layers work together: structural validation (schema), content validation (PII, harmful content), and semantic validation (does the output make sense in context)
- Validation failures should route to fallback paths, not crash the agent — the failure mode matters as much as the validation logic
Agent frameworks give you plenty of tools for controlling what agents do, but the harder problem in production is controlling what agents output. A code-generating agent that leaks API keys in its output, a customer support agent that includes another customer’s name in its reply, or a data extraction agent that returns malformed JSON that breaks your parsing pipeline — these are output failures, and they’re not caught by prompt-level guardrails.
This guide covers the engineering layer that catches output problems before they become user-facing failures, compliance incidents, or broken downstream systems.
The Three Layers of Output Validation
Agent output validation splits into three distinct concerns that require different techniques:
Structural validation — is the output in the expected format? Does the JSON match the schema? Are required fields present? Are types correct? This is the cheapest validation to run and catches the most common failure mode: models generating near-valid output that breaks strict parsers.
Content validation — does the output contain things it shouldn’t? PII from other users, credentials, harmful content, competitor names, or restricted information. This is more expensive than structural validation and requires pattern matching, entity recognition, or a secondary model pass.
Semantic validation — does the output make sense in context? Is a summarisation agent returning a summary of the wrong document? Is a classification agent returning a valid class label that doesn’t match the content? This is the hardest to automate and often requires lightweight model-as-judge approaches.
Run them in order — structural first because it’s fast and a malformed structure breaks content validation, then content, then semantic for high-stakes outputs only.
Structural Validation with JSON Schema
The most common agent output format is JSON. The most common failure mode is subtle JSON invalidity: trailing commas, unescaped quotes, truncated outputs when context windows overflow, or models generating creative variations on the requested schema.
import json
from jsonschema import validate, ValidationError
from pydantic import BaseModel, ValidationError as PydanticError
# Define expected output schema
invoice_extraction_schema = {
"type": "object",
"required": ["vendor", "amount", "currency", "invoice_date", "line_items"],
"properties": {
"vendor": {"type": "string", "minLength": 1},
"amount": {"type": "number", "minimum": 0},
"currency": {"type": "string", "enum": ["GBP", "USD", "EUR"]},
"invoice_date": {"type": "string", "format": "date"},
"line_items": {
"type": "array",
"minItems": 1,
"items": {
"type": "object",
"required": ["description", "quantity", "unit_price"],
"properties": {
"description": {"type": "string"},
"quantity": {"type": "number"},
"unit_price": {"type": "number"}
}
}
}
}
}
def validate_invoice_output(raw_output: str) -> dict:
# Step 1: parse JSON
try:
parsed = json.loads(raw_output)
except json.JSONDecodeError as e:
raise ValueError(f"Agent returned invalid JSON: {e}")
# Step 2: validate schema
try:
validate(instance=parsed, schema=invoice_extraction_schema)
except ValidationError as e:
raise ValueError(f"Output failed schema validation: {e.message}")
return parsedPrefer Pydantic for complex schemas — Pydantic gives you cleaner error messages, coercion where you want it (e.g., string “2026-01-15” coerced to datetime.date), and typed access throughout your codebase:
from pydantic import BaseModel, Field
from datetime import date
from typing import Literal
class LineItem(BaseModel):
description: str
quantity: float = Field(gt=0)
unit_price: float = Field(ge=0)
class InvoiceExtraction(BaseModel):
vendor: str = Field(min_length=1)
amount: float = Field(ge=0)
currency: Literal["GBP", "USD", "EUR"]
invoice_date: date
line_items: list[LineItem] = Field(min_length=1)
def parse_invoice_output(raw_output: str) -> InvoiceExtraction:
parsed = json.loads(raw_output)
return InvoiceExtraction.model_validate(parsed)Structured output API features help but don’t replace validation. Anthropic, OpenAI, and most major providers now offer constrained output generation (JSON mode or tool use with schemas). Use them — they reduce but don’t eliminate validation failures. Edge cases still slip through, and schema-constrained generation at the API level doesn’t catch content-level issues like PII.
Retry Logic for Structural Failures
When structural validation fails, the right default is a retry with the error context injected:
async def extract_invoice_with_retry(document: str, max_retries: int = 2) -> InvoiceExtraction:
messages = [
{"role": "user", "content": f"Extract invoice data from this document:\n\n{document}"}
]
for attempt in range(max_retries + 1):
response = await client.messages.create(
model="claude-opus-4-7",
max_tokens=2048,
system="Extract invoice data as valid JSON matching the schema exactly.",
messages=messages
)
raw_output = response.content[0].text
try:
return parse_invoice_output(raw_output)
except (json.JSONDecodeError, ValidationError) as e:
if attempt == max_retries:
raise
# Inject the error context for the next attempt
messages.extend([
{"role": "assistant", "content": raw_output},
{"role": "user", "content": f"That output was invalid: {e}. Please correct it and return valid JSON matching the schema."}
])
raise RuntimeError("Extraction failed after retries")Cap retries at 2-3. More than that and you’re likely hitting a systematic failure (wrong schema, wrong model, wrong document type) that a retry loop won’t fix.
PII Detection and Redaction
PII in agent outputs is the compliance risk that causes breach notifications. The scenarios:
- A customer support agent surfaces another user’s order history because a RAG retrieval pulled the wrong documents
- A document analysis agent includes the original author’s name and email in its summary output
- A code generation agent generates SQL queries containing hardcoded values from example data that was PII
- A legal research agent includes case details from confidential filings in a public-facing response
The two implementation approaches are regex/pattern matching (fast, high precision for known patterns) and NLP entity recognition (slower, higher recall for novel PII).
import re
from typing import NamedTuple
class PIIMatch(NamedTuple):
pii_type: str
value: str
start: int
end: int
# Fast pattern-based detection for structured PII
PII_PATTERNS = {
"email": re.compile(r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b'),
"uk_phone": re.compile(r'\b(?:0|\+44)\s?(?:\d\s?){9,10}\b'),
"uk_national_insurance": re.compile(r'\b[A-CEGHJ-PR-TW-Z]{1}[A-CEGHJ-NPR-TW-Z]{1}[0-9]{6}[A-D\s]\b', re.IGNORECASE),
"credit_card": re.compile(r'\b(?:\d{4}[\s-]?){3}\d{4}\b'),
"uk_postcode": re.compile(r'\b[A-Z]{1,2}\d[A-Z\d]?\s*\d[A-Z]{2}\b', re.IGNORECASE),
"aws_access_key": re.compile(r'\bAKIA[0-9A-Z]{16}\b'),
"generic_api_key": re.compile(r'\b(?:api[_-]?key|secret|token|password)\s*[:=]\s*["\']?[A-Za-z0-9+/]{20,}["\']?', re.IGNORECASE),
}
def detect_pii(text: str) -> list[PIIMatch]:
matches = []
for pii_type, pattern in PII_PATTERNS.items():
for match in pattern.finditer(text):
matches.append(PIIMatch(
pii_type=pii_type,
value=match.group(),
start=match.start(),
end=match.end()
))
return sorted(matches, key=lambda m: m.start)
def redact_pii(text: str, replacement: str = "[REDACTED]") -> tuple[str, list[PIIMatch]]:
matches = detect_pii(text)
if not matches:
return text, []
result = []
last_end = 0
for match in matches:
result.append(text[last_end:match.start])
result.append(f"[{match.pii_type.upper()}_REDACTED]")
last_end = match.end
result.append(text[last_end:])
return "".join(result), matchesFor higher-recall PII detection on unstructured text (names, addresses, organisation names), use spaCy with a NER model or a dedicated PII detection library like Microsoft Presidio:
from presidio_analyzer import AnalyzerEngine
from presidio_anonymizer import AnonymizerEngine
analyzer = AnalyzerEngine()
anonymizer = AnonymizerEngine()
def redact_pii_presidio(text: str, language: str = "en") -> str:
results = analyzer.analyze(
text=text,
language=language,
entities=["PERSON", "EMAIL_ADDRESS", "PHONE_NUMBER", "CREDIT_CARD", "UK_NHS", "LOCATION"]
)
anonymized = anonymizer.anonymize(text=text, analyzer_results=results)
return anonymized.textMatch your PII detection to your output type. Structured JSON outputs often need field-level PII checking (is this “customer_name” field supposed to be in the output at all, or was it retrieved by mistake?) rather than full-text scanning. Consider separating PII detection from redaction — detection tells you whether to redact, alert, or block; redaction is the action you take.
Harmful Content Filtering
For customer-facing agents, filtering potentially harmful outputs is a product requirement, not just an ethics consideration. Categories to filter vary by context but typically include:
- Discriminatory language about protected characteristics
- Instructions for illegal activities
- Competitor disparagement (legal risk)
- Medical, legal, or financial advice that exceeds the agent’s designated scope
The practical implementation is layered:
Layer 1: Block-list and regex — fast, deterministic, handles the obvious cases. A customer service bot should never output the word “lawsuit” or “lawyer up” regardless of context.
Layer 2: Classifier — a small, fast classification model that scores output toxicity or off-topic probability. Perspective API, OpenAI Moderation API, or a fine-tuned classifier. Run this on every output.
Layer 3: LLM-as-judge — a secondary model call that evaluates the output against specific criteria. Expensive; use selectively for high-stakes categories or when classifier confidence is low.
import anthropic
client = anthropic.Anthropic()
async def check_output_safety(
output: str,
agent_context: str,
fail_open: bool = False
) -> tuple[bool, str]:
"""Returns (is_safe, reason). fail_open=True passes uncertain cases through."""
# Layer 1: hard block list
blocked_phrases = ["sue us", "contact a lawyer", "legal action"]
for phrase in blocked_phrases:
if phrase.lower() in output.lower():
return False, f"Blocked phrase detected: '{phrase}'"
# Layer 2: LLM judge for ambiguous cases (sample 10% of outputs or all high-stakes)
judge_response = await client.messages.create(
model="claude-haiku-4-5-20251001",
max_tokens=128,
system="""You are a content safety classifier. Evaluate if the output is safe to show to a customer.
Respond with JSON: {"safe": true/false, "reason": "brief explanation"}
Context about the agent: """ + agent_context,
messages=[{"role": "user", "content": f"Evaluate this output:\n\n{output}"}]
)
try:
result = json.loads(judge_response.content[0].text)
return result["safe"], result.get("reason", "")
except (json.JSONDecodeError, KeyError):
return fail_open, "Judge response parsing failed"Audit Logging for Agent Outputs
Validation without audit logging is debugging without evidence. Log every output, every validation result, and every block or redaction — not for compliance theatre, but because agent failures are often subtle and invisible until something breaks downstream.
Minimum audit record per agent output:
from dataclasses import dataclass
from datetime import datetime, UTC
import uuid
@dataclass
class AgentOutputAuditRecord:
audit_id: str
agent_id: str
session_id: str
user_id: str | None
timestamp: str
raw_output_hash: str # hash of pre-validation output, not the raw content
validation_passed: bool
schema_valid: bool
pii_detected: list[str] # types detected, not values
pii_redacted: bool
safety_check_result: str # "passed", "blocked", "redacted"
final_output_delivered: bool
latency_ms: int
def create_audit_record(
agent_id: str,
session_id: str,
user_id: str | None,
raw_output: str,
validation_result: dict,
latency_ms: int
) -> AgentOutputAuditRecord:
import hashlib
return AgentOutputAuditRecord(
audit_id=str(uuid.uuid4()),
agent_id=agent_id,
session_id=session_id,
user_id=user_id,
timestamp=datetime.now(UTC).isoformat(),
raw_output_hash=hashlib.sha256(raw_output.encode()).hexdigest()[:16],
validation_passed=validation_result["passed"],
schema_valid=validation_result["schema_valid"],
pii_detected=validation_result.get("pii_types_detected", []),
pii_redacted=validation_result.get("pii_redacted", False),
safety_check_result=validation_result.get("safety_result", "unknown"),
final_output_delivered=validation_result.get("delivered", False),
latency_ms=latency_ms
)Store raw outputs separately with tighter access controls than audit records. Audit records can live in your main logging infrastructure; raw outputs (which may contain PII or harmful content) should be in an access-controlled store with retention limits.
Putting It Together: A Validation Pipeline
import time
async def validated_agent_output(
agent_fn, # async function that calls the model
schema_validator, # function that parses and validates structure
agent_context: str,
session_id: str,
user_id: str | None = None
) -> dict:
start = time.monotonic()
# Run the agent
raw_output = await agent_fn()
audit = {
"schema_valid": False,
"pii_types_detected": [],
"pii_redacted": False,
"safety_result": "pending",
"passed": False,
"delivered": False
}
# Layer 1: Structural validation
try:
validated = schema_validator(raw_output)
audit["schema_valid"] = True
except ValueError as e:
# Log and raise — structural failures block delivery
record = create_audit_record("agent-id", session_id, user_id, raw_output, audit,
int((time.monotonic() - start) * 1000))
log_audit(record)
raise
# Layer 2: PII detection on string representations
output_str = json.dumps(validated)
redacted_str, pii_matches = redact_pii(output_str)
if pii_matches:
audit["pii_types_detected"] = list({m.pii_type for m in pii_matches})
audit["pii_redacted"] = True
validated = json.loads(redacted_str)
# Layer 3: Safety check (sample or always, depending on risk tier)
is_safe, safety_reason = await check_output_safety(
json.dumps(validated), agent_context, fail_open=True
)
audit["safety_result"] = "passed" if is_safe else f"blocked: {safety_reason}"
if not is_safe:
audit["delivered"] = False
record = create_audit_record("agent-id", session_id, user_id, raw_output, audit,
int((time.monotonic() - start) * 1000))
log_audit(record)
raise ValueError(f"Output blocked by safety check: {safety_reason}")
audit["passed"] = True
audit["delivered"] = True
record = create_audit_record("agent-id", session_id, user_id, raw_output, audit,
int((time.monotonic() - start) * 1000))
log_audit(record)
return validatedFailure Modes and Fallback Paths
Validation failure should route to a fallback, not an unhandled exception visible to the user:
- Schema failure: retry with error context injected (up to 2 retries), then return a structured error response
- PII detected and redacted: deliver redacted output, log for review — the agent retrieved more than it should and that’s a signal to fix the retrieval, not a delivery blocker
- Safety blocked: return a safe fallback message (“I can’t help with that in this context”) and alert the ops team
- Semantic validation failed: the most nuanced case — often means the agent completed the task for the wrong input. Route to human review for high-stakes workflows; log and continue for low-stakes ones
Don’t conflate validation failure with model failure. A PII detection hit doesn’t mean the model misbehaved — it often means your retrieval or system prompt is exposing data the model correctly reproduced. Fix the data flow, not just the output filter.
What to Monitor in Production
Track these metrics across your agent output pipeline:
- Schema validation failure rate — should be under 2% for well-configured structured outputs. Higher means your prompts or schemas need work.
- PII detection rate — what percentage of outputs contain PII before redaction. Trending up means something in your context injection changed.
- Safety block rate — baseline this for your agent and alert on deviations. A sudden spike means either input distribution shifted or your system prompt broke.
- Retry success rate — how often does a retry after schema failure succeed? If it’s below 70%, the retry prompt needs improvement.
- End-to-end validation latency — validation adds latency. Know your p99 so you can set appropriate timeouts.
Output validation is not a one-time implementation — it’s a live signal about how your agents are actually behaving in production. The metrics tell you what to fix upstream.