API Decisions de OpenAI: Respuestas Tipadas, No Texto

Request: tres preguntas sobre un input

HTTP plano con el body documentado: model, input, questions. Seteá OPENAI_API_KEY. No corrí esto.

import os
import requests

URL = "https://api.openai.com/v1/decisions"
HEADERS = {
    "Authorization": "Bearer " + os.environ["OPENAI_API_KEY"],
    "Content-Type": "application/json",
}

QUESTIONS = [
    {"type": "predicate", "name": "refund_requested",
     "instructions": "Does the customer explicitly ask for a refund?"},
    {"type": "choice", "name": "team",
     "instructions": "Which team should handle this ticket?",
     "choices": [
         {"value": "billing", "description": "Payments, invoices, refunds."},
         {"value": "technical", "description": "Bugs, errors, outages."},
         {"value": "shipping", "description": "Delivery and damaged items."},
         {"value": "other", "description": "None of the above fits."},
     ]},
    {"type": "score", "name": "urgency",
     "instructions": "How urgent is this ticket?",
     "levels": [
         {"label": "Can wait", "description": "No time pressure."},
         {"label": "This week", "description": "Needs a reply soon."},
         {"label": "Today", "description": "Customer is blocked now."},
     ]},
]

def decide(text, timeout=10):
    body = {"model": "gpt-6-luna", "input": text, "questions": QUESTIONS}
    r = requests.post(URL, headers=HEADERS, json=body, timeout=timeout)
    r.raise_for_status()
    return {a["name"]: a for a in r.json()["answers"]}

Forma de la response (un boceto, no output real)

Las respuestas vienen en el orden de las preguntas. Todos los valores de abajo son placeholders, no resultados medidos.

{
  "answers": [          // same order as your questions
    { "type": "predicate", "name": "refund_requested", "probability": <0..1> },
    { "type": "choice", "name": "team", "choice": "<one of your values>",
      "confidence": <0..1>,
      "probabilities": [ { "value": "billing", "probability": <0..1> }, ... ] },
    { "type": "score", "name": "urgency", "score": <weighted level index, from 0>,
      "confidence": <0..1>,
      "probabilities": [ { "value": 0, "label": "Can wait", "probability": <0..1> }, ... ] }
    // any question may instead return { "type": "refusal", "name": "..." }
  ],
  "model": "<string>",
  "usage": { "input_tokens": <n>, "output_tokens": <n>, "total_tokens": <n>, ... }
}

Policy gate en tu código

Los umbrales son placeholders hasta que la sección 4 los fije con tus datos etiquetados.

def route(text, t_refund=0.85, t_team=0.80, t_urgent=1.5):
    a = decide(text)
    refund, team, urgency = a["refund_requested"], a["team"], a["urgency"]
    if "refusal" in (refund["type"], team["type"], urgency["type"]):
        return "human_review"            # a question can be refused
    if team["choice"] == "other" or team["confidence"] < t_team:
        return "human_review"            # closed set did not fit, or unsure
    if refund["probability"] >= t_refund:
        return "refund_checks"           # code verifies amount, identity, limits
    if urgency["score"] >= t_urgent:
        return team["choice"] + ":priority"
    return team["choice"]

Matriz de confusión y latencia

JSONL etiquetado de entrada, precision y recall por clase de salida. Las refusals cuentan como errores, que es como deberías tratarlas.

import json
import time
from collections import Counter, defaultdict

def collect(path):
    rows, lat = [], []
    for line in open(path, encoding="utf-8"):
        ex = json.loads(line)  # {"text": ..., "team": ..., "refund": true/false}
        t0 = time.perf_counter()
        rows.append((ex, decide(ex["text"])))
        lat.append((time.perf_counter() - t0) * 1000)
    lat.sort()
    print("p50 ms", lat[len(lat) // 2], "p95 ms", lat[int(len(lat) * 0.95)])
    return rows

def confusion(rows, labels):
    m = defaultdict(Counter)
    for ex, a in rows:
        t = a["team"]
        m[ex["team"]][t["choice"] if t["type"] == "choice" else "REFUSAL"] += 1
    for gold in labels:
        tp = m[gold][gold]
        fn = sum(m[gold].values()) - tp
        fp = sum(m[g][gold] for g in labels if g != gold)
        p = tp / (tp + fp) if tp + fp else 0.0
        r = tp / (tp + fn) if tp + fn else 0.0
        print(f"{gold:10s} precision={p:.2f} recall={r:.2f} n={tp + fn}")
    return m

Barrido de umbral y error de calibración

Elegí el umbral más bajo que cumpla tu precision objetivo en una mitad y reportá en la mitad reservada.

def sweep(rows, target_precision=0.95):
    best = None
    for k in range(1, 20):
        t = k / 20
        tp = fp = fn = 0
        for ex, a in rows:
            q = a["refund_requested"]
            hit = q["type"] == "predicate" and q["probability"] >= t
            tp += hit and ex["refund"]
            fp += hit and not ex["refund"]
            fn += (not hit) and ex["refund"]
        p = tp / (tp + fp) if tp + fp else 1.0
        r = tp / (tp + fn) if tp + fn else 0.0
        print(f"t={t:.2f} precision={p:.2f} recall={r:.2f}")
        if best is None and p >= target_precision:
            best = t
    return best  # lowest threshold that meets your precision target

def ece(pairs, bins=10):  # pairs: (stated probability, was it right?)
    buckets = [[] for _ in range(bins)]
    for conf, ok in pairs:
        buckets[min(int(conf * bins), bins - 1)].append((conf, ok))
    total = 0.0
    for b in buckets:
        if b:
            gap = abs(sum(c for c, _ in b) / len(b) - sum(o for _, o in b) / len(b))
            total += len(b) / len(pairs) * gap
    return total

# pairs for the choice question, built from rows:
# (max(p["probability"] for p in a["team"]["probabilities"]), a["team"]["choice"] == ex["team"])

Reintentos acotados y circuit breaker

Un boceto de los dos primeros escalones. El resto son tus adapters, y quien llama trata None como “bajá por la escalera”.

import random
import time
import requests

RETRYABLE = {429, 500, 502, 503, 504}
breaker = {"fails": 0, "until": 0.0}

def decide_resilient(text, tries=3):
    if time.time() < breaker["until"]:
        return None                      # breaker open: skip the call
    for attempt in range(tries):
        try:
            answers = decide(text, timeout=5)
            breaker["fails"] = 0
            return answers
        except requests.HTTPError as e:
            if e.response.status_code not in RETRYABLE:
                raise                    # 400/401/403: fix the request, do not retry
            wait = e.response.headers.get("Retry-After", "")
            time.sleep(float(wait) if wait.isdigit() else 2 ** attempt + random.random())
        except (requests.Timeout, requests.ConnectionError):
            time.sleep(2 ** attempt + random.random())
    breaker["fails"] += 1
    if breaker["fails"] >= 3:
        breaker.update(fails=0, until=time.time() + 60)
    return None  # caller falls to the next rung: secondary model, rules, human

También en este blog

• • Lanzamiento de TypeSafe Jev: un modelo de decisión hosteado con un set de preguntas similar (sí/no, choice y score)
• • GPT-6 Sol vs Luna: tiers, precios y dónde encaja Luna
• • Resumen de OpenAI DevDay 2026: qué más salió para devs