Output Parser in Machine Learning

import instructor
from openai import OpenAI
from pydantic import BaseModel, Field
from typing import List, Optional
from enum import Enum


class Sentiment(str, Enum):
    POSITIVE = "positive"
    NEGATIVE = "negative"
    NEUTRAL = "neutral"


class ProductReview(BaseModel):
    """Structured representation of a product review."""
    product_name: str = Field(description="Name of the product being reviewed")
    sentiment: Sentiment = Field(description="Overall sentiment of the review")
    rating: float = Field(ge=1.0, le=5.0, description="Rating from 1.0 to 5.0")
    key_pros: List[str] = Field(description="List of positive aspects mentioned")
    key_cons: List[str] = Field(description="List of negative aspects mentioned")
    purchase_verified: Optional[bool] = Field(
        default=None, description="Whether the purchase is verified"
    )
    summary: str = Field(
        max_length=200, description="One-sentence summary of the review"
    )


# Patch the OpenAI client with Instructor
client = instructor.from_openai(OpenAI())

# Extract structured data from free-text review
review_text = """
Bought the boAt Rockerz 450 headphones from Flipkart during the Big Billion Days
sale for INR 999. Sound quality is surprisingly good for the price - punchy bass
and clear mids. Battery lasts about 12 hours which is decent. The padding on the
ear cups could be better though, gets uncomfortable after 2 hours. Build quality
feels a bit plasticky. Overall great value for money if you're on a budget.
"""

review = client.chat.completions.create(
    model="gpt-4o-mini",
    response_model=ProductReview,
    max_retries=3,
    messages=[
        {
            "role": "user",
            "content": f"Extract a structured review from this text:\n\n{review_text}",
        }
    ],
)

print(review.model_dump_json(indent=2))
# {
#   "product_name": "boAt Rockerz 450",
#   "sentiment": "positive",
#   "rating": 3.8,
#   "key_pros": ["Good sound quality", "Punchy bass", "12-hour battery", "Great value"],
#   "key_cons": ["Uncomfortable after 2 hours", "Plasticky build quality"],
#   "purchase_verified": null,
#   "summary": "Solid budget headphones with good sound and battery, but comfort and build could improve."
# }

from openai import OpenAI
from pydantic import BaseModel
from typing import List

client = OpenAI()


class KYCExtraction(BaseModel):
    """Extracted KYC information from an identity document."""
    document_type: str  # e.g., "Aadhaar", "PAN", "Passport"
    full_name: str
    document_number: str
    date_of_birth: str  # ISO format YYYY-MM-DD
    address: str
    is_valid: bool
    confidence_score: float
    extraction_notes: List[str]


# Using the parse method for guaranteed structured output
completion = client.beta.chat.completions.parse(
    model="gpt-4o-2024-08-06",
    messages=[
        {
            "role": "system",
            "content": "You are a KYC document parser. Extract structured information from document text. Use ISO date format.",
        },
        {
            "role": "user",
            "content": """OCR text from document:
            INCOME TAX DEPARTMENT GOVT OF INDIA
            Permanent Account Number Card
            RAJESH KUMAR SHARMA
            BQAPS1234K
            Date of Birth: 15/08/1990
            Father's Name: SURESH KUMAR SHARMA""",
        },
    ],
    response_format=KYCExtraction,
)

kyc_data = completion.choices[0].message.parsed
print(f"Document: {kyc_data.document_type}")
print(f"Name: {kyc_data.full_name}")
print(f"PAN: {kyc_data.document_number}")
print(f"Valid: {kyc_data.is_valid}")
# Document: PAN
# Name: RAJESH KUMAR SHARMA
# PAN: BQAPS1234K
# Valid: True

from langchain_core.output_parsers import PydanticOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from pydantic import BaseModel, Field
from typing import List


class RestaurantRecommendation(BaseModel):
    """Structured restaurant recommendation."""
    restaurant_name: str = Field(description="Name of the restaurant")
    cuisine_type: str = Field(description="Primary cuisine type")
    price_range_inr: str = Field(description="Price range per person in INR")
    rating: float = Field(ge=1.0, le=5.0)
    must_try_dishes: List[str] = Field(description="Top 3 dishes to try")
    vegetarian_friendly: bool
    reasoning: str = Field(description="Why this restaurant is recommended")


# Create the parser
parser = PydanticOutputParser(pydantic_object=RestaurantRecommendation)

# Build prompt with auto-injected format instructions
prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a food recommendation expert for Indian cities."),
    ("user", "{query}\n\n{format_instructions}"),
])

# Chain: prompt -> LLM -> parser
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
chain = prompt | llm | parser

# Invoke
result = chain.invoke({
    "query": "Recommend a great biryani place in Hyderabad for a family dinner",
    "format_instructions": parser.get_format_instructions(),
})

print(result.restaurant_name)  # e.g., "Paradise Biryani"
print(result.price_range_inr)  # e.g., "INR 300-500"
print(result.must_try_dishes)  # e.g., ["Hyderabadi Dum Biryani", ...]

import { generateObject } from 'ai';
import { openai } from '@ai-sdk/openai';
import { z } from 'zod';

// Define the output schema with Zod
const OrderSchema = z.object({
  items: z.array(
    z.object({
      name: z.string().describe('Name of the food item'),
      quantity: z.number().int().positive(),
      priceINR: z.number().positive().describe('Price in Indian Rupees'),
      specialInstructions: z.string().optional(),
    })
  ),
  deliveryAddress: z.string(),
  estimatedTotalINR: z.number().positive(),
  isVegetarian: z.boolean(),
  paymentMethod: z.enum(['UPI', 'card', 'COD', 'wallet']),
});

// Type is automatically inferred from the schema
type Order = z.infer<typeof OrderSchema>;

async function parseOrder(userMessage: string): Promise<Order> {
  const { object } = await generateObject({
    model: openai('gpt-4o-mini'),
    schema: OrderSchema,
    prompt: `Extract a structured food order from this message: ${userMessage}`,
  });

  // 'object' is fully typed as Order
  return object;
}

// Usage
const order = await parseOrder(
  'I want 2 butter naans, 1 dal makhani, and 1 paneer tikka. ' +
  'Deliver to Koramangala 4th Block, Bangalore. Pay with UPI. ' +
  'Make the naan extra crispy please.'
);

console.log(order.items[0].name); // "Butter Naan"
console.log(order.paymentMethod); // "UPI"
console.log(order.estimatedTotalINR); // e.g., 650

import outlines
from pydantic import BaseModel, Field
from typing import List
from enum import Enum


class TicketPriority(str, Enum):
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"
    CRITICAL = "critical"


class SupportTicket(BaseModel):
    """Structured support ticket extracted from customer message."""
    category: str = Field(description="Issue category")
    priority: TicketPriority
    summary: str = Field(max_length=200)
    affected_service: str
    requires_escalation: bool
    suggested_tags: List[str]


# Load a local model with Outlines
model = outlines.models.transformers("mistralai/Mistral-7B-Instruct-v0.3")

# Create a generator that is GUARANTEED to produce valid SupportTicket JSON
generator = outlines.generate.json(model, SupportTicket)

# Generate -- output is always valid, no retries needed
prompt = """Extract a structured support ticket from this customer message:

Customer: My UPI payments through PhonePe keep failing since yesterday.
I've tried 5 times and each time it says 'transaction declined'.
My account has sufficient balance. This is urgent as I need to pay
my electricity bill before the deadline tonight.
"""

ticket = generator(prompt)
print(ticket.priority)     # TicketPriority.HIGH
print(ticket.category)     # "Payment Failure"
print(ticket.requires_escalation)  # True

import json
import json_repair
from pydantic import BaseModel, ValidationError
from openai import OpenAI
from typing import TypeVar, Type, Optional
import logging

logger = logging.getLogger(__name__)

T = TypeVar("T", bound=BaseModel)


def robust_parse(
    client: OpenAI,
    model: str,
    prompt: str,
    schema: Type[T],
    max_retries: int = 3,
    fix_with_llm: bool = True,
) -> Optional[T]:
    """Production-grade output parser with multi-layer fallback.

    Layer 1: Direct JSON parsing
    Layer 2: JSON repair (fix trailing commas, quotes, etc.)
    Layer 3: LLM-based output fixing
    Layer 4: Full retry with error context
    """
    schema_json = json.dumps(schema.model_json_schema(), indent=2)
    last_error = None

    for attempt in range(max_retries):
        # Build prompt with error context on retries
        messages = [{"role": "user", "content": prompt}]
        if attempt > 0 and last_error:
            messages.append({
                "role": "user",
                "content": (
                    f"Your previous response failed validation with error: "
                    f"{last_error}\n\nPlease fix and return valid JSON matching "
                    f"this schema:\n{schema_json}"
                ),
            })

        response = client.chat.completions.create(
            model=model,
            messages=messages,
            response_format={"type": "json_object"},
        )
        raw_text = response.choices[0].message.content

        # Layer 1: Direct parse
        try:
            data = json.loads(raw_text)
            return schema.model_validate(data)
        except json.JSONDecodeError as e:
            logger.warning(f"Attempt {attempt+1}: JSON parse failed: {e}")

            # Layer 2: JSON repair
            try:
                repaired = json_repair.loads(raw_text)
                return schema.model_validate(repaired)
            except (ValidationError, Exception) as repair_err:
                logger.warning(f"Attempt {attempt+1}: Repair failed: {repair_err}")
                last_error = str(repair_err)

        except ValidationError as e:
            logger.warning(f"Attempt {attempt+1}: Validation failed: {e}")

            # Layer 3: LLM-based fixing
            if fix_with_llm and attempt < max_retries - 1:
                try:
                    fix_response = client.chat.completions.create(
                        model=model,
                        messages=[
                            {
                                "role": "user",
                                "content": (
                                    f"Fix this JSON to match the schema.\n\n"
                                    f"JSON: {raw_text}\n\n"
                                    f"Errors: {e}\n\n"
                                    f"Schema: {schema_json}\n\n"
                                    f"Return ONLY valid JSON, no explanation."
                                ),
                            }
                        ],
                        response_format={"type": "json_object"},
                    )
                    fixed_data = json.loads(fix_response.choices[0].message.content)
                    return schema.model_validate(fixed_data)
                except Exception as fix_err:
                    logger.warning(f"LLM fix failed: {fix_err}")
                    last_error = str(fix_err)
            else:
                last_error = str(e)

    logger.error(f"All {max_retries} attempts failed for schema {schema.__name__}")
    return None  # Caller should handle None (dead letter queue, default, etc.)

# Production output parser configuration (YAML)
output_parser:
  primary_strategy: "provider_structured_output"  # or "constrained_decoding", "post_hoc_parsing"
  
  provider_config:
    openai:
      response_format: "json_schema"
      strict: true
    anthropic:
      tool_use_strict: true
    
  retry:
    max_attempts: 3
    backoff_ms: [100, 500, 2000]  # exponential backoff
    fix_with_llm: true
    fix_model: "gpt-4o-mini"  # cheaper model for fixing
    
  validation:
    framework: "pydantic_v2"  # or "zod", "json_schema"
    coerce_types: true
    strip_extra_fields: true
    
  fallback:
    strategy: "dead_letter_queue"  # or "default_value", "raise_exception"
    dlq_table: "failed_outputs"
    alert_threshold: 0.05  # alert if >5% failure rate
    
  monitoring:
    track_compliance_rate: true
    track_retry_count: true
    track_parse_latency: true
    log_failed_outputs: true