docs / Biblioteca PythonPython library
Biblioteca Python (minifmt 1.3.3)#
Implementación de referencia. Sin dependencias en tiempo de ejecución; tiktoken es opcional para contar tokens.
API pública#
from minifmt import (Contract, Field, MiniError, MiniValidationError, Document,
parse, dumps, detect_prefix, Registry, spec_block, parser_prompt,
canonical_equal, roundtrip_ok, __version__, SPEC_VERSION)
| Función | Qué hace |
|---|---|
Registry.load(dir=…) |
Descubre forks/*/contract.json; reg.get("a") devuelve el Contract. |
parse(text, contract, strict=True) |
Analiza y valida. Estricto: lanza MiniValidationError con todos los errores. Tolerante (strict=False): devuelve Document con records válidos y errors. |
Document.to_canonical() |
JSON canónico: {"prefix", "header", "<records_key>": [...]}. |
dumps(obj, contract) |
Canónico → .mini. |
spec_block(contract, lang="en"|"es") |
Bloque de especificación para el prompt del modelo. |
parser_prompt(contract, fixtures, lang) |
Prompt para que un modelo escriba un parser de la familia. |
roundtrip_ok(obj, contract) |
parse(dumps(obj)) == obj y serialización estable. |
detect_prefix(text) |
Prefijo declarado en la cabecera, sin validar. |
MiniError#
Campos: code, line (1-based, 0 = documento), message, field. str(e) produce E10 line 3 [level]: ….
Los códigos están en el índice de errores.
Modo tolerante#
doc = parse(texto, c, strict=False)
doc.records # solo los válidos, con tipos del contrato
doc.errors # lista de MiniError
doc.record_lines # línea física de cada registro aceptado
doc.diagnostics() # informe con las líneas a regenerar (mismo que `mini diagnose`)
Lectura en streaming#
from minifmt import read_records, create_reader
for item in read_records(fragmentos, c): # str o bytes UTF-8, partidos en cualquier punto
procesar(item.record, item.line) # se emite al cerrarse la línea del registro
read_records produce los mismos registros y errores que parse y no retiene los registros, por lo que la memoria
no crece con el tamaño del documento. create_reader(c) ofrece push(fragmento) y end() para integrarlo con la
respuesta en streaming de un modelo.
Contratos desde JSON Schema y Pydantic#
from minifmt import from_json_schema, to_json_schema, from_pydantic
c = from_json_schema(esquema, "tk") # registros de un nivel (SPEC §12)
esquema = to_json_schema(c) # ida y vuelta sin pérdida mediante anotaciones x-mini
c = from_pydantic(Modelo, "job") # pydantic es opcional
Las propiedades requeridas pasan al núcleo y las opcionales a las extensiones; format: date produce el tipo date.
Una estructura con más de un nivel de anidación se rechaza con E20, citando SPEC §12.
Reparación selectiva y proveedores#
from minifmt.ai import repair_request, merge_repair
from minifmt.ai.adapters import get_adapter
req = repair_request(respuesta, c, "es") # solo las líneas inválidas, con sus códigos
if req.needed:
modelo = get_adapter("openai", "gpt-5-mini") # también "anthropic", "groq" o "simulado"
reparada = modelo.generate(req.system, req.user, max_tokens=req.max_tokens_hint * 2, temperature=0)
fusion = merge_repair(respuesta, reparada["text"], c, req) # acepta cada corrección solo si es válida
Cambiar de proveedor no modifica el contrato ni el resto de la integración. Una demostración sin conexión del flujo
completo se ejecuta con python demo/sin-conexion/demo.py.
Python library (minifmt 1.3.3)#
Reference implementation. No runtime dependencies; tiktoken is optional for token counting.
Public API#
from minifmt import (Contract, Field, MiniError, MiniValidationError, Document,
parse, dumps, detect_prefix, Registry, spec_block, parser_prompt,
canonical_equal, roundtrip_ok, __version__, SPEC_VERSION)
| Function | What it does |
|---|---|
Registry.load(dir=…) |
Discovers forks/*/contract.json; reg.get("a") returns the Contract. |
parse(text, contract, strict=True) |
Parses and validates. Strict: raises MiniValidationError with every error. Lenient (strict=False): returns a Document with valid records and errors. |
Document.to_canonical() |
Canonical JSON: {"prefix", "header", "<records_key>": [...]}. |
dumps(obj, contract) |
Canonical → .mini. |
spec_block(contract, lang="en"|"es") |
Specification block for the model's prompt. |
parser_prompt(contract, fixtures, lang) |
Prompt for a model to write a parser for the family. |
roundtrip_ok(obj, contract) |
parse(dumps(obj)) == obj with stable serialization. |
detect_prefix(text) |
Prefix declared in the header, without validating. |
MiniError#
Fields: code, line (1-based, 0 = document), message, field. str(e) yields E10 line 3 [level]: ….
Codes are in the error index.
Lenient mode#
doc = parse(texto, c, strict=False)
doc.records # only the valid ones, with contract types
doc.errors # list of MiniError
doc.record_lines # physical line of each accepted record
doc.diagnostics() # report with the lines to regenerate (same as `mini diagnose`)
Streaming read#
from minifmt import read_records, create_reader
for item in read_records(chunks, c): # str or UTF-8 bytes, split anywhere
handle(item.record, item.line) # emitted when the record's line closes
read_records yields the same records and errors as parse and does not retain records, so memory does not grow
with the document. create_reader(c) offers push(chunk) and end() for a model's streaming response.
Contracts from JSON Schema and Pydantic#
from minifmt import from_json_schema, to_json_schema, from_pydantic
c = from_json_schema(schema, "tk") # one-level records (SPEC §12)
schema = to_json_schema(c) # lossless round trip through x-mini annotations
c = from_pydantic(Model, "job") # pydantic is optional
Required properties become core fields and optional ones extensions; format: date yields the date type. Deeper
nesting is rejected with E20, citing SPEC §12.
Selective repair and providers#
from minifmt.ai import repair_request, merge_repair
from minifmt.ai.adapters import get_adapter
req = repair_request(answer, c, "en") # only the invalid lines, with their codes
if req.needed:
model = get_adapter("openai", "gpt-5-mini") # also "anthropic", "groq" or "simulated"
repaired = model.generate(req.system, req.user, max_tokens=req.max_tokens_hint * 2, temperature=0)
merged = merge_repair(answer, repaired["text"], c, req) # accepts each correction only if valid
Switching providers changes neither the contract nor the rest of the integration. An offline demonstration of the
whole flow runs with python demo/sin-conexion/demo.py.