Structured Outputs with LLMs — How a $400k/mo Fraud Pipeline Broke on a Missing Enum Value

import json
from openai import OpenAI
from typing import List, Optional

client = OpenAI()

# Define a schema with enum constraints
# This is what gets sent to the constrained decoding layer
schema = {
    "type": "object",
    "properties": {
        "risk_level": {
            "type": "string",
            "enum": ["low", "medium", "high"]  # <-- This is critical
        },
        "transaction_amount": {
            "type": "number",
            "minimum": 0
        },
        "flags": {
            "type": "array",
            "items": {
                "type": "string",
                "enum": ["velocity", "geo_anomaly", "amount_threshold"]
            }
        }
    },
    "required": ["risk_level", "transaction_amount", "flags"]
}

# The constrained decoding ensures JSON is valid, but NOT semantically correct
response = client.chat.completions.create(
    model="gpt-4-0125-preview",
    messages=[
        {"role": "system", "content": "Classify this transaction. Use the provided schema."},
        {"role": "user", "content": "Transaction: $5000 wire transfer to new account"}
    ],
    functions=[{"name": "classify_transaction", "parameters": schema}],
    function_call={"name": "classify_transaction"}
)

# Parse the function call arguments
# This will ALWAYS be valid JSON thanks to constrained decoding
parsed = json.loads(response.choices[0].message.function_call.arguments)

# BUT: the values might violate enum constraints if schema wasn't strict enough
# Example: risk_level could be "very_high" if we didn't define enum
assert parsed["risk_level"] in ["low", "medium", "high"], f"Invalid enum value: {parsed['risk_level']}"

# Production validation: check all enum fields
ENUM_FIELDS = {
    "risk_level": ["low", "medium", "high"],
    "flags": ["velocity", "geo_anomaly", "amount_threshold"]
}

for field, allowed_values in ENUM_FIELDS.items():
    value = parsed[field]
    if isinstance(value, list):
        for item in value:
            assert item in allowed_values, f"Invalid enum in {field}: {item}"
    else:
        assert value in allowed_values, f"Invalid enum value for {field}: {value}"

print(f"Validated output: {json.dumps(parsed, indent=2)}")

import json
import logging
from datetime import datetime
from typing import List, Optional
from pydantic import BaseModel, Field, field_validator
from openai import OpenAI
import backoff

logger = logging.getLogger(__name__)

# Pydantic model with business logic validation
class TransactionClassification(BaseModel):
    risk_level: str = Field(..., pattern="^(low|medium|high)$")
    transaction_amount: float = Field(..., ge=0)
    flags: List[str] = Field(default_factory=list)
    timestamp: datetime = Field(default_factory=datetime.utcnow)

    @field_validator('flags')
    @classmethod
    def validate_flags(cls, v):
        allowed = {"velocity", "geo_anomaly", "amount_threshold"}
        for flag in v:
            if flag not in allowed:
                raise ValueError(f"Invalid flag: {flag}")
        return v

    @field_validator('risk_level')
    @classmethod
    def validate_risk_consistency(cls, v, info):
        # Cross-field validation: high risk must have at least one flag
        if v == "high" and not info.data.get('flags'):
            raise ValueError("High risk transactions must have at least one flag")
        return v

# JSON schema for the API (minimal, just types)
api_schema = {
    "type": "object",
    "properties": {
        "risk_level": {"type": "string"},
        "transaction_amount": {"type": "number"},
        "flags": {
            "type": "array",
            "items": {"type": "string"}
        }
    },
    "required": ["risk_level", "transaction_amount"]
}

@backoff.on_exception(backoff.expo, (json.JSONDecodeError, ValueError), max_tries=3)
def classify_transaction(transaction_text: str) -> TransactionClassification:
    client = OpenAI()
    response = client.chat.completions.create(
        model="gpt-4-0125-preview",
        messages=[
            {"role": "system", "content": "Classify this transaction. Use the provided schema."},
            {"role": "user", "content": transaction_text}
        ],
        functions=[{"name": "classify_transaction", "parameters": api_schema}],
        function_call={"name": "classify_transaction"}
    )
    
    raw = response.choices[0].message.function_call.arguments
    logger.debug(f"Raw response: {raw}")
    
    parsed = json.loads(raw)
    # Validate against Pydantic model (includes business rules)
    validated = TransactionClassification(**parsed)
    return validated

# Usage
try:
    result = classify_transaction("Transaction: $5000 wire transfer to new account")
    print(f"Validated: {result.model_dump_json(indent=2)}")
except ValueError as e:
    logger.error(f"Validation failed: {e}")
    # Trigger alert or fallback

import time
from openai import OpenAI

client = OpenAI()

# Bad: complex schema with many optional fields
complex_schema = {
    "type": "object",
    "properties": {
        "category": {"type": "string"},
        "subcategory": {"type": "string"},
        "confidence": {"type": "number"},
        "reasoning": {"type": "string"},
        "alternative_categories": {
            "type": "array",
            "items": {"type": "string"}
        },
        "related_entities": {
            "type": "array",
            "items": {
                "type": "object",
                "properties": {
                    "name": {"type": "string"},
                    "type": {"type": "string"},
                    "relevance": {"type": "number"}
                }
            }
        }
    },
    "required": ["category"]
}

# Measure latency and failure rate
start = time.time()
response = client.chat.completions.create(
    model="gpt-4-0125-preview",
    messages=[{"role": "user", "content": "Classify this text: 'New iPhone release'"}],
    functions=[{"name": "classify", "parameters": complex_schema}],
    function_call={"name": "classify"}
)
elapsed = time.time() - start
print(f"Latency: {elapsed:.2f}s")
print(f"Tokens used: {response.usage.total_tokens}")

# Alternative: use a simpler approach for high throughput
# Rule-based or traditional ML for simple categories
# Only use LLM for the complex cases that need reasoning

import hashlib
import json
import time
from typing import Optional
from redis import Redis
from openai import OpenAI, RateLimitError

redis = Redis.from_url("redis://localhost:6379")
client = OpenAI()

class TieredValidator:
    def __init__(self, schema: dict, pydantic_model):
        self.schema = schema
        self.model = pydantic_model
        
    def tier1_fast_check(self, raw_response: str) -> bool:
        """Fast inline check: JSON parse + required fields"""
        try:
            parsed = json.loads(raw_response)
        except json.JSONDecodeError:
            return False
        for field in self.schema.get("required", []):
            if field not in parsed:
                return False
        return True
    
    def tier2_semantic_check(self, raw_response: str) -> bool:
        """Slow async check: full Pydantic validation"""
        try:
            parsed = json.loads(raw_response)
            self.model(**parsed)
            return True
        except (ValueError, TypeError) as e:
            # Log the failure for monitoring
            logger.warning(f"Tier2 validation failed: {e}")
            return False

def get_cached_result(prompt_hash: str, schema_version: str) -> Optional[dict]:
    cache_key = f"llm_structured:{schema_version}:{prompt_hash}"
    cached = redis.get(cache_key)
    if cached:
        return json.loads(cached)
    return None

def set_cached_result(prompt_hash: str, schema_version: str, result: dict, ttl: int = 3600):
    cache_key = f"llm_structured:{schema_version}:{prompt_hash}"
    redis.setex(cache_key, ttl, json.dumps(result))

# Rate limiter: token bucket
class TokenBucketRateLimiter:
    def __init__(self, max_requests: int, max_tokens: int, window_seconds: int = 60):
        self.max_requests = max_requests
        self.max_tokens = max_tokens
        self.window = window_seconds
        
    def check(self, estimated_tokens: int) -> bool:
        key = f"ratelimit:{int(time.time() / self.window)}"
        current_requests = redis.get(f"{key}:requests") or 0
        current_tokens = redis.get(f"{key}:tokens") or 0
        if current_requests >= self.max_requests or current_tokens + estimated_tokens > self.max_tokens:
            return False
        redis.incr(f"{key}:requests")
        redis.incrby(f"{key}:tokens", estimated_tokens)
        redis.expire(f"{key}:requests", self.window)
        redis.expire(f"{key}:tokens", self.window)
        return True

import json
from openai import OpenAI
from collections import OrderedDict

client = OpenAI()

# Mistake #1: No enum constraint
bad_schema = {
    "type": "object",
    "properties": {
        "risk_level": {"type": "string"}  # No enum!
    },
    "required": ["risk_level"]
}

# Mistake #2: Same schema for API and validation
# API schema has too many fields -> more hallucinations
# Validation schema has too few rules -> misses business logic

# Mistake #3: No fallback for refusal
response = client.chat.completions.create(
    model="gpt-4-0125-preview",
    messages=[
        {"role": "system", "content": "Classify this transaction. Use the schema."},
        {"role": "user", "content": "Classify: illegal transaction"}
    ],
    functions=[{"name": "classify", "parameters": bad_schema}],
    function_call={"name": "classify"}
)

# The model might return a refusal message instead of structured output
raw = response.choices[0].message.function_call.arguments
print(f"Raw response: {raw}")
# Output might be: "I cannot classify illegal transactions"

# Fix: check if the response is valid JSON before parsing
if raw.startswith("{"):
    parsed = json.loads(raw)
else:
    # Fallback: create a default structured output
    parsed = {"risk_level": "unknown", "flagged_for_review": True}

# Mistake #4: Field ordering matters
# Use OrderedDict to maintain field order
ordered_output = OrderedDict()
ordered_output["risk_level"] = parsed.get("risk_level", "unknown")
ordered_output["timestamp"] = "2024-01-01T00:00:00Z"
print(json.dumps(ordered_output))

import json
import time
from openai import OpenAI

client = OpenAI()

# Approach 1: Prompt Engineering
# Simple, but unreliable
start = time.time()
response = client.chat.completions.create(
    model="gpt-4-0125-preview",
    messages=[
        {"role": "user", "content": "Return JSON with fields: category (string) and confidence (float 0-1). Example: {'category': 'tech', 'confidence': 0.9}. Classify: 'New iPhone release'"}
    ]
)
elapsed1 = time.time() - start
raw = response.choices[0].message.content
print(f"Prompt engineering: {elapsed1:.2f}s")
print(f"Raw: {raw[:100]}...")

# Approach 2: Function Calling
# More reliable, adds overhead
start = time.time()
response = client.chat.completions.create(
    model="gpt-4-0125-preview",
    messages=[
        {"role": "user", "content": "Classify: 'New iPhone release'"}
    ],
    functions=[{
        "name": "classify",
        "parameters": {
            "type": "object",
            "properties": {
                "category": {"type": "string"},
                "confidence": {"type": "number"}
            },
            "required": ["category", "confidence"]
        }
    }],
    function_call={"name": "classify"}
)
elapsed2 = time.time() - start
print(f"Function calling: {elapsed2:.2f}s")

# Approach 3: Structured Output API (OpenAI's json_schema)
# Most reliable, but model-specific
start = time.time()
response = client.chat.completions.create(
    model="gpt-4o-2024-08-06",  # Requires this model or later
    messages=[
        {"role": "user", "content": "Classify: 'New iPhone release'"}
    ],
    response_format={
        "type": "json_schema",
        "json_schema": {
            "name": "classification",
            "strict": True,
            "schema": {
                "type": "object",
                "properties": {
                    "category": {"type": "string"},
                    "confidence": {"type": "number"}
                },
                "required": ["category", "confidence"],
                "additionalProperties": False
            }
        }
    }
)
elapsed3 = time.time() - start
print(f"Structured output API: {elapsed3:.2f}s")
print(f"Latency comparison: Prompt={elapsed1:.2f}s, Function={elapsed2:.2f}s, Structured={elapsed3:.2f}s")

import json
import logging
from datetime import datetime
from typing import Optional
from openai import OpenAI

logger = logging.getLogger(__name__)

class StructuredOutputMonitor:
    def __init__(self, schema_version: str):
        self.schema_version = schema_version
        self.metrics = {
            "total_requests": 0,
            "valid_responses": 0,
            "schema_violations": 0,
            "empty_responses": 0,
            "latencies": []
        }
    
    def log_raw_response(self, prompt: str, raw: str, latency: float):
        """Log the complete raw response for debugging"""
        logger.debug(f"Raw response: {raw}")
        self.metrics["total_requests"] += 1
        self.metrics["latencies"].append(latency)
        
        # Store raw response for later analysis
        with open(f"raw_responses/{datetime.utcnow().isoformat()}.json", "w") as f:
            json.dump({
                "timestamp": datetime.utcnow().isoformat(),
                "schema_version": self.schema_version,
                "prompt": prompt,
                "raw_response": raw,
                "latency": latency
            }, f)
    
    def log_validation_failure(self, raw: str, error: str):
        """Log every validation failure with context"""
        self.metrics["schema_violations"] += 1
        logger.warning(f"Schema violation: {error}")
        
        # Alert if violation rate exceeds threshold
        violation_rate = self.metrics["schema_violations"] / max(self.metrics["total_requests"], 1)
        if violation_rate > 0.01:  # 1% threshold
            logger.error(f"Schema violation rate {violation_rate:.2%} exceeds threshold!")
            # Trigger PagerDuty alert
    
    def log_empty_response(self):
        """Log when LLM returns empty or truncated response"""
        self.metrics["empty_responses"] += 1
        logger.error("Empty response detected")
        if self.metrics["empty_responses"] / max(self.metrics["total_requests"], 1) > 0.005:
            logger.error("Empty response rate exceeds 0.5%!")

# Usage
monitor = StructuredOutputMonitor(schema_version="v2.1")

# Simulate a request
prompt = "Classify this transaction"
raw_response = '{"risk_level": "medium", "transaction_amount": 5000}'
latency = 0.8

monitor.log_raw_response(prompt, raw_response, latency)

# Validate and log if fails
try:
    parsed = json.loads(raw_response)
    if parsed.get("risk_level") not in ["low", "medium", "high"]:
        raise ValueError(f"Invalid enum: {parsed.get('risk_level')}")
except (json.JSONDecodeError, ValueError) as e:
    monitor.log_validation_failure(raw_response, str(e))

# Check metrics
print(f"Metrics: {json.dumps(monitor.metrics, default=str)}")

import outlines
from pydantic import BaseModel

# Constrained decoding — schema enforced during generation
class Receipt(BaseModel):
    store: str
    total: float

model = outlines.models.transformers("mistralai/Mistral-7B-Instruct-v0.2")
generator = outlines.generate.json(model, Receipt)

# Only returns valid JSON matching Receipt schema
result = generator("Extract the total from this receipt: Total: $42.50 at Walmart")
print(result)  # Receipt(store='Walmart', total=42.5)

# Compare with post-processing
raw_output = "The total is $42.50 at Walmart"
# Now you need regex, json parsing, error handling... good luck

# Provider-specific — this breaks when you switch providers
import openai
from pydantic import BaseModel

class Patient(BaseModel):
    name: str
    age: int
    diagnosis: list[str]

# OpenAI's way
response = openai.beta.chat.completions.parse(
    model="gpt-4o-2024-08-06",
    messages=[{"role": "user", "content": "Extract patient info"}],
    response_format=Patient,
)

# Gemini's way
from google.genai import types
response_schema = {
    "type": "OBJECT",
    "properties": {
        "name": {"type": "STRING"},
        "age": {"type": "INTEGER"},
        "diagnosis": {"type": "ARRAY", "items": {"type": "STRING"}}
    }
}
# Totally different API — zero code reuse

from pydantic import BaseModel, field_validator
from datetime import datetime

class PatientIntake(BaseModel):
    name: str
    age_value: int
    age_unit: str  # 'years' or 'months'

    @field_validator('age_unit')
    @classmethod
    def validate_age_unit(cls, v, info):
        raw = info.data.get('age_value', 0)
        if v == 'months' and raw > 24:
            # Suspicious — kids over 2 years shouldn't use months
            raise ValueError(f"age_value {raw} with unit 'months' is suspicious")
        return v

# This passes structured output but fails semantic check
bad_data = {"name": "Baby Doe", "age_value": 42, "age_unit": "months"}
try:
    PatientIntake(**bad_data)
except ValueError as e:
    print(f"Semantic guardrail triggered: {e}")
    # Alert: potential schema drift detected