Preguntas en paralelo
Ejecuta un informe regulatorio de 13 preguntas sobre el artículo de Wikipedia del RGPD, mostrando que agrupar todas las preguntas en una sola llamada a TypeSafe es 12.2x más barato y 10.0x más rápido sin cambios en las respuestas.
Tienes un documento y N preguntas sobre él. Puedes enviar una solicitud con las N preguntas, o N solicitudes con una pregunta cada una. Con TypeSafe las respuestas salen iguales de las dos formas: cada pregunta se puntúa por su cuenta contra el documento, así que su respuesta no depende de qué más haya en la solicitud.
Para comprobarlo, el cookbook hace cada pregunta varias veces de las dos formas —las N en una sola solicitud y una pregunta por solicitud— y compara la desviación estándar entre ejecuciones: cuánto se mueve una respuesta de una repetición a la siguiente. El ruido que tenga una pregunta lo tiene bajo ambas estrategias de procesamiento por lotes; el procesamiento por lotes no añade ninguno. La mayoría de las respuestas volvieron idénticas en las 5 repeticiones de las dos formas, el mismo valor en cada llamada, desviación estándar exactamente 0.0.
El coste y la velocidad sí cambian. El documento domina cada solicitud. N llamadas de una sola pregunta lo pagan N veces, en N viajes de ida y vuelta; la llamada por lotes lo paga una vez. Cuanto más grande es el documento, más se acerca ese ahorro a un Nx completo.
El caso aquí es un informe regulatorio. El documento es el artículo de Wikipedia sobre el RGPD (unos 54,000 caracteres, una carga de trabajo dominada por el documento, donde el documento es la mayor parte de cada solicitud), y un equipo de cumplimiento quiere que se comprueben 13 cosas: 8 preguntas Noul, 2 preguntas Choice y 3 preguntas Score.
Configuración
pip install ipython 'cooksafe>=0.2.0,<0.3.0'luego define TYPESAFE_API_KEY.
import json
import os
import urllib.request
from pathlib import Path
from statistics import mean, stdev
from time import perf_counter
from cooksafe import JsonCache, make_playground_link
from IPython.display import Markdown, display
from typesafe_sdk import Choice, ChoiceAnswer, Noul, NoulAnswer, Score, TypeSafeClient
TYPESAFE_MODEL = "jev-1.12"
PRICE = (
0.042,
0.00,
) # $ per 1M tokens (input, output); TypeSafe jev-1.12 as of 2026-09, see README
RUNS = 5 # repeats per batching strategy, to estimate each answer's run-to-run std dev
client = TypeSafeClient(api_key=os.environ["TYPESAFE_API_KEY"], timeout=120.0)
json_cache = JsonCache(Path("json_cache.json"))El documento: el artículo de Wikipedia sobre el RGPD
Se obtiene como texto plano de una revisión fijada del artículo y se guarda en caché en json_cache.json junto a las llamadas a la API, de modo que el documento y sus cifras permanecen fijos aunque el artículo en vivo se siga editando.
WIKIPEDIA_REVISION = 1363040264 # "General Data Protection Regulation", as of 2026-07
@json_cache
def fetch_article(revision_id: int) -> str:
url = (
"https://en.wikipedia.org/w/api.php?action=query&format=json"
f"&prop=extracts&explaintext=1&revids={revision_id}"
)
request = urllib.request.Request(
url, headers={"User-Agent": "typesafe-cookbook/1.0"}
)
with urllib.request.urlopen(request) as response:
pages = json.loads(response.read())["query"]["pages"]
return next(iter(pages.values()))["extract"]
DOCUMENT = {
"source": f"https://en.wikipedia.org/?oldid={WIKIPEDIA_REVISION}",
"text": fetch_article(WIKIPEDIA_REVISION),
}
print(f"{len(DOCUMENT['text']):,} characters")
display(Markdown(f"📄 [Read the pinned Wikipedia revision]({DOCUMENT['source']})"))53,777 characters📄 Lee la revisión fijada de Wikipedia
Las preguntas: 8 nouls + 2 choices + 3 scores
Un número registrado por respuesta, según el tipo:
Noul: la probabilidad de que la respuesta sea sí.Choice: la probabilidad máxima, la probabilidad de la etiqueta elegida.criteriaasigna a cada etiqueta su significado.Score: la puntuación normalizada a 0–1, la puntuación dividida por el nivel superior.criteriaenumera las descripciones de los niveles, desde el nivel 0 hacia arriba.
QUESTIONS = {
"breach_72h": Noul(
instructions="Must a personal data breach be reported to the supervisory authority within 72 hours?"
),
"applies_non_eu": Noul(
instructions="Does the regulation apply to organisations established outside the EU that offer goods or services to people in the EU?"
),
"dpo_all_orgs": Noul(
instructions="Must every organisation appoint a Data Protection Officer, regardless of what data it processes?"
),
"pre_ticked_consent": Noul(
instructions="Can valid consent be obtained through pre-ticked boxes or inactivity?"
),
"right_erasure": Noul(
instructions="Does the regulation grant individuals a right to erasure of their personal data?"
),
"data_portability": Noul(
instructions="Does the regulation include a right to data portability?"
),
"us_federal_law": Noul(instructions="Is the GDPR a United States federal law?"),
"criminal_penalties": Noul(
instructions="Does the GDPR itself impose criminal penalties such as imprisonment?"
),
"instrument_type": Choice(
instructions="What kind of EU legal instrument is the GDPR?",
criteria={
"Regulation": "Directly binding law in all member states, no national implementation needed.",
"Directive": "Sets goals that member states implement through national law.",
"Treaty": "An international treaty between states.",
"Recommendation": "Non-binding guidance.",
},
),
"max_fine": Choice(
instructions="What is the maximum administrative fine for the most serious infringements?",
criteria={
"TwentyM_or_4pct": "Up to EUR 20 million or 4% of annual worldwide turnover, whichever is greater.",
"TenM_or_2pct": "Up to EUR 10 million or 2% of annual worldwide turnover, whichever is greater.",
"FixedCap": "A fixed amount not tied to turnover.",
"NoFines": "The GDPR provides no administrative fines.",
},
),
"individual_rights": Score(
instructions="How strong are the rights the GDPR grants to individuals over their data?",
criteria=[
"None: individuals get no rights over their data.",
"Weak: a right to be informed, but little control.",
"Moderate: access and correction rights, but limited means to act on them.",
"Strong: access, erasure, portability, and objection rights, with enforcement behind them.",
],
),
"penalty_severity": Score(
instructions="How severe are the penalties the GDPR provides for non-compliance?",
criteria=[
"None: no penalties of any kind.",
"Symbolic: small fixed fines unlikely to change behavior.",
"Substantial: fines large enough to matter to most companies.",
"Severe: fines scaled to global revenue, material even to the largest companies.",
],
),
"compliance_burden": Score(
instructions="How heavy is the compliance burden the GDPR places on organisations?",
criteria=[
"Negligible: no meaningful obligations.",
"Light: a few notices and disclosures.",
"Moderate: documented processes and some dedicated roles for larger processors.",
"Heavy: records, impact assessments, officers, and breach procedures for many organisations.",
"Extreme: obligations so demanding that ordinary organisations cannot fully comply.",
],
),
}
N = len(QUESTIONS)
METRIC = { # question type -> the one number we track per answer
Noul: "p(yes)",
Choice: "max prob",
Score: "normalized score",
}Pregúntalo de dos formas, 5 veces cada una
ask() envía cualquier subconjunto de las preguntas junto con el documento y reduce cada respuesta a su único número registrado. El documento es idéntico byte a byte en cada llamada.
Ambas estrategias de procesamiento por lotes se ejecutan RUNS = 5 veces, lo que da a cada pregunta 5 respuestas por estrategia, suficiente para comparar la media (¿coinciden las dos?) y la desviación estándar (¿el procesamiento por lotes añade ruido?). Las llamadas se guardan en caché en json_cache.json, que viene con el cookbook, así que volver a renderizar es gratis; elimínalo para volver a ejecutar en vivo.
@json_cache
def ask(keys: tuple[str, ...], run: int):
"""One TypeSafe call -> ({key: tracked metric}, input_tokens, output_tokens, latency_s);
``run`` only forces a distinct live call per repeat."""
started = perf_counter()
response = client.system_one(
state={"article": DOCUMENT},
questions={key: QUESTIONS[key] for key in keys},
model=TYPESAFE_MODEL,
)
values = {}
for key in keys:
answer = response.answers[key]
if isinstance(answer, NoulAnswer):
values[key] = answer.noul
elif isinstance(answer, ChoiceAnswer):
values[key] = max(answer.probabilities.values())
else:
values[key] = answer.score / (len(QUESTIONS[key].criteria) - 1)
return (
values,
response.usage.input_tokens,
response.usage.output_tokens,
perf_counter() - started,
)
def priced(result):
"""({key: metric}, in_tokens, out_tokens, latency) -> ({key: metric}, cost_usd, latency)."""
values, input_tokens, output_tokens, latency = result
return values, input_tokens / 1e6 * PRICE[0] + output_tokens / 1e6 * PRICE[1], latency
# Price after cache retrieval, so a price change needs no new calls.
batched = [
priced(ask(tuple(QUESTIONS), run)) for run in range(RUNS)
] # all N in one call, x RUNS
singles = [
{key: priced(ask((key,), run)) for key in QUESTIONS} for run in range(RUNS)
] # N x 1, x RUNSEl procesamiento por lotes no cambia las respuestas
Por pregunta: la media y la desviación estándar de su número registrado a lo largo de las 5 ejecuciones, bajo cada estrategia de procesamiento por lotes. Si el procesamiento por lotes cambiara las respuestas, las columnas por lotes diferirían de las columnas individuales. Una media desplazada es sesgo. Una desviación estándar mayor es ruido.
print(
f"{'question':<22}{'metric':<18}{'batched mean':>13}{'single mean':>12}"
f"{'batched std':>13}{'single std':>12}"
)
for key, question in QUESTIONS.items():
batched_values = [values[key] for values, _cost, _latency in batched]
single_values = [singles[run][key][0][key] for run in range(RUNS)]
print(
f"{key:<22}{METRIC[type(question)]:<18}{mean(batched_values):>13.3f}"
f"{mean(single_values):>12.3f}{stdev(batched_values):>13.4f}{stdev(single_values):>12.4f}"
)question metric batched mean single mean batched std single std
breach_72h p(yes) 0.804 0.814 0.0055 0.0055
applies_non_eu p(yes) 0.990 0.990 0.0000 0.0000
dpo_all_orgs p(yes) 0.030 0.030 0.0000 0.0000
pre_ticked_consent p(yes) 0.040 0.040 0.0000 0.0000
right_erasure p(yes) 0.990 0.990 0.0000 0.0000
data_portability p(yes) 0.990 0.990 0.0000 0.0000
us_federal_law p(yes) 0.010 0.010 0.0000 0.0000
criminal_penalties p(yes) 0.108 0.108 0.0045 0.0084
instrument_type max prob 1.000 1.000 0.0000 0.0000
max_fine max prob 1.000 1.000 0.0000 0.0000
individual_rights normalized score 1.000 1.000 0.0000 0.0000
penalty_severity normalized score 1.000 1.000 0.0000 0.0000
compliance_burden normalized score 0.750 0.750 0.0000 0.0000Leyendo la tabla por tipo de pregunta:
- Los choices, los scores y seis de los ocho nouls vuelven idénticos en las 5 repeticiones: desviación estándar exactamente 0.0 bajo ambas estrategias de procesamiento por lotes, y cada llamada por lotes e individual devuelve el mismo número. Una llamada con N preguntas da las mismas respuestas que N llamadas con una pregunta cada una.
breach_72hycriminal_penaltiesarrastran algo de ruido de muestreo entre ejecuciones, y tiene el mismo tamaño bajo ambas estrategias de procesamiento por lotes, con las medias coincidiendo dentro de ese ruido. El ruido es una propiedad de la pregunta, no de cómo agrupas: el procesamiento por lotes ni desplaza la respuesta ni añade varianza.
De una forma u otra, no hay efecto del procesamiento por lotes: la respuesta de ninguna pregunta depende de las otras 12 preguntas que comparten su solicitud.
La única diferencia: coste y velocidad
Mismas respuestas, factura distinta. El artículo de unos 54,000 caracteres domina cada solicitud, así que:
- Coste: las 13 llamadas de una sola pregunta reenvían el artículo 13 veces; la llamada por lotes lo envía una vez. Este ahorro se mantiene sin importar cómo lances las llamadas.
- Velocidad: la cifra suma las 13 latencias de las llamadas individuales, así que da por supuesto que se ejecutan una tras otra. Lánzalas en paralelo y la diferencia se reduce, pero el coste de tokens 13x se mantiene.
Los recuentos de tokens y las latencias se guardan en caché junto a las respuestas; el coste se aplica después, y ambas cosas se promedian sobre las 5 ejecuciones.
batched_cost = mean(cost for _values, cost, _latency in batched)
batched_latency = mean(latency for _values, _cost, latency in batched)
singles_cost = mean(
sum(singles[run][key][1] for key in QUESTIONS) for run in range(RUNS)
)
singles_latency = mean(
sum(singles[run][key][2] for key in QUESTIONS) for run in range(RUNS)
)
print(f"{'batching':<24}{'calls':>6}{'cost':>12}{'total time':>12}")
print(
f"{f'one call, all {N}':<24}{1:>6}{'$' + format(batched_cost, '.6f'):>12}{format(batched_latency, '.2f') + 's':>12}"
)
print(
f"{f'{N} calls, one each':<24}{N:>6}{'$' + format(singles_cost, '.6f'):>12}{format(singles_latency, '.2f') + 's':>12}"
)
print(
f"\nbatching: {singles_cost / batched_cost:.1f}x cheaper, {singles_latency / batched_latency:.1f}x faster"
)batching calls cost total time
one call, all 13 1 $0.000497 0.27s
13 calls, one each 13 $0.006090 2.71s
batching: 12.2x cheaper, 10.0x fasterÁbrelo en el playground de TypeSafe
El mismo artículo y las mismas 13 preguntas, empaquetados en un enlace para compartir. Ábrelo para volver a ejecutar el informe en vivo; vuelven los mismos números.
playground_link = make_playground_link(
{"article": DOCUMENT}, QUESTIONS, models=[TYPESAFE_MODEL]
)
display(
Markdown(
f"🔗 [Open this article + questions in the TypeSafe playground]({playground_link})"
)
)