Controlling Generation

# script_id: day_006_sampling_reasoning_effort/naive_call
# fragment
# The old habit: no reasoning setting (so effort defaults to "medium") plus temperature=0
client.responses.create(model="gpt-6-luna", input="Summarize issue 42.", temperature=0)
# -> openai.BadRequestError, HTTP 400, naming 'temperature' as unsupported (exact wording: ⚠ verify)

# script_id: day_006_sampling_reasoning_effort/simulators
"""Offline sampling simulators: temperature, top-p and a seeded sampler.

Install: nothing (stdlib only) · Env: none · Services: none · No API key, no network.
"""
import math
import random
from collections import Counter

# Toy distribution after "Orbit retries failed jobs with exponential ..."
NEXT = {"backoff": 0.55, "delay": 0.20, "retries": 0.12, "jitter": 0.08, "sadness": 0.05}

def apply_temperature(probs: dict[str, float], temperature: float) -> dict[str, float]:
    """Reshape a distribution: T < 1 sharpens, T = 1 leaves it alone, T > 1 flattens."""
    if temperature < 0:
        raise ValueError("temperature must be >= 0")
    if temperature == 0:  # the limit as T -> 0: all weight on the top token (greedy)
        top = max(probs, key=probs.get)
        return {tok: float(tok == top) for tok in probs}
    # log(p) recovers the model's raw scores up to a constant; divide by T, then re-normalize
    scaled = {tok: math.log(p) / temperature for tok, p in probs.items() if p > 0}
    peak = max(scaled.values())  # subtract the max so exp() can't overflow
    weights = {tok: math.exp(z - peak) for tok, z in scaled.items()}
    total = sum(weights.values())
    return {tok: w / total for tok, w in weights.items()}

def show(probs: dict[str, float]) -> str:
    return ", ".join(f"{tok} {p:.0%}" for tok, p in probs.items())

for t in (0.1, 0.5, 1.0, 2.0):
    print(f"T={t}: {show(apply_temperature(NEXT, t))}")

T=0.1: backoff 100%, delay 0%, retries 0%, jitter 0%, sadness 0%
T=0.5: backoff 83%, delay 11%, retries 4%, jitter 2%, sadness 1%
T=1.0: backoff 55%, delay 20%, retries 12%, jitter 8%, sadness 5%
T=2.0: backoff 36%, delay 22%, retries 17%, jitter 14%, sadness 11%

# script_id: day_006_sampling_reasoning_effort/simulators
def apply_top_p(probs: dict[str, float], p: float) -> dict[str, float]:
    """Keep the smallest set of top tokens whose probability reaches p, renormalized."""
    if not 0 < p <= 1:
        raise ValueError("top_p must be in (0, 1]")
    kept: dict[str, float] = {}
    cumulative = 0.0
    for tok, prob in sorted(probs.items(), key=lambda kv: kv[1], reverse=True):
        kept[tok] = prob
        cumulative += prob
        if cumulative >= p:  # stop once the nucleus covers p
            break
    total = sum(kept.values())
    return {tok: prob / total for tok, prob in kept.items()}

print("top_p=0.9:", show(apply_top_p(NEXT, 0.9)))
print("top_p=0.5:", show(apply_top_p(NEXT, 0.5)))

top_p=0.9: backoff 58%, delay 21%, retries 13%, jitter 8%
top_p=0.5: backoff 100%

# script_id: day_006_sampling_reasoning_effort/simulators
def sample_next(probs: dict[str, float], *, temperature: float = 1.0,
                top_p: float = 1.0, rng: random.Random) -> str:
    """One draw: temperature first, then the top-p cut, then a weighted random pick."""
    dist = apply_top_p(apply_temperature(probs, temperature), top_p)
    tokens = list(dist)
    return rng.choices(tokens, weights=[dist[t] for t in tokens], k=1)[0]

for t in (0.2, 1.0, 1.5):
    rng = random.Random(42)  # a fixed seed makes the whole run reproducible
    counts = Counter(sample_next(NEXT, temperature=t, rng=rng) for _ in range(1000))
    print(f"T={t}: " + "  ".join(f"{tok} {counts[tok]:>3}" for tok in NEXT))

rng_a, rng_b = random.Random(7), random.Random(7)  # two generators, same seed
first = [sample_next(NEXT, rng=rng_a) for _ in range(5)]
again = [sample_next(NEXT, rng=rng_b) for _ in range(5)]
print("seed 7:", first, "| repeat matches:", first == again)

T=0.2: backoff 988  delay  10  retries   2  jitter   0  sadness   0
T=1.0: backoff 532  delay 211  retries 120  jitter  78  sadness  59
T=1.5: backoff 406  delay 223  retries 157  jitter 122  sadness  92
seed 7: ['backoff', 'backoff', 'delay', 'backoff', 'backoff'] | repeat matches: True

# script_id: day_006_sampling_reasoning_effort/live_sampling
"""Live sampling: temperature only works at reasoning effort "none".

Install: uv add "openai>=3.46,<4" · Env: OPENAI_API_KEY (required), ATLAS_MODEL (optional)
Model:   must offer effort "none" (gpt-6-luna, gpt-6.1-terra, gpt-6-sol as of 2026-11)
"""
import os

from openai import OpenAI

MODEL = os.environ.get("ATLAS_MODEL", "gpt-6-luna")
client = OpenAI()
PROMPT = "Suggest one name for Orbit's retry-backoff helper function. Reply with the name only."

def sample_names(temperature: float, n: int = 3) -> list[str]:
    names = []
    for _ in range(n):
        r = client.responses.create(
            model=MODEL,
            input=PROMPT,
            reasoning={"effort": "none"},  # the only mode that accepts sampling knobs
            temperature=temperature,        # 0-2; tune this OR top_p, not both
            max_output_tokens=50,
        )
        names.append(r.output_text.strip())
    return names

for t in (0.2, 1.2):
    print(f"T={t}: {sample_names(t)}")

Illustrative shape, not a recorded run (your names will differ):
T=0.2: ['<name A>', '<name A>', '<name A>']      # often the same, or nearly the same, name
T=1.2: ['<name B>', '<name C>', '<name D>']      # usually three different names

# script_id: day_006_sampling_reasoning_effort/sampling_errors
"""Two calls that fail with HTTP 400, caught and printed. Rejected requests generate nothing.

Install: uv add "openai>=3.46,<4" · Env: OPENAI_API_KEY (required), ATLAS_MODEL (optional)
"""
import os

import openai
from openai import OpenAI

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

BROKEN = {
    "temperature at effort low": {"model": MODEL, "reasoning": {"effort": "low"}, "temperature": 0},
    "effort none on gpt-6-astra": {"model": "gpt-6-astra", "reasoning": {"effort": "none"}},
}

for label, kwargs in BROKEN.items():
    try:
        client.responses.create(input="Say hi.", max_output_tokens=50, **kwargs)
        print(f"{label}: accepted (the rules changed; re-check the model page)")
    except openai.BadRequestError as e:
        # .param names the offending field when the server reports one
        print(f"{label}: HTTP {e.status_code} param={e.param!r}\n  {e.message}")
    except openai.APIStatusError as e:  # any other HTTP error, e.g. no access to that model
        print(f"{label}: HTTP {e.status_code} {e.message}")

Illustrative shape (exact server wording: ⚠ verify on your run):
temperature at effort low: HTTP 400 param='temperature'
  Error code: 400 - {... 'message': "<says temperature isn't supported with this reasoning setting>" ...}
effort none on gpt-6-astra: HTTP 400 param=<'reasoning.effort' or None>
  Error code: 400 - {... 'message': "<says 'none' isn't a supported effort for this model>" ...}

# script_id: day_006_sampling_reasoning_effort/effort_sweep
"""Sweep reasoning effort on one checkable task: correctness, latency, tokens and cost.

Install: uv add "openai>=3.46,<4" · Env: OPENAI_API_KEY (required), ATLAS_MODEL, SWEEP_EFFORTS
"""
import os
import time

from openai import OpenAI

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

# Compact copy of Day 4's PRICES_PER_1M + estimate_cost (with Day 5's >272K surcharge).
# USD per 1M tokens, Standard tier. 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},
    "gpt-6.1-terra": {"input": 1.00, "cached_input": 0.05, "output": 5.00},
}

def estimate_cost(input_tokens: int, output_tokens: int, model: str = MODEL) -> float:
    price = PRICES_PER_1M[model]  # KeyError on purpose: an unknown model must not cost $0
    long_prompt = input_tokens > 272_000  # whole request: 2x input, 1.5x output
    in_rate = price["input"] * (2 if long_prompt else 1)
    out_rate = price["output"] * (1.5 if long_prompt else 1)
    return (input_tokens * in_rate + output_tokens * out_rate) / 1_000_000

def run_with_effort(prompt: str, effort: str, **kwargs) -> dict:
    """One call at one effort level, reduced to the numbers you compare settings on."""
    model = kwargs.pop("model", MODEL)
    start = time.perf_counter()
    r = client.responses.create(model=model, input=prompt, reasoning={"effort": effort}, **kwargs)
    latency = time.perf_counter() - start
    u = r.usage
    return {
        "text": r.output_text, "status": r.status, "latency_s": latency,
        "output_tokens": u.output_tokens,  # includes reasoning tokens
        "reasoning_tokens": u.output_tokens_details.reasoning_tokens,
        "cost": estimate_cost(u.input_tokens, u.output_tokens, model),
    }

# script_id: day_006_sampling_reasoning_effort/effort_sweep
TASK = (
    "Orbit evaluates cron expressions in UTC. A job is scheduled with `0 9 * * MON`. "
    "The current time is Thursday 2026-11-05 08:30 UTC. When is the next run? "
    "Reply with only: YYYY-MM-DD HH:MM UTC"
)
EXPECTED = "2026-11-09 09:00"
EFFORTS = [e.strip() for e in os.environ.get("SWEEP_EFFORTS", "none,low,medium,high").split(",")]

print(f"{'effort':<8}{'correct':<9}{'latency s':>10}{'out tok':>9}{'reason tok':>11}{'cost $':>11}")
for effort in EFFORTS:
    row = run_with_effort(TASK, effort, max_output_tokens=25_000)  # headroom, see next section
    correct = EXPECTED in row["text"] and row["status"] == "completed"
    print(f"{effort:<8}{str(correct):<9}{row['latency_s']:>10.1f}{row['output_tokens']:>9}"
          f"{row['reasoning_tokens']:>11}{row['cost']:>11.6f}")

Shape of the table, not a recorded run (fill it in from yours; values vary run to run):
effort  correct  latency s  out tok  reason tok     cost $
none    <bool>       <s>      <n>        <n>        <$>
low     <bool>       <s>      <n>        <n>        <$>
medium  <bool>       <s>      <n>        <n>        <$>
high    <bool>       <s>      <n>        <n>        <$>

# script_id: day_006_sampling_reasoning_effort/incomplete
"""Hit the max_output_tokens cap on purpose, then handle incomplete responses explicitly.

Install: uv add "openai>=3.46,<4" · Env: OPENAI_API_KEY (required), ATLAS_MODEL (optional)
"""
import os

from openai import OpenAI
from openai.types.responses import Response

MODEL = os.environ.get("ATLAS_MODEL", "gpt-6-luna")
client = OpenAI()
TASK = ("Orbit evaluates cron in UTC. Schedule `0 9 * * MON`; now is Thursday 2026-11-05 08:30 "
        "UTC. When is the next run? Reply with only: YYYY-MM-DD HH:MM UTC")

r = client.responses.create(model=MODEL, input=TASK, reasoning={"effort": "high"},
                            max_output_tokens=64)  # far too small for high effort
reason = r.incomplete_details.reason if r.incomplete_details else None
print(f"status={r.status} reason={reason} text={r.output_text!r}")
print("reasoning tokens:", r.usage.output_tokens_details.reasoning_tokens)

Illustrative shape, not a recorded run (high effort usually needs far more than 64 tokens):
status=incomplete reason=max_output_tokens text=''
reasoning tokens: <up to 64>

# script_id: day_006_sampling_reasoning_effort/incomplete
def require_complete(resp: Response) -> str:
    """Return output_text only for a completed response; otherwise explain what to change."""
    if resp.status == "completed":
        return resp.output_text
    reason = resp.incomplete_details.reason if resp.incomplete_details else None
    if reason == "max_output_tokens":
        used = resp.usage.output_tokens_details.reasoning_tokens
        raise RuntimeError(
            f"cut off at max_output_tokens={resp.max_output_tokens} ({used} reasoning tokens): "
            "raise the cap (start near 25,000 for reasoning) or lower the effort"
        )
    raise RuntimeError(f"response not usable: status={resp.status} reason={reason}")

for cap in (64, 25_000):
    resp = client.responses.create(model=MODEL, input=TASK, reasoning={"effort": "high"},
                                   max_output_tokens=cap)
    try:
        print(f"cap={cap}: {require_complete(resp)}")
    except RuntimeError as e:
        print(f"cap={cap}: {e}")

Typical shape (your numbers will differ):
cap=64: cut off at max_output_tokens=64 (<n> reasoning tokens): raise the cap (start near 25,000 for reasoning) or lower the effort
cap=25000: 2026-11-09 09:00 UTC

# script_id: day_006_sampling_reasoning_effort/determinism
"""Same prompt, five calls, two modes: count how many distinct answers come back.

Install: uv add "openai>=3.46,<4" · Env: OPENAI_API_KEY (required), ATLAS_MODEL (optional)
"""
import os

from openai import OpenAI

MODEL = os.environ.get("ATLAS_MODEL", "gpt-6-luna")
client = OpenAI()
PROMPT = "In one sentence, explain what a dead-letter queue is for."

def distinct_answers(n: int, **settings) -> int:
    answers = {client.responses.create(model=MODEL, input=PROMPT, **settings).output_text
               for _ in range(n)}
    return len(answers)

print("effort none, T=0:", distinct_answers(5, reasoning={"effort": "none"}, temperature=0,
                                            max_output_tokens=200), "distinct of 5")
print("effort low:      ", distinct_answers(5, reasoning={"effort": "low"},
                                            max_output_tokens=25_000), "distinct of 5")

Illustrative shape (your counts will differ, and can change between runs):
effort none, T=0: <1 to 5> distinct of 5
effort low:       <1 to 5> distinct of 5

# script_id: day_006_sampling_reasoning_effort/settings_by_route
"""Per-route generation settings as data, validated offline before any request is sent.

Install: nothing (stdlib only) · Env: ATLAS_MODEL (optional) · No API key, no network.
"""
import os

MODEL = os.environ.get("ATLAS_MODEL", "gpt-6-luna")
# Models that offer effort "none" (as of 2026-11; check the model pages when this changes)
NONE_MODELS = {"gpt-6-luna", "gpt-6.1-terra", "gpt-6-sol"}
SAMPLING_KEYS = {"temperature", "top_p", "top_logprobs"}

ROUTES: dict[str, dict] = {
    "extract_issue":  {"effort": "none"},
    "label_component": {"effort": "low"},
    "next_cron_run":  {"effort": "medium", "max_output_tokens": 25_000},
    "name_suggestion": {"effort": "none", "temperature": 1.1, "max_output_tokens": 50},
}

def generation_params(route: str, model: str = MODEL) -> dict:
    """Return keyword arguments for responses.create(), or raise before the API would."""
    settings = dict(ROUTES[route])
    effort = settings.pop("effort")
    if effort == "minimal":
        raise ValueError(f"{route}: use effort 'low' instead of 'minimal'")
    if effort == "none" and model not in NONE_MODELS:
        raise ValueError(f"{route}: {model} doesn't offer effort 'none'; use 'low'")
    if effort != "none" and SAMPLING_KEYS & settings.keys():
        raise ValueError(f"{route}: sampling knobs need effort 'none', got {effort!r}")
    if not 0 <= settings.get("temperature", 1.0) <= 2:
        raise ValueError(f"{route}: temperature must be within 0-2")
    return {"model": model, "reasoning": {"effort": effort}, **settings}

for name in ROUTES:
    print(f"{name:<16} {generation_params(name)}")
try:
    generation_params("name_suggestion", model="gpt-6-astra")
except ValueError as e:
    print("caught before sending:", e)

extract_issue    {'model': 'gpt-6-luna', 'reasoning': {'effort': 'none'}}
label_component  {'model': 'gpt-6-luna', 'reasoning': {'effort': 'low'}}
next_cron_run    {'model': 'gpt-6-luna', 'reasoning': {'effort': 'medium'}, 'max_output_tokens': 25000}
name_suggestion  {'model': 'gpt-6-luna', 'reasoning': {'effort': 'none'}, 'temperature': 1.1, 'max_output_tokens': 50}
caught before sending: name_suggestion: gpt-6-astra doesn't offer effort 'none'; use 'low'