第 24 章：预解析值抽取

本文整理自 Datawhale 开源项目 datawhalechina/jev-cookbook（CC BY-NC-SA 4.0），源文件：main/04_实战指南/15_预解析值抽取.ipynb。

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

笔记本结构

章节	内容	实验
0. 准备	安装、客户端、连通性、离线回退	—
📖 理论速览	正则找候选 → Choice 挑一个 → 代码逐字复制	—
1. find / pick	邮箱、电话、金额的候选发现与选择	辅助函数
2. 中文邮件	多个地址里挑收据收件人 / 发件人	2 次调用
3. 电话 + 发票	联系手机与应付总额；Noul 区分 charge/credit	约 4 次调用

每个主题按固定节奏展开：原理 → 理论根基 → 定义数据 → 定义问题 → 调用 → 解读结果， 每个单元格只做一件事，可直接顺着跑完（约 6 次 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 本章离线示例数据

候选字符串本身会作为 Choice 的选项 key；离线答案因此绑定到下方中文演示文里会出现的具体片段。

def _ch(choice, confidence):
    return _FakeAnswer(
        "choice",
        choice=choice,
        confidence=confidence,
        probabilities={choice: confidence},
    )


# 邮件：收据 → 个人邮箱；发件人 → From
OFFLINE_RECEIPT = {"pick": _ch("dana.personal@gmail.com", 0.98)}
OFFLINE_SENDER = {"pick": _ch("dana.whit@xinghe.cn", 0.99)}

# 电话：挑手机号
OFFLINE_MOBILE = {"pick": _ch("138-0013-8000", 0.97)}

# 金额：应付总额 + 贷记；币种；Noul(是否 credit)
OFFLINE_TOTAL = {"pick": _ch("¥1,315.50", 0.96)}
OFFLINE_CREDIT = {"pick": _ch("¥50.00", 0.95)}
OFFLINE_CURRENCY = {"q": _ch("CNY", 0.92)}
OFFLINE_IS_CREDIT_TOTAL = {"q": _FakeAnswer("noul", noul=0.02)}
OFFLINE_IS_CREDIT_CREDIT = {"q": _FakeAnswer("noul", noul=0.97)}

---

📖 理论速览（精简）

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

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

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

---

1. 预解析值提取：原理

TypeSafe 的 Choice 只能从你给出的选项里挑，因此必须先在代码里找出候选：

1. 正则 / 解析器在文本中找出候选（宁多勿漏、去重、保持文档顺序）；
2. TypeSafe挑出问题所问的那一个，并可附带读属性（币种、是否贷记）；
3. 代码逐字复制选中片段并规范化——模型不会凭空发明数字。

📖 理论根基

出处：Pre-Parsed Value Extraction / 中文镜像。

概念	要点
选项 = 候选片段	choice 是某个 span 的原样复制（或逃生口 none）
代码拥有字符串	规范化（小写、E.164、Decimal）全在代码侧
none 逃生口	没有候选合适时显式承认，而不是硬选一个
可选 Noul	例如金额是 charge 还是 credit

1.1 定义正则与 find / pick / is_true

import re
from decimal import Decimal

NONE = "none"  # 每个挑选题的逃生口：没有候选合适

EMAIL_RE = re.compile(r"[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}")
PHONE_RE = re.compile(r"\(?\+?\d[\d\s()\-.]{6,}\d")
# 支持 $ / € / £ / ¥ 以及中文“元”前的数字
MONEY_RE = re.compile(r"(?:[$€£¥]\s?\d[\d,]*(?:\.\d{2})?|\d[\d,]*(?:\.\d{2})?\s?元)")


def find(pattern, text: str) -> list:
    """代码侧候选发现：高召回正则，去重，按文档顺序。"""
    seen = set()
    out = []
    for match in pattern.findall(text):
        span = match.strip()
        if span and span not in seen:
            seen.add(span)
            out.append(span)
    return out


def pick(document: str, candidates: list, question: str, offline_answers) -> dict:
    """TypeSafe 从候选片段中挑选扮演该角色的那一个。"""
    criteria = {c: None for c in candidates}
    criteria[NONE] = "这些候选都不符合问题所问的值。"
    questions = {"pick": Choice(instructions=question, criteria=criteria)}
    ans = ts.call(document, questions, offline_answers=offline_answers).answers["pick"]
    return {"choice": ans.choice, "confidence": ans.confidence}


def is_true(document: str, question: str, offline_answers) -> float:
    """是非题 Noul，返回 P(是)。"""
    questions = {"q": Noul(instructions=question)}
    return ts.call(document, questions, offline_answers=offline_answers).nouls["q"].noul


print("辅助函数已定义：find / pick / is_true")

1.2 中文邮件：按角色挑选邮箱

EMAIL_DOC = """From: 王丹 
To: billing@xinghe.cn
Cc: orders@xinghe.cn
Reply-To: dana.personal@gmail.com

各位好——这单请不要发到账单别名。收据请寄到我的个人邮箱。谢谢，王丹。"""

emails = find(EMAIL_RE, EMAIL_DOC)
print("candidates :", emails)

receipt = pick(
    EMAIL_DOC,
    emails,
    "发件人希望把收据寄到哪个邮箱地址？",
    OFFLINE_RECEIPT,
)
sender = pick(
    EMAIL_DOC,
    emails,
    "这封邮件的发件地址是哪个（From 行）？",
    OFFLINE_SENDER,
)

# 代码逐字复制并规范化（小写）；从不重新键入
print(f"receipt -> : {receipt['choice'].lower():<28} (conf {receipt['confidence']:.2f})")
print(f"sender  -> : {sender['choice'].lower():<28} (conf {sender['confidence']:.2f})")

观察要点（邮件）

• 正则找出四个地址；TypeSafe 根据正文意图挑出个人 Gmail 作为收据地址。
• sender 对应 From 行，与收据地址不同——同一批候选、不同角色问题。
• 返回值是候选列表中的原样复制，再由代码 .lower()。

1.3 电话：挑选联系手机号

PHONE_DOC = """星河科技上海办联系方式：前台总机 (021) 5555-0199，
传真 (021) 5555-0142，紧急请打我手机 138-0013-8000。"""

phones = find(PHONE_RE, PHONE_DOC)
print("candidates :", phones)

mobile = pick(
    PHONE_DOC,
    phones,
    "哪个号码是经办人的直接手机 / 移动电话？",
    OFFLINE_MOBILE,
)
print(f"mobile  -> : {mobile['choice']}  (conf {mobile['confidence']:.2f})")

# phonenumbers 可选：缺失时用简易清洗
try:
    import phonenumbers
    parsed = phonenumbers.parse(mobile["choice"], "CN")
    e164 = phonenumbers.format_number(parsed, phonenumbers.PhoneNumberFormat.E164)
    print(f"E.164   -> : {e164}")
except Exception:
    digits = re.sub(r"\D", "", mobile["choice"])
    if digits.startswith("86"):
        e164 = "+" + digits
    elif len(digits) == 11:
        e164 = "+86" + digits
    else:
        e164 = "+" + digits
    print(f"E.164   -> : {e164}  (简易正则规范化；安装 phonenumbers 可更稳)")

1.4 发票金额：挑总额 / 贷记，并用 Noul 判定 charge vs credit

MONEY_DOC = """发票 INV-2087
小计：¥1,200.00
税额：¥115.50
应付合计：¥1,315.50
上月已抵扣的善意贷记：¥50.00"""

amounts = find(MONEY_RE, MONEY_DOC)
print("candidates :", amounts)

currency_q = {
    "q": Choice(
        instructions="这些金额使用的是哪种货币？",
        criteria={"USD": None, "EUR": None, "GBP": None, "CNY": None, "JPY": None},
    )
}
currency_ans = ts.call(MONEY_DOC, currency_q, offline_answers=OFFLINE_CURRENCY).choices["q"]
currency = {"choice": currency_ans.choice, "confidence": currency_ans.confidence}

total = pick(MONEY_DOC, amounts, "哪个金额是客户必须支付的应付合计？", OFFLINE_TOTAL)
credit = pick(MONEY_DOC, amounts, "哪个金额是已抵扣的善意贷记？", OFFLINE_CREDIT)


def to_decimal(value: str) -> Decimal:
    """复制选中片段，在代码里解析数字（此处按千分位逗号 + 小数点）。"""
    return Decimal(re.sub(r"[^\d.]", "", value))


for label, chosen, offline_noul in [
    ("total due", total, OFFLINE_IS_CREDIT_TOTAL),
    ("credit", credit, OFFLINE_IS_CREDIT_CREDIT),
]:
    p_credit = is_true(
        MONEY_DOC,
        f"金额 {chosen['choice']} 对客户而言是贷记或退款，而不是扣款（charge）吗？",
        offline_noul,
    )
    kind = "credit" if p_credit > 0.5 else "charge"
    print(
        f"{label:<10}: {chosen['choice']:<12} -> {to_decimal(chosen['choice'])} "
        f"{currency['choice']} ({kind}, P(credit)={p_credit:.2f})"
    )

观察要点（电话与金额）

• 号码本身看不出谁是手机——周围的“手机 / 紧急”词语才是信号；Choice 读的是角色。
• 应付合计 ¥1,315.50 的 Noul P(credit) 应很低 → 标注 charge； 贷记 ¥50.00 则 P(credit) 很高 → 标注 credit。
• to_decimal 假定逗号分组、句点小数；若文档用欧洲写法，可另加 Noul 询问约定再分支。

---

小结

步骤	工具
找候选	re（可选 phonenumbers）
挑角色	TypeSafe Choice（选项 = 候选 + none）
属性 / 符号	Choice（币种）或 Noul（charge vs credit）
规范化	代码：lower / E.164 / Decimal

延伸阅读：Pre-Parsed Value Extraction。

知识补充

• 本篇是官方对"判断不生成"的最直接示范：解析器已经抽出候选值，模型只负责判"这个值对不对/该用哪个候选"——错误率集中且可测，比让模型自由生成再修罗盘稳得多。
• 失败回退：解析失败的片段可以转交大模型，但那是升级路径不是主路径；主路径永远是最便宜的可验证方法。
• 同类：14_日期抽取.ipynb（日期特化）、11_引用核查.ipynb（先字符串匹配）同一分层思想。