Prompting estructurado: JSON que se puede validar.

import json
from pydantic import (BaseModel, ConfigDict, Field,
                      ValidationError, model_validator)

class Prevision(BaseModel):
    """Previsión de ventas del trimestre siguiente."""
    model_config = ConfigDict(extra="forbid")
    trimestre: int | None = Field(ge=1, le=4)
    anio: int | None = Field(ge=2018, le=2026)
    bajo: float | None = Field(gt=0, description="millones de USD")
    alto: float | None = Field(gt=0, description="millones de USD")

    @model_validator(mode="after")
    def horquilla(self):
        if (self.bajo is None) != (self.alto is None):
            raise ValueError("bajo y alto: los dos o ninguno")
        if self.bajo is not None and self.bajo > self.alto:
            raise ValueError("bajo > alto")
        return self

pruebas = [
    '{"trimestre": 4, "anio": 2024, "bajo": 7230, "alto": 7650}',
    '{"trimestre": null, "anio": null, "bajo": null, "alto": null}',
    '{"trimestre": 4, "anio": 2024, "bajo": 7650, "alto": 7230}',
    '{"trimestre": "Q4", "anio": 2024, "bajo": 7230, "alto": 7650}',
    '{"trimestre": 4, "anio": 2024, "bajo": 7.230,0, "alto": 7650}',
    '{"trimestre": 4, "anio": 2024, "bajo": 1, "alto": 2, "x": 0}',
]
for t in pruebas:
    try:
        Prevision.model_validate_json(t)
        print("válido")
    except ValidationError as e:
        err = e.errors()[0]
        print(f"inválido · {err['type']} · {err['msg'][:44]}")

def sin_titulos(s):
    if isinstance(s, dict):
        return {k: sin_titulos(v) for k, v in s.items()
                if k != "title"}
    if isinstance(s, list):
        return [sin_titulos(v) for v in s]
    return s

def firma(esquema):
    """Resumen de una línea del esquema, para el prompt."""
    partes = []
    for campo in esquema["required"]:
        prop = esquema["properties"][campo]
        t = next(a for a in prop["anyOf"] if a["type"] != "null")
        rango = (f" {t['minimum']}-{t['maximum']}" if "minimum" in t
                 else f" > {t['exclusiveMinimum']}")
        partes.append(f'"{campo}": {t["type"]}{rango} o null')
    return "{" + ", ".join(partes) + "}"

ESQUEMA = sin_titulos(Prevision.model_json_schema())
TXT_ESQUEMA = json.dumps(ESQUEMA, ensure_ascii=False,
                         separators=(",", ":"))
FIRMA = firma(ESQUEMA)
print(TXT_ESQUEMA)
print(FIRMA)

import re
import numpy as np
import pandas as pd
import torch
from bs4 import BeautifulSoup
from transformers import AutoModelForCausalLM, AutoTokenizer

torch.manual_seed(0)
torch.use_deterministic_algorithms(True)
NOMBRE = "Qwen/Qwen2.5-0.5B-Instruct"
tok = AutoTokenizer.from_pretrained(NOMBRE)
tok.padding_side = "left"
llm = AutoModelForCausalLM.from_pretrained(
    NOMBRE, dtype=torch.float32).eval()

SISTEMA = (
    "Extrae la previsión de ventas del trimestre siguiente. "
    "Responde solo con un objeto JSON así:\n" + FIRMA +
    "\nImportes en millones de USD, sin separador de miles "
    "(1 billion = 1000). Sin previsión: null.")

fil = pd.read_parquet("filings6_documentos.parquet")
items = fil["items"].fillna("").astype(str)
cand = fil[fil["tipo"].isin(["8-K", "OIR", "HR"])
           & ~items.str.contains("4.02")]
muestra = cand.sample(20, random_state=3)

def comunicado(t):
    """Texto del comunicado: lo que sigue al Exhibit 99.1 (SEC)
    o a las magnitudes (CNMV), sin tablas ni formato."""
    t = re.split(r"Exhibit 99\.1</h2>|## Principales magnitudes",
                 t)[-1]
    t = re.sub(r"<table>.*?</table>", "", t, flags=re.S)
    t = "\n".join(l for l in t.splitlines()
                  if not l.startswith("|"))
    t = BeautifulSoup(t, "html.parser").get_text("\n")
    return re.sub(r"\n\s*\n+", "\n", t.replace("*", "")).strip()

docs = [comunicado(t) for t in muestra["texto"]]
n_tok = [len(tok(d)["input_ids"]) for d in docs]
print(muestra["tipo"].value_counts().to_dict())
print("tokens: sistema", len(tok(SISTEMA)["input_ids"]),
      "· esquema completo", len(tok(TXT_ESQUEMA)["input_ids"]),
      "· firma", len(tok(FIRMA)["input_ids"]),
      "· documento (mediana)", int(np.median(n_tok)))
print(docs[0])

verdad = pd.read_csv("documentos6_extraccion_verdad.csv")
g = verdad[verdad["tipo"] == "guidance"].set_index("id_doc")

def esperado(id_doc):
    if id_doc not in g.index:
        return Prevision(trimestre=None, anio=None, bajo=None,
                         alto=None)
    v = g.loc[id_doc]
    return Prevision(trimestre=int(v["periodo"][-1]),
                     anio=int(v["periodo"][:4]),
                     bajo=v["bajo"], alto=v["alto"])

ESPERADO = [esperado(i) for i in muestra["id_doc"]]

def acierta(p, e):
    """Trimestre y año exactos; extremos con un error < 0,5 %."""
    if p is None:
        return False
    if e.bajo is None:
        return p.bajo is None and p.alto is None
    if p.bajo is None or (p.trimestre, p.anio) != (e.trimestre,
                                                  e.anio):
        return False
    return (abs(p.bajo / e.bajo - 1) < 0.005
            and abs(p.alto / e.alto - 1) < 0.005)

print(sum(e.bajo is not None for e in ESPERADO),
      "de 20 documentos traen previsión")

def generar(lista_mensajes, max_tokens=48, **kw):
    textos = [tok.apply_chat_template(m, tokenize=False,
                                      add_generation_prompt=True)
              for m in lista_mensajes]
    b = tok(textos, return_tensors="pt", padding=True)
    with torch.inference_mode():
        o = llm.generate(**b, do_sample=False,
                         max_new_tokens=max_tokens,
                         pad_token_id=tok.eos_token_id, **kw)
    n = b["input_ids"].shape[1]
    return [tok.decode(r[n:], skip_special_tokens=True)
            for r in o]

def validar(texto):
    try:
        return Prevision.model_validate_json(texto), None
    except ValidationError as e:
        return None, e.errors()[0]["type"]

def resumen(nombre, salidas):
    res = [validar(s) for s in salidas]
    ok = sum(p is not None for p, _ in res)
    bien = sum(acierta(p, e) for (p, _), e in zip(res, ESPERADO))
    print(f"{nombre}: {ok}/20 válidos, {bien}/20 correctos")
    return res

mensajes = [[{"role": "system", "content": SISTEMA},
             {"role": "user", "content": d}] for d in docs]
libre = generar(mensajes)
res_libre = resumen("libre", libre)
print(pd.Series([e or "válido" for _, e in res_libre])
      .value_counts().to_dict())
for k in [0, 1, 5, 6, 13]:
    corto = libre[k].replace("`", "'").replace("\n", " ")
    print(f"  {muestra['id_doc'].iloc[k]} "
          f"{res_libre[k][1] or 'válido':>16} · {corto[:58]}")

malas = [k for k, (p, _) in enumerate(res_libre) if p is None]
segundo = [mensajes[k] + [
    {"role": "assistant", "content": libre[k]},
    {"role": "user", "content":
     f"Error de validación: {res_libre[k][1]}. Corrige y "
     "responde solo con el JSON."}] for k in malas]
corregidas = generar(segundo)
reintento = list(libre)
for k, s in zip(malas, corregidas):
    reintento[k] = s
res_reint = resumen("con reintento", reintento)
for k, s in list(zip(malas, corregidas))[:4]:
    corto = s.replace("`", "'").replace("\n", " ")
    print(f"  {muestra['id_doc'].iloc[k]} · {corto[:62]}")

CAMPOS = list(Prevision.model_fields)

def reparar(texto):
    m = re.search(r"\{.*?\}", texto, flags=re.S)
    if not m:
        return texto
    t = m.group(0)
    # 13.561.9 -> 13561.9: puntos de millar salvo el último
    t = re.sub(r"\d+(?:\.\d+){2,}",
               lambda x: x.group(0).replace(".", "", x.group(0)
                                            .count(".") - 1), t)
    try:
        d = json.loads(t)
    except json.JSONDecodeError:
        return t
    d = {k: v for k, v in d.items() if k in CAMPOS}
    for k, v in d.items():
        if isinstance(v, str):
            num = re.sub(r"[^\d.]", "", v.replace(",", "."))
            d[k] = float(num) if num else None
    return json.dumps(d)

reparado = [reparar(s) for s in reintento]
res_rep = resumen("con reparación", reparado)

V = len(tok)
PIEZA = tok.batch_decode([[i] for i in range(V)])
POR_TEXTO = {}
for i, s in enumerate(PIEZA):
    POR_TEXTO.setdefault(s, []).append(i)
DIGITOS = [i for d in "0123456789" for i in POR_TEXTO[d]]

def prefijos(resto):
    """Tokens que son un prefijo no vacío de `resto`."""
    return [i for n in range(1, len(resto) + 1)
            for i in POR_TEXTO.get(resto[:n], [])]

texto = tok.apply_chat_template(mensajes[0], tokenize=False,
                                add_generation_prompt=True)
x = tok(texto + '{"trimestre": ', return_tensors="pt")
with torch.inference_mode():
    z = llm(**x).logits[0, -1]
p = torch.softmax(z.double(), -1).numpy()
permitidos = ({i for i in DIGITOS if PIEZA[i] in "1234"}
              | set(prefijos("null")))
orden = np.argsort(-p)[:8]
pd.DataFrame({"token": [repr(PIEZA[i]) for i in orden],
              "prob_%": (100 * p[orden]).round(1),
              "permitido": [i in permitidos for i in orden]})

EOS = tok.eos_token_id
NUMERO = re.compile(r"null|\d+(\.\d*)?")

def plantilla(esquema):
    """Literales y huecos (tipo, mínimo, máximo) de un objeto
    plano, en el orden de `required`."""
    piezas = []
    for k, campo in enumerate(esquema["required"]):
        prop = esquema["properties"][campo]
        t = next(a for a in prop.get("anyOf", [prop])
                 if a["type"] != "null")
        piezas.append(("{" if k == 0 else ", ") + f'"{campo}": ')
        piezas.append((t["type"], t.get("minimum"),
                       t.get("maximum")))
    return piezas + ["}"]

def en_hueco(v, tipo, lo, hi, siguiente):
    """Tokens permitidos con el valor parcial `v` en un hueco."""
    ok = prefijos("null"[len(v):]) if "null".startswith(v) else []
    if v.startswith("n"):
        return ok
    if tipo == "integer" and lo is not None and hi is not None:
        validos = [str(i) for i in range(lo, hi + 1)]
        ok += [d for d in DIGITOS
               if any(s.startswith(v + PIEZA[d]) for s in validos)]
        return ok + (prefijos(siguiente) if v in validos else [])
    entero, _, dec = v.partition(".")
    if not v:                       # sin ceros a la izquierda
        ok += [d for d in DIGITOS if PIEZA[d] != "0"]
    elif len(entero) < 7 and "." not in v or (
            "." in v and len(dec) < 2):
        ok += DIGITOS
    if tipo == "number" and v and "." not in v:
        ok += POR_TEXTO["."]
    cerrado = v and not v.endswith(".")
    return ok + (prefijos(siguiente) if cerrado else [])

def permitidos_para(texto, piezas):
    pos = 0
    for k, pieza in enumerate(piezas):
        resto = texto[pos:]
        if isinstance(pieza, str):           # literal
            if resto.startswith(pieza):
                pos += len(pieza)
                continue
            return prefijos(pieza[len(resto):])
        m = NUMERO.match(resto)
        v = m.group(0) if m else resto
        if pos + len(v) < len(texto):        # hueco ya cerrado
            pos += len(v)
            continue
        return en_hueco(v, *pieza, piezas[k + 1])
    return [EOS]

def generar_restringido(lista_mensajes, esquema):
    piezas = plantilla(esquema)
    textos = [tok.apply_chat_template(m, tokenize=False,
                                      add_generation_prompt=True)
              for m in lista_mensajes]
    b = tok(textos, return_tensors="pt", padding=True)
    n = b["input_ids"].shape[1]

    def regla(_, ids):
        return permitidos_para(
            tok.decode(ids[n:], skip_special_tokens=True), piezas)

    with torch.inference_mode():
        o = llm.generate(**b, do_sample=False, max_new_tokens=60,
                         pad_token_id=EOS,
                         prefix_allowed_tokens_fn=regla)
    return [tok.decode(r[n:], skip_special_tokens=True)
            for r in o]

restringido = generar_restringido(mensajes, ESQUEMA)
res_restr = resumen("restringida", restringido)
for k in [0, 1, 5, 6, 13]:
    print(f"  {muestra['id_doc'].iloc[k]} · {restringido[k]}")

from openai import OpenAI

cliente = OpenAI()
r = cliente.responses.parse(
    model="gpt-6-astra",
    instructions="Extrae la previsión de ventas del trimestre "
                 "siguiente en millones de USD.",
    input=texto_del_comunicado,
    text_format=Prevision,
)
prevision = r.output_parsed          # una instancia de Prevision

import anthropic

cliente = anthropic.Anthropic()
r = cliente.messages.parse(
    model="claude-opus-5-5",
    max_tokens=1024,
    system="Extrae la previsión de ventas del trimestre "
           "siguiente en millones de USD.",
    messages=[{"role": "user", "content": texto_del_comunicado}],
    output_format=Prevision,
)
prevision = r.parsed_output          # una instancia de Prevision

import threading
from http.server import BaseHTTPRequestHandler
from http.server import ThreadingHTTPServer

class APIEstructurada(BaseHTTPRequestHandler):
    def log_message(self, *args):
        pass

    def do_POST(self):
        n = int(self.headers["Content-Length"])
        req = json.loads(self.rfile.read(n))
        if self.path == "/v1/chat/completions":
            esq = req["response_format"]["json_schema"]["schema"]
            msgs = req["messages"]
        else:                                  # /v1/messages
            esq = req["output_config"]["format"]["schema"]
            msgs = ([{"role": "system", "content": req["system"]}]
                    + req["messages"])
        print("  [servidor]", self.path, "· trimestre:",
              json.dumps(esq["properties"]["trimestre"]["anyOf"][0],
                         ensure_ascii=False))
        txt = generar_restringido([msgs], esq)[0]
        if self.path == "/v1/chat/completions":
            r = {"id": "chatcmpl-local-2",
                 "object": "chat.completion",
                 "created": 1767225600, "model": req["model"],
                 "choices": [{"index": 0, "logprobs": None,
                              "finish_reason": "stop",
                              "message": {"role": "assistant",
                                          "content": txt,
                                          "refusal": None}}],
                 "usage": {"prompt_tokens": 0,
                           "completion_tokens": 0,
                           "total_tokens": 0}}
        else:
            r = {"id": "msg_local_2", "type": "message",
                 "role": "assistant", "model": req["model"],
                 "content": [{"type": "text", "text": txt}],
                 "stop_reason": "end_turn", "stop_sequence": None,
                 "usage": {"input_tokens": 0, "output_tokens": 0}}
        cuerpo = json.dumps(r).encode()
        self.send_response(200)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(cuerpo)))
        self.end_headers()
        self.wfile.write(cuerpo)

srv = ThreadingHTTPServer(("127.0.0.1", 0), APIEstructurada)
threading.Thread(target=srv.serve_forever, daemon=True).start()
URL = f"http://127.0.0.1:{srv.server_port}"

from openai import OpenAI

INSTR = ("Extrae la previsión de ventas del trimestre siguiente "
         "en millones de USD (1 billion = 1000).")
oa = OpenAI(base_url=URL + "/v1", api_key="local", max_retries=0)
for k in [0, 6]:
    try:
        r = oa.chat.completions.parse(
            model="qwen2.5-0.5b-instruct",
            response_format=Prevision,
            messages=[{"role": "system", "content": INSTR},
                      {"role": "user", "content": docs[k]}])
        print(" ", repr(r.choices[0].message.parsed))
    except ValidationError as e:
        print("  ValidationError:", e.errors()[0]["msg"])
    print("  esperado:", repr(ESPERADO[k]))

import anthropic

an = anthropic.Anthropic(base_url=URL, api_key="local",
                         max_retries=0)
for k in [0]:
    try:
        r = an.messages.parse(
            model="qwen2.5-0.5b-instruct", max_tokens=200,
            system=INSTR, output_format=Prevision,
            messages=[{"role": "user", "content": docs[k]}])
        print(" ", repr(r.parsed_output))
    except ValidationError as e:
        print("  ValidationError:", e.errors()[0]["msg"])
srv.shutdown()