并行问题

你有一份文档和关于它的 N 个问题。你可以发送一个包含所有 N 个问题的请求，或者发送 N 个请求，每个请求包含一个问题。使用 TypeSafe，无论采用哪种方式，答案都是一致的：每个问题都是针对文档单独评分的，因此其答案不依赖于请求中的其他内容。

为了验证这一点，cookbook 以两种方式多次询问每个问题——所有 N 个问题在一个请求中，以及每个请求一个问题——并比较运行间的标准差：即答案在一次重复到下一次重复之间的变化幅度。无论一个问题存在多少噪声，在两种批处理策略下它都存在。批处理并未增加任何噪声。大多数答案在所有 5 次重复中完全相同，每次调用返回相同的值，标准差恰好为 0.0。

成本和速度确实会发生变化。文档主导了每个请求。N 个单问题调用需要支付 N 次文档费用，涉及 N 次往返；而批处理调用只需支付一次。文档越大，这种节省就越接近完全的 Nx 倍。

本案例的场景是一份监管简报。文档是维基百科上关于 GDPR 的文章（约 54,000 个字符，属于文档主导型工作负载，其中文档占据了每个请求的大部分），合规团队希望检查 13 项内容：8 个 Noul 问题，2 个 Choice 问题，以及 3 个 Score 问题。

设置

pip install ipython "typesafe-sdk>=0.5.7" cooksafe --extra-index-url https://pypi.typesafe.ai/

然后设置 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"))

文档：关于 GDPR 的维基百科文章

从该文章的固定版本中获取纯文本，并缓存在 json_cache.json 中，紧邻 API 调用，因此即使实时文章被编辑，文档及其数据也保持固定。

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

📄 查看置顶的维基百科修订版

问题：8 个 Noul + 2 个 Choice + 3 个 Score

按类型，每个答案跟踪一个数值：

• Noul：表示“是”的概率。
• Choice：最大概率，即被选中标签上的概率。criteria 将每个标签映射到其含义。
• Score：归一化到 0-1 的分数，即分数除以最高层级。criteria 从第 0 级开始列出各级别的描述。

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",
}

两种提问方式，各执行 5 次

ask() 将任意子集的问题与文档一起发送，并将每个答案简化为其唯一跟踪的数值。在每次调用中，文档的字节内容完全一致。

两种批处理策略均运行 RUNS = 5 次，为每个问题提供 5 个答案（每种策略一个），足以比较均值（两者是否一致？）和标准差（批处理是否引入噪声？）。调用结果会缓存到 json_cache.json 中，该文件随 cookbook 一起提供，因此重新渲染无需额外开销；删除该文件即可重新进行实时运行。

@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 RUNS

批处理不会改变答案

针对每个问题：在每种批处理策略下，其跟踪数值在 5 次运行中的均值和标准差。如果批处理改变了答案，则批处理列的值会与单列的值不同。均值偏移表示偏差，标准差增大表示噪声。

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.0000

按问题类型解读表格：

• 选项、得分以及八个 Noul 中的六个在 5 次重复中完全一致： 在两种批处理策略下，标准差均为 0.0，每次批处理和单次调用返回的数字都相同。一次调用包含 N 个问题与 N 次调用各包含一个问题，所得答案相同。
• breach_72h 和 criminal_penalties 存在轻微的逐次运行采样噪声，且在两种批处理策略下噪声大小相同，均值在该噪声范围内一致。这种噪声是问题本身的属性，而非批处理方式所致：批处理既不会改变答案，也不会增加方差。

无论哪种情况，均不存在批处理效应：没有任何问题的答案依赖于与其共享请求的其他 12 个问题。

唯一的区别：成本和速度

答案相同，账单不同。由于约 54,000 字符的文章主导了每个请求，因此：

• 成本：13 次单次问题调用会重新发送 13 次文章；而批处理调用仅发送一次。无论你如何发起调用，这一节省都成立。
• 速度：该数值汇总了 13 次单次调用的延迟，因此假设它们是依次运行的。如果并发发起调用，差距会缩小，但 13 倍的 token 成本保持不变。

Token 计数和延迟与答案一起缓存；成本随后应用，且两者均基于 5 次运行取平均值。

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

在 TypeSafe 游乐场中打开

相同的文章和相同的 13 个问题，已打包成一个分享链接。打开它以重新运行简报；结果数字将再次返回。

playground_link = make_playground_link(
    {"article": DOCUMENT}, QUESTIONS, models=[TYPESAFE_MODEL]
)
display(
    Markdown(
        f"🔗 [Open this article + questions in the TypeSafe playground]({playground_link})"
    )
)

在 TypeSafe 游乐场中打开本文 + 问题 →