--- title: "6.2.2 Prompting estructurado y JSON validado con Pydantic" description: "Esquemas JSON desde Pydantic, validación, reintentos, reparación y decodificación restringida, medidos en 20 filings con un LLM local y una API simulada." url: "https://pythonparatrading.com/curso-python-trading/nivel-6/prompting-estructurado-y-json-con-pydantic/" lang: "es" image: "https://pythonparatrading.com/og/curso-python-trading/nivel-6/prompting-estructurado-y-json-con-pydantic.png" --- # 6.2.2 · Prompting estructurado: JSON que se puede validar > Un dato extraído por un LLM solo sirve si un programa puede leerlo. Esta lección define el contrato con Pydantic, se lo pide a un modelo de 0,5B parámetros en 20 comunicados de resultados y mide cuántas respuestas son JSON válido, qué arreglan los reintentos y la reparación, y qué garantiza la decodificación restringida que usan las APIs con salidas estructuradas. - Curso: [Nivel 6: IA generativa, LLMs, agentes, big data y datos alternativos](https://pythonparatrading.com/curso-python-trading/nivel-6/) - Módulo: 6.2 · LLMs para investigación y trading - Duración: 75 min - Requisitos: [6.2.1 · Cómo funciona un LLM: del siguiente token a la API](https://pythonparatrading.com/curso-python-trading/nivel-6/como-funcionan-los-llm-apis-y-modelos-locales/), [0.4.2 · Tipado: type hints, mypy o ty y Pydantic para configuraciones](https://pythonparatrading.com/curso-python-trading/nivel-0/tipado-type-hints-y-pydantic/) ## Al terminar esta lección - Definir con Pydantic el contrato de una extracción (tipos, rangos, nulos, reglas entre campos) y generar su JSON Schema. - Escribir un prompt estructurado: rol, tarea, esquema, unidades, caso sin dato y texto delimitado. - Medir la tasa de JSON válido y la de valores correctos de un LLM frente a una verdad conocida, y clasificar los errores. - Implementar un bucle de validación con reintento y una reparación determinista, y medir lo que aporta cada uno. - Explicar e implementar la decodificación restringida, y usar las salidas estructuradas de los SDK de OpenAI y Anthropic. ## Del texto libre a un contrato En la lección 6.2.1 el modelo respondía con frases. Para un programa, una frase es un problema: «la compañía espera entre 7,23 y 7,65 mil millones» hay que convertirla en dos números, una unidad y un trimestre antes de que sirva para algo, y cada respuesta llega redactada de otra manera. Lo que un pipeline de research necesita es un **contrato**: un formato fijo con campos, tipos y rangos, que el modelo rellena y que un programa comprueba antes de aceptar el dato. El formato natural es JSON, y la herramienta natural para el contrato en Python es Pydantic (lección 0.4.2): define la estructura una vez y de ella salen el **JSON Schema** que se le enseña al modelo (o a la API) y la **validación** de lo que devuelve. El flujo completo de esta lección: ```mermaid flowchart TB A[Documento] --> B[Prompt y esquema] B --> C[LLM] C --> D[Validar] D -->|válido| E[Dato] D -->|inválido| F[Reintentar] F --> C D -->|inválido| G[Reparar] G --> D ``` La tarea es extraer la **previsión de ventas** (el guidance) del comunicado de resultados de una empresa: trimestre, año y los dos extremos de la horquilla en millones de USD. Los documentos son los 8-K tipo SEC y los comunicados tipo CNMV de `filings6_documentos.parquet`, y la respuesta correcta de cada uno está en `documentos6_extraccion_verdad.csv` (README de los datos del nivel 6). Son datos sintéticos: las empresas no existen y las frases salen de plantillas, de modo que el resultado mide al modelo en un caso fácil, no en filings reales. ## El contrato en Pydantic ```python 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]}") ``` ```output válido válido inválido · value_error · Value error, bajo > alto inválido · int_parsing · Input should be a valid integer, unable to p inválido · json_invalid · Invalid JSON: key must be a string at line 1 inválido · extra_forbidden · Extra inputs are not permitted ``` El contrato rechaza seis maneras distintas de equivocarse con un mensaje que dice cuál. Tres detalles del diseño: - **Los nulos son explícitos.** Un documento sin previsión no es un error: todos los campos van a `null`. Sin esa salida, el modelo tiende a inventar una cifra para rellenar el hueco (lección 6.2.1). - **Las unidades van en el contrato.** «Millones de USD» está en la descripción del campo y en el prompt; sin ella, una previsión de «7,23 billion» puede llegar como 7,23, como 7.230 o como 7.230.000.000. - **`extra="forbid"`.** Por defecto Pydantic ignora las claves que no conoce. Con un LLM eso es peligroso: un modelo pequeño copia a veces partes del esquema (`"required": [...]`) dentro de la respuesta, y sin `forbid` ese JSON pasaría como válido. La regla «bajo ≤ alto» relaciona dos campos y no se puede escribir en JSON Schema: el esquema que se le enseña al modelo no la incluye y solo la comprueba el validador. El esquema que genera Pydantic, sin los títulos (que no aportan nada al modelo y cuestan tokens), y una **firma** de una línea derivada de él: ```python 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) ``` ```output {"additionalProperties":false,"description":"Previsión de ventas del trimestre siguiente.","properties":{"trimestre":{"anyOf":[{"maximum":4,"minimum":1,"type":"integer"},{"type":"null"}]},"anio":{"anyOf":[{"maximum":2026,"minimum":2018,"type":"integer"},{"type":"null"}]},"bajo":{"anyOf":[{"exclusiveMinimum":0,"type":"number"},{"type":"null"}],"description":"millones de USD"},"alto":{"anyOf":[{"exclusiveMinimum":0,"type":"number"},{"type":"null"}],"description":"millones de USD"}},"required":["trimestre","anio","bajo","alto"],"type":"object"} {"trimestre": integer 1-4 o null, "anio": integer 2018-2026 o null, "bajo": number > 0 o null, "alto": number > 0 o null} ``` ## Un prompt estructurado Un prompt para extraer datos tiene las mismas piezas siempre, en el mismo orden: el **rol y la tarea** (qué extraer), el **formato** (el esquema y nada más que JSON), las **convenciones** (unidades, separadores, cómo convertir «billion»), el **caso sin dato** y el **texto**, separado de las instrucciones. Las instrucciones van en el mensaje de sistema y el documento en el del usuario: así el documento no puede hacerse pasar por instrucción (la lección 6.4.5 trata la inyección de prompts). ```python 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|## Principales magnitudes", t)[-1] t = re.sub(r".*?
", "", 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]) ``` ```output {'8-K': 14, 'OIR': 4, 'HR': 2} tokens: sistema 100 · esquema completo 144 · firma 51 · documento (mediana) 76 Revenue of $6.10 billion Net income of $5,157.5 million, or $13.67 per share Stores at quarter end: 324. We currently anticipate the fourth quarter of 2024 revenue of between $7.23 billion and $7.65 billion. ``` El JSON Schema completo ocupa 144 tokens; la firma de una línea, 51, y el mensaje de sistema entero, 100. El esquema completo es lo que reciben las APIs con salidas estructuradas (más abajo); en un prompt escrito a mano, una firma derivada del mismo modelo de Pydantic dice lo mismo con un tercio de los tokens, y ese ahorro se multiplica por cada documento (con una API comercial, literalmente). Los comunicados se recortan a su parte de texto, sin tablas ni etiquetas, para que la lección corra en la CPU; aun así conservan la trampa principal, la cifra de ventas del trimestre que acaba de cerrar («Revenue of \$6.10 billion»), que no es la previsión. La verdad de cada documento: ```python 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") ``` ```output 15 de 20 documentos traen previsión ``` ## Cuántas respuestas son JSON válido La medida tiene dos niveles: **validez** (el texto es JSON y cumple el contrato) y **exactitud** (además, los valores coinciden con la verdad). Las 20 peticiones van en un lote con decodificación voraz y un máximo de 48 tokens por respuesta, más que suficiente para el JSON esperado. ```python 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]}") ``` ```output libre: 2/20 válidos, 1/20 correctos {'greater_than': 10, 'json_invalid': 5, 'value_error': 3, 'válido': 2} D003403 value_error · { "trimestre": 4, "anio": null, "bajo": 5157.5, "a D002322 válido · { "trimestre": 3, "anio": null, "bajo": null, "alt D002189 json_invalid · '''json { "trimestre": 4, "anio": 2022, "bajo": 14.0 D001708 greater_than · { "trimestre": 3, "anio": null, "bajo": 0, "alto": D003459 json_invalid · '''json { "trimestre": 3, "anio": null, "bajo": 530, ``` Solo 2 de las 20 respuestas cumplen el contrato y solo 1 es correcta. Los errores no son aleatorios. La mitad (10) son `greater_than`: un importe que no es positivo, como el `"bajo": 0` de D001708, aunque la firma diga «number > 0»; 5 no son JSON (dos de las que muestra la celda empiezan con un bloque de Markdown, que aparece con comillas simples); y 3 fallan en las reglas entre campos del validador (bajo ≤ alto; los dos importes o ninguno): D003403, por ejemplo, toma el beneficio (5.157,5) como extremo bajo. Varias dejan el año en `null` aunque el texto lo dice. Las válidas tampoco aciertan necesariamente: D002322 es válida con los importes en `null` y ese documento sí trae previsión. *Fig. — Extracción libre con Qwen2.5-0.5B-Instruct en 20 comunicados: primer error de validación de Pydantic de cada respuesta.* Cuatro barras horizontales. 10 respuestas tienen un importe que no es positivo (destacado), 5 no son JSON válido, 3 tienen el extremo bajo mayor que el alto y 2 son válidas. La figura cuenta los tipos de error de la extracción libre. Lo que más falla no es la sintaxis, sino el contenido: un modelo de 0,5B no sigue con precisión una instrucción como «sin previsión: null» o «importes positivos». ## Reintentar con el mensaje de error La receta habitual con modelos grandes es devolverle al modelo su respuesta y el error del validador, y pedirle que lo corrija. Con los SDK de las APIs, el mismo bucle se escribe con unas pocas líneas y un número máximo de intentos. Aquí, un intento más para las respuestas inválidas: ```python 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]}") ``` ```output con reintento: 2/20 válidos, 1/20 correctos D003403 · null D003657 · null D000169 · null D003018 · null ``` El reintento no arregla nada: ninguna de las 18 respuestas inválidas pasa la validación en el segundo intento, y las cuatro que muestra la celda son un simple `null` (JSON válido, pero no un objeto `Prevision`). Un modelo de 0,5B no sabe usar el mensaje de error para corregir la respuesta y se limita a contestar algo corto. Con modelos grandes, el reintento con el error sí suele corregir una parte de los fallos; aquí el resultado se mide, no se supone, y cada reintento cuesta una llamada más con el historial entero. ## Reparar sin el modelo La otra vía es arreglar el texto con código determinista antes de validarlo: quedarse con el primer objeto JSON, quitar las claves que no están en el contrato, convertir a número un número escrito como texto y quitar los puntos de millar de una cifra como `13.561.9`. Es barata y no gasta tokens, pero cada regla es una suposición sobre el error. ```python 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) ``` ```output con reparación: 5/20 válidos, 1/20 correctos ``` La reparación sube la validez de 2 a 5 respuestas sin gastar un token, y la exactitud no se mueve: sigue en 1. Lo que se ha reparado es la forma (bloques de Markdown alrededor, separadores, claves de más), no el contenido. Ese es el riesgo de reparar: una respuesta reparada pasa la validación y entra en el pipeline con la misma apariencia que una correcta. En producción, cada reparación se registra y su resultado se contrasta con la fuente. ## Decodificación restringida: el formato por construcción Las dos vías anteriores actúan **después** de generar. La decodificación restringida actúa **durante** la generación: en cada paso, antes de elegir el token, pone a $-\infty$ el logit de todo token que no pueda continuar un texto válido según el esquema. El modelo sigue eligiendo el token más probable, pero solo entre los permitidos. Willard y Louf (2023) mostraron cómo hacerlo de forma eficiente convirtiendo una expresión regular o una gramática en una máquina de estados finitos sobre el vocabulario (es la base de la librería Outlines); vLLM usa los motores xgrammar o guidance, y las APIs de OpenAI y Anthropic ofrecen salidas estructuradas con un esquema. Para un objeto plano como `Prevision`, la máquina de estados es sencilla: literales fijos (`{"trimestre": `, `, "anio": `…) que el modelo tiene que escribir tal cual, y huecos para cada valor: ```mermaid stateDiagram-v2 [*] --> Literal Literal --> Valor: literal completo Valor --> Valor: dígito o punto Valor --> Literal: valor cerrado Literal --> [*]: llave final ``` En un hueco, los tokens permitidos son los dígitos que mantienen el número dentro de su tipo y su rango, el punto decimal si el campo es `number`, el comienzo de `null` y, cuando el número ya está completo, el comienzo del literal siguiente. Así se ve la máscara en el primer hueco del primer documento, después de que el modelo haya escrito `{"trimestre": `: ```python 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]}) ``` ```output token prob_% permitido 0 '4' 84.4 True 1 '3' 10.3 True 2 '1' 2.9 True 3 '2' 1.0 True 4 '0' 0.4 False 5 '5' 0.4 False 6 '6' 0.2 False 7 '8' 0.1 False ``` En ese paso el modelo da el 84,4 % al token `4` (la previsión es del cuarto trimestre) y reparte el resto entre `3`, `1` y `2`. Los tokens `0`, `5`, `6` y `8` suman un 1,1 %, poco, pero en 20 documentos y cuatro campos un 1 % de probabilidad acaba saliendo; la máscara los elimina. La figura dibuja el paso. *Fig. — Decodificación restringida: probabilidad de los ocho tokens más probables tras «{"trimestre": » en el primer documento, y cuáles deja pasar la máscara (1 a 4 o null).* Ocho barras horizontales. El token 4 tiene un 84,4 % (destacado) y está permitido; 3, 1 y 2 también (10,3, 2,9 y 1,0 %). Los tokens 0, 5, 6 y 8 suman un 1,1 % y la máscara los elimina. El decodificador completo lee la plantilla del propio JSON Schema (los campos en el orden de `required`, el tipo y el rango de cada uno) y se conecta a `generate` de transformers con `prefix_allowed_tokens_fn`, que en cada paso recibe lo generado hasta ahora y devuelve la lista de tokens permitidos: ```python 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]}") ``` ```output restringida: 13/20 válidos, 3/20 correctos D003403 · {"trimestre": 4, "anio": 2024, "bajo": 5157.5, "alto": 7650} D002322 · {"trimestre": 3, "anio": 2023, "bajo": 4.94, "alto": 5.15} D002189 · {"trimestre": 4, "anio": 2022, "bajo": 14.00, "alto": 14.73} D001708 · {"trimestre": 3, "anio": 2021, "bajo": 11.8, "alto": 43} D003459 · {"trimestre": 1, "anio": 2025, "bajo": 530, "alto": 555} ``` Con la máscara, 13 de las 20 respuestas son válidas y 3 correctas, frente a 2 y 1 de la extracción libre. Las 7 inválidas fallan todas por las reglas entre campos (bajo ≤ alto; los dos importes o ninguno), que no están en el JSON Schema y que la máscara, por tanto, no puede imponer (el ejercicio 2 lo comprueba). La exactitud mejora algo porque el modelo ya no puede escribir `null` en el año ni `0` en los importes y tiene que elegir una cifra del texto, pero las que elige muestran los mismos fallos de fondo: D003403 toma el beneficio (5.157,5) como extremo bajo; D002322 y D002189 copian las cifras en miles de millones sin convertirlas (4,94 y 5,15 donde la verdad es 4.936 y 5.147); D003459 acierta. El decodificador de la lección implementa solo una parte de JSON Schema (tipos, rangos de enteros, sin ceros a la izquierda); los motores de vLLM o de las APIs cubren mucho más, pero ninguno arregla que el modelo no sepa qué número es la previsión. ## Salidas estructuradas en las APIs Las APIs comerciales hacen esto en el servidor. El código real con los SDK oficiales, sin ejecutar (necesita clave y red); los dos aceptan directamente la clase de Pydantic: ```python skip title="salida_estructurada_openai.py" 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 ``` ```python skip title="salida_estructurada_anthropic.py" 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 ``` La documentación de OpenAI recomienda «always using Structured Outputs instead of JSON mode when possible»: el modo JSON solo garantiza que la respuesta sea JSON, no que cumpla el esquema. La de Anthropic explica que el SDK transforma el esquema del modelo de Pydantic, lo envía como `output_config.format`, valida la respuesta y la devuelve en `parsed_output`. Ollama acepta un JSON Schema en su parámetro `format` y vLLM en `response_format` (lección 6.2.1). Para ver qué viaja de verdad, la lección monta otra vez un servidor local con la forma de las dos APIs (como en 6.2.1), pero que, cuando la petición trae un esquema, genera con el decodificador restringido de la celda anterior y ese esquema. Es lo que hace un servidor con salidas estructuradas, con un modelo de 0,5B en vez de uno comercial. ```python 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}" ``` El recuento de tokens de `usage` va a cero porque no interesa aquí. Primero, el SDK de OpenAI con `chat.completions.parse`, que acepta la clase de Pydantic en `response_format`: ```python 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])) ``` ```output [servidor] /v1/chat/completions · trimestre: {"maximum": 4, "minimum": 1, "type": "integer"} Prevision(trimestre=1, anio=2024, bajo=7.23, alto=7.65) esperado: Prevision(trimestre=4, anio=2024, bajo=7228.0, alto=7648.0) [servidor] /v1/chat/completions · trimestre: {"maximum": 4, "minimum": 1, "type": "integer"} Prevision(trimestre=1, anio=2021, bajo=659.0, alto=692.0) esperado: Prevision(trimestre=4, anio=2021, bajo=659.0, alto=692.0) ``` El SDK de OpenAI ha enviado el esquema de `Prevision` como `response_format` de tipo `json_schema` con `"strict": true`, y los rangos han viajado: el servidor ve `minimum` 1 y `maximum` 4 en el trimestre y los aplica. La respuesta llega convertida en un objeto `Prevision` ya validado (`message.parsed`). Pero los valores están mal: el trimestre es 1 en los dos documentos (la previsión es del cuarto) y el primero da la horquilla en miles de millones (7,23 y 7,65). El servidor del simulador aplica el esquema como máscara pero no se lo enseña al modelo, y con la instrucción corta el modelo no sabe qué significa cada campo. Qué incluyen los servidores comerciales en el contexto del modelo no está documentado con ese detalle; lo que sí se ve aquí es que la máscara no sustituye a explicar la tarea. ```python 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() ``` ```output [servidor] /v1/messages · trimestre: {"type": "integer", "description": "{maximum: 4, minimum: 1}"} ValidationError: Input should be greater than or equal to 2018 ``` El SDK de Anthropic 2.1.0 envía otro esquema para el mismo modelo de Pydantic: ha movido los rangos a la descripción del campo (`"description": "{maximum: 4, minimum: 1}"`), así que el servidor ya no puede forzarlos al decodificar y el modelo escribe un año fuera de 2018-2026. El SDK valida la respuesta con la clase de Pydantic al recibirla y lanza `ValidationError`. Es exactamente el papel de la validación del cliente: lo que no viaja en el esquema solo se comprueba al final, y por eso nunca se omite, aunque el servidor prometa un formato. ## Validez no es exactitud | Método | JSON válido | Valores correctos | Coste | |---|---|---|---| | Extracción libre (firma en el prompt) | 2/20 | 1/20 | Una llamada por documento | | Más un reintento con el error | 2/20 | 1/20 | Una llamada más por cada respuesta inválida (18) | | Más reparación determinista | 5/20 | 1/20 | Ninguno en tokens; riesgo de validar un error | | Decodificación restringida | 13/20 | 3/20 | Una llamada; la máscara en cada paso | La figura de la cabecera dibuja esta tabla. Las técnicas de la lección actúan sobre la **forma**: la decodificación restringida lleva la validez de 2 a 13 respuestas (y a 20 si la regla entre campos no existiera), pero los valores correctos pasan solo de 1 a 3. Con un modelo de 0,5B, el cuello de botella es entender el documento, no escribir JSON. Con un modelo comercial y salidas estructuradas, la validez deja de ser el problema principal y la exactitud es lo que hay que medir; la lección 6.2.3 lo hace con previsiones, KPIs y tono, y compara el LLM con reglas, con FinBERT y con un diccionario. ## Ejercicios 1. Cambia el contrato para que el modelo devuelva también la `moneda` como texto con valores posibles `"USD"` o `"EUR"` (pista: `Literal["USD", "EUR"] | None`) y mira cómo aparece en el JSON Schema. ¿Podría el decodificador restringido de la lección manejar ese campo tal como está? **Solución.** ```python from typing import Literal class PrevisionMoneda(Prevision): moneda: Literal["USD", "EUR"] | None = None esq = sin_titulos(PrevisionMoneda.model_json_schema()) print(json.dumps(esq["properties"]["moneda"])) print(esq["required"]) ``` ```output {"anyOf": [{"enum": ["USD", "EUR"], "type": "string"}, {"type": "null"}], "default": null} ['trimestre', 'anio', 'bajo', 'alto'] ``` El campo aparece como `enum` dentro de `anyOf` y, al tener valor por defecto, no entra en `required`. El decodificador de la lección solo conoce huecos numéricos y recorre `required`: ignoraría el campo. Para manejarlo haría falta un hueco de tipo enumerado, cuyos tokens permitidos sean los prefijos de `"USD"`, `"EUR"` y `null`, la misma idea que el literal. Las salidas estructuradas estrictas de OpenAI exigen además que todos los campos estén en `required` (los opcionales se expresan con `null`), y su SDK reescribe el esquema para cumplirlo. 2. El decodificador restringido garantiza que `bajo` es un número positivo, pero no que `bajo ≤ alto` ni que vengan los dos importes o ninguno. Cuenta en las 20 respuestas restringidas cuántas fallan solo por esas reglas. **Solución.** ```python tipos = pd.Series([e or "válido" for _, e in res_restr]) tipos.value_counts() ``` ```output válido 13 value_error 7 Name: count, dtype: int64 ``` Las reglas entre campos (`model_validator`) no existen en JSON Schema y ninguna API las impone al decodificar. Por eso la validación con Pydantic se hace siempre al recibir la respuesta, aunque el servidor prometa un formato. 3. Escribe una función `extraer(doc, llamar, intentos=2)` que pida la extracción con la función `llamar` (que recibe los mensajes y devuelve el texto del modelo), valide, reintente con el error y, si sigue fallando, devuelva `None` con el motivo, sin reparar nada. Pruébala sin gastar un solo token: con una función falsa que primero responde con un JSON roto y después con uno válido. ¿Por qué conviene registrar el motivo además de devolver `None`? **Solución.** ```python def extraer(doc, llamar, intentos=2): msgs = [{"role": "system", "content": SISTEMA}, {"role": "user", "content": doc}] for _ in range(intentos): txt = llamar(msgs) p, err = validar(txt) if p is not None: return p, None msgs = msgs + [ {"role": "assistant", "content": txt}, {"role": "user", "content": f"Error de validación: {err}. Corrige y responde " "solo con el JSON."}] return None, err respuestas = iter([ '{"trimestre": 4, "anio": 2024, "bajo": 7.230,0}', '{"trimestre": 4, "anio": 2024, "bajo": 7230, "alto": 7650}']) print(extraer(docs[0], lambda m: next(respuestas))) print(extraer(docs[0], lambda m: "No hay previsión.")) ``` ```output (Prevision(trimestre=4, anio=2024, bajo=7230.0, alto=7650.0), None) (None, 'json_invalid') ``` Separar la llamada al modelo (`llamar`) del bucle permite probar la lógica de reintentos con respuestas fijas, en milisegundos y sin red; en producción `llamar` envuelve el SDK de la API o el modelo local. El motivo distingue un documento sin formato reconocible de un fallo sistemático (por ejemplo, todas las previsiones en miles de millones fallan por unidades). Contar motivos por tipo de documento, idioma o fecha es la manera de ver que un cambio en los documentos de origen o en el modelo ha roto la extracción. ## Autoevaluación 1. Con la decodificación restringida, las 20 respuestas del modelo de 0,5B cumplen el JSON Schema y 13 pasan además la validación de Pydantic. ¿Qué se puede concluir? - A. Que las 20 extracciones son correctas - B. Que el modelo ha entendido la tarea - C. Solo que el formato es correcto: tipos, campos y rangos del esquema; los valores pueden seguir siendo erróneos, como muestra la tasa de aciertos frente a la verdad - D. Que ya no hace falta validar con Pydantic *Respuesta C: Solo que el formato es correcto: tipos, campos y rangos del esquema; los valores pueden seguir siendo erróneos, como muestra la tasa de aciertos frente a la verdad.* La máscara impide escribir algo que no encaje en el esquema, pero no sabe qué cifra es la previsión: en la lección, 3 de 20 respuestas son correctas. Las 7 que no pasan Pydantic fallan por las reglas entre campos (bajo ≤ alto; los dos importes o ninguno), que no se expresan en JSON Schema y solo comprueba el validador. 2. En el reintento, el modelo recibe su respuesta y el mensaje de error de Pydantic. ¿Qué hace el modelo de 0,5B en la mayoría de los casos? - A. Corrige el campo que fallaba - B. Responde con algo corto que no es una previsión (en la lección, `null`) y ninguna de las 18 respuestas pasa la validación - C. Se niega a responder - D. Devuelve el documento original *Respuesta B: Responde con algo corto que no es una previsión (en la lección, `null`) y ninguna de las 18 respuestas pasa la validación.* Un modelo tan pequeño no sabe usar el mensaje de error para corregir la respuesta. El reintento con el error funciona mucho mejor con modelos grandes; con este, hay que medirlo, no suponerlo, y cada reintento cuesta otra llamada con el historial. 3. ¿Por qué el modelo de Pydantic de la lección usa `extra="forbid"`? - A. Para que el JSON ocupe menos tokens - B. Porque OpenAI no admite campos opcionales - C. Para que Pydantic convierta los textos a números - D. Para que un JSON con claves que no están en el contrato (el modelo copia a veces `required` o `title` del esquema) se rechace en vez de aceptarse en silencio; y para que el esquema lleve `additionalProperties: false`, que piden las salidas estructuradas estrictas *Respuesta D: Para que un JSON con claves que no están en el contrato (el modelo copia a veces `required` o `title` del esquema) se rechace en vez de aceptarse en silencio; y para que el esquema lleve `additionalProperties: false`, que piden las salidas estructuradas estrictas.* Por defecto Pydantic ignora las claves desconocidas, y una respuesta que trae `"required": [...]` dentro pasaría como válida. Con `forbid` el error queda a la vista. El SDK de OpenAI envía el esquema con `additionalProperties: false`, requisito de su modo estricto. 4. El SDK de Anthropic 2.1.0 envía el esquema de `Prevision` con los rangos (`minimum`, `maximum`) movidos a la descripción del campo. ¿Qué consecuencia tiene? - A. Ninguna: el servidor sigue aplicándolos - B. El servidor ya no puede forzar el rango al decodificar; el SDK valida la respuesta con el modelo de Pydantic al recibirla, y ahí es donde se detecta un trimestre 0 o 5 - C. La respuesta deja de ser JSON - D. El modelo recibe un error 400 *Respuesta B: El servidor ya no puede forzar el rango al decodificar; el SDK valida la respuesta con el modelo de Pydantic al recibirla, y ahí es donde se detecta un trimestre 0 o 5.* Lo que no viaja en el esquema no lo puede imponer la decodificación restringida: el servidor solo ve el texto de la descripción. La validación del cliente es la última barrera y no se puede omitir. 5. Un analista propone pedir al LLM la previsión en JSON y, si falla, «arreglar» el texto con expresiones regulares que quitan puntos de millar. ¿Cuál es el riesgo principal? - A. Que la reparación convierta en válido un JSON con un número mal interpretado (por ejemplo, una cifra en miles de millones leída como millones), que entonces pasa la validación y entra en el pipeline sin aviso - B. Que tarde demasiado - C. Que Pydantic no acepte números reparados - D. Ninguno: la reparación siempre mejora los datos *Respuesta A: Que la reparación convierta en válido un JSON con un número mal interpretado (por ejemplo, una cifra en miles de millones leída como millones), que entonces pasa la validación y entra en el pipeline sin aviso.* La reparación arregla la sintaxis, no el significado. En la lección sube la validez de 2 a 5 respuestas y los valores correctos siguen en 1. Toda reparación se registra y su resultado se contrasta con la fuente o con reglas de plausibilidad. ## Ideas clave - El contrato va primero: un modelo de Pydantic con tipos, rangos, nulos, `extra="forbid"` y reglas entre campos. De él salen el JSON Schema para el prompt o la API y la validación de la respuesta. - Un prompt estructurado dice el rol, la tarea, el esquema, las unidades, qué hacer si no hay dato y dónde empieza el texto. El esquema cuesta tokens en cada llamada: 144 el JSON Schema completo y 51 una firma de una línea derivada de él. - Con un modelo de 0,5B, 2 de 20 respuestas son válidas y 1 correcta; los errores típicos son importes a cero, Markdown alrededor del JSON, cifras de distinta magnitud mezcladas y miles de millones sin convertir. - El reintento con el mensaje de error no ayuda a un modelo tan pequeño; la reparación determinista sube la validez de 2 a 5 sin mejorar la exactitud, y puede volver válido un dato erróneo. - La decodificación restringida enmascara los tokens que no encajan en el esquema: 13 de 20 respuestas válidas (las demás fallan una regla entre campos) y solo 3 correctas. Validez no es exactitud. - Las APIs con salidas estructuradas (OpenAI `parse`, Anthropic `messages.parse`, Ollama `format`, vLLM `response_format`) aplican esa idea en el servidor; lo que el SDK no envía en el esquema solo lo comprueba Pydantic al recibir la respuesta. ## Recursos ### Documentación - [Pydantic: JSON Schema (`model_json_schema`)](https://pydantic.dev/docs/validation/latest/concepts/json_schema/) · pydantic.dev - [OpenAI: Structured Outputs](https://developers.openai.com/api/docs/guides/structured-outputs) · developers.openai.com - [Anthropic: Structured outputs](https://platform.claude.com/docs/en/build-with-claude/structured-outputs) · platform.claude.com - [vLLM: Structured Outputs](https://docs.vllm.ai/en/latest/features/structured_outputs/) · docs.vllm.ai ### Libros - [Willard, B. T. y Louf, R. (2023): Efficient Guided Generation for Large Language Models (arXiv:2307.09702)](https://arxiv.org/abs/2307.09702) · arxiv.org - [Tam, Z. R. et al. (2024): Let Me Speak Freely? A Study on the Impact of Format Restrictions on Performance of Large Language Models (arXiv:2408.02442)](https://arxiv.org/abs/2408.02442) · arxiv.org ### Datos - [documentos6_extraccion_verdad.csv: previsiones y KPI que contiene cada documento (datos sintéticos)](https://pythonparatrading.com/curso/datos/nivel-6/documentos6_extraccion_verdad.csv) · Descarga · documentos6_extraccion_verdad.csv Lección anterior: [6.2.1 · Cómo funciona un LLM: del siguiente token a la API](https://pythonparatrading.com/curso-python-trading/nivel-6/como-funcionan-los-llm-apis-y-modelos-locales/) Lección siguiente: [6.2.3 · Guidance, KPIs y tono: qué extrae bien un LLM](https://pythonparatrading.com/curso-python-trading/nivel-6/extraccion-de-guidance-kpis-y-tono-con-llm/)