OpenAI Structured Output with o-series Models

from openai import OpenAI
from pydantic import BaseModel, ConfigDict, Field

client = OpenAI()


# ✅ Pydantic model — extra="forbid" makes it strict-mode compatible
class BugReport(BaseModel):
    model_config = ConfigDict(extra="forbid")   # required for strict mode

    title: str = Field(description="Short bug title")
    severity: str = Field(description="One of: low, medium, high, critical")
    steps_to_reproduce: list[str] = Field(min_length=1)
    expected_behavior: str
    actual_behavior: str
    suspected_cause: str | None = Field(default=None, description="Optional cause hypothesis")


# ✅ Chat Completions with parse helper
resp = client.beta.chat.completions.parse(
    model="o3",
    messages=[
        {"role": "developer", "content": "Extract a structured bug report from the user's description."},
        {"role": "user",      "content": "The login page 500s when the password has emoji. I tried logging in with 🔒hello and got a server error. Should just log me in."},
    ],
    response_format=BugReport,
    max_completion_tokens=25_000,
    reasoning_effort="medium",
)

# ✅ CHECK REFUSAL FIRST
choice = resp.choices[0]
if choice.message.refusal:
    print(f"Model refused: {choice.message.refusal}")
else:
    report: BugReport = choice.message.parsed
    print(report.title, report.severity)


# ✅ Responses API with parse helper
resp_r = client.responses.parse(
    model="o3",
    input=bug_description,
    instructions="Extract a structured bug report.",
    text_format=BugReport,
    max_output_tokens=25_000,
    reasoning={"effort": "medium"},
)

# Refusal check on Responses is per-item
report: BugReport | None = resp_r.output_parsed
if report is None:
    # Look for refusal in output items
    for item in resp_r.output:
        if item.type == "message":
            for part in item.content:
                if part.type == "refusal":
                    print(f"Refused: {part.refusal}")
else:
    print(report.title)


# ✅ Complete safe wrapper
class ParseFailure(Exception):
    pass

class Refusal(Exception):
    pass

def parse_or_raise(model: str, prompt: str, schema: type[BaseModel], budget: int = 25_000):
    resp = client.beta.chat.completions.parse(
        model=model,
        messages=[{"role": "user", "content": prompt}],
        response_format=schema,
        max_completion_tokens=budget,
        reasoning_effort="medium",
    )
    choice = resp.choices[0]
    if choice.message.refusal:
        raise Refusal(choice.message.refusal)
    if choice.finish_reason == "length":
        raise ParseFailure(f"Truncated at {budget} tokens; reasoning={resp.usage.completion_tokens_details.reasoning_tokens}")
    if choice.message.parsed is None:
        raise ParseFailure("Model returned unparseable output")
    return choice.message.parsed

from openai import OpenAI

client = OpenAI()


# ❌ BUG — permissive schema fails in strict mode
bad_schema = {
    "type": "object",
    "properties": {
        "name": {"type": "string"},
        "age": {"type": "integer"},
        "email": {"type": "string"},           # "optional" — not in required
    },
    "required": ["name"],
    # Missing additionalProperties: false
}


# ✅ FIXED — strict-mode compliant
person_schema = {
    "type": "object",
    "additionalProperties": False,             # required
    "properties": {
        "name":  {"type": "string"},
        "age":   {"type": "integer"},
        "email": {"type": ["string", "null"]}, # optional expressed via null union
    },
    "required": ["name", "age", "email"],      # ALL properties listed
}

resp = client.chat.completions.create(
    model="o3",
    messages=[
        {"role": "user", "content": "Extract: John, 42, no email on file"}
    ],
    response_format={
        "type": "json_schema",
        "json_schema": {
            "name": "Person",
            "schema": person_schema,
            "strict": True,
        },
    },
    max_completion_tokens=15_000,
)


# ✅ Nested objects also need additionalProperties=false
company_schema = {
    "type": "object",
    "additionalProperties": False,
    "properties": {
        "name": {"type": "string"},
        "hq": {
            "type": "object",
            "additionalProperties": False,     # ← every nested object too
            "properties": {
                "city": {"type": "string"},
                "country": {"type": "string"},
            },
            "required": ["city", "country"],
        },
        "employees": {
            "type": "array",
            "items": person_schema,            # array of Person (already strict-compliant)
        },
    },
    "required": ["name", "hq", "employees"],
}


# ✅ Responses API equivalent — schema at text.format.schema
resp_r = client.responses.create(
    model="o3",
    input="Extract: John, 42, no email",
    text={
        "format": {
            "type": "json_schema",
            "name": "Person",                  # ← top-level in Responses format
            "schema": person_schema,
            "strict": True,
        },
    },
    max_output_tokens=15_000,
    reasoning={"effort": "medium"},
)

# Parse the JSON manually since we're not using the parse helper
import json
data = json.loads(resp_r.output_text)


# ✅ Unsupported keywords to avoid in strict mode
# - Complex format validators (email, uri) — often silently ignored
# - Recursive $ref — often rejected; flatten instead
# - allOf composition — limited support
# - Pattern constraints — accepted but not enforced strictly
# When in doubt, run a validation call before deploying the schema

def validate_schema(schema: dict) -> list[str]:
    """Quick sanity check for strict-mode compliance."""
    errors = []
    def check_object(s, path):
        if s.get("type") == "object":
            if not s.get("additionalProperties") is False:
                errors.append(f"{path}: missing additionalProperties: false")
            required = set(s.get("required", []))
            props = set(s.get("properties", {}).keys())
            missing_req = props - required
            if missing_req:
                errors.append(f"{path}: properties not in required: {missing_req}")
            for name, sub in s.get("properties", {}).items():
                check_object(sub, f"{path}.{name}")
        elif s.get("type") == "array":
            if "items" in s:
                check_object(s["items"], f"{path}[]")

    check_object(schema, "$")
    return errors


issues = validate_schema(person_schema)
if issues:
    for e in issues:
        print(f"  · {e}")

from openai import OpenAI
from pydantic import BaseModel, ConfigDict

client = OpenAI()


class MarketReport(BaseModel):
    model_config = ConfigDict(extra="forbid")

    company: str
    quarter: str
    revenue_millions: float
    yoy_growth_pct: float
    key_products: list[str]
    risks: list[str]
    outlook_summary: str


# ✅ Adaptive retry — bump budget if reasoning exhausted it
def extract_report(text: str, effort: str = "medium"):
    initial_budget = {"low": 8_000, "medium": 25_000, "high": 60_000}[effort]
    budget = initial_budget

    for attempt in range(3):
        resp = client.beta.chat.completions.parse(
            model="o3",
            messages=[
                {"role": "developer", "content": "Extract a structured market report from the document."},
                {"role": "user",      "content": text},
            ],
            response_format=MarketReport,
            max_completion_tokens=budget,
            reasoning_effort=effort,
        )
        choice = resp.choices[0]

        if choice.message.refusal:
            raise ValueError(f"Refused: {choice.message.refusal}")

        if choice.finish_reason == "stop" and choice.message.parsed:
            return choice.message.parsed

        # Truncated — check where the budget went
        reasoning = resp.usage.completion_tokens_details.reasoning_tokens
        visible = resp.usage.completion_tokens - reasoning

        if reasoning >= budget * 0.9:
            # Reasoning burned budget — retry with more
            budget = min(budget * 2, 100_000)
            continue

        if visible > 0 and visible < 100:
            # Reasoning succeeded but JSON was tiny/incomplete — model may be confused
            raise ValueError(f"Model produced only {visible} visible tokens; JSON incomplete")

        raise RuntimeError(f"Truncated at {budget}, reasoning={reasoning}, visible={visible}")

    raise RuntimeError("Repeated exhaustion — consider lower effort or simpler schema")


# ✅ Reduce budget pressure — simplify the schema
# Complex nested schemas force more reasoning to plan the structure.
# When budget is tight:
#   - Flatten nested objects
#   - Split into two calls (extract → refine)
#   - Use effort="low" if the task is straightforward
class SimplifiedReport(BaseModel):
    model_config = ConfigDict(extra="forbid")
    company: str
    revenue: float
    growth_pct: float


# ✅ Two-stage extraction — first pass simple, second pass detailed
def two_stage_extract(text: str) -> MarketReport:
    # Stage 1 — quick core facts (low effort)
    core = client.beta.chat.completions.parse(
        model="o3",
        messages=[{"role": "user", "content": f"Extract core facts:\n{text}"}],
        response_format=SimplifiedReport,
        max_completion_tokens=5_000,
        reasoning_effort="low",
    ).choices[0].message.parsed

    # Stage 2 — enrich with analysis (medium effort)
    enriched = client.beta.chat.completions.parse(
        model="o3",
        messages=[{"role": "user", "content": (
            f"Given these core facts: {core.model_dump()}\n\n"
            f"Original text: {text}\n\n"
            f"Produce the full MarketReport."
        )}],
        response_format=MarketReport,
        max_completion_tokens=20_000,
        reasoning_effort="medium",
    )
    return enriched.choices[0].message.parsed