Choosing a Model

# script_id: day_007_choosing_a_model/cost_at_scale
# What 10,000 triage calls a day cost on each tier. Pure arithmetic: no key, no network.
# Run: uv run cost_at_scale.py
# Prices as of 2026-11; verify at https://developers.openai.com/api/docs/pricing
import os

MODEL = os.environ.get("ATLAS_MODEL", "gpt-6-luna")

# USD per 1M tokens, Standard tier, prompts up to 272K input tokens (built on Day 4)
PRICES_PER_1M: dict[str, dict[str, float]] = {
    "gpt-6-luna":    {"input": 0.10, "cached_input": 0.01, "output": 0.50},
    "gpt-6.1-terra": {"input": 1.00, "cached_input": 0.05, "output": 5.00},
    "gpt-6.1-sol":   {"input": 2.00, "cached_input": 0.10, "output": 10.00},
    "gpt-6-astra":   {"input": 10.00, "cached_input": 1.00, "output": 50.00},
    # Claude comparison row; cache rates differ, see https://www.anthropic.com/pricing (verify)
    "claude-haiku-5-5": {"input": 0.10, "output": 0.50},
}

def estimate_cost(input_tokens: int, output_tokens: int, model: str = MODEL) -> float:
    """USD for one call. output_tokens must include reasoning tokens (they bill as output)."""
    if model not in PRICES_PER_1M:
        raise KeyError(f"no price for {model!r}; add it to PRICES_PER_1M")
    price = PRICES_PER_1M[model]
    return (input_tokens * price["input"] + output_tokens * price["output"]) / 1_000_000

CALLS_PER_DAY, IN_TOKENS, OUT_TOKENS = 10_000, 600, 80   # one triage call

for model in PRICES_PER_1M:
    per_call = estimate_cost(IN_TOKENS, OUT_TOKENS, model)
    print(f"{model:<17} ${per_call:.5f}/call  ${per_call * CALLS_PER_DAY:>7.2f}/day  "
          f"${per_call * CALLS_PER_DAY * 30:>8.2f}/month")

gpt-6-luna        $0.00010/call  $   1.00/day  $   30.00/month
gpt-6.1-terra     $0.00100/call  $  10.00/day  $  300.00/month
gpt-6.1-sol       $0.00200/call  $  20.00/day  $  600.00/month
gpt-6-astra       $0.01000/call  $ 100.00/day  $ 3000.00/month
claude-haiku-5-5  $0.00010/call  $   1.00/day  $   30.00/month

# script_id: day_007_choosing_a_model/one_item_all_tiers
# Same prompt, one Orbit issue, three tiers. Install: uv add "openai>=3.46,<4"
# Env: OPENAI_API_KEY; ORBIT_DATA (default datasets/orbit). Cost: under $0.01 (as of 2026-11, verify)
import json, os, time
from pathlib import Path
from openai import OpenAI

DATA_DIR = Path(os.environ.get("ORBIT_DATA", "datasets/orbit"))
PRICES = {"gpt-6-luna": (0.10, 0.50), "gpt-6.1-terra": (1.00, 5.00), "gpt-6.1-sol": (2.00, 10.00)}
client = OpenAI()
with (DATA_DIR / "issues" / "issues.jsonl").open(encoding="utf-8") as f:
    issue = json.loads(f.readline())   # issue #1, labelled comp:cli
prompt = "Name the one Orbit component this issue is about (for example cli, core, scheduler). One word."

for model, (p_in, p_out) in PRICES.items():
    t0 = time.perf_counter()
    r = client.responses.create(model=model, instructions=prompt, reasoning={"effort": "low"},
                                input=issue["title"] + "\n\n" + issue["body"])
    cost = (r.usage.input_tokens * p_in + r.usage.output_tokens * p_out) / 1_000_000
    print(f"{model:<14} → {r.output_text.strip():<10} · {time.perf_counter() - t0:.1f} s · ${cost:.5f}")

gpt-6-luna     → cli        · 1.1 s · $0.00012
gpt-6.1-terra  → cli        · 1.3 s · $0.00110
gpt-6.1-sol    → cli        · 1.9 s · $0.00230

# script_id: day_007_choosing_a_model/tier_vs_effort
# Tier and effort are two dials; they can cross over. Install: uv add "openai>=3.46,<4"
# Env: OPENAI_API_KEY; ORBIT_DATA. Cost: about $0.01 (as of 2026-11, verify)
import json, os, time
from pathlib import Path

import openai
from openai import OpenAI

DATA_DIR = Path(os.environ.get("ORBIT_DATA", "datasets/orbit"))
PRICES = {"gpt-6-luna": (0.10, 0.50), "gpt-6.1-sol": (2.00, 10.00), "gpt-6-astra": (10.00, 50.00)}
client = OpenAI()
with (DATA_DIR / "issues" / "issues.jsonl").open(encoding="utf-8") as f:
    issue = json.loads(f.readline())
QUESTION = "Which Orbit component is this about, and is it a regression? Answer in one line."

def run(model: str, effort: str) -> None:
    t0 = time.perf_counter()
    r = client.responses.create(model=model, reasoning={"effort": effort},
                                max_output_tokens=25_000,   # a ceiling for reasoning + answer
                                input=f"{QUESTION}\n\n{issue['title']}\n{issue['body']}")
    u = r.usage
    p_in, p_out = PRICES[model]
    cost = (u.input_tokens * p_in + u.output_tokens * p_out) / 1_000_000  # reasoning is inside output
    print(f"{model:<12} effort={effort:<6} reasoning={u.output_tokens_details.reasoning_tokens:>5} "
          f"output={u.output_tokens:>5} {time.perf_counter() - t0:5.1f} s ${cost:.5f}")

for model, effort in [("gpt-6-luna", "low"), ("gpt-6-luna", "high"), ("gpt-6.1-sol", "low")]:
    run(model, effort)

gpt-6-luna   effort=low    reasoning=   64 output=   98   1.2 s $0.00011
gpt-6-luna   effort=high   reasoning= 1856 output= 1890   9.8 s $0.00106
gpt-6.1-sol  effort=low    reasoning=   48 output=   80   1.7 s $0.00200

# script_id: day_007_choosing_a_model/tier_vs_effort
# Break it on purpose. Both calls fail fast with a typed error.
for model, effort in [("gpt-6-astra", "none"), ("gpt-6-lunna", "low")]:
    try:
        run(model, effort)
    except openai.BadRequestError as e:   # 400: astra accepts low…max, not "none" (Day 6)
        print(f"{model}: BadRequestError {e.status_code} code={e.code}")
    except openai.NotFoundError as e:     # 404: typo'd, retired or not-yet-enabled model ID
        print(f"{model}: NotFoundError {e.status_code} code={e.code}")

gpt-6-astra: BadRequestError 400 code=...
gpt-6-lunna: NotFoundError 404 code=...

# script_id: day_007_choosing_a_model/preflight
# Fail at startup, not mid-traffic: does the configured model exist, and is it retiring?
# Install: uv add "openai>=3.46,<4". Env: OPENAI_API_KEY, ATLAS_MODEL (default gpt-6-luna)
import os, sys

import openai
from openai import OpenAI

MODEL = os.environ.get("ATLAS_MODEL", "gpt-6-luna")
client = OpenAI()

def check_model_available(model: str) -> bool:
    """True if this API key can see `model`; typo'd and retired IDs raise 404."""
    try:
        client.models.retrieve(model)
        return True
    except openai.NotFoundError:
        return False

def shutdown_date(model: str) -> str | None:
    """'YYYY-MM-DD' once the provider schedules a shutdown, else None."""
    return client.models.retrieve(model).shutdown_date

if __name__ == "__main__":
    if not check_model_available(MODEL):
        sys.exit(f"ATLAS_MODEL={MODEL!r} is not available to this API key")
    if date := shutdown_date(MODEL):
        print(f"WARNING: {MODEL} shuts down on {date}; plan the migration (Day 21)")
    print(f"ok: {MODEL}")

ok: gpt-6-luna

# script_id: day_007_choosing_a_model/gold_items
# Look at your 10 test items before you score anything. No key needed.
# Data: ORBIT_DATA (default datasets/orbit): eval/extraction_gold.jsonl joined to issues/issues.jsonl
import json, os
from pathlib import Path

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

def read_jsonl(path: Path) -> list[dict]:
    with path.open(encoding="utf-8") as f:
        return [json.loads(line) for line in f if line.strip()]

def load_items(n: int = 10) -> list[dict]:
    """First n gold records, each joined to its issue (gold issue_id == issue id)."""
    issues = {i["id"]: i for i in read_jsonl(DATA_DIR / "issues" / "issues.jsonl")}
    gold = read_jsonl(DATA_DIR / "eval" / "extraction_gold.jsonl")
    return [{**issues[g["issue_id"]], "gold": g} for g in gold[:n]]

def normalize_label(s: str) -> str:
    """'comp:Core.', ' CORE ' and '`core`' all become 'core'."""
    return s.strip().strip(".`'\"").strip().lower().removeprefix("comp:")

for item in load_items():
    gold = item["gold"]
    print(f"#{item['id']:<4} {gold['type']:<9} {','.join(gold['components']):<13} {item['title'][:44]}")
for raw in ["comp:Core.", " CORE ", "`scheduler`"]:
    print(f"{raw!r:>14} → {normalize_label(raw)!r}")

#10   question  core          worker not picking up jobs
#25   question  core          ORB-1202 PayloadTooLarge al encolar un PDF
#62   bug       scheduler     cron jobs off by an hour since sunday
#63   question  broker-redis  Docker Compose में worker शुरू करते ही ORB-1
#74   feature   cli           CLI: respect NO_COLOR / don't print ANSI cod
#83   bug       core          workers stop working
#97   question  cli           Como migrar variáveis ORBIT_* para orbit.tom
#104  bug       core          ORB-1802 UnknownConfigKey for documented key
#109  bug       flows,cli     `orbit flows status` shows step as running f
#113  question  core,cli      orbit doctor fails: heartbeat_seconds (120)
  'comp:Core.' → 'core'
      ' CORE ' → 'core'
 '`scheduler`' → 'scheduler'

# script_id: day_007_choosing_a_model/compare_tiers
# Compare tiers on hand-checked Orbit issues: correct/N, latency, cost. Install: uv add "openai>=3.46,<4"
# Env: OPENAI_API_KEY. Optional: ATLAS_MODEL, ATLAS_TIER_<CHEAP|MID|STRONG|FLAGSHIP>, ATLAS_EFFORT,
#      ATLAS_N_ITEMS, ATLAS_MAX_SPEND (USD, default 0.50), ATLAS_INCLUDE_FLAGSHIP=1. Data: ORBIT_DATA.
# Cost: ≈ $0.05 without the flagship, ≈ $0.20 with it (as of 2026-11, verify)
import json, os, statistics, sys, time
from dataclasses import dataclass
from pathlib import Path

import openai
from openai import OpenAI

MODEL = os.environ.get("ATLAS_MODEL", "gpt-6-luna")
DATA_DIR = Path(os.environ.get("ORBIT_DATA", "datasets/orbit"))
EFFORT = os.environ.get("ATLAS_EFFORT", "low")         # every default tier accepts "low"
N_ITEMS = int(os.environ.get("ATLAS_N_ITEMS", "10"))
MAX_SPEND = float(os.environ.get("ATLAS_MAX_SPEND", "0.50"))
MAX_OUTPUT = 25_000                                     # ceiling for reasoning + answer

_DEFAULT_TIERS = {"cheap": MODEL, "mid": "gpt-6.1-terra", "strong": "gpt-6.1-sol", "flagship": "gpt-6-astra"}
TIERS: dict[str, str] = {t: os.environ.get(f"ATLAS_TIER_{t.upper()}", m) for t, m in _DEFAULT_TIERS.items()}

PRICES_PER_1M: dict[str, dict[str, float]] = {  # USD, as of 2026-11; verify on the pricing page
    "gpt-6-luna":    {"input": 0.10, "cached_input": 0.01, "output": 0.50},
    "gpt-6.1-terra": {"input": 1.00, "cached_input": 0.05, "output": 5.00},
    "gpt-6.1-sol":   {"input": 2.00, "cached_input": 0.10, "output": 10.00},
    "gpt-6-sol":     {"input": 2.00, "cached_input": 0.20, "output": 10.00},
    "gpt-6-astra":   {"input": 10.00, "cached_input": 1.00, "output": 50.00},
}
client = OpenAI()

def estimate_cost(input_tokens: int, output_tokens: int, model: str = MODEL) -> float:
    if model not in PRICES_PER_1M:
        raise KeyError(f"no price for {model!r}; add it to PRICES_PER_1M")
    p = PRICES_PER_1M[model]
    return (input_tokens * p["input"] + output_tokens * p["output"]) / 1_000_000

def check_model_available(model: str) -> bool:
    """True if this API key can see `model`; typo'd and retired IDs raise 404."""
    try:
        client.models.retrieve(model)
        return True
    except openai.NotFoundError:
        return False

# script_id: day_007_choosing_a_model/compare_tiers
COMPONENTS = ["core", "scheduler", "flows", "broker-redis", "broker-postgres", "broker-sqs", "cli",
              "dashboard", "retries", "results", "observability", "auth", "docs"]

@dataclass(frozen=True)
class Task:
    name: str
    instructions: str
    gold_field: str            # key in eval/extraction_gold.jsonl

COMPONENT_TASK = Task(
    "component",
    "You triage issues for Orbit, a Python background-jobs library. Reply with exactly one "
    f"component name from this list and nothing else: {', '.join(COMPONENTS)}.",
    "components",
)

def read_jsonl(path: Path) -> list[dict]:
    with path.open(encoding="utf-8") as f:
        return [json.loads(line) for line in f if line.strip()]

def load_items(n: int = N_ITEMS) -> list[dict]:
    issues = {i["id"]: i for i in read_jsonl(DATA_DIR / "issues" / "issues.jsonl")}
    gold = read_jsonl(DATA_DIR / "eval" / "extraction_gold.jsonl")
    return [{**issues[g["issue_id"]], "gold": g} for g in gold[:n]]

def normalize_label(s: str) -> str:
    return s.strip().strip(".`'\"").strip().lower().removeprefix("comp:")

def is_correct(label: str, gold_value: str | list[str] | None) -> bool:
    """Exact match after normalizing; a multi-component issue accepts any listed component."""
    golds = gold_value if isinstance(gold_value, list) else [gold_value]
    return label in {normalize_label(g) for g in golds if g}

# script_id: day_007_choosing_a_model/compare_tiers
def classify_component(issue: dict, model: str, effort: str = EFFORT,
                       task: Task = COMPONENT_TASK) -> dict:
    text = f"Title: {issue['title']}\n\n{issue['body'][:6000]}"   # cap the rare very long issue
    t0 = time.perf_counter()
    r = client.responses.create(model=model, instructions=task.instructions, input=text,
                                reasoning={"effort": effort}, max_output_tokens=MAX_OUTPUT)
    latency_s = time.perf_counter() - t0       # total latency; time to first token is Day 13
    u = r.usage
    return {
        "label": normalize_label(r.output_text) if r.status == "completed" else "",
        "latency_s": latency_s,
        "input_tokens": u.input_tokens,
        "output_tokens": u.output_tokens,       # includes reasoning tokens
        "cost": estimate_cost(u.input_tokens, u.output_tokens, model),
    }

# script_id: day_007_choosing_a_model/compare_tiers
def summarize(tier: str, model: str, effort: str, results: list[dict]) -> dict:
    lat, cost = [x["latency_s"] for x in results], sum(x["cost"] for x in results)
    return {"tier": tier, "model": model, "effort": effort, "n": len(results),
            "correct": sum(x["correct"] for x in results), "p50_s": statistics.median(lat),
            "max_s": max(lat), "usd": cost, "usd_per_1k": cost / len(results) * 1000}

def compare(models: dict[str, str], items: list[dict], effort: str = EFFORT,
            task: Task = COMPONENT_TASK) -> list[dict]:
    rows, spent = [], 0.0
    for tier, model in models.items():
        results = []
        for item in items:
            res = classify_component(item, model, effort=effort, task=task)
            res["correct"] = is_correct(res["label"], item["gold"][task.gold_field])
            results.append(res)
            spent += res["cost"]
            if spent >= MAX_SPEND:
                print(f"stopped: spend guard ${MAX_SPEND:.2f} reached (raise ATLAS_MAX_SPEND)")
                return rows + [summarize(tier, model, effort, results)]
        rows.append(summarize(tier, model, effort, results))
    return rows

def print_table(rows: list[dict]) -> None:
    print(f"{'tier':<9}{'model':<15}{'effort':<8}{'correct':>9}{'p50 s':>7}{'max s':>7}"
          f"{'$ total':>10}{'$/1k':>8}")
    for r in rows:
        print(f"{r['tier']:<9}{r['model']:<15}{r['effort']:<8}{r['correct']:>4}/{r['n']:<4}"
              f"{r['p50_s']:>7.1f}{r['max_s']:>7.1f}{r['usd']:>10.4f}{r['usd_per_1k']:>8.2f}")

if __name__ == "__main__":
    selected = {t: m for t, m in TIERS.items()
                if t != "flagship" or os.environ.get("ATLAS_INCLUDE_FLAGSHIP") == "1"}
    missing = [m for m in selected.values() if not check_model_available(m)]
    if missing:
        sys.exit(f"not available to this API key: {missing}")
    print_table(compare(selected, load_items()))

tier     model          effort    correct  p50 s  max s   $ total    $/1k
cheap    gpt-6-luna     low        7/10      0.9    2.1    0.0013    0.13
mid      gpt-6.1-terra  low        8/10      1.1    2.4    0.0121    1.21
strong   gpt-6.1-sol    low        8/10      1.6    3.0    0.0238    2.38

# script_id: day_007_choosing_a_model/claude_row
# Optional: the same 10 items on Claude, as one comparison row. Install: uv add anthropic
# Env: ANTHROPIC_API_KEY; ORBIT_DATA. Cost: under $0.01.
# Prices from Anthropic's reference cached 2026-10-06; verify at https://www.anthropic.com/pricing
import json, os, statistics, time
from pathlib import Path

import anthropic

DATA_DIR = Path(os.environ.get("ORBIT_DATA", "datasets/orbit"))
CLAUDE_MODEL = os.environ.get("ATLAS_CLAUDE_MODEL", "claude-haiku-5-5")
P_IN, P_OUT = 0.10, 0.50   # USD per 1M tokens for claude-haiku-5-5, prompts up to 100K tokens
COMPONENTS = ("core, scheduler, flows, broker-redis, broker-postgres, broker-sqs, cli, dashboard, "
              "retries, results, observability, auth, docs")
client = anthropic.Anthropic()

def read_jsonl(path: Path) -> list[dict]:
    with path.open(encoding="utf-8") as f:
        return [json.loads(line) for line in f if line.strip()]

issues = {i["id"]: i for i in read_jsonl(DATA_DIR / "issues" / "issues.jsonl")}
correct, latencies, cost = 0, [], 0.0
for g in read_jsonl(DATA_DIR / "eval" / "extraction_gold.jsonl")[:10]:
    issue = issues[g["issue_id"]]
    t0 = time.perf_counter()
    msg = client.messages.create(
        model=CLAUDE_MODEL,
        max_tokens=2_000,   # required; leaves room in case a thinking block comes first
        system=f"You triage issues for Orbit. Reply with exactly one component from: {COMPONENTS}.",
        messages=[{"role": "user", "content": f"Title: {issue['title']}\n\n{issue['body'][:6000]}"}],
    )
    latencies.append(time.perf_counter() - t0)
    text = "".join(b.text for b in msg.content if b.type == "text")   # skip thinking blocks
    correct += text.strip().strip(".`'\"").strip().lower().removeprefix("comp:") in g["components"]
    cost += (msg.usage.input_tokens * P_IN + msg.usage.output_tokens * P_OUT) / 1_000_000
print(f"{CLAUDE_MODEL:<17} {correct}/10  p50 {statistics.median(latencies):.1f} s  "
      f"max {max(latencies):.1f} s  ${cost:.4f}")

# script_id: day_007_choosing_a_model/pick_tier
# Turn a comparison table into a decision: the cheapest config that clears the bar. No key needed.
from typing import TypedDict

class Row(TypedDict):
    model: str
    effort: str
    correct: int
    p50_s: float
    usd_per_1k: float

def pick_tier(rows: list[Row], *, min_correct: int = 8, max_p50_s: float | None = None) -> Row | None:
    """Cheapest row meeting the bars. None means: rethink the task or prompt before buying a bigger model."""
    ok = [r for r in rows
          if r["correct"] >= min_correct and (max_p50_s is None or r["p50_s"] <= max_p50_s)]
    return min(ok, key=lambda r: r["usd_per_1k"], default=None)

ROWS: list[Row] = [  # illustrative values, NOT measurements: paste your own from compare_tiers
    {"model": "gpt-6-luna", "effort": "low", "correct": 7, "p50_s": 0.9, "usd_per_1k": 0.13},
    {"model": "gpt-6-luna", "effort": "high", "correct": 8, "p50_s": 6.5, "usd_per_1k": 0.95},
    {"model": "gpt-6.1-terra", "effort": "low", "correct": 8, "p50_s": 1.1, "usd_per_1k": 1.21},
    {"model": "gpt-6.1-sol", "effort": "low", "correct": 9, "p50_s": 1.6, "usd_per_1k": 2.38},
]
for bar, latency in [(7, None), (8, None), (8, 2.0), (10, None)]:
    best = pick_tier(ROWS, min_correct=bar, max_p50_s=latency)
    choice = f"{best['model']} @ {best['effort']}" if best else "nothing qualifies"
    print(f"need >= {bar}/10, p50 <= {latency or 'any'} s → {choice}")

need >= 7/10, p50 <= any s → gpt-6-luna @ low
need >= 8/10, p50 <= any s → gpt-6-luna @ high
need >= 8/10, p50 <= 2.0 s → gpt-6.1-terra @ low
need >= 10/10, p50 <= any s → nothing qualifies