第 12 章：并行提问

本文整理自 Datawhale 开源项目 datawhalechina/jev-cookbook（CC BY-NC-SA 4.0），源文件：main/04_实战指南/03_并行提问.ipynb。

本 notebook 把中文镜像站中的 Cookbook 改写成可以逐格运行、修改输入并观察结果的最小实验。 比较一次批量提问与逐条提问，并观察按 ID 取答案。

运行方式与 ../03_架构模式/01_架构模式.ipynb 一致：有有效的 TYPESAFE_API_KEY 时调用真实的 TypeSafe API；没有 Key 或返回 401 时使用内置的离线示例答案。后续代码不区分两种模式，便于先学习 控制流，再切换到真实模型观察概率和置信度。

学习提示：先顺序运行全部单元格，再回到“定义 state”或“定义问题”的单元格修改内容，重新运行后面的单元格。 API Key 只从环境变量读取，不能写进 notebook。

0. 准备

0.1 安装依赖

%pip install -q -U typesafe-sdk

0.2 创建客户端

import os
import statistics
import time
from pprint import pprint

from typesafe_sdk import (
    Choice,
    Score,
    Noul,
    TypeSafeClient,
    TypeSafeAuthenticationError,
)

API_KEY = os.environ.get("TYPESAFE_API_KEY", "")
client = TypeSafeClient(api_key=API_KEY, model="jev-latest") if API_KEY else None
print("客户端已创建：模型=jev-latest，Key=", "已配置" if API_KEY else "未配置（将使用离线示例）")

0.3 离线响应与统一调用入口

class _FakeAnswer:
    def __init__(self, type_, **values):
        self.type = type_
        for key, value in values.items():
            setattr(self, key, value)


class _FakeResponse:
    def __init__(self, answers):
        self.answers = answers
        self.nouls = {k: v for k, v in answers.items() if v.type == "noul"}
        self.choices = {k: v for k, v in answers.items() if v.type == "choice"}
        self.scores = {k: v for k, v in answers.items() if v.type == "score"}
        self.model = "jev-latest（离线示例）"
        self.usage = _FakeAnswer("usage", input_tokens=0, output_tokens=0)


class TS:
    offline = False
    _warned = False

    @classmethod
    def call(cls, state, questions, offline_answers):
        if client is None:
            cls.offline = True
            if not cls._warned:
                cls._warned = True
                print("⚠️ 未设置有效 TYPESAFE_API_KEY，以下输出使用内置离线示例。")
            return _FakeResponse(offline_answers)
        try:
            return client.system_one(state, questions)
        except TypeSafeAuthenticationError:
            cls.offline = True
            if not cls._warned:
                cls._warned = True
                print("⚠️ 未设置有效 TYPESAFE_API_KEY，以下输出使用内置离线示例。")
            return _FakeResponse(offline_answers)


def answer_line(name, answer):
    if answer.type == "noul":
        return f"{name}: noul={answer.noul:.2f}"
    if answer.type == "choice":
        return f"{name}: choice={answer.choice} confidence={answer.confidence:.2f}"
    return f"{name}: score={answer.score:.2f} confidence={answer.confidence:.2f}"


print("模式：", "离线示例" if TS.offline else "真实 API（首次调用后确定）")

0.4 连通性测试

if client is None:
    TS.offline = True
    print("⚠️ API Key 未设置，后续单元格使用离线示例。")
else:
    try:
        ping = client.system_one("你好", {"is_greeting": Noul(instructions="这段文字是在打招呼吗？")})
        print("✅ API 连通正常，后续单元格会使用真实结果。")
    except TypeSafeAuthenticationError:
        TS.offline = True
        print("⚠️ API Key 无效，后续单元格使用离线示例。")

1. 并行提问

把相互独立的问题放入一次 system_one 请求，再与逐条调用比较。批量请求的答案按 ID 返回，代码可以 只消费当前分支真正需要的字段。

1.1 定义文章状态与问题集合

ARTICLE = '公司宣布下季度把客服、退款和安全审计流程统一到一个工作台。'
QUESTIONS = {
    "about_refund": Noul(instructions="这段文字是否提到退款？"),
    "about_security": Noul(instructions="这段文字是否提到安全审计？"),
    "topic": Choice(
        instructions="这段文字的主要主题是什么？",
        criteria={"product": "产品变化", "operations": "运营流程", "policy": "政策公告"},
    ),
    "urgency": Score(
        instructions="这段文字表达的紧迫程度",
        criteria=["没有紧迫性", "需要近期关注", "需要立即处理"],
    ),
}
print("state/questions 已定义：文章长度=", len(ARTICLE), "，问题数=", len(QUESTIONS))

1.2 一次请求回答全部问题

OFFLINE = {
    "about_refund": _FakeAnswer("noul", noul=.81),
    "about_security": _FakeAnswer("noul", noul=.74),
    "topic": _FakeAnswer("choice", choice="operations", confidence=.72, probabilities={"operations": .72}),
    "urgency": _FakeAnswer("score", score=1.15, confidence=.68, probabilities={0: .15, 1: .70, 2: .15}),
}
batch_start = time.perf_counter()
batch = TS.call(ARTICLE, QUESTIONS, OFFLINE)
batch_elapsed = time.perf_counter() - batch_start
for name, answer in batch.answers.items():
    print(answer_line(name, answer))
print(f"批量请求耗时（离线时仅供参考）：{batch_elapsed * 1000:.1f} ms")

1.3 逐条调用对照

if TS.offline:
    print("当前为离线模式，跳过逐条网络计时；批量答案已经完整展示。")
else:
    single_start = time.perf_counter()
    for name, question in QUESTIONS.items():
        single = client.system_one(ARTICLE, {name: question})
        print(answer_line(name, single.answers[name]))
    single_elapsed = time.perf_counter() - single_start
    print(f"逐条调用耗时：{single_elapsed * 1000:.1f} ms")

观察：并行提问并不要求代码使用每个答案。先一次取回独立判断，再在 Python 中按业务分支消费结果，通常比串行追问更快。

知识补充

• System One 的核心卖点：一次调用里各问题并行、相互独立求值——加问题几乎不增延迟。把"逐条循环调用"重写成"一次多问"，是 Jev 落地最常见的性能优化。
• 投机式多问：不确定用不用得上的问题也可以先问（输出免费、输入按 token 计），代码端剪枝无关答案。智能家居 demo 一次捆绑 10 问就是这个思路的完整版（见 ../05_智能家居实验/01_智能家居实验.ipynb）。
• 注意：问题之间不能互相引用——每个问题必须独立读完 state 就能回答；需要"看上一个答案"的场景要拆两轮请求。

小结

这本 notebook 的边界很清楚：TypeSafe 只负责受限、可编程的判断；排序、阈值、分组、重建文本和 函数分派都由 Python 代码完成。修改输入或问题后重新运行，就能观察“模型答案 → 确定性代码”的变化。