Una aplicación ficticia traduce una solicitud corta a SELECT SQLite sobre un inventario local. Se entrena el mismo modelo con los mismos ejemplos y adaptadores, primero sobre una base de 16 bits y después sobre una base cuantificada a 4 bits. Ningún SQL producido se ejecuta con permisos de escritura.
Una solicitud da una consulta propuesta, validada sintácticamente y después evaluada en solo lectura sobre varios inventarios pequeños. Las tablas separan el efecto de la cuantificación sin entrenamiento y el efecto de cada adaptación.
Descargar el proyecto completo, revisión r002
El problema y sus datos
El esquema proporcionado es inventory(sku TEXT, zone TEXT, qty INTEGER). Una solicitud indica una zona, un umbral de cantidad y un resultado: SKU, SKU/cantidad con orden explícito, número de filas, suma, mínimo o máximo. Una solicitud de precio o proveedor debe obtener REFUSE porque estos campos no existen. El modelo nunca recibe la consulta objetivo en el prompt.
StockSQL-v1 contiene 300 familias de ocho formulaciones, es decir, 2400 solicitudes ficticias. Las familias 0–209 dan 1680 ejemplos de train, 210–254 dan 360 de validation y 255–299 dan 360 de test. El tipo de SELECT, la zona y el umbral definen 300 SQL familiares distintos. Las zonas de train van de A a G; el test usa I y J. Los tipos de consulta y patrones de lengua siguen compartidos. Un identificador Ticket Txxx-j hace trazables las solicitudes, pero no crea por sí solo novedad semántica. Las solicitudes fuera de esquema repiten las mismas intenciones de precio/proveedor entre particiones; no se afirma una generalización universal fuera de esquema.
Entre los 360 casos de test, 330 esperan SQL y 30 un rechazo. Se generan cinco inventarios independientes a partir de semillas fijas: cada uno tiene 300 SKU en diez zonas, con cantidades de 0 a 20. Los contenidos se escriben en cinco verdaderas bases SQLite y se vuelven a leer para comprobar la correspondencia con las filas JSON originales. Los SHA-256 antes/después de la evaluación comprueban su integridad.
La señal supervisada
El tokenizer y la plantilla de diálogo pertenecen a Qwen2.5-1.5B-Instruct. El prompt contiene la instrucción de sistema, el esquema, la solicitud y después la frontera de una respuesta de asistente. Los ID del prompt se obtienen mediante apply_chat_template; el objetivo SQL o REFUSE se codifica por separado y se añade EOS=151645. Este contrato explícito no se presenta como equivalente a retokenizar la cadena completa.
El prompt se limita a 256 tokens, la respuesta a 128 incluido EOS y el total a 384. Un exceso provoca un error identificado, nunca un truncamiento silencioso. Los labels del prompt y del padding son −100. Los demás labels supervisan cada token de respuesta y EOS. PAD=151643 y EOS=151645 son distintos; aun así, atención y supervisión dependen de posiciones reales.
La CE alinea logits[:, :-1] y labels[:, 1:], selecciona las posiciones activas y calcula en FP32. El paso por el modelo usa autocast BF16. Los ocho ejemplos de una actualización pueden tener respuestas de longitudes distintas: sus sumas de CE se dividen por el mismo número total de tokens de respuesta. Calcular una media de ocho medias cambiaría ese peso. La prueba compara los gradientes de una acumulación con la normalización única.
LoRA: aprender una corrección de bajo rango
Para una proyección W de forma salida × entrada, la corrección es (alpha/r) B A, con A de forma r × entrada y B de forma salida × r. La capa calcula xWᵀ + (alpha/r)xAᵀBᵀ. W permanece congelada; solo se optimizan A y B. El pequeño módulo LoRALinear muestra este cálculo y después compara salida y gradientes de entrada/A/B con una capa PEFT real en FP32. B no es nulo en esta comparación, para comprobar realmente el gradiente de A.
Para el entrenamiento real, A sigue una inicialización de Kaiming y B comienza en cero. En ese punto, la corrección vale cero y el primer gradiente de A puede ser nulo; no es una congelación involuntaria. Pasos posteriores deben modificar realmente A y B. Las comprobaciones se hacen sobre los tensors reales y sus hashes.
Los adaptadores cubren q_proj, k_proj, v_proj, o_proj, gate_proj, up_proj y down_proj en los 28 bloques: 196 proyecciones y 392 matrices A/B. Con r8 y alpha 16 representan 9 232 384 parámetros FP32. Con r16 y alpha 32 representan 18 464 768 parámetros FP32. Los archivos Safetensors añaden metadatos al tamaño lógico de estos valores; se mide su tamaño real.
BF16 y NF4: qué cambia realmente
La base congelada es Qwen/Qwen2.5-1.5B-Instruct, revisión 989aa7980e4cf806f80c7fef2b1adb7bc71aa306. Tras la carga se verifican anchura 1536, 28 bloques, 12 cabezas Q, dos cabezas K/V, MLP 8960 y vocabulario 151936. Debe conservarse el compartir entre embeddings de entrada y salida.
En la rama BF16, los pesos de las 196 proyecciones permanecen almacenados en BF16. En la rama NF4, bitsandbytes los cuantifica en NF4 con doble cuantificación; se auditan las clases Linear4bit, el estado cuantificado anidado, el almacenamiento y el dtype de cálculo BF16. La carga designa explícitamente CUDA 0 en el contenedor, limitado a la GPU elegida. No usa una colocación automática en varias tarjetas.
prepare_model_for_kbit_training se llama antes de instalar PEFT. Esta preparación convierte parámetros no cuantificados a FP32. Para evitar una diferencia adicional, la rama BF16 también conserva en FP32 embeddings/salida vinculada, normas y sesgos, dejando sus 196 pesos de proyección en BF16. Sería falso decir que todo el modelo usa cuatro bits. Los metadatos de cuantificación, activaciones, gradientes de adaptador y estados AdamW también ocupan memoria.
PEFT puede convertir adaptadores según el dtype de carga. Tras su instalación, reinicializamos explícitamente A/B en FP32 con los mismos generadores CPU por nombre de módulo en ambas ramas. Sus hashes iniciales deben ser exactamente iguales a igual rango; una primera inicialización redondeada en BF16 en una sola rama confundiría la comparación. Gradient checkpointing está desactivado en ambas ramas. No se usa ningún optimizador paginado: QLoRA designa aquí adaptadores sobre una base NF4, con las demás decisiones detalladas, no una reproducción completa del protocolo del artículo.
Un presupuesto comparable y una reanudación verificable
Las dos ramas usan la misma semilla 45, el mismo orden de ejemplos, dos épocas, microbatch 1 y acumulación de ocho ejemplos: 420 actualizaciones y 3360 presentaciones. AdamW tiene lr=0,0001 y weight_decay=0,01; clipping a 1; warm-up de 42 pasos y después descenso lineal hasta el 10 % de la tasa máxima. Solo los adaptadores FP32 entran en el optimizador. La validación de respuesta sola se mide tras cada época, sin elegir un checkpoint según el test.
Un pase forward/backward sobre un ejemplo train precalienta el cálculo sin paso de optimizador. Se borran los gradientes, se restauran los RNG y se comprueba que los pesos no han cambiado. Su duración y tokens adicionales se registran aparte. Después se reinician los picos allocated/reserved; reserved puede conservar la caché del asignador creada durante este precalentamiento. El tiempo de optimización se distingue del que incluye validación y del coste de carga/guardado. Las medidas del proceso y de nvidia-smi responden a preguntas distintas.
La interrupción ocurre al final de la primera época. La reanudación recarga la misma base verificada y después los adaptadores, AdamW, sampler, scheduler, historiales y RNG Python/NumPy/PyTorch CPU/CUDA. Todos los estados finales deben ser exactamente iguales a los del recorrido continuo de dos épocas. Los parámetros congelados y metadatos cuantificados deben conservar los mismos hashes. Este contrato no promete una reanudación en medio de un microbatch.
Los estados .pt se producen localmente y se escriben mediante archivo temporal y después renombrado atómico. No deben sustituirse por objetos no fiables. Los adaptadores finales también se guardan en Safetensors, con base/configuración/hashes explícitos y recarga estricta. No se duplica ningún peso congelado en estos adaptadores ni se realiza ninguna fusión en otra base.
Generar y controlar sin dar las claves al modelo
Las dos bases sin adaptación se evalúan por separado, y después los dos r8 y los dos r16. Cada variante genera libremente las mismas 360 solicitudes de test, mediante decodificación greedy, un beam, máximo 128 tokens nuevos y EOS=151645. Una solicitud fija de train sirve de precalentamiento, se registra pero se excluye de denominadores de test y del caudal. Se conservan ID, mensajes, texto bruto, motivo de parada y duraciones. La NLL de respuesta sola se calcula separadamente con los mismos objetivos de test y EOS y no sustituye la evaluación de generaciones libres.
Una salida detenida en el límite es TRUNCATED, aunque ya parezca un SELECT válido o REFUSE. El analizador no retira etiquetas Markdown ni repara nada. Fuera de espacios externos, REFUSE debe ser exacto. Una salida SQL debe pertenecer a una gramática completa y acotada: SELECT sobre inventory, columnas conocidas o un agregado COUNT(*)/SUM(qty)/MIN(qty)/MAX(qty), como máximo cuatro predicados AND sobre zone/qty y después un orden opcional. No se admiten comentarios, subconsultas, funciones libres, JOIN, UNION, PRAGMA, escrituras ni un segundo statement.
Las consultas admitidas se confían a un proceso Python separado, sin modelo. SQLite abre bases locales en URI mode=ro e immutable=1; se aplican query_only, extensión desactivada, autorizador estricto y plazo VM. Los PRAGMA de configuración son código de confianza ejecutado antes del autorizador; las entradas del modelo no pueden proporcionarlos. El autorizador acepta solo SELECT, lecturas de columnas inventory y cuatro agregados. El proceso tiene límite de CPU de dos segundos, memoria virtual de 256 Mio, 32 archivos abiertos y tamaño de archivo escrito nulo. El padre impone tres segundos de tiempo mural; las cinco consultas están limitadas por el control VM y 1000 filas de resultado.
Un timeout o un error del proceso conserva un fallo explícito. Si el proceso se detiene antes de publicar su JSON, sus posibles operaciones intermedias no se cuentan como resultados observados; no se inventa ningún resultado. Los inventarios permanecen en solo lectura y su integridad se vuelve a comprobar.
Medir lo correcto
El informe conserva 1800 ubicaciones de snapshots por variante: cinco para cada uno de los 360 casos. Las 30 solicitudes fuera de esquema dan 150 ubicaciones explícitamente ignoradas, cualquiera que sea la salida; evalúan el rechazo, no la ejecución de una consulta sin relación. Las salidas inválidas o rechazos inesperados de los otros 330 casos no se ejecutan, pero siguen siendo fallos en sus denominadores.
La exactitud SQL normalizada compara la estructura textual canónica, sin pretender resolver toda equivalencia. Una permutación de las dos condiciones AND puede fallar este exact match conservando los mismos resultados. La exactitud de ejecución exige las filas esperadas, en el orden esperado, en los cinco inventarios. Los resultados esperados también se calculan mediante un intérprete Python independiente de SQLite. COUNT sobre una población vacía vale cero; SUM/MIN/MAX vacíos valen null, como verifican las pruebas.
Se publican por separado: SQL válidos, truncados, rechazos, rechazos correctos sobre 30, SQL exactos sobre 330, resultados correctos sobre cinco inventarios sobre 330, exactitud de tarea sobre 360, ejecuciones realmente observadas y timeouts. Una consulta constante que acierta por casualidad en una base no basta. Superar las cinco fixtures no demuestra de forma general equivalencia SQL ni garantiza resultados sobre datos de negocio.
Ejercicio y límites
El ejercicio entrena nuevos r16 con alpha 32, por tanto el mismo alpha/r=2 que r8/alpha 16. El orden de solicitudes, número de actualizaciones, respuestas supervisadas y optimizador permanecen idénticos. Duplicar el rango duplica el número de parámetros de adaptador, no el número de parámetros de la base. Esto no borra automáticamente un error de cuantificación; calidad, tamaño, memoria y duración se comparan sin exigir una ganancia. Los r8 y resultados principales están protegidos mediante SHA-256.
Estas solicitudes, esquemas e inventarios son sintéticos y pequeños. Las formulaciones y tipos de consulta se comparten; una sola semilla de aprendizaje no mide toda la variabilidad. El rechazo por gramática reduce el espacio SQL admisible: una consulta SQLite legal pero fuera de gramática falla deliberadamente. El modelo no lanza ningún SQL de cliente, red de bases de datos, herramienta externa, servidor ni acción de negocio.
Crear solicitudes e inventarios reservados
Validez SQL y semántica
Error frecuente: Paráfrasis del mismo SQL se separan al azar: agrupar las familias antes de los splits.
Invariante: Familias disjuntas, consultas oracle válidas sobre sus snapshots.
Archivo src/lora_qlora_sql/data.py
import hashlib, random, sqlite3
from pathlib import Path
from .config import save_json, sha
SCHEMA = "inventory(sku TEXT, zone TEXT, qty INTEGER)"
def split_queries(family):
return "train" if family < 210 else "validation" if family < 255 else "test"
def query_spec(family):
kind = family % 10
zone = chr(65 + family // 30)
threshold = 10 + (family // 10) % 3
op = "<" if kind in [0, 2, 4, 6, 9] else ">="
projection = [
"sku",
"sku",
"sku, qty",
"sku, qty",
"COUNT(*)",
"COUNT(*)",
"SUM(qty)",
"SUM(qty)",
"MIN(qty)",
"MAX(qty)",
][kind]
order = (
" ORDER BY sku"
if kind in [0, 1]
else (
" ORDER BY qty, sku"
if kind == 2
else " ORDER BY qty DESC, sku" if kind == 3 else ""
)
)
sql = f"SELECT {projection} FROM inventory WHERE zone = '{zone}' AND qty {op} {threshold}{order};"
return {
"kind": kind,
"zone": zone,
"threshold": threshold,
"operator": op,
"sql": sql,
}
def request_text(spec, j):
kind = spec["kind"]
zone = spec["zone"]
threshold = spec["threshold"]
condition = (
f"fewer than {threshold} items"
if spec["operator"] == "<"
else f"at least {threshold} items"
)
outputs = [
"the SKUs sorted by SKU",
"the SKUs sorted by SKU",
"SKU and quantity sorted by quantity then SKU",
"SKU and quantity sorted by quantity descending then SKU",
"the number of matching rows",
"the number of matching rows",
"the sum of their quantities",
"the sum of their quantities",
"the smallest matching quantity",
"the largest matching quantity",
]
result = outputs[kind]
templates = [
f"For zone {zone}, return {result} for rows with {condition}.",
f"Find rows in zone {zone} with {condition}; report {result}.",
f"Using zone {zone} and {condition} as filters, give {result}.",
f"I need {result} among inventory rows in zone {zone} holding {condition}.",
f"Select entries with {condition} in zone {zone}, then provide {result}.",
f"In the inventory for zone {zone}, consider rows containing {condition}. Return {result}.",
f"Report {result}; restrict the stock to zone {zone} and rows with {condition}.",
f"Filter zone {zone} to stock rows with {condition}. What are {result}?",
]
return templates[j]
def generate_query_families():
pages = []
gold = []
family_sql = []
for family in range(300):
spec = query_spec(family)
family_sql.append(spec["sql"])
for j in range(8):
outside = family % 3 == 0 and j in [6, 7]
text = (
(
f"List the selling prices for SKUs in zone {spec['zone']}."
if j == 6
else f"Give the supplier names for stock in zone {spec['zone']}."
)
if outside
else request_text(spec, j)
)
# The family tag is an original ticket reference, not a SQL label.
# It keeps repeated unsupported requests traceable without claiming
# that unsupported semantic patterns are novel across partitions.
text = f"Ticket T{family:03d}-{j}. " + text
identifier = f"S{family:03d}-{j}"
pages.append(
{
"document_id": identifier,
"family": family,
"formulation": j,
"split": split_queries(family),
"request": text,
"content_sha256": hashlib.sha256(text.encode()).hexdigest(),
}
)
gold.append(
{
"document_id": identifier,
"outside_schema": outside,
"target_response": "REFUSE" if outside else spec["sql"],
"query_spec": None if outside else spec,
}
)
if len(set(family_sql)) != 300 or len({p["request"] for p in pages}) != 2400:
raise ValueError("Duplicate family SQL or ticket")
return pages, gold
def build_inventory_snapshots():
snapshots = []
for index in range(5):
generator = random.Random(45000 + index)
rows = []
for zone in range(10):
for offset in range(30):
rows.append(
{
"sku": f"K{zone*30+offset:04d}",
"zone": chr(65 + zone),
"qty": generator.randrange(21),
}
)
snapshots.append({"snapshot_id": index, "rows": rows})
return snapshots
def prepare(config):
root = Path("data/generated")
root.mkdir(parents=True, exist_ok=True)
if (root / "manifest.json").exists():
raise ValueError("Preserve generated corpus; use a fresh extraction")
pages, gold = generate_query_families()
snapshots = build_inventory_snapshots()
save_json(root / "input/pages.json", pages)
save_json(root / "labels/gold.json", gold)
save_json(root / "input/inventories.json", snapshots)
lookup = {g["document_id"]: g for g in gold}
audit = {}
for split in ["train", "validation", "test"]:
selected = [p for p in pages if p["split"] == split]
labels = [lookup[p["document_id"]] for p in selected]
save_json(root / f"{split}/pages.json", selected)
if split != "test":
save_json(root / f"labels/{split}.json", labels)
audit[split] = {
"documents": len(selected),
"families": len({p["family"] for p in selected}),
"outside_schema": sum(g["outside_schema"] for g in labels),
}
databases = []
directory = root / "databases"
directory.mkdir(parents=True, exist_ok=True)
for snapshot in snapshots:
path = directory / f"snapshot-{snapshot['snapshot_id']}.sqlite"
if path.exists():
raise ValueError("Never overwrite an inventory database")
connection = sqlite3.connect(path)
try:
connection.execute(
"CREATE TABLE inventory(sku TEXT PRIMARY KEY, zone TEXT NOT NULL, qty INTEGER NOT NULL)"
)
connection.executemany(
"INSERT INTO inventory(sku,zone,qty) VALUES(?,?,?)",
[(r["sku"], r["zone"], r["qty"]) for r in snapshot["rows"]],
)
connection.commit()
actual = connection.execute(
"SELECT sku,zone,qty FROM inventory ORDER BY sku"
).fetchall()
finally:
connection.close()
if actual != [(r["sku"], r["zone"], r["qty"]) for r in snapshot["rows"]]:
raise ValueError("SQLite fixture differs from generated rows")
databases.append(
{
"snapshot_id": snapshot["snapshot_id"],
"file": str(path),
"sha256": sha(path),
"bytes": path.stat().st_size,
"rows": len(actual),
}
)
save_json(
"outputs/database-audit.json",
{"schema": SCHEMA, "databases": databases, "generated_only": True},
)
manifest = {
"dataset": "StockSQL-v1",
"rights": "CC0-1.0",
"seed": config["seed"],
"split_audit": audit,
"snapshots": 5,
"inventory_rows_per_snapshot": 300,
"unique_family_sql": 300,
"files": {
str(f.relative_to(root)): sha(f) for f in root.rglob("*") if f.is_file()
},
}
save_json(root / "manifest.json", manifest)
save_json("outputs/data-audit.json", manifest)
return audit
Mostrar y verificar la señal SFT
Solo respuesta y control de frontera
Error frecuente: El prompt proporciona por accidente la consulta objetivo: auditar los campos realmente formateados.
Invariante: Las ramas reciben los mismos ID y labels.
Archivo src/lora_qlora_sql/collator.py
import torch
from .data import SCHEMA
SYSTEM = "Translate the request into one read-only SELECT over the supplied SQLite schema. Return SQL only, with no explanation or code fence. If the request needs a column absent from the schema, return exactly REFUSE."
def format_sql_prompt(tokenizer, request):
messages = [
{"role": "system", "content": SYSTEM},
{"role": "user", "content": "Schema: " + SCHEMA + "\nRequest: " + request},
]
ids = tokenizer.apply_chat_template(
messages, tokenize=True, add_generation_prompt=True
)
return messages, ids
def make_sql_labels(tokenizer, request, response, config, identifier):
messages, prompt = format_sql_prompt(tokenizer, request)
target = tokenizer.encode(response, add_special_tokens=False) + [151645]
if (
not prompt
or len(prompt) > config["max_prompt_tokens"]
or not 1 < len(target) <= config["max_response_tokens_including_eos"]
or len(prompt) + len(target) > config["context"]
):
raise ValueError(identifier + ": token budget exceeded; no truncation")
return {
"document_id": identifier,
"messages": messages,
"input_ids": prompt + target,
"labels": [-100] * len(prompt) + target,
"prompt_tokens": len(prompt),
"response_tokens": len(target),
}
def collate(rows, device="cuda", pad_to=None):
if not rows:
raise ValueError("Empty supervised batch")
width = max(len(r["input_ids"]) for r in rows) if pad_to is None else pad_to
ids = torch.full((len(rows), width), 151643, dtype=torch.long, device=device)
mask = torch.zeros_like(ids, dtype=torch.bool)
labels = torch.full_like(ids, -100)
for i, row in enumerate(rows):
n = len(row["input_ids"])
if n > width:
raise ValueError("Do not truncate a response while padding")
ids[i, :n] = torch.tensor(row["input_ids"], device=device)
mask[i, :n] = True
labels[i, :n] = torch.tensor(row["labels"], device=device)
return ids, mask, labels
def response_loss(logits, labels):
active = labels[:, 1:] != -100
if not active.any():
raise ValueError("No active response token")
targets = labels[:, 1:][active]
selected = logits[:, :-1][active].float()
values = torch.nn.functional.cross_entropy(selected, targets, reduction="none")
return values.sum(), int(active.sum()), values, selected, targets
def inspect_sql_batch(config):
from .adapters import load_tokenizer, compare_peft_gradients
from .config import save_json
tokenizer = load_tokenizer()
request = "List the SKUs in zone B with fewer than 5 items, sorted by SKU."
response = "SELECT sku FROM inventory WHERE zone = 'B' AND qty < 5 ORDER BY sku;"
row = make_sql_labels(tokenizer, request, response, config, "illustrative-query")
ids, mask, labels = collate([row], "cpu")
save_json(
"outputs/mechanism.json",
{
"illustrative_query": True,
"request": request,
"target_response": response,
"row": row,
"tokens": tokenizer.convert_ids_to_tokens(row["input_ids"]),
"labels": labels[0].tolist(),
"attention_mask": mask[0].tolist(),
"eos": 151645,
"pad": 151643,
"prompt_and_response_encoded_separately": True,
"lora": compare_peft_gradients(),
},
)
Codificar LoRA y cargar las dos bases
Bajo rango y cuantificación
Error frecuente: La carga k-bit también congela adaptadores añadidos demasiado pronto: comprobar el orden de las operaciones y requires_grad.
Invariante: Control de capa superado y mediciones de las dos bases sin entrenamiento registradas.
Archivo src/lora_qlora_sql/adapters.py
import base64, hashlib, json, math, os
from pathlib import Path
import numpy as np
import torch
from torch import nn
import bitsandbytes as bnb
from peft import (
LoraConfig,
get_peft_model,
prepare_model_for_kbit_training,
get_peft_model_state_dict,
set_peft_model_state_dict,
)
from peft.tuners.lora.layer import Linear as PeftLinear
from transformers import Qwen2ForCausalLM, AutoTokenizer, BitsAndBytesConfig
from safetensors.torch import save_file, load_file
from .config import save_json, read_json, sha, seed_all
TARGETS = ["q_proj", "k_proj", "v_proj", "o_proj", "gate_proj", "up_proj", "down_proj"]
class LoRALinear(nn.Module):
def __init__(self, weight, rank, alpha):
super().__init__()
if rank < 1 or alpha <= 0:
raise ValueError("Rank and scale must be positive")
self.weight = nn.Parameter(weight.clone(), requires_grad=False)
self.A = nn.Parameter(torch.empty(rank, weight.shape[1]))
self.B = nn.Parameter(torch.zeros(weight.shape[0], rank))
self.scale = alpha / rank
nn.init.kaiming_uniform_(self.A, a=math.sqrt(5))
def forward(self, x):
return torch.nn.functional.linear(
x, self.weight
) + self.scale * torch.nn.functional.linear(
torch.nn.functional.linear(x, self.A), self.B
)
def compare_peft_gradients():
torch.manual_seed(45)
weight = torch.randn(3, 5)
coded = LoRALinear(weight, 2, 4)
coded.B.data.normal_(0, 0.2)
base = nn.Linear(5, 3, bias=False)
base.weight.data.copy_(weight)
base.requires_grad_(False)
reference = PeftLinear(base, "default", r=2, lora_alpha=4, lora_dropout=0.0)
reference.lora_A["default"].weight.data.copy_(coded.A)
reference.lora_B["default"].weight.data.copy_(coded.B)
x = torch.randn(4, 5, requires_grad=True)
y = x.detach().clone().requires_grad_(True)
probe = torch.randn(4, 3)
a = coded(x)
b = reference(y)
(a * probe).sum().backward()
(b * probe).sum().backward()
pairs = {
"forward": (a, b),
"input_gradient": (x.grad, y.grad),
"A_gradient": (coded.A.grad, reference.lora_A["default"].weight.grad),
"B_gradient": (coded.B.grad, reference.lora_B["default"].weight.grad),
}
errors = {}
for name, (first, second) in pairs.items():
errors[name] = float((first - second).abs().max())
if not torch.allclose(first, second, atol=1e-6, rtol=1e-5):
raise ValueError("Coded LoRA differs from PEFT: " + name)
if coded.weight.grad is not None or reference.base_layer.weight.grad is not None:
raise ValueError("Frozen teaching base got gradients")
return {
"illustrative_fp32_layer": True,
"weight": weight.tolist(),
"A": coded.A.detach().tolist(),
"B": coded.B.detach().tolist(),
"scale": 2.0,
"input": x.detach().tolist(),
"upstream_gradient": probe.tolist(),
"output": a.detach().tolist(),
"input_gradient": x.grad.tolist(),
"A_gradient": coded.A.grad.tolist(),
"B_gradient": coded.B.grad.tolist(),
"maximum_errors": errors,
"atol": 1e-6,
"rtol": 1e-5,
"base_frozen": True,
}
def tensor_bytes(tensor):
return (
tensor.detach()
.cpu()
.contiguous()
.reshape(-1)
.view(torch.uint8)
.numpy()
.tobytes()
)
def fingerprint(state):
rows = []
for name, tensor in sorted(state.items()):
raw = tensor_bytes(tensor)
rows.append(
{
"name": name,
"shape": list(tensor.shape),
"dtype": str(tensor.dtype),
"bytes": len(raw),
"sha256": hashlib.sha256(raw).hexdigest(),
}
)
digest = hashlib.sha256(
json.dumps(rows, sort_keys=True, separators=(",", ":")).encode()
).hexdigest()
return {"sha256": digest, "tensors": rows}
def frozen_fingerprint(model):
return fingerprint(
{
k: v
for k, v in model.state_dict().items()
if "lora_A" not in k and "lora_B" not in k
}
)
def adapter_state(model):
return get_peft_model_state_dict(
model, adapter_name="default", save_embedding_layers=False
)
def adapter_fingerprint(model):
return fingerprint(adapter_state(model))
def load_tokenizer():
tokenizer = AutoTokenizer.from_pretrained(
"assets/model", local_files_only=True, trust_remote_code=False, use_fast=True
)
if not tokenizer.is_fast or tokenizer.eos_token_id != 151645:
raise ValueError("Unexpected instruct tokenizer")
tokenizer.pad_token_id = 151643
tokenizer.padding_side = "left"
return tokenizer
def target_modules(model):
result = {
name: module
for name, module in model.named_modules()
if name.split(".")[-1] in TARGETS
}
expected = {}
for i in range(28):
for name in TARGETS:
scope = (
"self_attn"
if name in ["q_proj", "k_proj", "v_proj", "o_proj"]
else "mlp"
)
shape = (
(1536, 1536)
if name in ["q_proj", "o_proj"]
else (
(256, 1536)
if name in ["k_proj", "v_proj"]
else (
(8960, 1536)
if name in ["gate_proj", "up_proj"]
else (1536, 8960)
)
)
)
expected[f"model.layers.{i}.{scope}.{name}"] = shape
if set(result) != set(expected):
raise ValueError("Unexpected target module names")
for name, module in result.items():
if (module.out_features, module.in_features) != expected[name]:
raise ValueError("Target dimensions changed")
return result
def load_precision_branch(config, precision):
if precision not in ["bf16", "nf4"] or not torch.cuda.is_bf16_supported():
raise ValueError("Expected BF16-capable CUDA device and explicit precision")
seed_all(config["seed"])
kwargs = {
"local_files_only": True,
"trust_remote_code": False,
"torch_dtype": torch.bfloat16,
"attn_implementation": "eager",
"output_loading_info": True,
}
if precision == "nf4":
kwargs.update(
{
"device_map": {"": 0},
"quantization_config": BitsAndBytesConfig(
load_in_4bit=True,
bnb_4bit_quant_type="nf4",
bnb_4bit_use_double_quant=True,
bnb_4bit_compute_dtype=torch.bfloat16,
bnb_4bit_quant_storage=torch.uint8,
),
}
)
model, info = Qwen2ForCausalLM.from_pretrained("assets/model", **kwargs)
if (
info.get("unexpected_keys")
or info.get("mismatched_keys")
or info.get("error_msgs")
or set(info.get("missing_keys", [])) - {"lm_head.weight"}
):
raise ValueError("Incomplete original checkpoint loading: " + str(info))
c = model.config
if (
c.hidden_size,
c.num_hidden_layers,
c.num_attention_heads,
c.num_key_value_heads,
c.intermediate_size,
c.vocab_size,
) != (1536, 28, 12, 2, 8960, 151936):
raise ValueError("Unexpected model architecture")
modules = target_modules(model)
if precision == "nf4":
model = prepare_model_for_kbit_training(model, use_gradient_checkpointing=False)
for name, module in modules.items():
if (
not isinstance(module, bnb.nn.Linear4bit)
or module.weight.quant_state is None
or module.weight.quant_type != "nf4"
or not module.weight.quant_state.nested
or module.compute_dtype != torch.bfloat16
):
raise ValueError("NF4/double quantization not active: " + name)
else:
# Match the non-quantized FP32 policy of prepare_model_for_kbit_training
# without promoting the BF16 transformer linear weights themselves.
linear_weights = {id(module.weight) for module in modules.values()}
for parameter in model.parameters():
parameter.requires_grad_(False)
if id(parameter) not in linear_weights:
parameter.data = parameter.data.float()
model = model.cuda()
if (
any(p.requires_grad for p in model.parameters())
or model.is_gradient_checkpointing
):
raise ValueError("Frozen base/checkpointing policy changed")
if model.lm_head.weight is not model.model.embed_tokens.weight:
raise ValueError("Tied embedding/output lost")
model.config.use_cache = False
model.eval()
audit = {
"precision": precision,
"target_modules": {
n: {
"in_features": m.in_features,
"out_features": m.out_features,
"class": type(m).__name__,
"weight_dtype": str(m.weight.dtype),
"weight_shape": list(m.weight.shape),
"bias_dtype": str(m.bias.dtype) if m.bias is not None else None,
"nested_quantization": (
bool(m.weight.quant_state.nested) if precision == "nf4" else False
),
}
for n, m in modules.items()
},
"embedding_dtype": str(model.model.embed_tokens.weight.dtype),
"head_dtype": str(model.lm_head.weight.dtype),
"compute_dtype": "torch.bfloat16",
"all_base_frozen": True,
"gradient_checkpointing": False,
"model_file_sha256": sha("assets/model/model.safetensors"),
"tokenizer_sha256": sha("assets/model/tokenizer.json"),
"unique_parameter_storage_bytes_excluding_external_quant_state": sum(
p.numel() * p.element_size() for p in model.parameters()
),
"library_memory_footprint_bytes": model.get_memory_footprint(),
"loading_info": info,
}
return model, audit
def install_adapters(model, config, rank):
alpha = 2 * rank
peft_config = LoraConfig(
r=rank,
lora_alpha=alpha,
lora_dropout=0.0,
target_modules=TARGETS,
bias="none",
task_type="CAUSAL_LM",
init_lora_weights=True,
use_rslora=False,
use_dora=False,
)
model = get_peft_model(model, peft_config, autocast_adapter_dtype=True)
modules = [
(name, module)
for name, module in model.named_modules()
if hasattr(module, "lora_A") and "default" in module.lora_A
]
if len(modules) != 196:
raise ValueError("Expected196adapted projections")
for index, (name, module) in enumerate(sorted(modules)):
A = module.lora_A["default"].weight
B = module.lora_B["default"].weight
if A.dtype != torch.float32 or B.dtype != torch.float32:
raise ValueError("Adapter masters must be FP32")
generator = torch.Generator(device="cpu").manual_seed(
config["seed"] * 1000 + index
)
initial = torch.empty(A.shape, dtype=torch.float32, device="cpu")
nn.init.kaiming_uniform_(initial, a=math.sqrt(5), generator=generator)
with torch.no_grad():
A.copy_(initial)
B.zero_()
trainable = {
name: parameter
for name, parameter in model.named_parameters()
if parameter.requires_grad
}
if (
any("lora_A" not in name and "lora_B" not in name for name in trainable)
or len(trainable) != 392
or sum(p.numel() for p in trainable.values()) != 1154048 * rank
):
raise ValueError("Unexpected trainable adapter state")
return model
def restore_adapter(model, state):
expected = adapter_state(model)
if expected.keys() != state.keys() or any(
expected[k].shape != state[k].shape or state[k].dtype != torch.float32
for k in state
):
raise ValueError("Adapter keys/shapes/dtypes changed")
result = set_peft_model_state_dict(model, state, adapter_name="default")
if result.unexpected_keys or any(
"lora_A" in k or "lora_B" in k for k in result.missing_keys
):
raise ValueError("Incomplete adapter load")
if fingerprint(adapter_state(model)) != fingerprint(state):
raise ValueError("Adapter reload mismatch")
def save_adapter(model, path, identity):
path = Path(path)
if path.exists():
raise ValueError("Preserve existing adapter")
state = adapter_state(model)
temporary = path.with_suffix(".pending")
save_file(
{k: v.detach().cpu().contiguous() for k, v in state.items()}, str(temporary)
)
os.replace(temporary, path)
save_json(
path.with_suffix(".json"),
{
"identity": identity,
"sha256": sha(path),
"bytes": path.stat().st_size,
"adapter": fingerprint(state),
},
)
def export_adapter_json(model, path):
state = adapter_state(model)
entries = {}
for name, tensor in state.items():
raw = (
tensor.detach()
.cpu()
.contiguous()
.numpy()
.astype("<f4", copy=False)
.tobytes()
)
entries[name] = {
"shape": list(tensor.shape),
"dtype": "float32-little-endian",
"bytes": len(raw),
"sha256": hashlib.sha256(raw).hexdigest(),
"base64": base64.b64encode(raw).decode(),
}
save_json(
path,
{
"weights": entries,
"adapter": fingerprint(state),
"parameters": sum(t.numel() for t in state.values()),
},
)
Entrenar las dos adaptaciones comparables
Mismos datos, presupuestos y optimizador
Error frecuente: Se carga una base antigua de la caché durante la reanudación: rechazarla si el SHA no coincide.
Invariante: Las dos ramas han consumido el presupuesto previsto y los pesos congelados no han cambiado.
Archivo src/lora_qlora_sql/train.py
import math, time
from pathlib import Path
import numpy as np
import os
import torch
from .config import read_json, save_json, sha, seed_all, rng_state, restore_rng
from .adapters import (
load_tokenizer,
load_precision_branch,
install_adapters,
adapter_state,
adapter_fingerprint,
frozen_fingerprint,
restore_adapter,
save_adapter,
export_adapter_json,
)
from .collator import make_sql_labels, collate
def equal(a, b):
if isinstance(a, torch.Tensor):
return isinstance(b, torch.Tensor) and torch.equal(a, b)
if isinstance(a, np.ndarray):
return isinstance(b, np.ndarray) and np.array_equal(a, b)
if isinstance(a, dict):
return (
isinstance(b, dict)
and a.keys() == b.keys()
and all(equal(a[k], b[k]) for k in a)
)
if isinstance(a, (list, tuple)):
return (
type(a) == type(b)
and len(a) == len(b)
and all(equal(x, y) for x, y in zip(a, b))
)
return a == b
def save_resume_state(path, state):
path = Path(path)
path.parent.mkdir(parents=True, exist_ok=True)
if path.exists():
raise ValueError("Existing training state is preserved")
temporary = path.with_suffix(".pending")
torch.save(state, temporary)
os.replace(temporary, path)
def encode_split(config, tokenizer, split):
pages = read_json(f"data/generated/{split}/pages.json")
labels = {
r["document_id"]: r for r in read_json(f"data/generated/labels/{split}.json")
}
if any(p["split"] != split for p in pages):
raise ValueError("Incorrect learning partition")
return [
make_sql_labels(
tokenizer,
p["request"],
labels[p["document_id"]]["target_response"],
config,
p["document_id"],
)
for p in pages
]
def active_response_logits(model, ids, mask, labels):
active = labels[:, 1:] != -100
if not active.any():
raise ValueError("No supervised response token")
with torch.autocast("cuda", dtype=torch.bfloat16):
full = model(
input_ids=ids, attention_mask=mask, use_cache=False, return_dict=True
).logits
logits = full[:, :-1][active]
targets = labels[:, 1:][active]
values = torch.nn.functional.cross_entropy(
logits.float(), targets, reduction="none"
)
return values.sum(), len(targets), values, logits, targets
def learning_rate(config, step, total):
warm = config["warmup_updates"]
peak = config["learning_rate"]
if step < warm:
return peak * (step + 1) / warm
return peak * (
1
- min(1.0, (step - warm) / max(1, total - warm - 1))
* (1 - config["minimum_lr_ratio"])
)
@torch.no_grad()
def validate_response(model, rows):
model.eval()
total = 0.0
count = 0
for row in rows:
_, n, values, _, _ = active_response_logits(model, *collate([row]))
total += float(values.double().sum())
count += n
model.train()
return {
"documents": len(rows),
"response_tokens": count,
"response_nll_nats_per_token": total / count,
}
def warm_up_training(model, row):
before = adapter_fingerprint(model)
rng = rng_state()
model.train()
torch.cuda.synchronize()
begun = time.perf_counter()
loss, n, _, _, _ = active_response_logits(model, *collate([row]))
(loss / n).backward()
torch.cuda.synchronize()
seconds = time.perf_counter() - begun
model.zero_grad(set_to_none=True)
restore_rng(rng)
if adapter_fingerprint(model) != before:
raise ValueError("Warm-up modified adapter weights")
return {
"document_id": row["document_id"],
"response_tokens": n,
"seconds": seconds,
"optimizer_steps": 0,
"rng_restored": True,
"excluded_from_optimization_budget": True,
}
def train_adapters(config, precision="bf16", rank=8, mode="continuous"):
if (
Path("data/generated/test").exists()
or Path("data/generated/labels/gold.json").exists()
):
raise ValueError("Learning service must not expose test inputs or gold")
if mode not in ["interrupt", "continuous", "resume"] or rank not in [
config["rank"],
config["exercise_rank"],
]:
raise ValueError("Unknown mode/rank")
seed_all(config["seed"])
tokenizer = load_tokenizer()
rows = encode_split(config, tokenizer, "train")
validation = encode_split(config, tokenizer, "validation")
if len(rows) != 1680 or len(validation) != 360:
raise ValueError("Expected1680/360 frozen population")
model, base_audit = load_precision_branch(config, precision)
model = install_adapters(model, config, rank)
initial = adapter_fingerprint(model)
frozen = frozen_fingerprint(model)
parameters = [p for p in model.parameters() if p.requires_grad]
optimizer = torch.optim.AdamW(
parameters, lr=config["learning_rate"], weight_decay=config["weight_decay"]
)
sampler = torch.Generator().manual_seed(config["seed"] + 100)
total = math.ceil(len(rows) / config["accumulation_examples"]) * config["epochs"]
branch = f"{precision}-r{rank}"
root = Path("assets/checkpoints")
destination = root / f"{branch}-{mode}.pt"
if destination.exists():
raise ValueError("Existing checkpoint is preserved")
identity = {
"branch": branch,
"precision": precision,
"rank": rank,
"alpha": rank * 2,
"config": config,
"initial_adapter": initial,
"frozen_base": frozen,
"model_sha256": sha("assets/model/model.safetensors"),
"tokenizer_sha256": sha("assets/model/tokenizer.json"),
"lock_sha256": sha("locks/models.lock.json"),
"implementation_sha256": {
name: sha("src/lora_qlora_sql/" + name)
for name in ["train.py", "adapters.py", "collator.py"]
},
"data_sha256": {
name: sha("data/generated/" + name)
for name in [
"train/pages.json",
"validation/pages.json",
"labels/train.json",
"labels/validation.json",
]
},
}
start = updates = 0
history = []
validation_history = []
if mode == "resume":
state = torch.load(
root / f"{branch}-interrupt.pt", map_location="cpu", weights_only=False
)
if state["identity"] != identity:
raise ValueError("Resume identity changed")
restore_adapter(model, state["adapter"])
optimizer.load_state_dict(state["optimizer"])
sampler.set_state(state["sampler"])
restore_rng(state["rng"])
start = state["epoch"]
updates = state["updates"]
history = state["history"]
validation_history = state["validation_history"]
if state["scheduler"] != {
"next_update": updates,
"next_lr": learning_rate(config, updates, total),
}:
raise ValueError("Resume scheduler changed")
del state
warmup = warm_up_training(model, rows[0])
torch.cuda.reset_peak_memory_stats()
torch.cuda.synchronize()
begun = time.perf_counter()
optimization_seconds = 0.0
stop = config["interrupt_epoch"] if mode == "interrupt" else config["epochs"]
model.train()
for epoch in range(start, stop):
permutation = torch.randperm(len(rows), generator=sampler).tolist()
torch.cuda.synchronize()
epoch_start = time.perf_counter()
for offset in range(0, len(rows), config["accumulation_examples"]):
selected = permutation[offset : offset + config["accumulation_examples"]]
denominator = sum(rows[i]["response_tokens"] for i in selected)
lr = learning_rate(config, updates, total)
for group in optimizer.param_groups:
group["lr"] = lr
optimizer.zero_grad(set_to_none=True)
summed = 0.0
for i in selected:
loss, n, values, _, _ = active_response_logits(
model, *collate([rows[i]])
)
if n != rows[i]["response_tokens"] or not torch.isfinite(loss):
raise ValueError("Incorrect/nonfinite response loss")
(loss / denominator).backward()
summed += float(values.detach().double().sum())
if any(
p.dtype != torch.float32
or p.grad is None
or p.grad.dtype != torch.float32
for p in parameters
):
raise ValueError("Expected FP32 adapter masters and gradients")
norm = torch.nn.utils.clip_grad_norm_(
parameters, config["gradient_clip"], error_if_nonfinite=True
)
optimizer.step()
updates += 1
history.append(
{
"update": updates,
"epoch": epoch + 1,
"document_ids": [rows[i]["document_id"] for i in selected],
"response_tokens": denominator,
"nll_nats_per_response_token": summed / denominator,
"learning_rate": lr,
"gradient_norm_before_clip": float(norm),
}
)
if updates % 42 == 0:
print(history[-1], flush=True)
torch.cuda.synchronize()
optimization_seconds += time.perf_counter() - epoch_start
observed = validate_response(model, validation)
validation_history.append({"epoch": epoch + 1, **observed})
print(validation_history[-1], flush=True)
torch.cuda.synchronize()
elapsed = time.perf_counter() - begun
peak_allocated = torch.cuda.max_memory_allocated()
peak_reserved = torch.cuda.max_memory_reserved()
final = adapter_fingerprint(model)
frozen_after = frozen_fingerprint(model)
initial_map = {r["name"]: r for r in initial["tensors"]}
changed = [
r["name"]
for r in final["tensors"]
if r["sha256"] != initial_map[r["name"]]["sha256"]
]
if (
frozen_after != frozen
or not any("lora_A" in n for n in changed)
or not any("lora_B" in n for n in changed)
):
raise ValueError("Frozen base changed or A/B learning absent")
state = {
"identity": identity,
"epoch": stop,
"updates": updates,
"adapter": adapter_state(model),
"optimizer": optimizer.state_dict(),
"sampler": sampler.get_state(),
"scheduler": {
"next_update": updates,
"next_lr": learning_rate(config, updates, total),
},
"rng": rng_state(),
"history": history,
"validation_history": validation_history,
}
save_resume_state(destination, state)
if mode == "resume":
save_adapter(model, root / f"{branch}-resume.safetensors", identity)
export_adapter_json(model, f"outputs/adapter-{branch}.json")
save_json(
f"outputs/train-{branch}-{mode}.json",
{
"branch": branch,
"mode": mode,
"identity": identity,
"base_audit": base_audit,
"epochs": stop,
"updates": updates,
"history": history,
"validation_history": validation_history,
"tokenization": rows,
"initial_adapter": initial,
"final_adapter": final,
"changed_parameters": changed,
"frozen_base_unchanged": True,
"trainable_parameters": sum(p.numel() for p in parameters),
"checkpoint_sha256": sha(destination),
"checkpoint_bytes": destination.stat().st_size,
"warmup": warmup,
"optimization_seconds_this_process": optimization_seconds,
"training_and_validation_seconds_this_process": elapsed,
"max_torch_allocated_bytes": peak_allocated,
"max_torch_reserved_bytes": peak_reserved,
"reserved_includes_warmup_allocator_cache": True,
"test_visible": False,
},
)
def verify_resume(precision, rank):
branch = f"{precision}-r{rank}"
root = Path("assets/checkpoints")
a = torch.load(
root / f"{branch}-continuous.pt", map_location="cpu", weights_only=False
)
b = torch.load(root / f"{branch}-resume.pt", map_location="cpu", weights_only=False)
checks = {k: equal(a[k], b[k]) for k in a}
if a.keys() != b.keys() or not all(checks.values()):
raise ValueError("Exact adapter resume differs: " + str(checks))
save_json(
f"outputs/resume-{branch}.json",
{
"status": "PASS",
"all_state_exact": True,
"checks": checks,
"boundary": "complete epoch; adapter/optimizer/sampler/scheduler/all RNG; frozen base reloaded by verified identity",
},
)
Ejecutar los SELECT en un control independiente
Interpretación segura y memoria medida
Error frecuente: Una cadena con dos statements supera un simple startswith SELECT: usar un verdadero analizador/autorizador.
Invariante: Cada caso tiene cinco resultados o un rechazo/fallo explícito; ninguna escritura posible.
Archivo src/lora_qlora_sql/evaluate.py
import json, math, subprocess, sys, tempfile, time
from pathlib import Path
import torch
from safetensors.torch import load_file
from transformers import GenerationConfig
from .config import read_json, save_json, sha, seed_all
from .adapters import (
load_tokenizer,
load_precision_branch,
install_adapters,
restore_adapter,
adapter_fingerprint,
frozen_fingerprint,
)
from .collator import format_sql_prompt, make_sql_labels, collate
from .train import active_response_logits
from .sql_sandbox import validate_readonly_sql, normalize_sql
def parse_completion(text, finish):
if finish != "eos":
return {"status": "TRUNCATED", "sql": None, "tree": None}
text = text.strip()
if text == "REFUSE":
return {"status": "REFUSE", "sql": None, "tree": None}
try:
tree = validate_readonly_sql(text)
except ValueError as error:
return {"status": "INVALID_SQL", "sql": None, "tree": None, "error": str(error)}
return {"status": "VALID_SQL", "sql": normalize_sql(text), "tree": tree}
def load_variant(config, precision, rank):
model, audit = load_precision_branch(config, precision)
identity = {
"precision": precision,
"rank": rank,
"model_sha256": sha("assets/model/model.safetensors"),
}
if rank:
model = install_adapters(model, config, rank)
path = Path(f"assets/checkpoints/{precision}-r{rank}-resume.safetensors")
metadata = read_json(path.with_suffix(".json"))
if (
sha(path) != metadata["sha256"]
or metadata["identity"]["config"] != config
or metadata["identity"]["precision"] != precision
or metadata["identity"]["rank"] != rank
or metadata["identity"]["frozen_base"] != frozen_fingerprint(model)
):
raise ValueError("Learned adapter/base identity changed")
restore_adapter(model, load_file(str(path), device="cpu"))
if adapter_fingerprint(model) != metadata["adapter"]:
raise ValueError("Adapter manifest differs")
identity.update(
{
"adapter_sha256": sha(path),
"adapter_bytes": path.stat().st_size,
"adapter": metadata["adapter"],
}
)
return model.eval(), audit, identity
@torch.no_grad()
def generate_one(config, model, tokenizer, page, generation):
messages, ids = format_sql_prompt(tokenizer, page["request"])
if len(ids) > config["max_prompt_tokens"]:
raise ValueError("Generation prompt over token cap; do not truncate")
tensor = torch.tensor([ids], device="cuda")
torch.cuda.synchronize()
begun = time.perf_counter()
with torch.autocast("cuda", dtype=torch.bfloat16):
output = (
model.generate(
input_ids=tensor,
attention_mask=torch.ones_like(tensor),
generation_config=generation,
)[0, len(ids) :]
.cpu()
.tolist()
)
torch.cuda.synchronize()
seconds = time.perf_counter() - begun
finish = "eos" if output and output[-1] == 151645 else "length"
content = output[:-1] if finish == "eos" else output
text = tokenizer.decode(
content, skip_special_tokens=False, clean_up_tokenization_spaces=False
)
return {
"document_id": page["document_id"],
"messages": messages,
"input_ids": ids,
"output_ids": output,
"text": text,
"finish_reason": finish,
"parsed": parse_completion(text, finish),
"input_tokens": len(ids),
"output_tokens": len(output),
"seconds": seconds,
}
@torch.no_grad()
def predict(config, precision, rank=0):
if Path("data/generated/labels").exists():
raise ValueError("Blind generation must not mount labels")
seed_all(config["seed"])
tokenizer = load_tokenizer()
model, audit, identity = load_variant(config, precision, rank)
pages = read_json("data/generated/input/pages.json")
test = [p for p in pages if p["split"] == "test"]
warmup_page = next(p for p in pages if p["split"] == "train")
variant = f"{precision}-r{rank}"
if len(test) != 360:
raise ValueError("Keep all360heldout cases")
generation = GenerationConfig(
do_sample=False,
num_beams=1,
max_new_tokens=config["max_new_tokens"],
use_cache=True,
eos_token_id=151645,
pad_token_id=151643,
bos_token_id=151643,
)
warmup = generate_one(config, model, tokenizer, warmup_page, generation)
torch.cuda.reset_peak_memory_stats()
torch.cuda.synchronize()
begun = time.perf_counter()
rows = []
for index, page in enumerate(test):
rows.append(generate_one(config, model, tokenizer, page, generation))
if (index + 1) % 60 == 0:
print(
{"variant": variant, "generated": index + 1, "total": 360}, flush=True
)
torch.cuda.synchronize()
seconds = time.perf_counter() - begun
tokens = sum(r["output_tokens"] for r in rows)
save_json(
f"outputs/predictions-{variant}.json",
{
"variant": variant,
"identity": identity,
"base_audit": audit,
"tokenizer_sha256": sha("assets/model/tokenizer.json"),
"input_sha256": sha("data/generated/input/pages.json"),
"generation": generation.to_dict(),
"warmup": warmup,
"warmup_excluded": True,
"rows": rows,
"seconds": seconds,
"generated_tokens_including_eos": tokens,
"generated_tokens_per_second": tokens / seconds,
"max_torch_allocated_bytes": torch.cuda.max_memory_allocated(),
"max_torch_reserved_bytes": torch.cuda.max_memory_reserved(),
"reserved_includes_warmup_allocator_cache": True,
"gold_visible": False,
},
)
@torch.no_grad()
def response_nll(config, precision, rank=0):
seed_all(config["seed"])
tokenizer = load_tokenizer()
model, _, identity = load_variant(config, precision, rank)
variant = f"{precision}-r{rank}"
pages = [
p for p in read_json("data/generated/input/pages.json") if p["split"] == "test"
]
gold = {g["document_id"]: g for g in read_json("data/generated/labels/gold.json")}
rows = []
probes = []
for index, page in enumerate(pages):
row = make_sql_labels(
tokenizer,
page["request"],
gold[page["document_id"]]["target_response"],
config,
page["document_id"],
)
_, n, losses, logits, targets = active_response_logits(model, *collate([row]))
rows.append(
{
"document_id": page["document_id"],
"input_ids": row["input_ids"],
"prompt_tokens": row["prompt_tokens"],
"response_tokens": n,
"response_targets": targets.cpu().tolist(),
"response_nll": losses.cpu().tolist(),
}
)
if index == 0:
for position in [0, n - 1]:
probes.append(
{
"document_id": page["document_id"],
"response_position": position,
"target": int(targets[position]),
"logits": logits[position].float().cpu().tolist(),
"nll": float(losses[position]),
}
)
count = sum(r["response_tokens"] for r in rows)
total = math.fsum(v for row in rows for v in row["response_nll"])
save_json(
f"outputs/response-nll-{variant}.json",
{
"variant": variant,
"identity": identity,
"documents": len(rows),
"response_tokens": count,
"response_nll_nats_per_token": total / count,
"rows": rows,
"probes": probes,
},
)
def reference_result(tree, inventory):
def matches(row):
for p in tree["predicates"]:
a = row[p["column"]]
b = p["value"]
op = p["operator"]
ok = {
"=": a == b,
"<>": a != b,
"<": a < b,
">": a > b,
"<=": a <= b,
">=": a >= b,
}[op]
if not ok:
return False
return True
selected = [r for r in inventory if matches(r)]
aggregate = tree["aggregate"]
if aggregate:
values = [r["qty"] for r in selected]
fn = aggregate["function"]
value = (
len(values)
if fn == "COUNT"
else (
(
sum(values)
if fn == "SUM"
else min(values) if fn == "MIN" else max(values)
)
if values
else None
)
)
return [[value]]
for order in reversed(tree["order"]):
selected.sort(
key=lambda r: r[order["column"]], reverse=order["direction"] == "DESC"
)
return [[r[c] for c in tree["columns"]] for r in selected]
def run_sql_worker(config, sql):
validate_readonly_sql(sql)
with tempfile.TemporaryDirectory(prefix="sql-request-") as directory:
path = Path(directory) / "request.json"
path.write_text(
json.dumps(
{
"sql": sql,
"vm_seconds": config["sqlite_vm_seconds"],
"max_rows": config["sqlite_max_rows"],
}
)
)
begun = time.perf_counter()
try:
result = subprocess.run(
[sys.executable, "scripts/sql_worker.py", "--request", str(path)],
capture_output=True,
text=True,
timeout=config["sqlite_worker_timeout_seconds"],
check=False,
)
except subprocess.TimeoutExpired:
return {
"status": "PROCESS_TIMEOUT",
"results": [],
"seconds": time.perf_counter() - begun,
}
if result.returncode:
return {
"status": "WORKER_ERROR",
"results": [],
"returncode": result.returncode,
"stderr": result.stderr[:2000],
"seconds": time.perf_counter() - begun,
}
try:
payload = json.loads(result.stdout)
except ValueError:
return {
"status": "WORKER_ERROR",
"results": [],
"error": "Worker output was not JSON",
"seconds": time.perf_counter() - begun,
}
if (
not isinstance(payload, dict)
or set(payload) != {"results"}
or not isinstance(payload["results"], list)
or len(payload["results"]) != 5
or [r.get("snapshot_id") for r in payload["results"]] != list(range(5))
):
raise ValueError("Malformed SQL worker result")
return {
"status": "COMPLETED",
"results": payload["results"],
"seconds": time.perf_counter() - begun,
}
def evaluate_variants(config, variants, exercise=False):
pages = {
p["document_id"]: p
for p in read_json("data/generated/input/pages.json")
if p["split"] == "test"
}
gold = {g["document_id"]: g for g in read_json("data/generated/labels/gold.json")}
snapshots = read_json("data/generated/input/inventories.json")
databases = read_json("outputs/database-audit.json")["databases"]
before = {d["file"]: sha(d["file"]) for d in databases}
result = {}
if before != {d["file"]: d["sha256"] for d in databases}:
raise ValueError("Original inventory changed before evaluation")
for variant in variants:
predictions = read_json(f"outputs/predictions-{variant}.json")
records = predictions["rows"]
if len(records) != 360 or {r["document_id"] for r in records} != set(pages):
raise ValueError("Every test case must be retained exactly once")
counts = {
"documents": 360,
"eligible_sql": 330,
"outside_schema": 30,
"valid_sql": 0,
"truncated": 0,
"refusal": 0,
"correct_refusal": 0,
"normalized_sql_exact": 0,
"execution_correct_all_five": 0,
"task_correct": 0,
"snapshot_slots": 1800,
"outside_schema_skipped_slots": 150,
"actual_snapshot_executions": 0,
"successful_snapshot_executions": 0,
"worker_timeouts": 0,
}
details = []
for row in records:
label = gold[row["document_id"]]
parsed = parse_completion(row["text"], row["finish_reason"])
if parsed != row["parsed"]:
raise ValueError("Recorded SQL parse changed")
status = parsed["status"]
counts["valid_sql"] += status == "VALID_SQL"
counts["truncated"] += status == "TRUNCATED"
counts["refusal"] += status == "REFUSE"
outside = label["outside_schema"]
correct_refusal = outside and status == "REFUSE"
counts["correct_refusal"] += correct_refusal
exact = False
execution = False
worker = {
"status": (
"OUTSIDE_SCHEMA_SKIP" if outside else "INVALID_OR_REFUSED_SKIP"
),
"results": [],
}
comparisons = []
if not outside and status == "VALID_SQL":
exact = parsed["sql"] == normalize_sql(label["target_response"])
worker = run_sql_worker(config, parsed["sql"])
counts["worker_timeouts"] += worker["status"] == "PROCESS_TIMEOUT"
counts["actual_snapshot_executions"] += len(worker["results"])
expected_tree = validate_readonly_sql(label["target_response"])
for observed in worker["results"]:
expected = reference_result(
expected_tree, snapshots[observed["snapshot_id"]]["rows"]
)
correct = (
observed["status"] == "OK" and observed["rows"] == expected
)
comparisons.append(
{
"snapshot_id": observed["snapshot_id"],
"expected_rows": expected,
"correct": correct,
}
)
counts["successful_snapshot_executions"] += (
observed["status"] == "OK"
)
execution = len(comparisons) == 5 and all(
c["correct"] for c in comparisons
)
counts["normalized_sql_exact"] += exact
counts["execution_correct_all_five"] += execution
counts["task_correct"] += bool(correct_refusal or execution)
details.append(
{
"document_id": row["document_id"],
"outside_schema": outside,
"parsed_status": status,
"correct_refusal": correct_refusal,
"normalized_sql_exact": exact,
"execution_correct_all_five": execution,
"worker": worker,
"snapshot_comparisons": comparisons,
}
)
counts.update(
{
"normalized_sql_exact_rate_over_330": counts["normalized_sql_exact"]
/ 330,
"execution_accuracy_over_330": counts["execution_correct_all_five"]
/ 330,
"refusal_accuracy_over_30": counts["correct_refusal"] / 30,
"task_accuracy_over_360": counts["task_correct"] / 360,
}
)
result[variant] = {"metrics": counts, "rows": details}
print({"variant": variant, **counts}, flush=True)
after = {path: sha(path) for path in before}
if before != after:
raise ValueError("Read-only evaluation modified an inventory")
save_json(
"outputs/" + ("exercise-evaluation.json" if exercise else "evaluation.json"),
{
"results": result,
"database_sha256_before": before,
"database_sha256_after": after,
"database_unchanged": True,
"quality_gain_required": False,
"execution_equivalence_on_five_fixtures_is_not_a_general_SQL_proof": True,
},
)
Probar la capacidad de rango sin cambiar la escala
Ablación de un factor
Error frecuente: Comparar un r16 entrenado más tiempo falsea el efecto del rango: conservar los tokens supervisados.
Invariante: alpha/r idéntico y nuevo tamaño de adaptador calculado.
Archivo src/lora_qlora_sql/exercise.py
from pathlib import Path
from .config import read_json, save_json, sha
def main_artifacts():
paths = ["outputs/evaluation.json"]
for precision in ["bf16", "nf4"]:
paths += [
f"assets/checkpoints/{precision}-r8-resume.safetensors",
f"outputs/resume-{precision}-r8.json",
f"outputs/adapter-{precision}-r8.json",
]
for rank in [0, 8]:
paths += [
f"outputs/predictions-{precision}-r{rank}.json",
f"outputs/response-nll-{precision}-r{rank}.json",
]
return paths
def preserve_main():
path = Path("outputs/exercise/protected.json")
if path.exists():
raise ValueError("Preserve existing exercise snapshot")
save_json(path, {name: sha(name) for name in main_artifacts()})
def compare_ranks():
before = read_json("outputs/exercise/protected.json")
if before != {name: sha(name) for name in main_artifacts()}:
raise ValueError("Rank exercise altered a main artifact")
comparisons = []
for rank in [8, 16]:
a = read_json(f"outputs/train-bf16-r{rank}-continuous.json")
b = read_json(f"outputs/train-nf4-r{rank}-continuous.json")
if a["initial_adapter"] != b["initial_adapter"]:
raise ValueError("Precision branches did not start from the same adapters")
if [
(r["document_ids"], r["response_tokens"], r["learning_rate"])
for r in a["history"]
] != [
(r["document_ids"], r["response_tokens"], r["learning_rate"])
for r in b["history"]
]:
raise ValueError("Precision branches used different learning schedules")
comparisons.append(
{
"rank": rank,
"alpha": 2 * rank,
"initial_adapters_exact": True,
"example_order_and_token_budget_exact": True,
"trainable_parameters": a["trainable_parameters"],
"updates": a["updates"],
"presentations": sum(len(r["document_ids"]) for r in a["history"]),
"response_tokens": sum(r["response_tokens"] for r in a["history"]),
}
)
save_json(
"outputs/exercise/rank-comparison.json",
{
"protected_sha256": before,
"main_artifacts_unchanged": True,
"fresh_rank16_both_precisions": True,
"same_scale_alpha_over_rank": 2.0,
"comparisons": comparisons,
"quality_gain_required": False,
"rank16_has_more_parameters": True,
},
)
Fuentes
LoRA, QLoRA, Qwen2.5-1.5B-Instruct, preparación k-bit de PEFT 0.17.1, capas LoRA de PEFT 0.17.1 y capas bitsandbytes 0.47.0. Las decisiones locales, incluida la ausencia de optimizador paginado y gradient checkpointing, se explicitan en este curso.
Lo observado durante el desarrollo
El recorrido completo superó sus 20 pruebas públicas. La auditoría independiente recalculó las decisiones y resultados SQL sobre los cinco inventarios, los denominadores, las pérdidas de respuesta, las identidades de adaptadores y los controles de reanudación. Esta validación técnica no impone una ganancia de calidad.
Las seis condiciones reciben las mismas 360 solicitudes de test: 330 dentro del esquema y 30 que solicitan un campo ausente. Una consulta solo se considera correcta si sus resultados coinciden con la referencia en los cinco inventarios. Los cinco inventarios no constituyen una prueba de equivalencia sobre toda base posible.
| Condición | SQL correcto en las cinco bases | Rechazo correcto | Tarea correcta | Respuestas truncadas |
|---|---|---|---|---|
| bf16-r0 | 0/330 | 0/30 | 0/360 | 0 |
| nf4-r0 | 10/330 | 0/30 | 10/360 | 0 |
| bf16-r8 | 325/330 | 30/30 | 355/360 | 0 |
| nf4-r8 | 324/330 | 30/30 | 354/360 | 0 |
| bf16-r16 | 318/330 | 30/30 | 348/360 | 0 |
| nf4-r16 | 329/330 | 30/30 | 359/360 | 0 |
La puntuación corresponde al contrato completo «solo SQL o REFUSE». Se rechazan un bloque Markdown, una explicación o un filtro fuera de gramática; por tanto, la puntuación no aísla un conocimiento abstracto de SQL. Para la primera solicitud S255-0 se conservan a continuación las respuestas exactas observadas.
{
"bf16-r0": {
"document_id": "S255-0",
"text": "```sql\nSELECT COUNT(*) FROM inventory WHERE zone = 'I' AND qty >= 1\n```",
"status": "INVALID_SQL"
},
"nf4-r0": {
"document_id": "S255-0",
"text": "```sql\nSELECT COUNT(*) FROM inventory WHERE zone = 'I' AND qty >= 11;\n```",
"status": "INVALID_SQL"
},
"bf16-r8": {
"document_id": "S255-0",
"text": "SELECT COUNT(*) FROM inventory WHERE zone = 'I' AND qty >= 11;",
"status": "VALID_SQL"
},
"nf4-r8": {
"document_id": "S255-0",
"text": "SELECT COUNT(*) FROM inventory WHERE zone = 'I' AND qty >= 11;",
"status": "VALID_SQL"
},
"bf16-r16": {
"document_id": "S255-0",
"text": "SELECT COUNT(*) FROM inventory WHERE zone = 'I' AND qty >= 11;",
"status": "VALID_SQL"
},
"nf4-r16": {
"document_id": "S255-0",
"text": "SELECT COUNT(*) FROM inventory WHERE zone = 'I' AND qty >= 11;",
"status": "VALID_SQL"
}
}
La pérdida siguiente trata de las mismas respuestas de referencia, con su EOS, y no únicamente de las respuestas generadas que tuvieron éxito. Su unidad es nat por token de respuesta. Una pérdida menor no garantiza una consulta correcta en generación libre.
| Condición | NLL por token de respuesta |
|---|---|
| bf16-r0 | 0.2835442012 |
| nf4-r0 | 0.2712743014 |
| bf16-r8 | 0.003191479554 |
| nf4-r8 | 0.002342658512 |
| bf16-r16 | 0.005727622692 |
| nf4-r16 | 0.0006154038 |
Los tiempos siguientes pertenecen a los caminos continuos de entrenamiento de esta ejecución. Cada condición tiene 420 actualizaciones. Los caminos interrumpidos y reanudados, adquisiciones, validaciones, inferencias y exportaciones también tienen un coste, conservado en los registros; esta tabla no es el coste total del recorrido.
| Condición | Parámetros entrenados | Optimización (s) | Optimización y validación (s) | Pico PyTorch allocated (bytes) | Pico reserved (bytes) |
|---|---|---|---|---|---|
| bf16-r8 | 9232384 | 759.159 | 821.304 | 4780346880 | 5114953728 |
| nf4-r8 | 9232384 | 992.322 | 1080.726 | 3044170752 | 3351248896 |
| bf16-r16 | 18464768 | 766.232 | 829.986 | 4946857472 | 5284823040 |
| nf4-r16 | 18464768 | 980.464 | 1068.256 | 3210681344 | 3521118208 |
Allocated y reserved son contadores PyTorch; no designan toda la VRAM del proceso observada por el controlador. La figura de memoria de la lección mide las generaciones de test, por tanto otra fase distinta de esta tabla. Su caché de asignación puede incluir el precalentamiento declarado. Una sola semilla, un corpus procedimental y una sola máquina no permiten prometer estos tiempos ni esta calidad en otras solicitudes.
Las salidas completas permiten examinar errores sin corregir después el analizador, el conjunto de test ni los hiperparámetros. El ejercicio de rango 16 conserva los resultados de rango 8 y repite las dos precisiones. No se deduce ninguna ventaja sistemática de rango, NF4 o LoRA de un simple PASS.
Observaciones de desarrollo
Valores recalculados a partir de las salidas brutas de este recorrido. No son umbrales de ganancia; el recorrido desde el ZIP los recalcula sobre las mismas poblaciones.
{
"status": "PASS",
"documents": 2400,
"test_documents": 360,
"snapshots": 5,
"rank8_parameters": 9232384,
"rank16_parameters": 18464768,
"metrics": {
"bf16-r0": {
"documents": 360,
"eligible_sql": 330,
"outside_schema": 30,
"valid_sql": 0,
"truncated": 0,
"refusal": 0,
"correct_refusal": 0,
"normalized_sql_exact": 0,
"execution_correct_all_five": 0,
"task_correct": 0,
"snapshot_slots": 1800,
"outside_schema_skipped_slots": 150,
"actual_snapshot_executions": 0,
"successful_snapshot_executions": 0,
"worker_timeouts": 0,
"normalized_sql_exact_rate_over_330": 0.0,
"execution_accuracy_over_330": 0.0,
"refusal_accuracy_over_30": 0.0,
"task_accuracy_over_360": 0.0
},
"nf4-r0": {
"documents": 360,
"eligible_sql": 330,
"outside_schema": 30,
"valid_sql": 10,
"truncated": 0,
"refusal": 0,
"correct_refusal": 0,
"normalized_sql_exact": 8,
"execution_correct_all_five": 10,
"task_correct": 10,
"snapshot_slots": 1800,
"outside_schema_skipped_slots": 150,
"actual_snapshot_executions": 50,
"successful_snapshot_executions": 50,
"worker_timeouts": 0,
"normalized_sql_exact_rate_over_330": 0.024242424242424242,
"execution_accuracy_over_330": 0.030303030303030304,
"refusal_accuracy_over_30": 0.0,
"task_accuracy_over_360": 0.027777777777777776
},
"bf16-r8": {
"documents": 360,
"eligible_sql": 330,
"outside_schema": 30,
"valid_sql": 325,
"truncated": 0,
"refusal": 30,
"correct_refusal": 30,
"normalized_sql_exact": 325,
"execution_correct_all_five": 325,
"task_correct": 355,
"snapshot_slots": 1800,
"outside_schema_skipped_slots": 150,
"actual_snapshot_executions": 1625,
"successful_snapshot_executions": 1625,
"worker_timeouts": 0,
"normalized_sql_exact_rate_over_330": 0.9848484848484849,
"execution_accuracy_over_330": 0.9848484848484849,
"refusal_accuracy_over_30": 1.0,
"task_accuracy_over_360": 0.9861111111111112
},
"nf4-r8": {
"documents": 360,
"eligible_sql": 330,
"outside_schema": 30,
"valid_sql": 324,
"truncated": 0,
"refusal": 30,
"correct_refusal": 30,
"normalized_sql_exact": 324,
"execution_correct_all_five": 324,
"task_correct": 354,
"snapshot_slots": 1800,
"outside_schema_skipped_slots": 150,
"actual_snapshot_executions": 1620,
"successful_snapshot_executions": 1620,
"worker_timeouts": 0,
"normalized_sql_exact_rate_over_330": 0.9818181818181818,
"execution_accuracy_over_330": 0.9818181818181818,
"refusal_accuracy_over_30": 1.0,
"task_accuracy_over_360": 0.9833333333333333
},
"bf16-r16": {
"documents": 360,
"eligible_sql": 330,
"outside_schema": 30,
"valid_sql": 318,
"truncated": 0,
"refusal": 30,
"correct_refusal": 30,
"normalized_sql_exact": 318,
"execution_correct_all_five": 318,
"task_correct": 348,
"snapshot_slots": 1800,
"outside_schema_skipped_slots": 150,
"actual_snapshot_executions": 1590,
"successful_snapshot_executions": 1590,
"worker_timeouts": 0,
"normalized_sql_exact_rate_over_330": 0.9636363636363636,
"execution_accuracy_over_330": 0.9636363636363636,
"refusal_accuracy_over_30": 1.0,
"task_accuracy_over_360": 0.9666666666666667
},
"nf4-r16": {
"documents": 360,
"eligible_sql": 330,
"outside_schema": 30,
"valid_sql": 329,
"truncated": 0,
"refusal": 30,
"correct_refusal": 30,
"normalized_sql_exact": 329,
"execution_correct_all_five": 329,
"task_correct": 359,
"snapshot_slots": 1800,
"outside_schema_skipped_slots": 150,
"actual_snapshot_executions": 1645,
"successful_snapshot_executions": 1645,
"worker_timeouts": 0,
"normalized_sql_exact_rate_over_330": 0.996969696969697,
"execution_accuracy_over_330": 0.996969696969697,
"refusal_accuracy_over_30": 1.0,
"task_accuracy_over_360": 0.9972222222222222
}
},
"response_nll": {
"bf16-r0": 0.2835442011762571,
"nf4-r0": 0.2712743014305074,
"bf16-r8": 0.003191479553683838,
"nf4-r8": 0.002342658512238993,
"bf16-r16": 0.005727622691669397,
"nf4-r16": 0.0006154038000183541
},
"quality_gain_required": false,
"scope": "Data-only independent SQL grammar/results on all five fixtures, tokenizer byte decoding, adapter bytes/gradients, frozen base metadata, exact budget/resume evidence and all-case metrics; not BPE re-encoding or full quantized Transformer reimplementation"
}
