并行提问

同一份文档，N 个问题。你可以发一次请求把 N 个问题一起问，也可以发 N 次请求、每次一个问题。对象信一号来说，两种方式得到的答案是一样的：每个问题都是在文档之上各自独立求值的，一个问题的答案不取决于同一请求里还有哪些问题（原理见推测式扇出）。

为了验证这一点，本文把每个问题用两种方式各问 5 次：一次请求问完全部 13 题，以及每次请求只问 1 题；然后比较每个答案的均值和重复间标准差。结果：13 个问题的结论（是/否、选中项、最可能档位）在全部 10 次调用中完全一致，其中 11 个问题连概率都逐位相同；另外 2 个 Score 题在两种方式之间差了约 0.003（归一化分数），但各自内部 5 次重复毫无波动。

变的是成本和速度。文档占了每次请求的大头：13 次单问要为文档付 13 次钱、走 13 趟往返，合并请求只付一次。实测合并请求便宜 11.3 倍（每次 ¥0.000230 对 ¥0.002602）；耗时在交错测量下快 3.3 倍（测量时服务端正被其他任务占满，绝对耗时偏高，见下文）。

本文的场景是一份合规速查。文档是《中华人民共和国个人信息保护法》全文（9,260 字，文档占了每次请求的绝大部分 token），合规团队想确认 13 件事：8 个 Noul、2 个 Choice、3 个 Score。

环境准备

pip install xiangxin-sdk
export XIANGXIN_API_KEY=sk-xx-...

import json
from pathlib import Path
from statistics import mean, stdev
from time import perf_counter

from xiangxin import Choice, ChoiceAnswer, Noul, NoulAnswer, Score, XiangxinClient

MODEL = "xiangxin-latest"
PRICE_PER_M = 0.042  # 元 / 百万输入 token，输出免费
RUNS = 5             # 每种方式重复 5 次，用来估计每个答案的重复间波动
client = XiangxinClient(timeout=120.0)

数据：《个人信息保护法》全文

正文取自维基文库「中华人民共和国个人信息保护法」页面的固定版本（revid 2612274），用 MediaWiki API 导出纯文本并存成 data/pipl.txt，这样即使页面日后被编辑，本文的数字也不会变。全文 8 章 74 条，9,260 字。依《著作权法》第五条，法律文本不受著作权保护。

DOCUMENT = {
    "标题": "中华人民共和国个人信息保护法",
    "来源": "https://zh.wikisource.org/w/index.php?oldid=2612274",
    "正文": Path("data/pipl.txt").read_text(),
}

开头几行：

== 第一章　总则 ==
第一条　为了保护个人信息权益，规范个人信息处理活动，促进个人信息合理利用，根据宪法，制定本法。
第二条　自然人的个人信息受法律保护，任何组织、个人不得侵害自然人的个人信息权益。
第三条　在中华人民共和国境内处理自然人个人信息的活动，适用本法。
……

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

每个答案只跟踪一个数：

• Noul：回答「是」的概率；
• Choice：最大概率，即选中项的概率；
• Score：归一化分数，即期望分除以最高档位，落在 0–1 之间。

问题里有意放了几道「陷阱题」：72 小时报告时限、罚款上限「二千万或营业额 4%」都是欧盟 GDPR 的规定，不是本法的。

QUESTIONS = {
    "breach_72h": Noul(
        instructions="本法是否明确要求：发生个人信息泄露后，必须在 72 小时内向监管部门报告？"
    ),
    "applies_abroad": Noul(
        instructions="境外机构以向中国境内自然人提供产品或服务为目的处理其个人信息，是否也适用本法？"
    ),
    "dpo_all_orgs": Noul(
        instructions="是否所有个人信息处理者，无论处理多少个人信息，都必须指定个人信息保护负责人？"
    ),
    "refuse_service": Noul(
        instructions="个人不同意处理其非必要的个人信息时，处理者能否以此为由拒绝提供产品或者服务？"
    ),
    "right_deletion": Noul(instructions="本法是否赋予个人请求删除其个人信息的权利？"),
    "data_portability": Noul(
        instructions="个人能否请求将其个人信息转移至其指定的个人信息处理者？"
    ),
    "eu_regulation": Noul(instructions="本法是欧盟制定的法规吗？"),
    "criminal_penalties": Noul(
        instructions="本法条文本身是否直接规定了有期徒刑、拘役等具体刑罚？"
    ),
    "instrument_type": Choice(
        instructions="本法属于哪一类规范性文件？",
        criteria={
            "法律": "由全国人民代表大会或其常务委员会制定。",
            "行政法规": "由国务院制定。",
            "部门规章": "由国务院各部委制定。",
            "推荐性国家标准": "不具强制力的技术标准。",
        },
    ),
    "max_fine": Choice(
        instructions="对情节严重的违法处理个人信息行为，本法规定的最高罚款是多少？",
        criteria={
            "五千万或营业额5%": "五千万元以下或者上一年度营业额百分之五以下。",
            "一百万": "一百万元以下。",
            "二千万或营业额4%": "二千万元以下或者上一年度营业额百分之四以下。",
            "无罚款": "本法不设罚款。",
        },
    ),
    "individual_rights": Score(
        instructions="本法赋予个人对其个人信息的权利有多强？",
        criteria=[
            "没有：个人对自己的信息没有任何权利。",
            "弱：只有知情权，几乎没有控制手段。",
            "中等：可以查阅、更正，但缺少行使权利的途径。",
            "强：查阅、复制、更正、删除、转移、拒绝等权利齐全，并有申请受理和诉讼途径保障。",
        ],
    ),
    "penalty_severity": Score(
        instructions="本法对违法行为规定的处罚有多重？",
        criteria=[
            "没有：不设任何处罚。",
            "象征性：小额固定罚款，难以改变企业行为。",
            "可观：罚款金额对多数企业都有分量。",
            "严厉：罚款与营业额挂钩，即使对最大的企业也有实质影响。",
        ],
    ),
    "compliance_burden": Score(
        instructions="本法给个人信息处理者带来的合规负担有多重？",
        criteria=[
            "可忽略：几乎没有义务。",
            "轻：发布几份告知和说明即可。",
            "中等：需要成文的制度流程，较大的处理者要设专门岗位。",
            "重：内部制度、影响评估、负责人、合规审计、泄露应急等义务覆盖大量处理者。",
            "极重：义务苛刻到普通组织根本无法完全合规。",
        ],
    ),
}
N = len(QUESTIONS)

两种问法，各跑 5 次

ask() 把任意一组问题连同文档发出去，并把每个答案压成它要跟踪的那个数。每次请求里的文档逐字节相同。

两种方式各跑 RUNS = 5 次，于是每个问题在每种方式下都有 5 个答案：比较均值（两种方式是否一致），也比较标准差（合并提问是否引入额外波动）。

def tracked(key, answer):
    if isinstance(answer, NoulAnswer):
        return answer.noul
    if isinstance(answer, ChoiceAnswer):
        return max(answer.probabilities.values())
    return answer.score / (len(QUESTIONS[key].criteria) - 1)


def ask(keys):
    started = perf_counter()
    resp = client.system_one(
        state={"法规": DOCUMENT},
        questions={k: QUESTIONS[k] for k in keys},
        model=MODEL,
    )
    return {
        "values": {k: tracked(k, resp.answers[k]) for k in keys},
        "input_tokens": resp.usage.input_tokens,
        "wall_s": perf_counter() - started,
        "model_ms": resp.model_ms,  # 响应头 x-xiangxin-model-ms
    }


batched = [ask(tuple(QUESTIONS)) for _ in range(RUNS)]                   # 1 次请求问 13 题，× 5
singles = [{k: ask((k,)) for k in QUESTIONS} for _ in range(RUNS)]       # 13 次请求各问 1 题，× 5

完整脚本还多做了两件事

• 把每次调用的结果缓存到 results/calls.json，重新运行不会重复花钱；
• 关掉 SDK 自动重试，遇到 429/5xx 时由脚本等待后重新计时，保证耗时数字里不含排队重试的时间；另外用 AsyncXiangxinClient 把 13 个单题请求并发发出，再测一组墙钟时间。

合并提问不改变答案

每个问题在两种方式下各有 5 个答案，下表列出它们的均值和标准差。如果合并提问会改变答案，「合并」两列和「单问」两列就会对不上：均值偏了是偏差，标准差变大是噪声。

for key, q in QUESTIONS.items():
    b = [r["values"][key] for r in batched]
    s = [singles[run][key]["values"][key] for run in range(RUNS)]
    print(f"{key:<20}{mean(b):>9.3f}{mean(s):>9.3f}{stdev(b):>9.4f}{stdev(s):>9.4f}")

真实输出（results/output.txt，模型 xiangxin-1.0.0）：

问题	指标	合并均值	单问均值	合并 std	单问 std	10 次调用的结论
breach_72h	p(是)	0.130	0.130	0.0000	0.0000	否
applies_abroad	p(是)	0.980	0.980	0.0000	0.0000	是
dpo_all_orgs	p(是)	0.070	0.070	0.0000	0.0000	否
refuse_service	p(是)	0.270	0.270	0.0000	0.0000	否
right_deletion	p(是)	0.990	0.990	0.0000	0.0000	是
data_portability	p(是)	0.980	0.980	0.0000	0.0000	是
eu_regulation	p(是)	0.030	0.030	0.0000	0.0000	否
criminal_penalties	p(是)	0.230	0.230	0.0000	0.0000	否
instrument_type	最大概率	0.960	0.960	0.0000	0.0000	法律
max_fine	最大概率	0.990	0.990	0.0000	0.0000	五千万或营业额5%
individual_rights	归一化分数	0.987	0.987	0.0000	0.0000	第 3 档（强）
penalty_severity	归一化分数	0.857	0.853	0.0000	0.0000	第 3 档（严厉）
compliance_burden	归一化分数	0.665	0.667	0.0000	0.0000	第 3 档（重）

按题型看：

• 8 个 Noul 和 2 个 Choice：两种方式、10 次调用返回的概率逐位相同，标准差为 0。两道 GDPR「陷阱题」都答对了：72 小时时限 p(是) = 0.13，「二千万或营业额 4%」没有被选中（max_fine 把 0.99 的概率给了本法第六十六条的「五千万或营业额 5%」）。
• 3 个 Score：individual_rights 完全相同；penalty_severity 合并时是 0.857、单问时是 0.853，compliance_burden 是 0.665 对 0.667。差距来自某一档概率上 0.01 的移动（API 概率保留两位小数），而且在每种方式内部 5 次重复毫无波动——也就是说它不是随机噪声，而是同一请求布局下稳定复现的一点数值差，推测来自一次前向里同批问题的排布不同。它小到不改变任何结论（最可能档位都是第 3 档），但严格说，「一题一请求」和「十三题一请求」并不保证逐位相同。

另一个值得注意的点：本次运行里象信一号 1.0 的答案没有任何重复间波动，TypeSafe 原文中有两道题出现了约 0.005 的抽样噪声，我们这里没有观察到。

所以合并提问不会让答案变差：没有一道题的结论取决于同一请求里另外 12 道题是什么。

真正的差别：成本与速度

答案一样，账单不一样。9,260 字的法条占了每次请求的绝大部分 token：

• 成本：13 次单问把法条重复发送 13 次，合并请求只发一次。这部分节省与你如何发请求（顺序还是并发）无关。
• 速度：合并请求一趟往返；13 次顺序单问要走 13 趟。并发发出可以缩短墙钟时间，但省不下 token。

b_cost = mean(r["input_tokens"] for r in batched) / 1e6 * PRICE_PER_M
s_cost = mean(sum(singles[run][k]["input_tokens"] for k in QUESTIONS) for run in range(RUNS)) / 1e6 * PRICE_PER_M
print(f"便宜 {s_cost / b_cost:.1f} 倍")

方式	请求数	输入 token	成本（元）
一次请求问完 13 题	1	5,474	0.000230
13 次请求，每次 1 题	13	61,946	0.002602

合并提问便宜 11.3 倍。 单问请求每次 4,731–4,831 个 token，其中文档约占 4,700 个；合并请求只比单问多出 13 道题本身的 token。文档越长、问题越多，这个倍数越接近 N。

耗时

耗时数字要打一个折扣：本次测量时，象信服务同时承载着其他多个实战手册的批量脚本，模型后端几乎满载，请求普遍要排队。所以下面的绝对耗时远高于空闲时的正常水平，只能看相对比例。

为了让两种方式处在同样的负载下，timing.py 做了交错测量：每一轮先后发出 1 次合并请求和 13 次顺序单问，轮与轮之间交换先后顺序；遇到 429/5xx 时等待后重新计时，等待不计入耗时（results/timing_output.txt）：

轮	先后顺序	合并（秒）	13 次顺序单问（秒）	倍数
0	合并 → 单问	14.80	49.89	3.4
1	单问 → 合并	15.53	88.71	5.7
2	合并 → 单问	17.92	45.44	2.5
3	单问 → 合并	12.43	32.67	2.6
4	合并 → 单问	12.91	28.07	2.2

均值：合并 14.72 秒，13 次顺序单问合计 48.96 秒，合并快 3.3 倍；只看响应头 x-xiangxin-model-ms 的模型耗时，也是 3.2 倍（14,412 ms 对 46,586 ms）。单个单问请求耗时的中位数是 2.62 秒，最快 0.99 秒、最慢 33.26 秒，负载波动之大可见一斑。TypeSafe 原文在空闲服务上测得 10.0 倍，我们在这种负载下测不出这么干净的数字，只能如实报告 3.3 倍。

主脚本 main.py 还试了把 13 个单问并发发出（AsyncXiangxinClient + asyncio.gather）。5 轮平均墙钟时间 40.89 秒，仍比合并请求慢；而且每轮都撞上了速率限制或 5xx，分别重试了 64、37、32、64、51 次（这些等待计入了墙钟时间）。一次性打出 13 个带整篇法条的请求，在共享的速率额度下很容易被限流；合并请求只是一个请求，没有这个问题。

讨论

为什么合并不影响答案。 象信一号读一遍 state，在其上对每个问题各自求值，问题之间互不可见。所以把「可能用得上的问题」都塞进一个请求是安全的，这正是推测式扇出的前提。本次实验给出了一个更精确的边界：结论完全不变，概率在个别 Score 题上可能差 0.01 这一级。

省钱的倍数取决于文档占比。 本例文档约 4,700 token、每道题几十 token，所以接近 13 倍。如果 state 很短而问题很长（比如每道题都带一大段选项说明），合并带来的节省就小得多。

什么时候不该合并。

• 后一个问题真正依赖前一个答案，比如沿类目树逐层下钻（见 进阶：结构化），只能分步问；
• 合并后超出上下文限制：单请求 64k token，state 加最长的一道题不超过 32k token（见 API 参考）；
• 你需要对不同问题用不同的 state（比如每题只附上相关条款），那就不是「同一份文档」的场景了。

关于 1.0 的局限。 refuse_service（0.27）和 criminal_penalties（0.23）虽然都答对了「否」，但把握不如其他 Noul 高：前者要把第十六条「不得以个人不同意……为由拒绝提供产品或者服务」和问题的反问句式对上，后者要区分第七十一条的「依法追究刑事责任」（转致刑法）与「本法直接规定刑罚」。这类需要细读条文措辞的问题，建议结合置信度设定阈值，把低把握的答案交给人工复核；9B 模型的能力边界见象信一号 1.0 已知短板。

本文的调用量：主实验 135 次成功调用（5 次合并 + 65 次顺序单问 + 65 次并发单问），交错计时 70 次。所有原始结果在 results/calls.json、results/timing.json。

完整代码

github.com/xiangxinai/xiangxin-cookbooks/tree/main/parallel_questions