Agent Structured Output

Why This Matters

Structured output transforms free-form LLM responses into predictable, machine-readable formats. This is essential for integrating agents with downstream systems, databases, and APIs. Without structured output, agents produce unparseable text that breaks automation pipelines and requires expensive post-processing.

Real-World Analogy

Structured output is like filling out a standardized form versus writing a letter. A form (schema) has specific fields, required data types, and validation rules. When you receive a completed form, you know exactly where to find each piece of information. Free-form text is like a letter—you might get the information, but you have to read and interpret everything manually.

Structured Output Architecture

JSON Schema Validation

import json
import re
import logging
from dataclasses import dataclass, field
from typing import Any, Optional, get_type_hints
from enum import Enum
import asyncio

logger = logging.getLogger(__name__)

class SchemaType(Enum):
    STRING = "string"
    NUMBER = "number"
    INTEGER = "integer"
    BOOLEAN = "boolean"
    ARRAY = "array"
    OBJECT = "object"
    NULL = "null"

@dataclass
class SchemaField:
    name: str
    type: SchemaType
    required: bool = True
    default: Any = None
    description: str = ""
    min_length: Optional[int] = None
    max_length: Optional[int] = None
    pattern: Optional[str] = None
    enum: Optional[list] = None
    items: Optional["SchemaField"] = None
    properties: Optional[list["SchemaField"]] = None

class JSONSchemaValidator:
    def __init__(self):
        self.schemas: dict[str, list[SchemaField]] = {}

    def register_schema(self, name: str, fields: list[SchemaField]):
        self.schemas[name] = fields

    def validate(self, data: dict, schema_name: str) -> tuple[bool, list[str]]:
        schema = self.schemas.get(schema_name)
        if not schema:
            return False, [f"Schema '{schema_name}' not found"]

        errors = []
        for field_def in schema:
            if field_def.required and field_def.name not in data:
                errors.append(f"Missing required field: {field_def.name}")
                continue
            
            if field_def.name in data:
                value = data[field_def.name]
                field_errors = self._validate_field(value, field_def)
                errors.extend(field_errors)

        return len(errors) == 0, errors

    def _validate_field(self, value: Any, field_def: SchemaField) -> list[str]:
        errors = []
        
        if value is None and not field_def.required:
            return errors
        
        type_errors = self._validate_type(value, field_def.type)
        errors.extend(type_errors)
        
        if not type_errors and field_def.type == SchemaType.STRING:
            if field_def.min_length and len(str(value)) < field_def.min_length:
                errors.append(f"{field_def.name}: length {len(str(value))} < {field_def.min_length}")
            if field_def.max_length and len(str(value)) > field_def.max_length:
                errors.append(f"{field_def.name}: length {len(str(value))} > {field_def.max_length}")
            if field_def.pattern and not re.match(field_def.pattern, str(value)):
                errors.append(f"{field_def.name}: doesn't match pattern {field_def.pattern}")
        
        if not type_errors and field_def.enum and value not in field_def.enum:
            errors.append(f"{field_def.name}: value '{value}' not in enum {field_def.enum}")
        
        if not type_errors and field_def.type == SchemaType.ARRAY and field_def.items:
            if not isinstance(value, list):
                errors.append(f"{field_def.name}: expected array")
            else:
                for i, item in enumerate(value):
                    item_errors = self._validate_field(item, field_def.items)
                    for err in item_errors:
                        errors.append(f"{field_def.name}[{i}]: {err}")
        
        return errors

    def _validate_type(self, value: Any, expected_type: SchemaType) -> list[str]:
        type_map = {
            SchemaType.STRING: str,
            SchemaType.NUMBER: (int, float),
            SchemaType.INTEGER: int,
            SchemaType.BOOLEAN: bool,
            SchemaType.ARRAY: list,
            SchemaType.OBJECT: dict,
        }
        
        if expected_type == SchemaType.NULL:
            if value is not None:
                return [f"Expected null, got {type(value).__name__}"]
            return []
        
        expected = type_map.get(expected_type)
        if expected and not isinstance(value, expected):
            return [f"Expected {expected_type.value}, got {type(value).__name__}"]
        
        return []

Function Calling System

import json
from dataclasses import dataclass, field
from typing import Any, Callable, Optional
import inspect
import asyncio
import logging

logger = logging.getLogger(__name__)

@dataclass
class FunctionParameter:
    name: str
    type: str
    description: str = ""
    required: bool = True
    default: Any = None
    enum: Optional[list] = None

@dataclass
class FunctionDefinition:
    name: str
    description: str
    parameters: list[FunctionParameter]
    handler: Callable = None

class FunctionCallingSystem:
    def __init__(self):
        self.functions: dict[str, FunctionDefinition] = {}

    def register_function(
        self,
        name: str,
        description: str,
        parameters: list[FunctionParameter],
        handler: Callable,
    ):
        self.functions[name] = FunctionDefinition(
            name=name,
            description=description,
            parameters=parameters,
            handler=handler,
        )

    def get_function_schemas(self) -> list[dict]:
        schemas = []
        for func in self.functions.values():
            properties = {}
            required = []
            for param in func.parameters:
                properties[param.name] = {
                    "type": param.type,
                    "description": param.description,
                }
                if param.enum:
                    properties[param.name]["enum"] = param.enum
                if param.required:
                    required.append(param.name)
            
            schemas.append({
                "type": "function",
                "function": {
                    "name": func.name,
                    "description": func.description,
                    "parameters": {
                        "type": "object",
                        "properties": properties,
                        "required": required,
                    },
                },
            })
        return schemas

    async def execute_function(
        self,
        function_name: str,
        arguments: dict,
    ) -> Any:
        func_def = self.functions.get(function_name)
        if not func_def:
            raise ValueError(f"Function '{function_name}' not found")
        
        if not func_def.handler:
            raise ValueError(f"Handler not registered for '{function_name}'")
        
        validated_args = self._validate_arguments(func_def, arguments)
        return await func_def.handler(**validated_args)

    def _validate_arguments(
        self,
        func_def: FunctionDefinition,
        arguments: dict,
    ) -> dict:
        validated = {}
        for param in func_def.parameters:
            if param.name in arguments:
                validated[param.name] = arguments[param.name]
            elif param.required:
                raise ValueError(f"Missing required parameter: {param.name}")
            elif param.default is not None:
                validated[param.name] = param.default
        return validated

    def format_for_llm(self) -> list[dict]:
        return self.get_function_schemas()

Pydantic Integration

from pydantic import BaseModel, Field, validator
from typing import Optional, List, Any
from enum import Enum
import json

class ResponseStatus(str, Enum):
    SUCCESS = "success"
    ERROR = "error"
    PARTIAL = "partial"

class AgentResponse(BaseModel):
    status: ResponseStatus
    content: str
    metadata: dict = Field(default_factory=dict)
    confidence: float = Field(ge=0.0, le=1.0, default=1.0)
    sources: List[str] = Field(default_factory=list)
    
    @validator("content")
    def content_not_empty(cls, v):
        if not v.strip():
            raise ValueError("Content cannot be empty")
        return v

class ToolCall(BaseModel):
    tool_name: str
    arguments: dict = Field(default_factory=dict)
    result: Optional[Any] = None
    success: bool = True
    error: Optional[str] = None

class StructuredAgentOutput(BaseModel):
    response: AgentResponse
    tool_calls: List[ToolCall] = Field(default_factory=list)
    reasoning: Optional[str] = None
    next_action: Optional[str] = None
    
    class Config:
        json_schema_extra = {
            "example": {
                "response": {
                    "status": "success",
                    "content": "Here is the analysis...",
                    "confidence": 0.95,
                },
                "tool_calls": [],
                "reasoning": "The user asked for analysis...",
            }
        }

class SchemaGenerator:
    @staticmethod
    def from_pydantic(model: type[BaseModel]) -> dict:
        return model.model_json_schema()
    
    @staticmethod
    def from_dict(data: dict) -> dict:
        return SchemaGenerator._infer_schema(data)
    
    @staticmethod
    def _infer_schema(data: Any) -> dict:
        if isinstance(data, dict):
            properties = {}
            for key, value in data.items():
                properties[key] = SchemaGenerator._infer_schema(value)
            return {"type": "object", "properties": properties}
        elif isinstance(data, list):
            if data:
                return {"type": "array", "items": SchemaGenerator._infer_schema(data[0])}
            return {"type": "array"}
        elif isinstance(data, str):
            return {"type": "string"}
        elif isinstance(data, bool):
            return {"type": "boolean"}
        elif isinstance(data, int):
            return {"type": "integer"}
        elif isinstance(data, float):
            return {"type": "number"}
        else:
            return {"type": "string"}

Structured Output Generator

import json
import asyncio
from dataclasses import dataclass
from typing import Any, Callable, Optional
from enum import Enum
import logging

logger = logging.getLogger(__name__)

class OutputFormat(Enum):
    JSON = "json"
    FUNCTION_CALL = "function_call"
    CONSTRAINED = "constrained"

@dataclass
class StructuredOutputConfig:
    format: OutputFormat = OutputFormat.JSON
    schema: dict = None
    retry_count: int = 3
    retry_delay: float = 1.0
    temperature: float = 0.0

class StructuredOutputGenerator:
    def __init__(self, llm_client=None, validator=None):
        self.llm_client = llm_client
        self.validator = validator or JSONSchemaValidator()
        self.generation_history: list[dict] = []

    async def generate(
        self,
        prompt: str,
        schema: dict = None,
        config: StructuredOutputConfig = None,
    ) -> tuple[Any, bool]:
        config = config or StructuredOutputConfig()
        schema = schema or config.schema

        for attempt in range(config.retry_count):
            try:
                if config.format == OutputFormat.JSON:
                    result = await self._generate_json(prompt, schema, config)
                elif config.format == OutputFormat.FUNCTION_CALL:
                    result = await self._generate_function_call(prompt, schema, config)
                else:
                    result = await self._generate_constrained(prompt, schema, config)

                if schema:
                    is_valid, errors = self.validator.validate(result, schema)
                    if is_valid:
                        self._log_generation(prompt, result, True, attempt)
                        return result, True
                    else:
                        prompt = f"{prompt}\n\nPrevious attempt had errors: {errors}\nPlease fix these issues."
                        self._log_generation(prompt, result, False, attempt)
                else:
                    return result, True

            except Exception as e:
                self._log_generation(prompt, None, False, attempt, str(e))
                if attempt < config.retry_count - 1:
                    await asyncio.sleep(config.retry_delay * (attempt + 1))

        return None, False

    async def _generate_json(
        self,
        prompt: str,
        schema: dict,
        config: StructuredOutputConfig,
    ) -> dict:
        schema_prompt = f"{prompt}\n\nRespond with valid JSON matching this schema:\n{json.dumps(schema, indent=2)}"
        
        if self.llm_client:
            response = await self.llm_client.complete(
                schema_prompt,
                temperature=config.temperature,
            )
            return json.loads(response)
        
        return {"status": "success", "content": "Generated response"}

    async def _generate_function_call(
        self,
        prompt: str,
        schema: dict,
        config: StructuredOutputConfig,
    ) -> dict:
        return await self._generate_json(prompt, schema, config)

    async def _generate_constrained(
        self,
        prompt: str,
        schema: dict,
        config: StructuredOutputConfig,
    ) -> dict:
        return await self._generate_json(prompt, schema, config)

    def _log_generation(
        self,
        prompt: str,
        result: Any,
        success: bool,
        attempt: int,
        error: str = None,
    ):
        self.generation_history.append({
            "prompt_length": len(prompt),
            "result": result,
            "success": success,
            "attempt": attempt,
            "error": error,
        })

Mathematical Foundation

Schema Coverage Score:

Validation Success Rate:

Retry Efficiency:

Schema Complexity (for constrained decoding):

Where:

— Number of fields in schema
— Number of constraints on field f

Token Overhead (schema in prompt):

Information Gain (structured vs. unstructured):

Where is entropy of unstructured output and is conditional entropy given the schema constraint.

Performance Considerations

Approach	Latency	Cost	Accuracy	Best For
JSON Mode	+50-100ms	+Schema tokens	85-95%	Simple structures
Function Calling	+100-200ms	+Tool definitions	90-98%	Tool integration
Constrained Decoding	+200-500ms	+Compute	99-100%	Strict schemas
Prompt + Validation	+100-300ms	+Retry tokens	80-95%	Flexible needs

Security Considerations

Schema injection: Validate schemas before use to prevent prompt injection
Data exposure: Structured output may expose sensitive fields
Type coercion: Be cautious with automatic type conversion
Nested validation: Deep schemas can be exploited for resource exhaustion
Output sanitization: Validate and sanitize all structured outputs before use

Interview Questions

1. What is the difference between JSON mode and function calling?

Answer: JSON mode ensures the model outputs valid JSON but doesn't enforce a specific schema. Function calling uses tool definitions with parameter schemas, producing structured output matching the function signature. Key differences: JSON mode is simpler but requires post-validation; function calling provides stronger guarantees and integrates with tool systems. Function calling is better when you need specific parameters extracted; JSON mode is better for free-form structured responses. Most modern APIs support both.

2. How does constrained decoding work for structured output?

Answer: Constrained decoding restricts token generation to only those tokens that maintain valid output according to a grammar or schema: 1) Define a grammar (e.g., JSON Schema as a finite state machine), 2) At each generation step, mask invalid tokens, 3) Only sample from valid tokens. Benefits: guaranteed valid output, no retry needed, deterministic structure. Drawbacks: requires special serving infrastructure, may reduce output quality, adds latency. Libraries: Outlines, LMQL, guidance. Best for strict schemas where invalid output is unacceptable.

3. How do you handle schema evolution in structured output?

Answer: Schema evolution strategies: 1) Version schemas — Include version field, support multiple versions, 2) Backward compatibility — New fields have defaults, never remove fields, 3) Schema migration — Transform old outputs to new schema, 4) Validation layers — Accept old formats, convert to new, 5) Gradual rollout — Deploy schema changes before code changes. For agents: version the prompt with schema, maintain compatibility matrix, and implement adapter patterns for old formats. Test with contract testing between producers and consumers.

4. What are the tradeoffs between different structured output approaches?

Answer: Tradeoffs: JSON Mode — Simple, model-native, but weak schema enforcement. Function Calling — Rich parameter support, tool integration, but requires API support. Constrained Decoding — Guaranteed valid, but needs special serving. Prompt + Validation — Most flexible, works everywhere, but requires retry logic. Choose based on: schema strictness (constrained for strict, prompt for flexible), infrastructure (function calling if available), reliability needs (constrained for zero-failure), and latency requirements (constrained for no retry overhead).

5. How do you validate complex nested schemas?

Answer: Use recursive validation: 1) Define validators for each type, 2) Recursively validate nested objects and arrays, 3) Handle circular references, 4) Validate cross-field constraints, 5) Use schema composition ($ref, allOf, oneOf). Implementation: walk the schema tree, validate each node, collect all errors. Libraries: Pydantic (Python), Zod (TypeScript), JSON Schema validators. For performance: compile schemas to validators, cache validation results, and validate incrementally.

6. How would you implement retry logic for structured output failures?

Answer: Retry strategy: 1) Parse error → retry with explicit format instructions, 2) Schema error → include schema in retry prompt, 3) Type error → add type coercion examples, 4) Missing fields → prompt for completeness. Implementation: exponential backoff, different prompts for different error types, track error patterns, and adjust prompts based on failure modes. Consider: maximum retry count, cost budget, and fallback to less structured output when retries exhausted. Log all failures for prompt improvement.

7. What is the role of temperature in structured output generation?

Answer: Temperature controls randomness: 0 = deterministic (highest structure compliance), 1 = most random. For structured output: use low temperature (0-0.3) to maximize schema compliance, higher temperature reduces validity rate. Tradeoff: low temperature may produce repetitive or boring output; high temperature may violate schema. Best practice: start with temperature 0, increase only if output quality is poor. Some APIs support logit bias to further constrain output. Monitor: schema compliance rate vs. temperature.

8. How do you evaluate the quality of structured output?

Answer: Metrics: 1) Schema compliance — % of outputs matching schema, 2) Field completeness — % of required fields present, 3) Type accuracy — % of fields with correct types, 4) Value validity — % of values within constraints, 5) Semantic accuracy — Content correctness (requires human evaluation), 6) Latency — Time to generate valid output, 7) Retry rate — % requiring retries. Benchmarks: test on diverse queries, measure across difficulty levels, compare approaches. Use A/B testing to optimize prompts and validation logic.

Common Pitfalls

Pitfall	Solution
Invalid JSON output	Use JSON mode or constrained decoding
Missing required fields	Include schema in prompt, retry with specifics
Type mismatches	Add type coercion and validation
Schema too complex	Simplify or decompose schemas
High retry rate	Improve prompts with examples
Schema versioning issues	Version schemas and maintain compatibility
Nested validation failures	Use recursive validation
Performance overhead	Cache validated outputs, compile schemas

KnowledgeCheck

What is the main advantage of constrained decoding over JSON mode?
- a) Faster generation
- b) Guaranteed valid output
- c) Lower cost
- d) Simpler implementation
How does function calling differ from JSON mode?
- a) Function calling is faster
- b) Function calling uses tool definitions with parameter schemas
- c) Function calling doesn't need validation
- d) Function calling only works with GPT models
Why is low temperature recommended for structured output?
- a) To increase creativity
- b) To maximize schema compliance
- c) To reduce latency
- d) To lower costs
What should retry logic do when schema validation fails?
- a) Retry with same prompt
- b) Include schema and error details in retry prompt
- c) Increase temperature
- d) Switch to unstructured output
How should schema evolution be handled?
- a) Remove old fields immediately
- b) Maintain backward compatibility with defaults
- c) Never change schemas
- d) Deploy code before schema changes
What metric measures the percentage of outputs matching the schema?
- a) Latency
- b) Schema compliance rate
- c) Token usage
- d) Cost per request

Answers: 1-b, 2-b, 3-b, 4-b, 5-b, 6-b

Agent Structured Output: JSON Mode, Function Calling & Schema Validation