Parallele Fragen
Führt ein regulatorisches Briefing mit 13 Fragen über den Wikipedia-Artikel zur DSGVO aus und zeigt, dass das Bündeln aller Fragen in einen einzigen TypeSafe-Aufruf 12.2x günstiger und 10.0x schneller ist, ohne dass sich die Antworten ändern.
Du hast ein Dokument und N Fragen dazu. Du kannst eine Anfrage mit allen N Fragen senden oder N Anfragen mit je einer Frage. Bei TypeSafe kommen die Antworten in beiden Fällen gleich heraus: Jede Frage wird für sich gegen das Dokument bewertet, ihre Antwort hängt also nicht davon ab, was sonst noch in der Anfrage steht.
Um das zu prüfen, stellt das Cookbook jede Frage mehrmals auf beide Arten – alle N in einer Anfrage und eine Frage pro Anfrage – und vergleicht die Standardabweichung zwischen den Durchläufen: wie weit sich eine Antwort von einer Wiederholung zur nächsten bewegt. Welches Rauschen eine Frage hat, hat sie unter beiden Batching-Strategien; das Batching fügt keines hinzu. Die meisten Antworten kamen in beiden Fällen über alle 5 Wiederholungen identisch zurück, derselbe Wert bei jedem Aufruf, Standardabweichung exakt 0.0.
Kosten und Geschwindigkeit ändern sich dagegen schon. Das Dokument dominiert jede Anfrage. N Einzelfragen-Aufrufe zahlen N-mal dafür, in N Roundtrips; der gebündelte Aufruf zahlt einmal. Je größer das Dokument, desto näher kommt diese Ersparnis an ein volles Nx heran.
Der Fall hier ist ein regulatorisches Briefing. Das Dokument ist der Wikipedia-Artikel
zur DSGVO (rund 54,000 Zeichen, eine dokumentdominierte Arbeitslast, bei der das Dokument
den größten Teil jeder Anfrage ausmacht), und ein Compliance-Team möchte 13 Dinge geprüft
haben: 8 Noul-Fragen, 2 Choice-Fragen und 3 Score-Fragen.
Einrichtung
pip install ipython 'cooksafe>=0.2.0,<0.3.0'Lege dann TYPESAFE_API_KEY fest.
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"))Das Dokument: der Wikipedia-Artikel zur DSGVO
Als reiner Text aus einer festgeschriebenen Revision des Artikels abgerufen und in
json_cache.json neben den API-Aufrufen zwischengespeichert, damit das Dokument und
seine Zahlen fest bleiben, auch wenn der live stehende Artikel weiter bearbeitet wird.
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📄 Lies die festgeschriebene Wikipedia-Revision
Die Fragen: 8 nouls + 2 choices + 3 scores
Pro Antwort wird genau eine Zahl erfasst, je nach Typ:
Noul: die Wahrscheinlichkeit für „Ja“.Choice: die maximale Wahrscheinlichkeit, also die Wahrscheinlichkeit des gewählten Labels.criteriaordnet jedem Label seine Bedeutung zu.Score: der auf 0–1 normalisierte Score, also der Score geteilt durch die oberste Stufe.criterialistet die Beschreibungen der Stufen auf, von Stufe 0 an aufwärts.
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",
}Auf zwei Wegen fragen, jeweils 5-mal
ask() sendet eine beliebige Teilmenge der Fragen zusammen mit dem Dokument und reduziert
jede Antwort auf ihre eine erfasste Zahl. Das Dokument ist bei jedem Aufruf byte-identisch.
Beide Batching-Strategien laufen RUNS = 5-mal, was jeder Frage 5 Antworten pro Strategie
gibt — genug, um den Mittelwert (stimmen die beiden überein?) und die Standardabweichung
(fügt das Batching Rauschen hinzu?) zu vergleichen. Die Aufrufe werden in json_cache.json
zwischengespeichert, das mit dem Cookbook ausgeliefert wird, sodass ein erneutes Rendern
kostenlos ist; lösche die Datei, um alles live neu auszuführen.
@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 RUNSBatching ändert die Antworten nicht
Pro Frage: der Mittelwert und die Standardabweichung ihrer erfassten Zahl über die 5 Durchläufe, unter jeder Batching-Strategie. Wenn das Batching die Antworten änderte, würden die gebündelten Spalten von den einzelnen Spalten abweichen. Ein verschobener Mittelwert ist Bias. Eine größere Standardabweichung ist Rauschen.
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.0000Die Tabelle nach Fragetyp gelesen:
- Choices, Scores und sechs der acht nouls kommen über die 5 Wiederholungen identisch zurück: Standardabweichung exakt 0.0 unter beiden Batching-Strategien, wobei jeder gebündelte und jeder einzelne Aufruf dieselbe Zahl liefert. Ein Aufruf mit N Fragen liefert dieselben Antworten wie N Aufrufe mit je einer Frage.
breach_72hundcriminal_penaltiestragen ein wenig Sampling-Rauschen von Durchlauf zu Durchlauf, und es ist unter beiden Batching-Strategien gleich groß, wobei die Mittelwerte innerhalb dieses Rauschens übereinstimmen. Das Rauschen ist eine Eigenschaft der Frage, nicht der Art des Bündelns: Batching verschiebt die Antwort weder noch fügt es Varianz hinzu.
So oder so gibt es keinen Batching-Effekt: Die Antwort keiner Frage hängt von den 12 anderen Fragen ab, die sich ihre Anfrage teilen.
Der einzige Unterschied: Kosten und Geschwindigkeit
Gleiche Antworten, andere Rechnung. Der rund 54,000 Zeichen lange Artikel dominiert jede Anfrage, also:
- Kosten: Die 13 Einzelfragen-Aufrufe senden den Artikel 13-mal erneut; der gebündelte Aufruf sendet ihn einmal. Diese Ersparnis gilt, egal wie du die Aufrufe startest.
- Geschwindigkeit: Die Zahl summiert die Latenzen der 13 Einzelaufrufe und nimmt also an, dass sie nacheinander laufen. Startest du sie nebenläufig, schrumpft die Lücke, aber die 13-fachen Token-Kosten bleiben.
Token-Zahlen und Latenzen werden neben den Antworten zwischengespeichert; die Kosten werden danach berechnet, und beides wird über die 5 Durchläufe gemittelt.
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 fasterIm TypeSafe-Playground öffnen
Derselbe Artikel und dieselben 13 Fragen, in einen Share-Link gepackt. Öffne ihn, um das Briefing live neu auszuführen; dieselben Zahlen kommen zurück.
playground_link = make_playground_link(
{"article": DOCUMENT}, QUESTIONS, models=[TYPESAFE_MODEL]
)
display(
Markdown(
f"🔗 [Open this article + questions in the TypeSafe playground]({playground_link})"
)
)