Cascata de SDE

Usa uma cascata de extração de dados estruturados em 2 etapas (mini → verificar → raciocinar) para obter a maior parte da qualidade de um grande modelo de raciocínio por uma fração do custo.

Visão geral
- modelos grandes de raciocínio extraem dados estruturados bem, mas são lentos e caros
- modelos pequenos são baratos, mas cometem erros
- uma cascata obtém a maior parte da qualidade por uma fração do custo
- os modelos que usamos, e seu preço ($ por 1M tokens, entrada / saída; tarifas padrão consultadas em 15 de setembro de 2026):
  - degrau 0 (mini): gpt-5.4-mini a $0.75 / $4.50
  - degrau 1 (raciocínio): gpt-5.5 a $5.00 / $30.00 (mais ou menos 7x o mini)
  - verificador: jev-1.12 do TypeSafe a $0.042 / $0.00 (tokens de saída são grátis; preço publicado da Jev)
Algoritmo
1. Extraia com um modelo barato/pequeno.
2. Verifique com as primitivas do TypeSafe: uma pergunta de sim/não (“pergunta Noul”) por campo
  - (ex.: “este valor está ausente da fonte?”, “ele foi retirado de um texto não relacionado?”), cada uma retornando P(algo está errado).
3. Escale para um modelo de raciocínio caro se um sinal do verificador disparar; caso contrário, fique com a resposta barata.
Este Cookbook
- percorre um exemplo real de ponta a ponta e depois mostra o trade-off em 100 prompts
- nota: os dois degraus de extração usam a OpenAI em modo texto
- nós não usamos saídas estruturadas, chamadas de ferramentas ou modo json, porque:
  - um erro de seguir o schema não é o erro que esperamos que um LLM cometa (é fácil gerar dados sintéticos para isso)
  - se um LLM de fato falha em seguir o schema, quase sempre está muito confuso, então a decodificação restrita não resolve o problema de fundo
  - mas incentivamos você a testá-los!

Configuração

instale as dependências (o client do verificador TypeSafe é servido pelo índice de pacotes do TypeSafe):

pip install openai datasets jsonschema ipython 'cooksafe>=0.2.0,<0.3.0'

depois defina OPENAI_API_KEY e TYPESAFE_API_KEY no seu ambiente

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)

Passo 1: os dados

Escolhemos um dataset do huggingface chamado scrapegraphai

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://events.nyu.edu/live/resource/image/_i/themes/global/images/icons/facebook.rev.1773448757.svg)](https://facebook.com/)
  * [![](https://events.nyu.edu/live/resource/image/_i/themes/global/images/icons/linkedin.rev.1773448758.svg)](https://linkedin.com/)
  * [![](https://events.nyu.edu/live/resource/image/_i/themes/global/images/icons/x.rev.1773448757.svg)](https://x.com/)
  * [![](https://events.nyu.edu/live/resource/image/_i/themes/global/images/icons/instagram.rev.1773448757.svg)](https://instagram.com/)
  * [![](https://events.nyu.edu/live/resource/image/_i/themes/global/images/icons/youtube.rev.1773448758.svg)](https://youtube.com/)

Esta linha é uma página de calendário de eventos da NYU (“Fall 2024 Census Date”):
- o schema pede apenas dois campos: registration_open_date e description
- o scrape do prompt capturou apenas a navegação do calendário e texto padrão: não há data de inscrição nem descrição
- observe que o campo description do schema chega a incluir um valor de exemplo (“Registration opens for the fall semester”) na própria descrição do campo
então um extrator bem-comportado deveria recusar inventar os campos que a página não contém
vamos ver se o modelo pequeno faz a coisa certa!

Passo 2: extraia com o modelo mini (modo texto)

nota: gpt-5.4-mini é muito estocástico nesta entrada – mesmo com temperature=0 ele inventa um description diferente em quase toda execução. Para um passo a passo reproduzível, nós fixamos no código a única fabricação canônica que o resto deste notebook explica (e que o verificador sinaliza com P(wrong) > 0.8). Um pipeline real simplesmente usaria extract(MINI, prompt, schema, content, temperature=0) direto.

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

O registro é válido para o schema (a linha acima imprime True), mas ainda está errado:
- registration_open_date fica em branco, o que corresponde à página: ela não informa data alguma
- mas description é fabricado: a página nunca descreve uma data de inscrição, então o mini inventa uma plausível. Ele pode papagaiar o próprio exemplo do schema, “Registration opens for the fall semester”, ou narrar “…was not found in the document”
- uma verificação de JSON-Schema não consegue ver isso. Um modelo barato produz fabricações confiantes e que satisfazem o schema desse tipo, e pegá-las é o trabalho de um verificador semântico

Passo 3: verifique com o TypeSafe

o verificador é o TypeSafe; para cada campo construímos uma pergunta Noul:
- um sim/não estreito, formulado de modo que true = algo está errado (escalonar)
o TypeSafe retorna um noul calibrado = P(true) por pergunta, em uma única chamada system_one
o conjunto de perguntas:
- uma cabeça holística __overall__::judge (“este registro deveria ser escalonado?”). Nós a calculamos e exibimos para contrastar um juízo de registro inteiro com as cabeças por campo, mas o gate do Passo 4 não a usa – o escalonamento é guiado pela bateria por campo.
- uma bateria por campo
  - campos não vazios recebem o conjunto completo de cabeças
  - campos vazios (null / “” / []) recebem apenas a cabeça absence_wrong
- (o pipeline completo também tem uma cabeça spurious para contêineres inteiros e um score difficulty geral; não mostrado aqui, para manter este passo a passo nas duas cabeças que governam o gate)
A Forma TypeSafe: Decomposição
- Repare como tudo é decomposto programaticamente; essa é a forma TypeSafe.
- A decomposição maximiza a inteligência de cada prompt e torna o algoritmo ajustável e interpretável.

# 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)
    }

Execute a bateria inteira sobre a extração do mini

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

Abra esta verificação no playground do TypeSafe →

O TypeSafe concentra o sinal nos campos que estão de fato errados.
Nossos resultados são calibrados: alto no campo que está errado, baixo no campo que está correto, médio em um campo que parece estranho sem ser claramente errado
É isso que um verificador do typesafe compra para você, em vez de um juiz cru do tipo “isso tudo está bom?”

Passo 4: o gate de escalonamento

agora fazemos o gate em any_flag: escalone se qualquer sinalizador de campo ultrapassar FIRE_T (0.7, definido acima e compartilhado com o marcador <== FIRES do Passo 3)
este é um gate no estilo max (escalone se qualquer campo dispara), não uma média, então um sinalizador vermelho confiante já basta, em vez de ser diluído no silêncio

# 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)

Passo 5: escale para o modelo de raciocínio

Como um sinal disparou, nós pagamos pelo modelo forte (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'  ->  ''

A melhoria
- O modelo de raciocínio descarta o description fabricado, retornando ""
- Ele reconheceu que a página nunca descreve uma data de inscrição e se recusou a inventar uma
- A cascata transformou uma fabricação confiante e válida para o schema em um campo vazio honesto
- E só gastou dólares de modelo de raciocínio neste item porque o verificador mandou

Passo 6: como isso se parece em 100 prompts

Estes são resultados internos do TypeSafe, produzidos com o método geral acima:
- o mesmo laço extract → verify → escalate, gpt-5.4-mini → gpt-5.5-reasoning, gate any_flag sobre as cabeças por campo, rodado em 100 prompts do scrapegraphai
- a extração do degrau barato de cada item é pontuada pelo TypeSafe; o limiar do gate (“cut”) é varrido de 0→1, e cada configuração resultante é plotada no espaço (custo, qualidade)
- o gráfico é um instantâneo histórico; seus custos não foram recalculados à tarifa atual da Jev listada acima

resultados internos: fronteira de custo/qualidade em 100 prompts

como ler:
- losangos pretos = os quatro modelos rodando sozinhos (o custo sobe com a capacidade; o mais forte, gpt-5.5-reasoning, fica no canto superior direito com ≈0.81 de qualidade por ≈$0.10/extração)
- pontos azuis = a cascata em muitos limiares de gate; a linha tracejada é a fronteira de Pareto
- a fronteira da cascata fica acima e à esquerda de todos os modelos individuais: varrer o gate compra para você a maior parte da qualidade do modelo principal por uma fração do custo
- o degrau barato resolve os itens fáceis por quase nada, e só os itens sinalizados pagam pelo modelo de raciocínio

Apêndice A: o que faz um bom sinal de verificador

a cascata só é tão boa quanto seu verificador; o que separa um sinal útil de um inútil:
- Estreito e fundamentado.
  - um sim/não verificável sobre um campo em relação à fonte (ex.: “este valor está ausente da fonte?”), e não um vago “esta extração está boa?”
  - perguntas vagas dão scores moles e não calibrados
- Ruim = TRUE, com critérios explícitos.
  - formule cada pergunta de modo que o caso de escalonar seja o caso true, e declare o que true/false significam
- Por campo, depois agregue com max.
  - um sinalizador por campo localiza o erro e permanece esparso e forte
  - max (“qualquer sinalizador dispara”) garante que um sinalizador vermelho confiante escale, em vez de ser diluído no silêncio
- Independente e barato.
  - um verificador dedicado (aqui, o TypeSafe) julgando a saída pega os pontos cegos do próprio extrator
  - ele precisa ser barato, ou não sobra economia a capturar
- Separador / calibrado.
  - um bom sinal é alto em erros reais e baixo nos corretos, então um único limiar separa de forma limpa aceitar de escalonar
  - é essa separação que empurra a curva de Pareto para cima e para a esquerda