Cascata SDE

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

Panorâmica
- os grandes modelos de raciocínio extraem bem dados estruturados, mas são lentos e caros
- os modelos pequenos são baratos, mas cometem erros
- uma cascata obtém a maior parte da qualidade a uma fração do custo
- os modelos que usamos, e o seu preço ($ por 1M de tokens, entrada / saída; tarifas padrão verificadas em 15 de setembro de 2026):
  - nível 0 (mini): gpt-5.4-mini a $0.75 / $4.50
  - nível 1 (raciocínio): gpt-5.5 a $5.00 / $30.00 (cerca de 7x o mini)
  - verificador: TypeSafe jev-1.12 a $0.042 / $0.00 (os tokens de saída são grátis; preço publicado do Jev)
Algoritmo
1. Extrai com um modelo barato/pequeno.
2. Verifica com as primitivas do TypeSafe: uma pergunta sim/não por campo (“pergunta Noul”)
  - (p. ex. “este valor está ausente da fonte?”, “foi retirado de texto não relacionado?”), cada uma devolvendo P(algo está errado).
3. Escala para um modelo de raciocínio caro se um sinal do verificador disparar; caso contrário, mantém 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 níveis de extração usam o OpenAI em modo texto
- não usamos saídas estruturadas, chamadas de ferramentas, nem json mode, porque:
  - um erro de seguimento do schema não é o erro que esperamos que um LLM cometa (é fácil criar dados sintéticos para isto)
  - se um LLM falha mesmo em seguir o schema, está quase sempre muito confuso, por isso a descodificação restrita não resolve o problema de fundo
  - mas encorajamos-te a experimentá-los!

Configuração

instala as dependências (o cliente do verificador TypeSafe é servido a partir do índice de pacotes do TypeSafe):

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

depois define OPENAI_API_KEY e TYPESAFE_API_KEY no teu 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 conjunto de dados 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 padronizado: não há data de registo, nem descrição
- nota que o campo description do schema até traz um valor de exemplo (“Registration opens for the fall semester”) na descrição do próprio campo
por isso um extrator bem-comportado deve recusar-se a inventar os campos que a página não contém
vamos ver se o modelo pequeno faz a coisa certa!

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

nota: o gpt-5.4-mini é muito estocástico neste input – mesmo a temperature=0 inventa uma description diferente em quase todas as execuções. Para um percurso reproduzível fixamos no código a única fabricação canónica que o resto deste notebook explica (e que o verificador sinaliza a P(wrong) > 0.8). Um pipeline real limitar-se-ia a usar extract(MINI, prompt, schema, content, temperature=0) diretamente.

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 registo é válido segundo o schema (a linha acima imprime True), mas está errado:
- registration_open_date fica em branco, o que corresponde à página: não indica data nenhuma
- mas a description é fabricada: a página nunca descreve uma data de registo, por isso o mini inventa uma plausível. Pode papaguear o exemplo do próprio 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 isto. Um modelo barato produz fabricações confiantes e que satisfazem o schema deste tipo, e apanhá-las é o trabalho de um verificador semântico

Passo 3: verificar 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 (escalar)
o TypeSafe devolve um noul calibrado = P(true) por pergunta, numa só chamada system_one
o conjunto de perguntas:
- uma cabeça holística __overall__::judge (“este registo deve ser escalado?”). Calculamo-la e mostramo-la para contrastar um julgamento de registo 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 tem também uma cabeça spurious para contentores inteiros e um score difficulty global; não mostrados aqui, para manter este percurso nas duas cabeças de gate)
A forma TypeSafe: decomposição
- Repara como tudo é programaticamente decomposto, esta é 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)
    }

Correr 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

Open this verification in the TypeSafe playground →

O TypeSafe concentra o sinal nos campos que estão de facto errados.
Os nossos resultados são calibrados: alto no campo que está errado, baixo no campo que está correto, médio num campo que parece estranho sem estar claramente errado
É isto que um verificador typesafe te dá em vez de um juiz rombudo de “isto no seu todo está bom?”

Passo 4: o gate de escalonamento

agora fazemos gating sobre any_flag: escala se qualquer sinalização de campo exceder FIRE_T (0.7, definido acima e partilhado com o marcador <== FIRES do Passo 3)
este é um gate de estilo max (escala se qualquer campo disparar), não uma média, por isso uma bandeira vermelha confiante basta, em vez de ser diluída 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: escalar para o modelo de raciocínio

Como um sinal disparou, 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 a description fabricada, devolvendo ""
- Reconheceu que a página nunca descreve uma data de registo, e recusou-se a inventar uma
- A cascata transformou uma fabricação confiante e válida segundo o schema num campo vazio honesto
- E só gastou dólares de modelo de raciocínio neste único item porque o verificador lhe disse para o fazer

Passo 6: como isto se parece em 100 prompts

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

resultados internos: fronteira custo/qualidade em 100 prompts

como lê-lo:
- losangos pretos = os quatro modelos corridos sozinhos (o custo sobe com a capacidade; o mais forte, gpt-5.5-reasoning, fica em cima à direita a ≈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 cada modelo: varrer o gate dá-te a maior parte da qualidade do modelo de topo a uma fração do seu custo
- o nível barato trata dos itens fáceis quase de graça, e só os itens sinalizados pagam pelo modelo de raciocínio

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

a cascata vale apenas o que vale o seu verificador; o que separa um sinal útil de um inútil:
- Estreito e fundamentado.
  - um sim/não verificável sobre um campo face à fonte (p. ex. “este valor está ausente da fonte?”), não um vago “esta extração está boa?”
  - perguntas vagas dão scores moles e não calibrados
- Mau = TRUE, com critérios explícitos.
  - formula cada pergunta de modo que o caso de escalonar seja o caso true, e indica o que true/false significam
- Por campo, depois agregar com max.
  - uma sinalização por campo localiza o erro e mantém-se esparsa e forte
  - max (“qualquer sinalização dispara”) garante que uma bandeira vermelha confiante escala, em vez de ser diluída no silêncio
- Independente e barato.
  - um verificador dedicado (aqui, o TypeSafe) a julgar a saída apanha os pontos cegos do próprio extrator
  - tem de ser barato, senão não sobra poupança nenhuma para capturar
- Separador / calibrado.
  - um bom sinal é alto em erros reais e baixo em acertos, por isso um único limiar separa limpidamente aceitar vs escalar
  - é essa separação que empurra a curva de pareto para cima e para a esquerda