You ask the model to return JSON. It comes back wrapped in a Markdown code block with extra explanation text around it. Sound familiar?
When building Python agent systems in Antigravity, there's always a point where you need to pass LLM output to downstream processing. Relying on regex to extract JSON or silently swallowing json.loads() errors gives you code that appears to work — until it quietly fails in production at the worst possible moment.
Google Gen AI SDK v1.0 includes a feature called Structured Output: pass a Pydantic model directly as the response schema, and the model will generate only JSON that conforms to that schema. The gap between getting the sample code running and building a stable production pipeline around it, however, is wider than the documentation suggests.
This guide is based on what I've learned building multiple data extraction projects in Antigravity. Not just the happy path — the edge cases, the production gotchas, and the patterns that actually hold up under real load.
Structured Output vs. JSON Mode: Why the Difference Matters
Google Gen AI SDK v1.0 gives you two approaches for structured data. Setting response_mime_type to "application/json" activates "JSON Mode," while passing a Pydantic model to response_schema activates "Structured Output Mode."
They look similar on the surface, but the underlying mechanics are completely different.
JSON Mode adds an instruction to the model to produce JSON. What comes back is "JSON as the model interprets it" — missing fields, type mismatches, and unexpected additional keys are common. Structured Output uses constrained decoding: the set of tokens the model can generate is restricted to only those that conform to your schema. The structure is guaranteed.
# config.py — Basic Structured Output setup
from google import genai
from google.genai import types
from pydantic import BaseModel, Field
from typing import Literal, Optional
import os
client = genai.Client(api_key=os.getenv("GEMINI_API_KEY"))
class InvoiceItem(BaseModel):
name: str = Field(description="Product or service name")
quantity: int = Field(ge=1, description="Quantity (must be 1 or more)")
unit_price: float = Field(ge=0, description="Unit price excluding tax")
class Invoice(BaseModel):
invoice_number: str = Field(description="Invoice number (e.g., INV-2026-001)")
vendor_name: str = Field(description="Name of the issuing vendor")
total_amount: float = Field(ge=0, description="Total amount including tax")
items: list[InvoiceItem] = Field(description="List of line items")
currency: Literal["JPY", "USD", "EUR"] = Field(default="USD")
def extract_invoice(document_text: str) -> Invoice:
response = client.models.generate_content(
model="gemini-2.5-flash",
contents=f"Extract information from the following invoice:\n\n{document_text}",
config=types.GenerateContentConfig(
response_mime_type="application/json",
response_schema=Invoice,
),
)
# response.parsed returns a Pydantic model instance automatically
return response.parsedThe key detail here is response.parsed. Rather than accessing response.text and parsing it yourself, response.parsed has the SDK automatically parse and validate the JSON into a Pydantic model instance. Full type completion works throughout — which makes development in Antigravity noticeably smoother.
Pydantic Schema Design: Patterns That Actually Work
Why Field Descriptions Control Output Quality
Once you start using Structured Output, the first frustration you'll hit isn't a broken schema — it's a structurally valid schema where the values are wrong. vendor_name contains an address. total_amount contains a subtotal. The structure is right; the content isn't.
The root cause is almost always insufficient Field(description=...). The model uses field names and descriptions to decide what to put where. Ambiguous field names without context lead to independent model interpretation.
from datetime import date
# Poor: missing descriptions, values frequently wrong
class ContractPoor(BaseModel):
party_a: str
party_b: str
amount: float
date: str
# Well-designed: explicit descriptions guide content decisions
class ContractWell(BaseModel):
party_a: str = Field(
description="The name of Party A in the contract (company or individual name)"
)
party_b: str = Field(
description="The name of Party B in the contract (company or individual name)"
)
amount: float = Field(
ge=0,
description="Contract value (tax-inclusive). Set to 0.0 if not specified"
)
contract_date: date = Field(
description="Date the contract was signed (return in YYYY-MM-DD format)"
)Using the date type has a practical benefit: the SDK automatically converts string output to a datetime.date object. Specifying the format explicitly in the description stabilizes output considerably.
Constraining Choices with Enum and Literal
For fields where only specific values are valid, Literal and Enum constraints are remarkably effective at eliminating freeform model interpretation.
from enum import Enum
class ContractStatus(str, Enum):
DRAFT = "draft"
ACTIVE = "active"
EXPIRED = "expired"
TERMINATED = "terminated"
class ContractAnalysis(BaseModel):
status: ContractStatus = Field(
description="Current contract status. Use 'active' if expiry date is unknown"
)
risk_level: Literal["low", "medium", "high", "critical"] = Field(
description="Risk assessment. Use 'high' or 'critical' for broad indemnification clauses"
)
auto_renewal: bool = Field(
description="Whether the contract contains an automatic renewal clause"
)
penalty_exists: bool = Field(
description="Whether the contract contains a penalty or liquidated damages clause"
)
summary: str = Field(
max_length=500,
description="Key contract points summarized in 200 words or fewer"
)The str, Enum inheritance lets you both compare with enum members (ContractStatus.ACTIVE) and use the value as a string directly — no extra conversion needed.
Optional Fields and the Null Instruction Pattern
Not every field can always be extracted. The key pattern for Optional fields is the explicit null instruction in the description.
class PersonalInfo(BaseModel):
full_name: str = Field(description="Full name")
email: Optional[str] = Field(
default=None,
description="Email address. Set to null if not present in the document"
)
phone: Optional[str] = Field(
default=None,
description="Phone number with hyphens. Set to null if not present"
)
department: Optional[str] = Field(
default=None,
description="Department or team name. Set to null for individuals"
)"Set to null if not present" in the description is not optional — it's load-bearing. Without it, models will frequently hallucinate plausible values for missing fields rather than returning null.
Production Pitfalls You Will Encounter
After running Structured Output pipelines in production across several projects, three failure modes come up repeatedly.
Pitfall 1: response.parsed Returns None
response.parsed returns None most often when the input text contains no information corresponding to the schema, or when the schema is too complex for constrained decoding to apply reliably.
from pydantic import ValidationError
import logging, json
logger = logging.getLogger(__name__)
def safe_extract(document_text: str, schema_class: type[BaseModel]) -> BaseModel | None:
"""Structured Output call with fallback and proper error handling"""
try:
response = client.models.generate_content(
model="gemini-2.5-flash",
contents=f"Extract information from the following text:\n\n{document_text}",
config=types.GenerateContentConfig(
response_mime_type="application/json",
response_schema=schema_class,
),
)
if response.parsed is not None:
return response.parsed
# Fallback: attempt manual parse from response.text
if response.text:
try:
raw_data = json.loads(response.text)
return schema_class.model_validate(raw_data)
except (json.JSONDecodeError, ValidationError) as e:
logger.warning(f"Fallback parse failed: {e}")
logger.error("Model returned an empty response")
return None
except Exception as e:
logger.error(f"Structured extraction failed: {type(e).__name__}: {e}")
raise RuntimeError(f"Structured extraction failed: {e}") from ePitfall 2: Accuracy Drops with Deep Nesting
Schema nesting beyond three levels tends to degrade accuracy. The practical fix is to flatten the schema or use a two-phase extraction approach:
def extract_in_phases(document: str) -> tuple[BaseModel, BaseModel]:
"""Two-phase extraction to avoid deep nesting accuracy issues"""
top_level = safe_extract(document, TopLevelSchema)
if top_level is None:
raise ValueError("Phase 1 extraction failed")
# Ground Phase 2 with Phase 1 results for better accuracy
detail_prompt = (
f"Extract detailed information from the document below.\n"
f"High-level context already extracted: {top_level.model_dump_json()}\n\n"
f"Document:\n{document}"
)
detail_level = safe_extract(detail_prompt, DetailLevelSchema)
return top_level, detail_levelPitfall 3: Long Document Accuracy Degradation
Even with Gemini 2.5 Flash's large context window, accuracy drops on very long documents. Chunk splitting with overlap prevents information loss at chunk boundaries:
def extract_from_long_document(
document: str,
schema_class: type[BaseModel],
chunk_size: int = 4000,
overlap: int = 500,
) -> list[BaseModel]:
"""Split long documents into overlapping chunks for extraction"""
results = []
start = 0
while start < len(document):
end = min(start + chunk_size, len(document))
chunk = document[start:end]
try:
result = safe_extract(chunk, schema_class)
if result is not None:
results.append(result)
except RuntimeError as e:
logger.warning(f"Chunk extraction failed (start={start}): {e}")
# Overlap preserves context across chunk boundaries
start = end - overlap if end < len(document) else len(document)
return resultsThree Production-Ready Use Cases
Use Case 1: Invoice and Receipt Parsing
Embedding domain validation logic in Pydantic validators lets you catch model computation errors at ingestion time — before bad data propagates downstream.
from pydantic import field_validator
class TaxInfo(BaseModel):
tax_rate: float = Field(ge=0, le=1, description="Tax rate (e.g., 0.10 for 10%)")
tax_amount: float = Field(ge=0, description="Tax amount")
class LineItem(BaseModel):
description: str = Field(description="Item or service description")
quantity: float = Field(ge=0, description="Quantity")
unit: Optional[str] = Field(default=None, description="Unit (each, hours, etc.)")
unit_price: float = Field(ge=0, description="Unit price excluding tax")
subtotal: float = Field(ge=0, description="Subtotal (quantity × unit_price)")
class ReceiptData(BaseModel):
document_type: Literal["invoice", "receipt", "estimate"] = Field(
description="Document type: invoice, receipt, or estimate"
)
issue_date: date = Field(description="Issue date (YYYY-MM-DD format)")
vendor_name: str = Field(description="Name of the vendor or seller")
client_name: Optional[str] = Field(default=None, description="Client name. Null if absent")
items: list[LineItem] = Field(description="Line item list")
subtotal: float = Field(ge=0, description="Pre-tax subtotal")
tax: TaxInfo = Field(description="Tax breakdown")
total: float = Field(ge=0, description="Total including tax")
@field_validator("total")
@classmethod
def validate_total_consistency(cls, v: float, info) -> float:
"""Catch model arithmetic errors at validation time (1 unit rounding tolerance)"""
data = info.data
if "subtotal" in data and "tax" in data and data["tax"] is not None:
expected = data["subtotal"] + data["tax"].tax_amount
if abs(v - expected) > 1.0:
logger.warning(
f"Total inconsistency: total={v}, expected={expected:.2f}"
)
return vUse Case 2: Automated Code Review Scoring
class CodeIssue(BaseModel):
severity: Literal["error", "warning", "info"] = Field(
description="Severity: error=must fix, warning=recommended, info=informational"
)
category: Literal["security", "performance", "maintainability", "correctness", "style"]
line_range: Optional[str] = Field(default=None, description="Line range (e.g., '12-15')")
description: str = Field(max_length=200, description="Issue description (100 words max)")
suggestion: str = Field(max_length=300, description="Improvement suggestion with code if applicable")
class CodeReviewResult(BaseModel):
overall_score: int = Field(ge=0, le=100, description="Quality score (0-100)")
issues: list[CodeIssue] = Field(description="Detected issues (max 10)")
strengths: list[str] = Field(description="Code strengths (max 3)")
summary: str = Field(max_length=300, description="Overall summary (100 words max)")
def review_code(code: str, language: str = "python") -> CodeReviewResult:
response = client.models.generate_content(
model="gemini-2.5-pro", # Pro for better security vulnerability detection
contents=(
f"Review this {language} code for security, performance, readability, "
f"and correctness. Return a detailed review with improvement suggestions.\n\n"
f"```{language}\n{code}\n```"
),
config=types.GenerateContentConfig(
response_mime_type="application/json",
response_schema=CodeReviewResult,
),
)
if response.parsed is None:
raise ValueError("Code review structured output was not returned")
return response.parsedUse Case 3: Multilingual Document Summarization
class DocumentSection(BaseModel):
heading: str = Field(description="Section heading")
key_points: list[str] = Field(description="Key points for this section (max 3)")
class DocumentSummary(BaseModel):
title: str = Field(description="Document title or main subject")
document_language: str = Field(description="Source language (e.g., ja, en, zh)")
document_type: Literal["technical", "legal", "business", "academic", "news", "other"]
sections: list[DocumentSection] = Field(description="Major sections (max 5)")
action_items: list[str] = Field(
description="Action items arising from this document (max 5). Empty list if none"
)
one_line_summary: str = Field(
max_length=150,
description="Single-sentence summary (50 words or fewer)"
)The document_language field enables automatic routing in multilingual pipelines — you can branch translation or downstream analysis logic based on detected language without a separate classification step.
Retry Logic and Production Resilience
No matter how well your schema is designed, API calls will fail. Rate limits will be hit. Building retry logic from the start is far less painful than retrofitting it after the first production incident.
import time
def extract_with_retry(
document_text: str,
schema_class: type[BaseModel],
model: str = "gemini-2.5-flash",
max_retries: int = 3,
initial_delay: float = 1.0,
) -> BaseModel | None:
"""Structured Output extraction with exponential backoff"""
delay = initial_delay
last_error = None
for attempt in range(max_retries):
try:
response = client.models.generate_content(
model=model,
contents=f"Extract information from:\n\n{document_text}",
config=types.GenerateContentConfig(
response_mime_type="application/json",
response_schema=schema_class,
),
)
return response.parsed
except Exception as e:
last_error = e
error_str = str(e)
if "RESOURCE_EXHAUSTED" in error_str or "429" in error_str:
# Rate limits need aggressive backoff
wait_time = min(delay * 4, 120.0)
logger.warning(f"Rate limit — waiting {wait_time:.1f}s (attempt {attempt + 1})")
time.sleep(wait_time)
elif "503" in error_str or "UNAVAILABLE" in error_str:
logger.warning(f"Service unavailable — retrying in {delay:.1f}s")
time.sleep(delay)
delay = min(delay * 2, 30.0)
else:
# Non-retryable errors should fail fast
raise
raise RuntimeError(f"Failed after {max_retries} attempts: {last_error}") from last_errorThe distinction between rate limit errors and general failures matters: rate limit exhaustion needs a 4x delay multiplier capped at 120 seconds; service unavailability uses a moderate exponential backoff. Non-retryable errors (invalid API keys, malformed requests) should raise immediately rather than burning through retry budget.
Async Processing for High-Throughput Pipelines
When processing large document volumes, synchronous extraction becomes a bottleneck. The Gen AI SDK supports async operations, and combining them with asyncio.gather() and a semaphore gives you controlled concurrency:
import asyncio
async_client = genai.AsyncClient(api_key=os.getenv("GEMINI_API_KEY"))
async def extract_async(
document_text: str,
schema_class: type[BaseModel],
semaphore: asyncio.Semaphore,
) -> BaseModel | None:
"""Async Structured Output extraction with concurrency control"""
async with semaphore:
try:
response = await async_client.aio.models.generate_content(
model="gemini-2.5-flash",
contents=f"Extract information from:\n\n{document_text}",
config=types.GenerateContentConfig(
response_mime_type="application/json",
response_schema=schema_class,
),
)
return response.parsed
except Exception as e:
logger.error(f"Async extraction failed: {e}")
return None
async def process_document_batch(
documents: list[str],
schema_class: type[BaseModel],
max_concurrent: int = 5,
) -> list[BaseModel | None]:
"""Process a batch concurrently with a concurrency cap"""
semaphore = asyncio.Semaphore(max_concurrent)
tasks = [extract_async(doc, schema_class, semaphore) for doc in documents]
return list(await asyncio.gather(*tasks, return_exceptions=False))Start with max_concurrent=5 and increase gradually based on observed rate limit errors. For Gemini 2.5 Flash, 10–20 concurrent requests is typically achievable depending on your quota tier.
Cost Optimization and Model Selection
Model selection directly impacts cost at scale. Here's the decision framework I've settled on through production experience:
Gemini 2.5 Flash works well for simple schemas (10 or fewer fields, 2 levels of nesting max). Invoice parsing, form extraction, and basic document classification sit comfortably in Flash territory.
Gemini 2.5 Pro is warranted for complex schemas, ambiguous text, and high-judgment tasks like code review. The accuracy difference on difficult inputs isn't marginal.
Pre-filtering with a lightweight screen cuts costs significantly on mixed-quality document batches:
class DocumentRelevance(BaseModel):
is_relevant: bool = Field(
description="Whether the document contains data extractable by the target schema"
)
confidence: float = Field(ge=0, le=1, description="Confidence (0.0-1.0)")
def filter_and_extract(
document: str,
target_schema: type[BaseModel],
confidence_threshold: float = 0.7,
) -> BaseModel | None:
"""Two-stage processing: screen with Flash, extract only relevant documents"""
screen_response = client.models.generate_content(
model="gemini-2.5-flash",
contents=(
f"Does this document contain '{target_schema.__name__}' data? "
f"First 1000 characters:\n\n{document[:1000]}"
),
config=types.GenerateContentConfig(
response_mime_type="application/json",
response_schema=DocumentRelevance,
),
)
relevance = screen_response.parsed
if relevance is None or not relevance.is_relevant:
return None
if relevance.confidence < confidence_threshold:
return None
return safe_extract(document, target_schema)In high-volume pipelines, this filtering alone can reduce API costs by 30–50% when a meaningful fraction of incoming documents are irrelevant.
Development Workflow in Antigravity
Pydantic's explicit type annotations make Antigravity's AI completion significantly more precise for schema development. The IDE can suggest correct field names, types, and Field(description=...) content when it understands the surrounding schema context.
The most valuable testing habit is schema-only validation before calling the model. Running schema_class.model_validate(test_json) catches schema design errors without any API calls:
import pytest
from pydantic import ValidationError
class TestInvoiceSchema:
def test_valid_invoice_passes(self):
data = {
"invoice_number": "INV-2026-001",
"vendor_name": "Acme Corp",
"total_amount": 1100.0,
"items": [{"name": "Design work", "quantity": 1, "unit_price": 1000.0}],
"currency": "USD"
}
invoice = Invoice.model_validate(data)
assert invoice.currency == "USD"
def test_zero_quantity_fails(self):
"""Quantity of zero should raise ValidationError"""
with pytest.raises(ValidationError) as exc_info:
InvoiceItem(name="Test item", quantity=0, unit_price=100.0)
assert "quantity" in str(exc_info.value)
def test_optional_fields_default_none(self):
"""Optional fields should default to None when omitted"""
info = PersonalInfo(full_name="Jane Smith")
assert info.email is None
assert info.phone is NoneFor deeper coverage of testing strategies across agent pipelines, the LangGraph Stateful Agent Design Guide covers end-to-end testing patterns that complement what's shown here. For foundational Gen AI SDK setup, the Google Gen AI SDK Python Production Guide covers environment configuration in detail.
The First Step Worth Taking Today
Of everything in this guide, the single change most worth making to an existing project today is the safe_extract function. Find one place in your codebase where you're accessing response.text and parsing it manually, and replace it with safe_extract. That one change eliminates the most common category of silent production failure: unhandled null responses and malformed JSON propagating into downstream systems.
Structured Output isn't about forcing the model to comply with a format. It's about establishing a contract between your type system and the model — defining the exact shape of data that can flow between them. Once you think about it that way, combining it with Pydantic becomes less of a technique and more of a natural architectural decision. The schema is the interface. The model is the implementation.