Structured Outputs

# script_id: day_016_structured_outputs/schemas
# schemas.py: compact copy of Day 15's atlas/schemas.py + atlas/data.py.
# Install: uv add "openai>=3.46,<4" pydantic tiktoken
# Env: ORBIT_DATA (optional, path to datasets/orbit). No API key needed for this file.
import json
import os
from pathlib import Path
from typing import Literal

from pydantic import BaseModel, ConfigDict, Field

DATA_DIR = Path(os.environ.get("ORBIT_DATA", "datasets/orbit"))

IssueType = Literal["bug", "feature", "question", "docs"]
Severity = Literal["S1", "S2", "S3", "S4"]
Component = Literal["core", "scheduler", "flows", "broker-redis", "broker-postgres", "broker-sqs",
                    "cli", "dashboard", "retries", "results", "observability", "auth", "docs"]


class Comment(BaseModel):
    author: str
    created_at: str
    body: str


class Issue(BaseModel):
    """One raw row of issues/issues.jsonl. Labels and version fields are ground truth."""
    id: int
    title: str
    body: str
    author: str
    created_at: str
    state: Literal["open", "closed"]
    labels: list[str]
    orbit_version: str | None
    python_version: str | None
    error_codes: list[str]
    duplicate_of: int | None
    comments: list[Comment]
    closed_at: str | None

# script_id: day_016_structured_outputs/schemas
class IssueRecord(BaseModel):
    """The facts Atlas extracts from one issue. Fields mirror eval/extraction_gold.jsonl."""
    model_config = ConfigDict(extra="forbid")

    type: IssueType = Field(description="bug, feature, question or docs")
    components: list[Component] = Field(description="Every Orbit component the issue is about")
    severity: Severity | None = Field(description="Bugs only: S1 outage or data loss, S2 major "
                                      "breakage, S3 has a workaround, S4 cosmetic. null otherwise")
    orbit_version: str | None = Field(description="Full Orbit version, e.g. 3.1.2. null when not stated")
    python_version: str | None = Field(description="Python version as written, e.g. 3.12.6. "
                                       "null when not stated")
    error_codes: list[str] = Field(description="ORB-NNNN codes quoted in the issue or its comments")
    summary: str = Field(description="One or two English sentences: symptom, trigger, version")


def _jsonl(path: Path) -> list[dict]:
    return [json.loads(line) for line in path.read_text(encoding="utf-8").splitlines() if line.strip()]


def load_issues(path: Path | None = None) -> list[Issue]:
    return [Issue.model_validate(row) for row in _jsonl(path or DATA_DIR / "issues" / "issues.jsonl")]


def load_gold() -> dict[int, IssueRecord]:
    """extraction_gold.jsonl keyed by issue id. Day 20 uses the first 20 ids as the dev set."""
    rows = _jsonl(DATA_DIR / "eval" / "extraction_gold.jsonl")
    return {row.pop("issue_id"): IssueRecord.model_validate(row) for row in rows}


if __name__ == "__main__":
    print(f"{len(load_issues())} issues · {len(load_gold())} gold records")

240 issues · 30 gold records

# script_id: day_016_structured_outputs/tokens
# tokens.py: compact copy of Day 4's atlas/tokens.py, plus truncate_tokens() (new today).
# Install: uv add tiktoken   (downloads the o200k_base file once, then works offline)
import tiktoken

# GPT-6's tokenizer isn't published, so these counts are approximate (Day 4).
_ENC = tiktoken.get_encoding("o200k_base")


def count_tokens(text: str) -> int:
    return len(_ENC.encode(text))


def truncate_tokens(text: str, max_tokens: int, head_share: float = 0.7) -> str:
    """Keep the first ~70% and the last ~30% of the budget, and mark the cut."""
    ids = _ENC.encode(text)
    if len(ids) <= max_tokens:
        return text
    head = int(max_tokens * head_share)
    tail = max_tokens - head
    cut = len(ids) - max_tokens
    return _ENC.decode(ids[:head]) + f"\n\n[... {cut:,} tokens cut ...]\n\n" + _ENC.decode(ids[-tail:])


if __name__ == "__main__":
    print(count_tokens("ORB-1101 LeaseExpired"))

6

# script_id: day_016_structured_outputs/llm
# llm.py: compact copy of Day 12's atlas/llm.py (client, CostTracker, log_call).
# Install: uv add "openai>=3.46,<4"
# Env: OPENAI_API_KEY (required), ATLAS_MODEL, ATLAS_LOG_DIR (optional)
import json
import os
import time
from pathlib import Path

from openai import OpenAI

MODEL = os.environ.get("ATLAS_MODEL", "gpt-6-luna")
LOG_DIR = Path(os.environ.get("ATLAS_LOG_DIR", "logs"))
# USD per 1M tokens, as of 2026-11; verify at https://developers.openai.com/api/docs/pricing
PRICES_PER_1M = {"gpt-6-luna": {"input": 0.10, "cached_input": 0.01, "output": 0.50}}

client = OpenAI(timeout=60, max_retries=3)

# script_id: day_016_structured_outputs/llm
def cost_from_usage(usage, model: str = MODEL) -> float:
    p, cached = PRICES_PER_1M[model], usage.input_tokens_details.cached_tokens
    return ((usage.input_tokens - cached) * p["input"] + cached * p["cached_input"]
            + usage.output_tokens * p["output"]) / 1_000_000   # reasoning is inside output_tokens


class CostTracker:
    def __init__(self) -> None:
        self.calls, self.total_usd = 0, 0.0

    def record(self, response, *, label: str = "", latency_s: float | None = None) -> float:
        cost = cost_from_usage(response.usage)
        self.calls, self.total_usd = self.calls + 1, self.total_usd + cost
        return cost

    def summary(self) -> str:
        return f"{self.calls} calls · ${self.total_usd:.4f} (prices as of 2026-11, verify)"


tracker = CostTracker()


def log_call(response, *, label: str, latency_s: float, cost_usd: float, extra: dict | None = None) -> None:
    """One JSON line per call. No prompts and no keys in the log."""
    LOG_DIR.mkdir(parents=True, exist_ok=True)
    row = {"ts": time.time(), "label": label, "model": response.model, "id": response.id,
           "status": response.status, "input_tokens": response.usage.input_tokens,
           "output_tokens": response.usage.output_tokens, "latency_s": round(latency_s, 3),
           "cost_usd": round(cost_usd, 6), **(extra or {})}
    with open(LOG_DIR / "calls.jsonl", "a", encoding="utf-8") as f:
        f.write(json.dumps(row) + "\n")

# script_id: day_016_structured_outputs/prompted_json_baseline
# The naive way: ask for JSON in the prompt, then validate the reply. ≈ $0.005.
# Install: uv add "openai>=3.46,<4" pydantic tiktoken
# Env: OPENAI_API_KEY, ORBIT_DATA. Imports the lesson-local llm.py and schemas.py.
# reasoning effort "low" = little hidden thinking, cheapest; Day 6 covers the knob.
from collections import Counter

from pydantic import ValidationError

from llm import MODEL, client
from schemas import IssueRecord, load_gold, load_issues

PROMPT = """Read the GitHub issue below and reply in JSON with the keys type, components,
severity, orbit_version, python_version, error_codes and summary.

{issue}"""

gold = load_gold()
issues = [i for i in load_issues() if i.id in gold][:10]
failures: Counter[str] = Counter()
for issue in issues:
    resp = client.responses.create(model=MODEL, reasoning={"effort": "low"},
                                   input=PROMPT.format(issue=f"# {issue.title}\n\n{issue.body}"))
    try:
        IssueRecord.model_validate_json(resp.output_text)
    except ValidationError as err:
        failures[err.errors()[0]["type"]] += 1
print(f"parsed {len(issues) - sum(failures.values())}/{len(issues)} · failures: {dict(failures)}")

parsed 6/10 · failures: {'json_invalid': 2, 'literal_error': 1, 'missing': 1}

# script_id: day_016_structured_outputs/failure_modes
# Four replies prompted JSON really produces, run through the Day 15 schema. Offline, $0.
# Install: uv add "openai>=3.46,<4" pydantic tiktoken · Env: ORBIT_DATA (optional)
from pydantic import ValidationError

from schemas import IssueRecord

GOOD = ('{"type": "bug", "components": ["core"], "severity": "S3", "orbit_version": "3.1.2", '
        '"python_version": "3.11", "error_codes": ["ORB-1101"], "summary": "Export runs twice."}')
replies = {
    "prose prefix": "Sure! Here's the JSON:\n" + GOOD,
    "markdown fence": "```json\n" + GOOD + "\n```",
    "word severity": GOOD.replace('"S3"', '"high"'),
    "missing field": GOOD.replace('"python_version": "3.11", ', ""),
}
for name, text in replies.items():
    try:
        IssueRecord.model_validate_json(text)
        print(f"{name:<15} ok")
    except ValidationError as err:
        e = err.errors()[0]
        print(f"{name:<15} {e['type']}: {e['msg']}")

prose prefix    json_invalid: Invalid JSON: expected value at line 1 column 1
markdown fence  json_invalid: Invalid JSON: expected value at line 1 column 1
word severity   literal_error: Input should be 'S1', 'S2', 'S3' or 'S4'
missing field   missing: Field required

# script_id: day_016_structured_outputs/strict_schema
# See exactly what responses.parse() sends. No key, no network, $0.
# Install: uv add "openai>=3.46,<4" pydantic tiktoken  (httpx2, the SDK's HTTP library, comes with openai>=3)
import json

import httpx2
from openai import OpenAI

from schemas import IssueRecord

sent: dict = {}
CANNED = ('{"type":"bug","components":["core"],"severity":"S3","orbit_version":null,'
          '"python_version":null,"error_codes":[],"summary":"stub"}')


def fake_api(request: httpx2.Request) -> httpx2.Response:
    sent.update(json.loads(request.content))          # keep the request body
    return httpx2.Response(200, json={"id": "resp_fake", "status": "completed", "output": [
        {"type": "message", "role": "assistant", "content": [{"type": "output_text", "text": CANNED}]}]})


client = OpenAI(api_key="sk-test-FAKE", http_client=httpx2.Client(transport=httpx2.MockTransport(fake_api)))
resp = client.responses.parse(model="gpt-6-luna", input="(issue text)", text_format=IssueRecord)

fmt = sent["text"]["format"]
print({k: v for k, v in fmt.items() if k != "schema"})
print("required:", fmt["schema"]["required"])
print("additionalProperties:", fmt["schema"]["additionalProperties"])
print("severity:", json.dumps(fmt["schema"]["properties"]["severity"]))
print("parsed:", type(resp.output_parsed).__name__, resp.output_parsed.severity)

{'type': 'json_schema', 'strict': True, 'name': 'IssueRecord'}
required: ['type', 'components', 'severity', 'orbit_version', 'python_version', 'error_codes', 'summary']
additionalProperties: False
severity: {"anyOf": [{"enum": ["S1", "S2", "S3", "S4"], "type": "string"}, {"type": "null"}], "description": "Bugs only: S1 outage or data loss, S2 major breakage, S3 has a workaround, S4 cosmetic. null otherwise", "title": "Severity"}
parsed: IssueRecord S3

# script_id: day_016_structured_outputs/parse_and_break
# One issue → one typed IssueRecord; the next section breaks it two ways. ≈ $0.005 in all.
# Install: uv add "openai>=3.46,<4" pydantic tiktoken
# Env: OPENAI_API_KEY, ORBIT_DATA. Imports the lesson-local llm.py and schemas.py.
from llm import MODEL, client
from schemas import IssueRecord, load_issues

issue = next(i for i in load_issues() if i.id == 176)          # the lease-expiry report
resp = client.responses.parse(
    model=MODEL, reasoning={"effort": "low"},
    instructions="Extract the facts of this Orbit issue. Use null when a fact is not stated.",
    input=f"# {issue.title}\n\n{issue.body}",
    text_format=IssueRecord,                                    # → strict JSON Schema
)
print(resp.status)
print(resp.output_parsed.model_dump_json(indent=2))

completed
{
  "type": "bug",
  "components": [
    "core"
  ],
  "severity": "S3",
  "orbit_version": "3.1.2",
  "python_version": "3.11",
  "error_codes": [
    "ORB-1101"
  ],
  "summary": "A nightly export job runs twice: the first attempt hits ORB-1101 LeaseExpired after 300 s because heartbeat_seconds is 240, so a second worker starts it."
}

# script_id: day_016_structured_outputs/parse_and_break
# Knob 1: no descriptions + no null rule → count invented versions.
from pydantic import BaseModel, ValidationError

from schemas import Component, IssueRecord, IssueType, Severity, load_gold, load_issues


class BareRecord(BaseModel):          # the same fields as IssueRecord, with no descriptions
    type: IssueType
    components: list[Component]
    severity: Severity | None
    orbit_version: str | None
    python_version: str | None
    error_codes: list[str]
    summary: str


gold = load_gold()
no_version = [i for i in load_issues() if i.id in gold and gold[i.id].orbit_version is None]
for schema, rule in [(BareRecord, ""), (IssueRecord, " Use null when a fact is not stated.")]:
    invented = 0
    for issue in no_version:
        resp = client.responses.parse(model=MODEL, instructions="Extract the issue's facts." + rule,
                                      input=f"# {issue.title}\n\n{issue.body}", text_format=schema,
                                      reasoning={"effort": "low"})
        invented += resp.output_parsed is not None and resp.output_parsed.orbit_version is not None
    print(f"{schema.__name__:<12} invented versions: {invented}/{len(no_version)}")

BareRecord   invented versions: 3/5
IssueRecord  invented versions: 0/5

# script_id: day_016_structured_outputs/parse_and_break
# Knob 2: a tiny max_output_tokens. The call ends `incomplete`, and you are still billed.
issue = next(i for i in load_issues() if i.id == 176)
try:
    resp = client.responses.parse(model=MODEL, input=issue.body, text_format=IssueRecord,
                                  reasoning={"effort": "low"}, max_output_tokens=20)
    print(resp.status, resp.incomplete_details, resp.output_parsed)
except ValidationError as err:        # the JSON was cut off mid-way: parse() raises before you see it
    print("truncated JSON:", err.errors()[0]["type"])

incomplete IncompleteDetails(reason='max_output_tokens') None

truncated JSON: json_invalid

# script_id: day_016_structured_outputs/extract
# atlas/extract.py: issue → IssueRecord with structured outputs. The core of Capstone R1 (Day 22).
# Install: uv add "openai>=3.46,<4" pydantic tiktoken
# Env: OPENAI_API_KEY (required), ATLAS_MODEL, ORBIT_DATA, ATLAS_LOG_DIR (optional)
import time
from typing import Literal

from openai.types.responses import ParsedResponse
from pydantic import ValidationError

from llm import MODEL, client, log_call, tracker
from schemas import Issue, IssueRecord, load_gold, load_issues
from tokens import truncate_tokens

EXTRACT_INSTRUCTIONS = """You extract facts from GitHub issues filed against Orbit, an open-source
Python background-jobs library. Fill every field of the schema from the issue text only.

Rules:
- Use null when a fact is not stated. Never guess a version from dates or context.
- severity is for bugs only; use null for features, questions and docs.
- error_codes: every ORB-NNNN code that appears anywhere, including logs and comments. [] if none.
- components: usually one; list two only when the issue really spans both.
- summary: one or two sentences in English, even when the issue is in another language.

Component glossary:
core = queues, workers, leases, job execution | scheduler = cron schedules and delays
flows = Flows, steps and step state | broker-redis, broker-postgres, broker-sqs = that broker
cli = the orbit command | dashboard = the web UI | retries = retry policy and backoff
results = the result backend | observability = metrics and tracing
auth = dashboard login and OIDC | docs = the documentation itself"""


def issue_to_input(issue: Issue, max_tokens: int = 3000) -> str:
    """Title, body and comments, within a token budget. Never the labels: they're the answer key."""
    parts = [f"# {issue.title}", issue.body]
    parts += [f"--- comment by {c.author} ---\n{c.body}" for c in issue.comments]
    return truncate_tokens("\n\n".join(parts), max_tokens)

# script_id: day_016_structured_outputs/extract
class ExtractionError(Exception):
    """The API answered, but not with a usable record."""

    def __init__(self, reason: Literal["refusal", "incomplete"],
                 response: ParsedResponse | None = None, detail: str = "") -> None:
        super().__init__(f"{reason}: {detail}")
        self.reason, self.response = reason, response


def extract_issue(issue: Issue, *, model: str = MODEL) -> IssueRecord:
    start = time.perf_counter()
    try:
        resp = client.responses.parse(
            model=model, instructions=EXTRACT_INSTRUCTIONS, input=issue_to_input(issue),
            text_format=IssueRecord, reasoning={"effort": "low"},
            max_output_tokens=25_000)        # headroom for reasoning + JSON (reasoning guide)
    except ValidationError as err:
        # Strict mode fixes the shape, so a parse failure here means the JSON was cut off.
        raise ExtractionError("incomplete", detail=f"issue {issue.id}: truncated JSON") from err
    latency = time.perf_counter() - start
    cost = tracker.record(resp, label="extract_issue", latency_s=latency)
    log_call(resp, label="extract_issue", latency_s=latency, cost_usd=cost, extra={"issue_id": issue.id})

    if resp.status == "incomplete":
        raise ExtractionError("incomplete", resp, str(resp.incomplete_details.reason))
    refusals = [part.refusal for item in resp.output if item.type == "message"
                for part in item.content if part.type == "refusal"]
    if refusals:
        raise ExtractionError("refusal", resp, refusals[0])
    if resp.output_parsed is None:
        raise ExtractionError("incomplete", resp, "no final answer")
    return resp.output_parsed

# script_id: day_016_structured_outputs/extract
if __name__ == "__main__":
    gold = load_gold()
    issues = [i for i in load_issues() if i.id in gold][:10]      # 10 of the 20 dev issues
    hits = {"type": 0, "severity": 0, "error_codes": 0}
    for issue in issues:
        try:
            rec = extract_issue(issue)
        except ExtractionError as err:
            print(f"#{issue.id:<4} skipped ({err})")
            continue
        want = gold[issue.id]
        hits["type"] += rec.type == want.type
        hits["severity"] += rec.severity == want.severity
        hits["error_codes"] += sorted(rec.error_codes) == sorted(want.error_codes)
        print(f"#{issue.id:<4} {rec.type:<9} {rec.severity or '-':<3} {rec.error_codes}")
    print(" · ".join(f"{k} {v}/{len(issues)}" for k, v in hits.items()), "·", tracker.summary())

#10   question  -   []
#25   question  -   ['ORB-1202']
#62   bug       S3  []
#63   question  -   ['ORB-1001']
#74   feature   -   []
#83   bug       S2  []
#97   question  -   []
#104  bug       S2  ['ORB-1802']
#109  bug       S2  ['ORB-1502', 'ORB-1301']
#113  bug       S3  []
type 9/10 · severity 8/10 · error_codes 10/10 · 10 calls · $0.0042 (prices as of 2026-11, verify)