第 22 章

第 22 章：SDE 级联

本文整理自 Datawhale 开源项目 datawhalechina/jev-cookbook（CC BY-NC-SA 4.0），源文件：main/04_实战指南/13_SDE级联.ipynb。

针对官方文档 实战指南 · SDE Cascade 的可运行实验笔记，用真实 TypeSafe API（Jev 模型）复刻核心流程并中文化。中文翻译版见 datawhalechina.github.io/jev-cookbook。

笔记本结构

章节	内容	实验
0. 准备	安装、客户端、连通性、离线回退	—
📖 理论速览	级联 = mini 提取 → TypeSafe 验证 → 条件升级	—
1. 模拟 mini 提取	硬编码中文发票/片段的 JSON（不依赖 OpenAI）	3 份文档
2. 逐字段 Noul 验证	hallucinated / mismatch / missing 等信号	2–3 次调用
3. any_flag 门控	任一 P(wrong)>FIRE_T → 升级，否则保留 mini	决策打印

每个主题按固定节奏展开：原理 → 理论根基 → 定义数据 → 定义问题 → 调用 → 解读结果，每个单元格只做一件事，可直接顺着跑完（约 3–5 次 API 调用）。

运行要求

Python ≥ 3.10（官方 SDK 要求；macOS 系统自带 python3 是 3.9，装不上 SDK）
一个 TypeSafe API Key（console.typesafe.ai/keys 获取）

推荐：一键创建本地环境（在本 notebooks 目录下）

./setup_env.sh                                  # 创建 .venv：Python 3.12 + 全部依赖
export TYPESAFE_API_KEY=你的key
.venv/bin/jupyter lab <本文件>.ipynb

或者手动创建：python3.12 -m venv .venv && .venv/bin/pip install -r requirements.txt

🔑 API Key 安全提示：本笔记从环境变量 TYPESAFE_API_KEY 读取密钥，不要把 Key 硬编码进笔记本（尤其打算提交到公开仓库时）。

🈶 关于语言：实验全部使用中文 state 与中文提示词。三种原语的选项 key （如 billing、verified）属于代码标识符，保持英文以便代码分支判断；它们的描述文字（criteria 值）均为中文，模型据此理解语义。

0. 准备

0.1 安装所需的库

如果已经用 ./setup_env.sh 创建过环境，本节通常显示“依赖已满足”；在其他环境里首次运行时会自动安装。

%pip install -q -U typesafe-sdk          # 本笔记本必需（要求 Python ≥ 3.10）
# %pip install -q -U jupyterlab         # 如本机还没有 Jupyter，取消注释运行一次
# %pip install -q -U nbformat nbclient  # 仅在需要重新生成/批量执行笔记本时安装

0.2 导入库并创建客户端

Choice / Score / Noul：三种问题原语的构造器（对应官方文档原语一章）；
TypeSafeClient：同步客户端，model="jev-latest" 表示使用官方旗舰模型的最新别名；
Key 从环境变量 TYPESAFE_API_KEY 读取；临时调试也可以直接赋值给 API_KEY（不要提交）。

import os
import time

from typesafe_sdk import (
    Choice,                       # 选择题：从命名选项中选一个，返回 choice + probabilities + confidence
    Score,                        # 打分题：按有序量表打分，返回期望分 + probabilities + confidence
    Noul,                         # 是非题：返回"是"的概率（0~1），本身就是概率所以没有 confidence 字段
    TypeSafeClient,
    TypeSafeAuthenticationError,  # 401 鉴权失败时抛出
)

API_KEY = os.environ.get("TYPESAFE_API_KEY", "")
# API_KEY = "apikey_..."   # ← 仅在临时调试时使用，注意不要提交到公开仓库

# Key 为空时不构造客户端：SDK 在无 Key 时抛 TypeSafeError（非 401 的
# TypeSafeAuthenticationError），不会被下面的回退捕获，会让整本笔记本中断。
client = TypeSafeClient(api_key=API_KEY, model="jev-latest") if API_KEY else None

0.3 离线回退用的两个替身类

真实 API 不可用时，我们需要一个与官方 SDK 响应对象同构的替身，让后续分析代码不用改。官方 SystemOneResponse 的访问方式是：

访问方式	返回
`resp.answers["名称"]`	全部答案（按问题名）
`resp.choices["名称"]`	选择题答案：`.choice` `.confidence` `.probabilities`
`resp.scores["名称"]`	打分题答案：`.score` `.confidence` `.legend` `.probabilities`
`resp.nouls["名称"]`	是非题答案：`.noul`（本身就是概率，无 confidence）
`resp.usage.input_tokens`	本次请求计费的 input token 数

下面的替身类暴露完全相同的属性，仅用于离线模式。

class _FakeAnswer:
    """单个答案的替身：按需挂属性（choice/score/noul/confidence/...）。"""

    def __init__(self, type_, **kw):
        self.type = type_
        for k, v in kw.items():
            setattr(self, k, v)


class _FakeResponse:
    """整个响应的替身：与 SystemOneResponse 同构（answers/nouls/choices/scores/usage）。"""

    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)

0.4 调用助手 `ts.call()`

统一入口：优先请求真实 API；只有当鉴权失败（401）时，才回退到各实验预置的离线示例数据，并在第一次回退时给出显著警告。这样拿到无效 Key 也能跑通全流程，而有效 Key 下全程真实。

class TS:
    real_calls = 0     # 成功的真实调用计数
    offline = False    # 一旦回退过就置 True，后续单元格据此跳过真实计时等逻辑
    _warned = False    # 完整警告只打印一次，避免刷屏

    def call(self, state, questions, offline_answers=None):
        if client is None:          # 未配置 Key：直接走离线示例，不触碰任何客户端
            TS.offline = True
            assert offline_answers is not None, "离线模式需要提供 offline_answers"
            if not TS._warned:
                TS._warned = True
                print("⚠️  离线示例模式：未设置 TYPESAFE_API_KEY，以下为内置示例数据而非真实模型结果；"
                      "设置有效的 TYPESAFE_API_KEY 后重跑本笔记本即可得到真实输出。")
            else:
                print("⚠️ （本次为离线示例数据，非真实 API 输出）")
            return _FakeResponse(offline_answers)
        try:
            resp = client.system_one(state, questions)
            TS.real_calls += 1
            return resp
        except TypeSafeAuthenticationError:
            TS.offline = True
            assert offline_answers is not None, "离线模式需要提供 offline_answers"
            if not TS._warned:
                TS._warned = True
                print("⚠️  离线示例模式：API Key 无效(401)，以下输出为内置示例数据而非真实模型结果；"
                      "设置有效的 TYPESAFE_API_KEY 后重跑本笔记本即可得到真实输出。")
            else:
                print("⚠️ （本次为离线示例数据，非真实 API 输出）")
            return _FakeResponse(offline_answers)


ts = TS()

0.5 连通性测试

用一条最简单的是非题试连官方 API：Key 有效则提示通过；若返回 401，后面的实验会自动切换到离线示例模式（见下一节说明），流程照样能走通。

if client is None:
    TS.offline = True
    print("⚠️  未设置 TYPESAFE_API_KEY，以下实验以【离线示例模式】运行：")
    print("    代码路径与真实调用完全一致，仅数据换成本笔记内置的示例值；")
    print("    在有效 Key 下重跑本笔记本即可得到真实模型输出。")
else:
  try:
    ping = client.system_one(
        "你好",
        {"is_greeting": Noul(instructions="这段文字是在打招呼吗？")},
    )
    print("✅ API 连通正常，Key 有效。将进行真实实验。")
  except TypeSafeAuthenticationError:
    TS.offline = True
    print("⚠️  API Key 无效（401）。以下实验将以【离线示例模式】运行：")
    print("    代码路径与真实调用完全一致，仅数据换成本笔记内置的示例值；")
    print("    在有效 Key 下重跑本笔记本即可得到真实模型输出。")

0.6 本章离线示例数据

Key 无效时，ts.call() 会回退到下列字典。这里预置了两条路径：通过（全部 P(wrong) 低）与 触发升级（某字段幻觉信号高）。

# 离线示例：通过（保留 mini）vs 触发（升级）
OFFLINE_PASS = {
    "vendor::hallucinated": _FakeAnswer("noul", noul=0.08),
    "vendor::name_desc_mismatch": _FakeAnswer("noul", noul=0.05),
    "total::hallucinated": _FakeAnswer("noul", noul=0.12),
    "total::missing_in_source": _FakeAnswer("noul", noul=0.04),
    "invoice_date::hallucinated": _FakeAnswer("noul", noul=0.10),
    "__overall__::judge": _FakeAnswer("noul", noul=0.15),
}

OFFLINE_FIRE = {
    "vendor::hallucinated": _FakeAnswer("noul", noul=0.18),
    "vendor::name_desc_mismatch": _FakeAnswer("noul", noul=0.09),
    "total::hallucinated": _FakeAnswer("noul", noul=0.92),  # 触发
    "total::missing_in_source": _FakeAnswer("noul", noul=0.71),  # 触发
    "invoice_date::hallucinated": _FakeAnswer("noul", noul=0.22),
    "notes::hallucinated": _FakeAnswer("noul", noul=0.88),  # 触发
    "__overall__::judge": _FakeAnswer("noul", noul=0.61),
}

# 第三份文档：字段缺失场景
OFFLINE_MISSING = {
    "contact_phone::missing_in_source": _FakeAnswer("noul", noul=0.11),
    "contact_phone::hallucinated": _FakeAnswer("noul", noul=0.06),
    "amount::hallucinated": _FakeAnswer("noul", noul=0.14),
    "amount::name_desc_mismatch": _FakeAnswer("noul", noul=0.07),
    "__overall__::judge": _FakeAnswer("noul", noul=0.12),
}

📖 理论速览（精简）

本笔记本默认你已读过 System One / 原语 / 置信度（中文镜像站有对应页）。此处只提醒实验会反复用到的三点：

请求模型：同一个 state + 一组问题 → 类型化答案；问题彼此独立、按 ID 返回；
三种原语：Choice 选命名选项；Score 返回概率加权期望分（可为小数）；Noul 返回“是”的概率（无 confidence）；
代码掌控制权：阈值、路由、合并答案都在你的代码里；模型只回答狭窄、原子的判断。

💡 问题 ID 不会发给模型——完整语义写在 instructions / criteria 里。

1. SDE 级联：原理

大型推理模型擅长结构化数据提取（SDE），但贵且慢；小模型便宜，却会幻觉、错位或漏字段。级联的做法是：

提取（mini）：用便宜模型（或本笔记里的硬编码模拟）产出结构化记录；
验证（TypeSafe）：对每个字段发一组狭窄的 Noul——“这个值有问题吗？”→ 得到 P(有问题)；
升级门控：若任一字段的 P(wrong) 超过阈值 FIRE_T，才升级到贵模型；否则直接保留 mini 结果。

本笔记不调用 OpenAI：mini 提取用硬编码 JSON 模拟；真实调用只发生在 TypeSafe 验证层。

📖 理论根基

出处：实战指南 · SDE Cascade / 中文镜像。

概念	要点
分解验证	不问“整条记录好不好”，而问逐字段的原子是非题
`Noul`	返回 P(是)=P(有问题)；没有 confidence 字段
`any_flag` 门控	取 max 而非均值：一个自信红旗就足以升级
代码掌控制权	阈值、是否升级、如何合并答案都写在你的代码里

💡 整体头 __overall__::judge 可用于对照，但官方演示的升级门控不依赖它——升级由逐字段信号驱动。

步骤说明

下面准备 3 份中文文档 + 对应的“mini 提取”结果（含一份故意掺幻觉的），再定义验证问题、调用 TypeSafe、打印级联决策。

1.1 定义文档与模拟 mini 提取结果

FIRE_T = 0.7  # 任一字段 P(wrong) 超过此阈值 → 升级

# 文档 A：干净发票（mini 提取应通过验证）
DOC_A = """电子发票
销售方：星河科技有限公司
发票号码：ACCT-000017
开票日期：2026年3月12日
价税合计：人民币 1280.00 元
备注：含软件服务费"""

MINI_A = {
    "vendor": "星河科技有限公司",
    "invoice_no": "ACCT-000017",
    "invoice_date": "2026-03-12",
    "total": "1280.00",
}

# 文档 B：同一发票，但 mini 幻觉了总额与备注
DOC_B = DOC_A  # 源文本相同

MINI_B = {
    "vendor": "星河科技有限公司",
    "invoice_no": "INV-20260312-8841",
    "invoice_date": "2026-03-12",
    "total": "9800.00",           # 幻觉：源文本是 1280.00
    "notes": "含硬件采购与运费",  # 幻觉：源文本未写硬件
}

# 文档 C：通知片段，联系电话在文中
DOC_C = """【付款提醒】请于本周五前将 350.00 元汇至对公账户。
如有疑问请致电经办人手机 138-0013-8000，勿拨前台总机。"""

MINI_C = {
    "amount": "350.00",
    "contact_phone": "138-0013-8000",
}

DOCUMENTS = [
    ("A_干净发票", DOC_A, MINI_A, "pass"),
    ("B_幻觉总额", DOC_B, MINI_B, "fire"),
    ("C_付款提醒", DOC_C, MINI_C, "pass"),
]

for name, doc, mini, kind in DOCUMENTS:
    print(f"=== {name}（期望门控: {kind}）===")
    print("源文本预览:", doc.splitlines()[0], "...")
    print("mini 提取:", mini)
    print()

1.2 定义逐字段验证问题（Noul）

from typesafe_sdk import NoulCriteria

# 精简版验证指标（中文 criteria；官方完整版见 cookbook）
VERIFY_METRICS = {
    "hallucinated": (
        "extracted_field 是否未被源文本支持、属于幻觉？",
        NoulCriteria(
            true="该值是幻觉——源文本不支持或不存在",
            false="该值有源文本依据",
        ),
    ),
    "name_desc_mismatch": (
        "extracted_field 是否与字段名/字段说明不符？",
        NoulCriteria(
            true="取值与字段名或说明不匹配",
            false="取值与字段名及说明匹配",
        ),
    ),
    "missing_in_source": (
        "按字段说明，源文本中本应有值，但 extracted_field 是否缺失或为空？",
        NoulCriteria(
            true="源文本有信息却被漏提或留空",
            false="空值合理，或字段已正确填写",
        ),
    ),
}


def build_verify_questions(extraction: dict) -> dict:
    """为每个非空字段挂一组 Noul；另加整体 judge 头（仅展示，不参与 any_flag）。"""
    questions = {
        "__overall__::judge": Noul(
            instructions=(
                "整条提取记录是否不正确（有幻觉、错位或漏提），因而应升级到更强模型？"
            ),
            criteria=NoulCriteria(
                true="记录不正确，应升级",
                false="记录正确，可保留",
            ),
        ),
    }
    for field, value in extraction.items():
        for metric, (q, criteria) in VERIFY_METRICS.items():
            questions[f"{field}::{metric}"] = Noul(
                instructions={
                    "field_name": field,
                    "extracted_field": value,
                    "main_question": q,
                },
                criteria=criteria,
            )
    return questions


# 预览：文档 A 会发出多少个问题
qs_a = build_verify_questions(MINI_A)
print(f"文档 A 验证问题数: {len(qs_a)}")
print("问题 ID 示例:", list(qs_a.keys())[:5], "...")

1.3 对每份文档：验证 → any_flag 门控

def any_flag(checks: dict, threshold: float = FIRE_T) -> dict:
    """排除 __overall__ 后，收集 P(wrong) > threshold 的字段信号。"""
    return {
        qid: p
        for qid, p in checks.items()
        if not str(qid).startswith("__overall__") and p > threshold
    }


def verify_and_gate(doc_name, source_text, extraction, offline_key):
    state = {
        "task": "验证结构化提取是否忠实于源文本",
        "source_text": source_text,
        "extraction": extraction,
    }
    questions = build_verify_questions(extraction)
    offline_map = {
        "pass": OFFLINE_PASS,
        "fire": OFFLINE_FIRE,
        "missing": OFFLINE_MISSING,
    }
    # 按字段裁剪离线答案，避免多余 key
    base = offline_map[offline_key]
    offline = {k: v for k, v in base.items() if k in questions or k.startswith("__")}
    # 确保每个问题都有离线值
    for qid in questions:
        if qid not in offline:
            offline[qid] = _FakeAnswer("noul", noul=0.10)

    resp = ts.call(state, questions, offline_answers=offline)
    checks = {qid: resp.nouls[qid].noul for qid in questions}
    fired = any_flag(checks)
    escalate = bool(fired)

    print(f"{'=' * 60}")
    print(f"文档: {doc_name}")
    print(f"{'qid':<40}{'P(wrong)':>9}")
    print("-" * 50)
    for qid, p in sorted(checks.items(), key=lambda c: -c[1]):
        flag = "  <== FIRES" if (not qid.startswith("__overall__") and p > FIRE_T) else ""
        print(f"{qid:<40}{p:>9.2f}{flag}")
    decision = "ESCALATE → 推理模型" if escalate else "ACCEPT → 保留 mini"
    print(f"\nany_flag 门控 (FIRE_T={FIRE_T}): {decision}")
    for qid, p in sorted(fired.items(), key=lambda c: -c[1]):
        print(f"  fired: {qid}  (P={p:.2f})")
    return {"escalate": escalate, "fired": fired, "checks": checks}


# 离线 key：A/C 用 pass，B 用 fire
offline_keys = {"A_干净发票": "pass", "B_幻觉总额": "fire", "C_付款提醒": "pass"}
results = []
for name, doc, mini, _kind in DOCUMENTS:
    results.append(verify_and_gate(name, doc, mini, offline_keys[name]))

观察要点

文档 A：字段均有源文本依据 → 各 P(wrong) 应低于 FIRE_T → ACCEPT。
文档 B：total / notes 与源文本不符 → hallucinated / missing_in_source 类信号应 FIRES → ESCALATE。
文档 C：金额与电话均可在源文本定位 → 通常 ACCEPT。
对比 __overall__::judge 与逐字段 max：整体头可能中等，但一个字段 0.9 就足以升级——这正是 any_flag 的设计意图。

小结

行为	本笔记观察
mini 提取	用硬编码 JSON 模拟（生产中可换成 gpt-mini / 本地小模型）
TypeSafe 验证	逐字段 `Noul` → P(wrong)
门控	`any(P > FIRE_T)` → 升级；否则保留便宜结果
成本直觉	多数干净文档停在验证层；只有红旗文档才付推理价

延伸阅读：SDE Cascade · 原语 Noul · 架构模式 · 置信度门控。

若输出中出现“离线示例”，说明当前 Key 无效；设置有效 TYPESAFE_API_KEY 后重跑即可。

知识补充

级联思想：小模型先干便宜活，Jev 逐字段校验（noul 逐个 flag），不一致才升级大模型——每层只做自己最划算的事。这是官方 Patterns 里意图路由在工程上的镜像。
校验粒度：逐字段而非整文档，错误可定位、可分诊（哪个字段错升级哪个领域的模型）。
进阶阅读：完整路由模式见 ../03_架构模式/01_架构模式.ipynb；置信度三路分流见 18_基于置信度的分类.ipynb。

第 22 章：SDE 级联

第 22 章：SDE 级联

笔记本结构

运行要求

0. 准备

0.1 安装所需的库

0.2 导入库并创建客户端

0.3 离线回退用的两个替身类

0.4 调用助手 ts.call()

0.5 连通性测试

0.6 本章离线示例数据

📖 理论速览（精简）

1. SDE 级联：原理

📖 理论根基

步骤说明

1.1 定义文档与模拟 mini 提取结果

1.2 定义逐字段验证问题（Noul）

1.3 对每份文档：验证 → any_flag 门控

观察要点

小结

知识补充

📑 Jev Cookbook：System One 判断模型实战教程

0.4 调用助手 `ts.call()`