When building systems on Antigravity agents, you'll eventually encounter output validation failures: the agent returns subtly different JSON formats on each run, downstream processing breaks because a required field is missing, or output quality is inconsistent enough to cause problems in production.
AI agent output is probabilistic. You can't eliminate variation entirely, but you can build systems that catch and recover from it reliably. This guide covers the three categories of validation errors and practical implementation patterns for each.
Three Categories of Agent Output Validation Errors
Format errors: The output can't be parsed as JSON, required fields are missing, or data types don't match. These break downstream systems immediately.
Quality errors: The format is correct but the content doesn't meet requirements — response too short, information inaccurate, or task requirements not fulfilled.
Consistency errors: Results vary significantly across multiple runs, or multiple agents in a pipeline produce contradictory outputs.
Handling Format Errors: Enforcing Structured Output
Prompt-level Format Specification
Making the JSON schema explicit in your prompt significantly reduces format errors.
def create_structured_prompt(task: str, schema: dict) -> str:
"""Generate a prompt that enforces structured output"""
schema_str = json.dumps(schema, indent=2)
return f"""
Complete the following task and output ONLY valid JSON matching the schema below.
Task: {task}
Output schema:
```json
{schema_str}Important:
- Output JSON only — no explanatory text
- Include all required fields
- Use empty strings or empty arrays instead of null
JSON output: """
Usage example
schema = { "title": "string (required)", "summary": "string (required, max 100 words)", "key_points": ["string (required, 3-5 items)"], "confidence_score": "number (required, 0.0-1.0)" }
prompt = create_structured_prompt("Analyze the latest AI trends", schema)
### Parse Error Handling and Auto-Repair
```python
import json
import re
from typing import Optional
def parse_agent_output(raw_output: str) -> Optional[dict]:
"""Parse agent output with progressive fallback strategies"""
# 1. Direct parse
try:
return json.loads(raw_output)
except json.JSONDecodeError:
pass
# 2. Strip markdown code blocks
code_block_pattern = r'```(?:json)?\s*([\s\S]*?)\s*```'
match = re.search(code_block_pattern, raw_output)
if match:
try:
return json.loads(match.group(1))
except json.JSONDecodeError:
pass
# 3. Extract JSON object from surrounding text
json_pattern = r'\{[\s\S]*\}'
match = re.search(json_pattern, raw_output)
if match:
try:
return json.loads(match.group())
except json.JSONDecodeError:
pass
return None
def validate_output(parsed: dict, schema: dict) -> tuple[bool, list[str]]:
"""Validate parsed output against schema. Returns (is_valid, error_list)"""
errors = []
for field, field_type in schema.items():
if field not in parsed:
errors.append(f"Missing required field: {field}")
continue
value = parsed[field]
if "string" in str(field_type) and not isinstance(value, str):
errors.append(f"Type error: {field} must be a string (got {type(value).__name__})")
elif "number" in str(field_type) and not isinstance(value, (int, float)):
errors.append(f"Type error: {field} must be a number")
elif isinstance(field_type, list) and not isinstance(value, list):
errors.append(f"Type error: {field} must be a list")
return len(errors) == 0, errors
Auto-Retry Until Valid
from antigravity import Agent # hypothetical Antigravity SDK
async def get_validated_output(
agent: Agent,
task: str,
schema: dict,
max_retries: int = 3
) -> Optional[dict]:
"""Retry with corrective feedback until output validates"""
for attempt in range(max_retries):
prompt = create_structured_prompt(task, schema)
raw_output = await agent.run(prompt)
parsed = parse_agent_output(raw_output)
if parsed is None:
if attempt < max_retries - 1:
task = f"""
Your previous output couldn't be parsed as JSON.
Previous output: {raw_output[:500]}
Please try again: {task}
"""
continue
is_valid, errors = validate_output(parsed, schema)
if is_valid:
return parsed
if attempt < max_retries - 1:
error_list = "\n".join(f"- {e}" for e in errors)
task = f"""
Your previous output had these validation errors:
{error_list}
Please fix and try again: {task}
"""
else:
print(f"Validation failed after {max_retries} attempts: {errors}")
return None
return NoneHandling Quality Errors: Self-Checking Pattern
Having an agent evaluate its own output (or using a separate checker agent) catches quality issues before they reach your users.
QUALITY_CHECK_PROMPT = """
Evaluate the quality of the following agent output.
Task: {task}
Output: {output}
Return a JSON assessment using this schema:
{{
"meets_requirements": true/false,
"quality_score": 0.0 to 1.0,
"issues": ["list of specific problems"],
"needs_revision": true/false,
"revision_guidance": "specific instructions for improvement if needed"
}}
Evaluation criteria:
1. Does it fully address the task requirements?
2. Is the information specific and actionable?
3. Are there any contradictions or inaccuracies?
4. Does it contain the necessary level of detail?
"""
async def get_quality_checked_output(
primary_agent: Agent,
checker_agent: Agent,
task: str,
schema: dict,
min_quality_score: float = 0.7
) -> Optional[dict]:
"""Generate output with quality gate"""
for attempt in range(3):
output = await get_validated_output(primary_agent, task, schema)
if output is None:
continue
check_prompt = QUALITY_CHECK_PROMPT.format(
task=task,
output=json.dumps(output)
)
check_raw = await checker_agent.run(check_prompt)
check_result = parse_agent_output(check_raw)
if check_result is None:
return output # If checker itself fails, use the output as-is
quality_score = check_result.get("quality_score", 0)
needs_revision = check_result.get("needs_revision", False)
if not needs_revision and quality_score >= min_quality_score:
return output
if attempt < 2:
guidance = check_result.get("revision_guidance", "Please improve quality")
issues = check_result.get("issues", [])
task = f"{task}\n\nRevision required:\n{guidance}\nIssues: {', '.join(issues)}"
else:
# Return with quality warning after final attempt
output["_quality_warning"] = {
"score": quality_score,
"issues": check_result.get("issues", [])
}
return output
return NoneHandling Consistency Errors: Consensus Aggregation
When output varies too much across runs or across multiple agents, use voting to produce a stable result.
from collections import Counter
from typing import List
async def get_consensus_output(
agent: Agent,
task: str,
num_samples: int = 3,
schema: dict = None
) -> Optional[dict]:
"""
Run the agent multiple times and aggregate results by majority vote.
Reduces variance in probabilistic outputs.
"""
outputs = []
for _ in range(num_samples):
output = await get_validated_output(agent, task, schema or {})
if output:
outputs.append(output)
if not outputs:
return None
if len(outputs) == 1:
return outputs[0]
consensus = {}
all_keys = set().union(*[o.keys() for o in outputs])
for key in all_keys:
if key.startswith("_"):
continue
if isinstance(outputs[0].get(key), (int, float)):
# Average numeric values
numeric_values = [o[key] for o in outputs if key in o and isinstance(o[key], (int, float))]
consensus[key] = sum(numeric_values) / len(numeric_values)
elif isinstance(outputs[0].get(key), list):
# For lists, rank items by frequency
all_items = [item for o in outputs for item in o.get(key, [])]
item_counts = Counter(all_items)
consensus[key] = [item for item, _ in item_counts.most_common()]
else:
# For strings, use the most common value
values = [str(o.get(key, "")) for o in outputs if key in o]
if values:
consensus[key] = Counter(values).most_common(1)[0][0]
return consensusMonitoring Validation Metrics in Production
from dataclasses import dataclass, field
@dataclass
class ValidationMetrics:
total_attempts: int = 0
format_errors: int = 0
quality_errors: int = 0
success_count: int = 0
retry_counts: list = field(default_factory=list)
@property
def success_rate(self) -> float:
return self.success_count / self.total_attempts if self.total_attempts > 0 else 0
@property
def avg_retries(self) -> float:
return sum(self.retry_counts) / len(self.retry_counts) if self.retry_counts else 0
def alert_if_needed(self):
if self.success_rate < 0.8:
print(f"⚠️ Output validation success rate low: {self.success_rate:.1%}")
if self.avg_retries > 1.5:
print(f"⚠️ High average retry count: {self.avg_retries:.1f}")
metrics = ValidationMetrics()Wrapping Up
Antigravity agent output validation errors fall into three categories: format errors (fixed with auto-repair parsing and retry loops), quality errors (caught with self-checking patterns), and consistency errors (mitigated with consensus aggregation across multiple runs).
Perfect output every time isn't realistic with probabilistic AI systems. But combining these three layers of validation gives you production-grade reliability without needing to manually inspect every output. We hope this helps you build more dependable agent workflows.