parser_native_structured_output.py

"""
Parser Model — Structured Output Debug
=======================================

Proves that:
  - Primary model call: response_format is None (no structured output params)
  - Parser model call:  response_format contains the full JSON schema (native structured output)
"""

import json
import random
from functools import wraps
from typing import List

from agno.agent import Agent, RunOutput  # noqa
from agno.models.openai import OpenAIResponses
from pydantic import BaseModel, Field
from rich.pretty import pprint  # noqa

# ---------------------------------------------------------------------------
# Monkey-patch OpenAIResponses.get_request_params to log response_format
# ---------------------------------------------------------------------------
_original_get_request_params = OpenAIResponses.get_request_params
_call_counter = 0


@wraps(_original_get_request_params)
def _patched_get_request_params(self, **kwargs):
    global _call_counter
    _call_counter += 1
    response_format = kwargs.get("response_format")

    print(f"\n{'='*70}")
    print(f"  API CALL #{_call_counter}  (model: {self.id})")
    print(f"{'='*70}")

    if response_format is None:
        print("  response_format: None  (no structured output params sent)")
    elif isinstance(response_format, type) and issubclass(response_format, BaseModel):
        print(f"  response_format: <Pydantic model: {response_format.__name__}>")
        print(f"  --> Will be converted to native json_schema format by the provider")
        schema = response_format.model_json_schema()
        print(f"  --> Schema fields: {list(schema.get('properties', {}).keys())}")
    elif isinstance(response_format, dict):
        print(f"  response_format: {json.dumps(response_format, indent=4, default=str)}")
    else:
        print(f"  response_format: {response_format}")

    result = _original_get_request_params(self, **kwargs)

    # Show what actually ends up in the request
    text_param = result.get("text", {})
    if isinstance(text_param, dict) and "format" in text_param:
        fmt = text_param["format"]
        print(f"\n  ACTUAL request['text']['format']:")
        print(f"    type: {fmt.get('type', 'N/A')}")
        if fmt.get("type") == "json_schema":
            print(f"    name: {fmt.get('name', 'N/A')}")
            print(f"    strict: {fmt.get('strict', 'N/A')}")
            schema = fmt.get("schema", {})
            print(f"    schema properties: {list(schema.get('properties', {}).keys())}")
    else:
        print(f"\n  ACTUAL request: NO format/structured-output params in request")

    print(f"{'='*70}\n")
    return result


OpenAIResponses.get_request_params = _patched_get_request_params


# ---------------------------------------------------------------------------
# Schema
# ---------------------------------------------------------------------------
class NationalParkAdventure(BaseModel):
    park_name: str = Field(..., description="Name of the national park")
    best_season: str = Field(
        ...,
        description="Optimal time of year to visit this park (e.g., 'Late spring to early fall')",
    )
    signature_attractions: List[str] = Field(
        ...,
        description="Must-see landmarks, viewpoints, or natural features in the park",
    )
    recommended_trails: List[str] = Field(
        ...,
        description="Top hiking trails with difficulty levels (e.g., 'Angel's Landing - Strenuous')",
    )
    wildlife_encounters: List[str] = Field(
        ..., description="Animals visitors are likely to spot, with viewing tips"
    )
    photography_spots: List[str] = Field(
        ...,
        description="Best locations for capturing stunning photos, including sunrise/sunset spots",
    )
    camping_options: List[str] = Field(
        ..., description="Available camping areas, from primitive to RV-friendly sites"
    )
    safety_warnings: List[str] = Field(
        ..., description="Important safety considerations specific to this park"
    )
    hidden_gems: List[str] = Field(
        ..., description="Lesser-known spots or experiences that most visitors miss"
    )
    difficulty_rating: int = Field(
        ...,
        ge=1,
        le=5,
        description="Overall park difficulty for average visitor (1=easy, 5=very challenging)",
    )
    estimated_days: int = Field(
        ...,
        ge=1,
        le=14,
        description="Recommended number of days to properly explore the park",
    )
    special_permits_needed: List[str] = Field(
        default=[],
        description="Any special permits or reservations required for certain activities",
    )


# ---------------------------------------------------------------------------
# Create Agent
# ---------------------------------------------------------------------------
agent = Agent(
    model=OpenAIResponses(id="gpt-4.1"),
    description="You help people plan amazing national park adventures and provide detailed park guides.",
    output_schema=NationalParkAdventure,
    parser_model=OpenAIResponses(id="gpt-5.2"),
)


# ---------------------------------------------------------------------------
# Run Agent
# ---------------------------------------------------------------------------
national_parks = [
    "Yellowstone National Park",
    "Yosemite National Park",
    "Grand Canyon National Park",
    "Zion National Park",
    "Grand Teton National Park",
]

if __name__ == "__main__":
    park = national_parks[random.randint(0, len(national_parks) - 1)]
    print(f"\nAsking about: {park}\n")

    run: RunOutput = agent.run(park)

    print(f"\n{'='*70}")
    print("  FINAL RESULT")
    print(f"{'='*70}")
    pprint(run.content)