SDE 级联
采用两阶段结构化数据提取级联(mini → verify → reasoning),以极低的成本获得大型推理模型的大部分质量。
本文由机器翻译自 en,未经人工校对,仅供快速参考。

- 概述
- 大型推理模型能够很好地提取结构化数据,但速度慢且成本高
- 小型模型成本低,但容易出错
- 一种级联(cascade)策略能以极低的成本获得大部分质量
- 我们使用的模型及其价格(每百万 token 的价格,输入/输出;标准费率,
截至 2026 年 9 月 15 日核查):
- 第 0 级(mini):
gpt-5.4-mini价格为 $0.75 / $4.50 - 第 1 级(推理):
gpt-5.5价格为 $5.00 / $30.00(约为 mini 的 7 倍) - 验证器:TypeSafe
jev-1.12,价格为 $0.042 / $0.00(输出 token 免费; 已发布的 Jev 定价)
- 第 0 级(mini):
- 算法
- 使用廉价/小型模型进行提取(Extract)。
- 使用 TypeSafe 原语进行验证(Verify):针对每个字段的是/否问题(“Noul 问题”)
- (例如,“该值是否不在源数据中?”,“它是否是从无关文本中提取的?”),每个问题返回 P(出错)。
- 如果验证器信号触发,则升级(Escalate)到昂贵的推理模型;否则保留廉价模型的答案。
- 本 Cookbook
- 逐步演示一个真实示例的完整流程,然后展示在 100 个提示词上的权衡
- 注意:两个提取层级均使用文本模式的 OpenAI
- 我们不使用结构化输出、工具调用或 JSON 模式,原因如下:
- 遵循模式(schema following)的错误并不是我们期望 LLM 犯的错误(为此生成合成数据很容易)
- 如果 LLM 未能遵循模式,通常意味着它处于极度困惑的状态,因此约束解码无法解决根本问题
- 我们鼓励您尝试这些功能!
设置
- 安装依赖项(TypeSafe 验证器客户端由 TypeSafe 的软件包索引提供服务):
pip install openai datasets jsonschema ipython "typesafe-sdk>=0.5.7" cooksafe --extra-index-url https://pypi.typesafe.ai/
- 然后在你的环境中设置
OPENAI_API_KEY和TYPESAFE_API_KEY
import json
import os
from pathlib import Path
import jsonschema
from cooksafe import JsonCache, make_playground_link
from datasets import load_dataset
from IPython.display import Markdown, display
from openai import OpenAI
from typesafe_sdk import Noul, NoulCriteria, TypeSafeClient
MINI = "gpt-5.4-mini" # rung 0: cheap + fast
REASONING = "gpt-5.5" # rung 1: strong, run with reasoning_effort="high"
TS_MODEL = "jev-1.12" # the TypeSafe verifier model
FIRE_T = 0.7 # escalate if any per-field P(wrong) exceeds this; also the "<== FIRES" display marker
oai = OpenAI()
ts = TypeSafeClient(api_key=os.environ["TYPESAFE_API_KEY"], timeout=30.0)
第一步:数据
我们选择了一个名为 scrapegraphai 的 HuggingFace 数据集
SCRAPEGRAPHAI_REVISION = "4bb9fba1dff9181c5acdb60a5a26fea62fa54fe9"
row = load_dataset(
"scrapegraphai/scrapegraphai-100k",
revision=SCRAPEGRAPHAI_REVISION,
split="train",
)[516]
schema = json.loads(row["schema"])
prompt = row["prompt"]
content = row["content"]
print(
f"""
PROMPT
===========
{prompt}
SCHEMA
===========
{json.dumps(schema, indent=2)}
CONTENT
===========
{content}
""".strip()
)
PROMPT
===========
Find registration open date fall semester for New York University in New York, NY for the 2024-2025 school year.
SCHEMA
===========
{
"properties": {
"registration_open_date": {
"description": "The date that registration opens for the fall semester. MUST be in the format mm/dd/yyyy. For example, for a college in the 2024-2025 school year, it might be something like 09/05/2024. Return a blank string if you are unsure.",
"title": "Registration Open Date",
"type": "string"
},
"description": {
"description": "A brief description of the registration open date. For example, 'Registration opens for the fall semester'.",
"title": "Description",
"type": "string"
}
},
"required": [
"registration_open_date",
"description"
],
"title": "RegistrationOpen",
"type": "object"
}
CONTENT
===========
Skip to content Skip to current page navigation
[ ](https://www.nyu.edu/)
Search Site
[ ](https://www.nyu.edu/)
* [ Academics](https://www.nyu.edu/academics.html)
* [ Admissions](https://www.nyu.edu/admissions.html)
* [ Research](https://www.nyu.edu/research.html)
* [ University Life](https://www.nyu.edu/life.html)
* [ About](https://www.nyu.edu/about.html)
All NYU
# Mobile Navigation
[ ](https://www.nyu.edu/)
Search Site
* [Academics](https://www.nyu.edu/academics.html)
* [Admissions](https://www.nyu.edu/admissions.html)
* [Research](https://www.nyu.edu/research.html)
* [University Life](https://www.nyu.edu/life.html)
* [About](https://www.nyu.edu/about.html)
All NYU
Info for
* Back to main menu
* Info for
* [Students](https://www.nyu.edu/students.html)
* [Faculty](https://www.nyu.edu/faculty.html)
* [Alumni](https://www.nyu.edu/alumni.html)
* [Employees](https://www.nyu.edu/employees.html)
* [Community](https://www.nyu.edu/community.html)
[Log In](http://home.nyu.edu/)
Info for
* [Students](https://www.nyu.edu/students.html)
* [Faculty](https://www.nyu.edu/faculty.html)
* [Alumni](https://www.nyu.edu/alumni.html)
* [Employees](https://www.nyu.edu/employees.html)
* [Community](https://www.nyu.edu/community.html)
[Log In](https://home.nyu.edu/)
Search Site Search
# Events Calendar
Search Events
Apply Reset
* [About the Events Calendar ](https://www.nyu.edu/employees/resources-and-services/media-and-communications/digital-communications/university-events-calendar.html)
* [Events Calendar Tutorial ](https://www.nyu.edu/employees/resources-and-services/media-and-communications/digital-communications/university-events-calendar/tutorials.html)
* [Report issue or provide feedback ](https://nyu.service-now.com/sp?id=sc_cat_item&sys_id=7698dd2a98bcf4004c8c03063d84e274)
Search Filters Calendar
New York University
Equal Opportunity and Non-Discrimination at NYU - New York University is committed to maintaining an environment that encourages and fosters respect for individual values and appropriate conduct among all persons. In all University spaces--physical and digital--programming, activities, and events are carried out in accordance with applicable law as well as University policy, which includes but is not limited to its Non-Discrimination and Anti-Harassment Policy.
Unless otherwise noted, all content copyright New York University. All rights reserved.
* [Search](https://search.nyu.edu/)
* [Campus Map](https://www.nyu.edu/map.html)
* [Events](https://events.nyu.edu/)
* [Contact Us](https://www.nyu.edu/contact-us.html)
* [Give](https://www.nyu.edu/about/giving.html)
* [Copyright & Fair Use](https://www.nyu.edu/copyright-and-fair-use.html)
* [Privacy](https://www.nyu.edu/privacy.html)
* [Accessibility](https://www.nyu.edu/accessibility.html)
* [Feedback](https://www.nyu.edu/#feedback.html)
* [New York Campus](https://www.nyu.edu/)
* [Abu Dhabi Campus](https://nyuad.nyu.edu/)
* [Shanghai Campus](https://shanghai.nyu.edu/)
* [](https://facebook.com/)
* [](https://linkedin.com/)
* [](https://x.com/)
* [](https://instagram.com/)
* [](https://youtube.com/)
- 这一行是一个 NYU 事件日历页面(“2024 年秋季普查日期”):
- 该模式仅要求两个字段:
registration_open_date和description - 提示词抓取仅捕获了日历导航和模板文本:不存在注册日期或描述
- 注意,模式的
description字段甚至在其自身的字段描述中提供了一个示例值(“秋季学期注册开放”)
- 该模式仅要求两个字段:
- 因此,一个行为规范的提取器应当拒绝编造页面中不存在的字段
- 让我们看看小模型是否能做出正确的判断!
步骤 2:使用迷你模型进行提取(文本模式)
- 注意:
gpt-5.4-mini对此输入的随机性很强——即使在temperature=0的情况下,它几乎每次运行都会编造出不同的description。为了进行可复现的分步演示,我们硬编码了本笔记本其余部分所解释的(以及验证器在 P(wrong) > 0.8 时标记的)唯一一种典型编造结果。在实际管道中,只需直接调用extract(MINI, prompt, schema, content, temperature=0)即可。
EXTRACT_SYSTEM = (
"You extract structured data from documents. Return only values supported by the text. "
"Follow any value format specified by the schema or its field descriptions."
)
# LLM and TypeSafe calls are cached to ``json_cache.json``, which ships with the cookbook, so
# re-rendering reproduces the published results with no API spend; delete the file to re-run live.
json_cache = JsonCache(Path("json_cache.json"))
@json_cache
def extract(
model: str,
prompt: str,
schema: dict,
content: str,
*,
reasoning_effort: str | None = None,
temperature: float | None = None,
) -> dict:
user = (
f"{prompt}\n\nReturn ONLY a JSON object matching this JSON Schema:\n"
f"{json.dumps(schema, indent=2)}\n\nDocument:\n{content}"
)
kwargs = {
"model": model,
"messages": [
{"role": "system", "content": EXTRACT_SYSTEM},
{"role": "user", "content": user},
],
}
if reasoning_effort:
kwargs["reasoning_effort"] = reasoning_effort
if temperature is not None:
kwargs["temperature"] = temperature
text = oai.chat.completions.create(**kwargs).choices[0].message.content
# The prompt asks for ONLY a JSON object, so parse the reply as-is -- no regex fishing a
# substring out of a malformed reply. If ``json.loads`` fails, treat it as an empty extraction
# (the record-level analog of NaN): every field reads as absent, which the verifier flags and the
# gate escalates -- the safe direction. Schema-following errors are rare here (see the overview).
try:
return json.loads(text)
except (ValueError, json.JSONDecodeError):
return {}
# Hard-coded canonical fabrication (see note above); a real pipeline would use extract(MINI, prompt, schema, content, temperature=0).
mini_record = {
"registration_open_date": "",
"description": "Registration opens for the fall semester",
}
print("mini extraction:\n", json.dumps(mini_record, indent=2))
# The record is a perfect fit for the JSON Schema -- and still wrong. Schema validation is necessary
# but not sufficient: it catches structural errors, never semantic ones. That gap is the whole point.
print("\nschema-valid:", jsonschema.Draft202012Validator(schema).is_valid(mini_record))
mini extraction:
{
"registration_open_date": "",
"description": "Registration opens for the fall semester"
}
schema-valid: True
- 该记录是 schema-valid(上一行打印
True),但它是错误的:registration_open_date为空,这与页面一致:页面未注明日期- 但
description是虚构的:页面从未描述过注册日期,因此 mini 编造了一个看似合理的日期。它可能会复述 schema 中的示例“秋季学期注册开放”,或者叙述“未在文档中找到…” - JSON-Schema 检查无法发现此问题。廉价模型会生成此类自信且满足 schema 的虚构内容,而捕捉它们正是语义验证器的职责
步骤 3:使用 TypeSafe 进行验证
- 验证器是 TypeSafe;对于每个字段,我们构建一个
Noul问题:- 一个狭窄的是/否问题,其表述方式使得
true= 出现问题(升级)
- 一个狭窄的是/否问题,其表述方式使得
- TypeSafe 在一次 system_one 调用中为每个问题返回校准后的
noul=P(true) - 问题集:
- 一个整体的
__overall__::judge头部(“是否应升级此记录?”)。我们计算并显示它,以对比整体记录判断与按字段划分的头部,但步骤 4 中的门控不使用它——升级由按字段划分的电池驱动。 - 按字段的电池
- 非空字段获得完整的头部集
- 空字段(null / “” / [])仅获得
absence_wrong头部
- (完整管道还有一个用于整个容器的
spurious头部和一个整体的difficulty分数;此处未显示,以保持本演练仅聚焦于两个门控头部)
- 一个整体的
- TypeSafe 方式:分解
- 注意一切是如何以编程方式分解的,这就是 TypeSafe 的方式。
- 分解最大化了每个提示的智能,并使算法可调和可解释。
-
# metric -> (question, NoulCriteria)
MAIN_QUESTIONS = {
"name_desc_mismatch": (
"Does the `extracted_field` fail to match the field at `path` or the `description` in the "
"`field_spec`? If the `description` is empty, judge against the `path` alone.",
NoulCriteria(
true="the `extracted_field` does not match the field name or its `description`",
false="the `extracted_field` matches the field name and `description`",
),
),
"type_mismatch": (
"Does the `extracted_field` violate the `type` declared in the `field_spec`?",
NoulCriteria(
true="the `extracted_field` violates the declared `type`",
false="the `extracted_field` conforms to the declared `type`",
),
),
"unreasonable": (
"Is the `extracted_field` one that a reasonable person would not have extracted for this "
"`field_spec`?",
NoulCriteria(
true="a reasonable person would not have extracted this value",
false="the extraction is reasonable",
),
),
"hallucinated": (
"Is the `extracted_field` unsupported by, or absent from, the source text?",
NoulCriteria(
true="the `extracted_field` is a hallucination -- not supported by, or absent "
"from, the source text",
false="the `extracted_field` is supported by the source text",
),
),
"off_target": (
"Does the source text fail to genuinely report the thing the `field_spec` describes, so the "
"value was pulled from incidental text?",
NoulCriteria(
true="the source does not genuinely provide this field -- the value was pulled "
"from incidental text",
false="the source genuinely reports this field",
),
),
"incomplete": (
"Does the `extracted_field` fail to capture a value the source supports (note whether the "
"`field_spec` is `required`)?",
NoulCriteria(
true="the field is wrongly empty, null, or missing a value the source supports",
false="the field captures the value the source supports",
),
),
"format_violation": (
"Does the `extracted_field` violate the format or constraints implied by the `description`, "
"the schema `type`, and the extraction instructions (e.g. date format, units, enum membership)?",
NoulCriteria(
true="the `extracted_field` violates the implied format or constraints",
false="the `extracted_field` satisfies the format and constraints",
),
),
}
ABSENCE_QUESTION = (
"The `extracted_field` is empty, null, or an empty collection. Does the source text contain the "
"information the `field_spec` describes, making the empty result wrong?"
)
ABSENCE_CRITERIA = NoulCriteria(
true="a value was wrongly omitted", false="returning nothing is correct"
)
# The pipeline also asks one holistic, whole-record head: "should this be escalated?"
OVERALL_JUDGE = (
"Is this extracted record an incorrect extraction -- some value unsupported by the source or "
"not conforming to the schema, required information missing or wrong, or some field hallucinated -- "
"so it should be escalated to a smarter model?"
)
OVERALL_JUDGE_CRITERIA = NoulCriteria(
true="the record is an incorrect extraction",
false="the record is a correct extraction",
)
def is_empty(v) -> bool:
return v is None or (isinstance(v, (str, list, dict)) and len(v) == 0)
def field_spec(name: str) -> dict:
"""Minimal spec pulled from the schema (unwrapping anyOf/null for optional fields)."""
p = schema["properties"][name]
branches = p.get("anyOf") or []
typ = p.get("type") or next(
(b["type"] for b in branches if b.get("type") != "null"), "unknown"
)
return {
"path": name,
"type": typ,
"description": p.get("description", ""),
"required": name in schema.get("required", []),
}
def build_questions(record: dict) -> dict[str, Noul]:
"""The verify question set: one holistic ``__overall__::judge`` head plus a per-field battery,
keyed ``field::metric`` (mirrors build_verify_prompts)."""
questions: dict[str, Noul] = {
"__overall__::judge": Noul(
instructions=OVERALL_JUDGE, criteria=OVERALL_JUDGE_CRITERIA
),
}
for name, value in record.items():
spec = field_spec(name)
if is_empty(value):
questions[f"{name}::absence_wrong"] = Noul(
instructions={
"field_spec": spec,
"extracted_field": value,
"main_question": ABSENCE_QUESTION,
},
criteria=ABSENCE_CRITERIA,
)
continue
for metric, (question, criteria) in MAIN_QUESTIONS.items():
if metric == "type_mismatch" and spec["type"] == "unknown":
continue
questions[f"{name}::{metric}"] = Noul(
instructions={
"field_spec": spec,
"extracted_field": value,
"main_question": question,
},
criteria=criteria,
)
return questions
@json_cache
def verify(record: dict) -> dict[str, float | str]:
"""Run the whole Noul battery over a record in one TypeSafe call; return ``{field::metric: P(true)}``."""
state = {
"system_message": EXTRACT_SYSTEM,
"instruction": "Extract the structured record from this document",
"source_text": row["content"],
"schema": schema,
"extraction": record,
}
questions = build_questions(record)
answers = ts.system_one(state=state, questions=questions, model=TS_MODEL).answers
return {qid: ans.noul for qid, ans in answers.items()} | {
"playground_link": make_playground_link(state, questions)
}
对整个电池组进行迷你提取
checks = verify(mini_record)
playground_link = checks.pop("playground_link")
display(
Markdown(
f"🔗 [Open this verification in the TypeSafe playground]({playground_link})"
)
)
print(f"{'qid':<40}{'P(wrong)':>9}")
print("-" * 50)
for fld, p in sorted(checks.items(), key=lambda c: -c[-1]):
flag = " <== FIRES" if p > FIRE_T else ""
print(f"{fld:<40}{p:>9.2f}{flag}")
qid P(wrong)
--------------------------------------------------
description::hallucinated 0.95 <== FIRES
description::off_target 0.85 <== FIRES
description::unreasonable 0.58
__overall__::judge 0.56
description::incomplete 0.16
registration_open_date::absence_wrong 0.14
description::format_violation 0.10
description::name_desc_mismatch 0.08
description::type_mismatch 0.02
- TypeSafe 将信号集中在实际出错的字段上。
- 我们的结果经过校准:在出错的字段上置信度高,在正确的字段上置信度低,在看起来异常但并非明显错误的字段上置信度中等。
- 这就是类型安全验证器相较于模糊的“整体是否良好?”评判器所带来的优势。
第 4 步:升级门控
- 现在我们对
any_flag进行门控控制:如果 任何 字段标志超过FIRE_T(0.7,在上方设定,并与第 3 步中的<== FIRES标记共享),则进行升级。 - 这是一个
max风格的门控(如果 任何 字段触发则升级),而非平均值,因此一个高置信度的红色标志就足以触发升级,而不会被平均化导致静默。
# any_flag is a per-field gate: the holistic __overall__ head is shown above but not part of it
fired = {
qid: p
for qid, p in checks.items()
if not qid.startswith("__overall__") and p > FIRE_T
}
escalate = bool(fired)
print(
f"any_flag gate (threshold {FIRE_T}): {'ESCALATE' if escalate else 'ACCEPT cheap result'}"
)
for qid, p in sorted(fired.items(), key=lambda c: -c[1]):
print(f" fired: {qid} (P={p:.2f})")
any_flag gate (threshold 0.7): ESCALATE
fired: description::hallucinated (P=0.95)
fired: description::off_target (P=0.85)
第 5 步:升级至推理模型
由于触发了信号,我们调用强模型(gpt-5.5,reasoning_effort="high")
final_record = (
extract(REASONING, prompt, schema, content, reasoning_effort="high")
if escalate
else mini_record
)
print("mini :", json.dumps(mini_record))
print("reasoning :", json.dumps(final_record))
print("\nfield-level diff (mini -> final):")
for name in mini_record:
if mini_record[name] != final_record.get(name):
print(f" {name}: {mini_record[name]!r} -> {final_record.get(name)!r}")
mini : {"registration_open_date": "", "description": "Registration opens for the fall semester"}
reasoning : {"description": "", "registration_open_date": ""}
field-level diff (mini -> final):
description: 'Registration opens for the fall semester' -> ''
- 改进之处
- 推理模型放弃了虚构的
description字段,返回"" - 它识别出页面从未描述过注册日期,因此拒绝编造一个
- 级联机制将一个自信且符合模式规范的虚构值转化为了诚实的空字段
- 并且,它仅在此单项上花费了推理模型的算力成本,因为验证器指示它这样做
- 推理模型放弃了虚构的
第 6 步:在 100 个提示词上的表现
- 这些是内部 TypeSafe 结果,采用上述通用方法生成:
- 相同的
extract → verify → escalate(提取 → 验证 → 升级)循环,gpt-5.4-mini → gpt-5.5-reasoning, 基于每个字段头部的any_flag门控,在 100 个 scrapegraphai 提示词上运行 - 每个项目的低成本层级提取结果由 TypeSafe 进行评分;门控阈值(“cut”) 在 0→1 范围内扫描,每个生成的配置都在(成本,质量)空间中绘制
- 该图表是历史快照;其成本未按照上述当前 Jev 费率重新计算
- 相同的
- 如何解读:
- 黑色菱形 = 单独运行的四个模型(成本随能力上升;最强的
gpt-5.5-reasoning位于右上角,质量约为 0.81,每次提取成本约为 $0.10) - 蓝色点 = 不同门控阈值下的级联结果;虚线是 pareto 前沿
- 级联前沿位于 所有单个模型的左上方:扫描门控让你以极低的成本获得大部分顶级模型的质量
- 低成本层级以近乎免费的成本处理简单项目,只有被标记的项目才需要支付推理模型的费用
- 黑色菱形 = 单独运行的四个模型(成本随能力上升;最强的
附录 A:什么是良好的验证器信号
- 级联的质量取决于其验证器;区分有用信号与无用信号的关键在于:
- 狭窄且基于事实。
- 针对源数据中的单个字段进行一项可验证的是/否检查(例如,“该值在源数据中是否缺失?”),而不是模糊的“此提取结果是否良好?”
- 模糊的问题会导致得分模糊且未校准
- 错误情况设为 TRUE,并明确标准。
- 构建每个问题,使 升级 的情况为
true情况,并明确说明true/false的含义
- 构建每个问题,使 升级 的情况为
- 按字段独立,然后使用
max聚合。- 按字段的标志将错误局部化,并保持稀疏和强效
max(“任何标志触发”)确保一个自信的红色标志触发升级,而不是被平均化为沉默
- 独立且低成本。
- 专用验证器(此处为 TypeSafe)评估输出结果,可以捕捉提取器自身的盲点
- 它必须低成本,否则就没有剩余节省空间可供捕获
- 分离/校准良好。
- 良好的信号在真实错误上得分高,在正确结果上得分低,因此单个阈值可以清晰地区分接受与升级
- 这种分离正是推动 pareto 曲线向左上方移动的原因
- 狭窄且基于事实。