Un'applicazione fittizia traduce una richiesta breve in SELECT SQLite su un inventario locale. Addestriamo lo stesso modello con gli stessi esempi e adattatori, prima su una base a 16 bit, poi su una base quantizzata a 4 bit. Nessun SQL prodotto viene eseguito con diritti di scrittura.
Una richiesta produce una query proposta, validata sintatticamente poi valutata in sola lettura su più piccoli inventari. Le tabelle separano l'effetto della quantizzazione senza addestramento e l'effetto di ogni adattamento.
Scaricare il progetto completo, revisione r002
Il problema e i suoi dati
Lo schema fornito è inventory(sku TEXT, zone TEXT, qty INTEGER). Una richiesta precisa una zona, una soglia di quantità e un risultato: SKU, SKU/quantità con ordine esplicito, numero di righe, somma, minimo o massimo. Una richiesta di prezzo o fornitore deve ottenere REFUSE, poiché questi campi non esistono. Il modello non riceve mai la query target nel prompt.
StockSQL-v1 contiene 300 famiglie di otto formulazioni, cioè 2400 richieste fittizie. Le famiglie 0–209 forniscono 1680 esempi train, 210–254 forniscono 360 validation, 255–299 forniscono 360 test. Il tipo di SELECT, la zona e la soglia definiscono 300 SQL familiari distinti. Le zone train sono dalla A alla G; il test usa I e J. I tipi di query e i modelli linguistici restano condivisi. Un identificativo Ticket Txxx-j rende tracciabili le richieste, ma non crea da solo una novità semantica. Le richieste fuori schema ripetono le stesse intenzioni prezzo/fornitore tra le partizioni; non viene rivendicata alcuna generalizzazione universale fuori schema.
Tra i 360 casi test, 330 attendono SQL e 30 un rifiuto. Cinque inventari indipendenti sono generati da semi fissi: ciascuno possiede 300 SKU in dieci zone, con quantità da 0 a 20. I contenuti sono scritti in cinque vere basi SQLite e riletti per verificare la corrispondenza con le righe JSON originali. Gli SHA-256 prima/dopo valutazione ne verificano l'integrità.
Il segnale supervisionato
Il tokenizer e il template di dialogo appartengono a Qwen2.5-1.5B-Instruct. Il prompt contiene l'istruzione di sistema, lo schema, la richiesta, poi il confine di una risposta assistant. Gli ID del prompt sono ottenuti con apply_chat_template; il target SQL o REFUSE è codificato separatamente, poi viene aggiunto EOS=151645. Questo contratto esplicito non è presentato come equivalente a una ritokenizzazione della stringa completa.
Il prompt è limitato a 256 token, la risposta a 128 incluso EOS, il totale a 384. Un superamento provoca un errore identificato, mai un troncamento silenzioso. Le label del prompt e del padding sono −100. Le altre label supervisionano ogni token della risposta e EOS. PAD=151643 ed EOS=151645 sono distinti; attenzione e supervisione dipendono comunque dalle posizioni reali.
La CE allinea logits[:, :-1] e labels[:, 1:], poi seleziona le posizioni attive e calcola in FP32. Il passaggio del modello usa autocast BF16. Gli otto esempi di un aggiornamento possono avere risposte di lunghezze diverse: le loro somme di CE sono divise per lo stesso numero totale di token risposta. Fare una media di otto medie cambierebbe questo peso. Il test confronta i gradienti di un'accumulazione con la normalizzazione unica.
LoRA: imparare una correzione a basso rango
Per una proiezione W di forma uscita × ingresso, la correzione è (alpha/r) B A, con A di forma r × ingresso e B di forma uscita × r. Il livello calcola xWᵀ + (alpha/r)xAᵀBᵀ. W resta congelato; vengono ottimizzati solo A e B. Il piccolo modulo LoRALinear mostra questo calcolo, poi confronta output e gradienti di input/A/B con un vero livello PEFT in FP32. B è non nullo in questo confronto, affinché il gradiente di A venga effettivamente testato.
Per l'addestramento reale, A segue un'inizializzazione Kaiming e B parte da zero. A questo punto la correzione vale zero e il primo gradiente di A può essere nullo; non è un congelamento involontario. Passi successivi devono davvero modificare A e B. I controlli riguardano i veri tensor e i loro hash.
Gli adattatori coprono q_proj, k_proj, v_proj, o_proj, gate_proj, up_proj e down_proj nei 28 blocchi: 196 proiezioni e 392 matrici A/B. Con r8 e alpha 16, rappresentano 9 232 384 parametri FP32. Con r16 e alpha 32, rappresentano 18 464 768 parametri FP32. I file Safetensors aggiungono metadati alla dimensione logica di questi valori; viene misurata la loro dimensione reale.
BF16 e NF4: ciò che cambia davvero
La base congelata è Qwen/Qwen2.5-1.5B-Instruct, revisione 989aa7980e4cf806f80c7fef2b1adb7bc71aa306. Larghezza 1536, 28 blocchi, 12 teste Q, due teste K/V, MLP 8960 e vocabolario 151936 sono verificati dopo il caricamento. La condivisione tra embedding di input e output deve essere conservata.
Nel ramo BF16, i pesi delle 196 proiezioni restano memorizzati in BF16. Nel ramo NF4, bitsandbytes li quantizza in NF4 con doppia quantizzazione; vengono sottoposti ad audit le classi Linear4bit, lo stato quantizzato annidato, lo storage e il dtype di calcolo BF16. Il caricamento designa esplicitamente CUDA 0 nel contenitore, a sua volta limitato alla GPU scelta. Non usa un posizionamento automatico su più schede.
prepare_model_for_kbit_training viene chiamato prima dell'installazione PEFT. Questa preparazione converte in FP32 i parametri non quantizzati. Per evitare una differenza supplementare, il ramo BF16 conserva anch'esso in FP32 embedding/output condiviso, norme e bias, lasciando i suoi 196 pesi di proiezione in BF16. Dire che tutto il modello usa quattro bit sarebbe dunque falso. Metadati di quantizzazione, attivazioni, gradienti dell'adattatore e stati AdamW occupano anch'essi memoria.
PEFT può convertire gli adattatori in base al dtype di caricamento. Dopo la sua installazione, reinizializziamo esplicitamente A/B in FP32 con gli stessi generatori CPU per nome di modulo in entrambi i rami. I loro hash iniziali devono essere esattamente uguali a parità di rango; una prima inizializzazione arrotondata in BF16 in un solo ramo confonderebbe il confronto. Gradient checkpointing è disattivato in entrambi i rami. Non viene usato alcun ottimizzatore paginato: QLoRA designa qui gli adattatori su una base NF4, con le altre scelte dettagliate, non una riproduzione completa del protocollo dell'articolo.
Un budget comparabile e una ripresa verificabile
I due rami usano lo stesso seme 45, lo stesso ordine di esempi, due epoche, microbatch 1 e accumulazione di otto esempi: 420 aggiornamenti, 3360 presentazioni. AdamW ha lr=0,0001 e weight_decay=0,01; clipping a 1; warm-up di 42 passi poi decrescita lineare fino al 10% del tasso massimo. Solo gli adattatori FP32 entrano nell'ottimizzatore. La validazione della sola risposta è misurata dopo ogni epoca, senza scegliere un checkpoint in base al test.
Un passaggio forward/backward su un esempio train preriscalda il calcolo senza passo di ottimizzatore. I gradienti vengono cancellati, gli RNG ripristinati e i pesi verificati invariati. La sua durata e i token supplementari sono registrati a parte. I picchi allocated/reserved vengono poi reimpostati; reserved può conservare la cache dell'allocatore creata durante questo preriscaldamento. La durata dell'ottimizzazione è distinta dal tempo che include la validazione e dal costo di caricamento/salvataggio. Le misure del processo e di nvidia-smi rispondono a domande diverse.
L'interruzione avviene alla fine della prima epoca. La ripresa ricarica la stessa base verificata, poi adattatori, AdamW, sampler, scheduler, cronologie e RNG Python/NumPy/PyTorch CPU/CUDA. Tutti gli stati finali devono essere esattamente uguali a quelli del percorso continuo di due epoche. I parametri congelati e i metadati quantizzati devono conservare gli stessi hash. Questo contratto non promette una ripresa nel mezzo di un microbatch.
Gli stati .pt sono prodotti localmente e scritti tramite file temporaneo poi rinomina atomica. Non devono essere sostituiti da oggetti non affidabili. Gli adattatori finali sono anche salvati in Safetensors, con base/configurazione/hash espliciti e ricaricamento rigoroso. Nessun peso congelato viene duplicato in questi adattatori e non viene effettuata alcuna fusione in un'altra base.
Generare poi controllare, senza dare le chiavi al modello
Le due basi senza adattamento sono valutate separatamente, poi i due r8 e i due r16. Ogni variante genera liberamente le stesse 360 richieste test, con decodifica greedy, una beam, al massimo 128 nuovi token ed EOS=151645. Una richiesta train fissa serve da preriscaldamento, registrata ma esclusa dai denominatori test e dal throughput. Sono conservati ID, messaggi, testo grezzo, ragione di arresto e durate. La NLL della sola risposta viene calcolata separatamente con gli stessi target test ed EOS e non sostituisce la valutazione di generazioni libere.
Un output fermato al limite è TRUNCATED, anche se assomiglia già a un SELECT valido o a REFUSE. Il parser non rimuove tag Markdown e non ripara nulla. Esclusi gli spazi esterni, REFUSE deve essere esatto. Un output SQL deve appartenere a una grammatica completa e delimitata: SELECT su inventory, colonne conosciute o un aggregato COUNT(*)/SUM(qty)/MIN(qty)/MAX(qty), al massimo quattro predicati AND su zone/qty, poi un ordine facoltativo. Nessun commento, sottoquery, funzione libera, JOIN, UNION, PRAGMA, scrittura o secondo statement.
Le query ammesse sono affidate a un processo Python separato, senza modello. SQLite apre basi locali in URI mode=ro e immutable=1; vengono applicati query_only, estensioni disattivate, autorizzatore rigoroso e limite VM. I PRAGMA di configurazione sono codice fidato eseguito prima dell'autorizzatore; gli input del modello non possono fornirli. L'autorizzatore accetta soltanto SELECT, letture delle colonne inventory e quattro aggregati. Il processo ha un limite CPU di due secondi, memoria virtuale di 256 Mio, 32 file aperti e dimensione di file scritta nulla. Il padre impone tre secondi di tempo reale; le cinque query sono ciascuna limitate dal controllo VM e a 1000 righe di risultato.
Un timeout o un errore del processo conserva un fallimento esplicito. Se il processo viene fermato prima di pubblicare il suo JSON, le eventuali operazioni intermedie non vengono contate come risultati osservati; non viene inventato alcun risultato. Gli inventari restano in sola lettura e la loro integrità viene ricontrollata.
Misurare la cosa giusta
Il rapporto conserva 1800 spazi di snapshot per variante: cinque per ciascuno dei 360 casi. Le 30 richieste fuori schema producono 150 spazi esplicitamente ignorati, qualunque sia l'output; valutano il rifiuto, non l'esecuzione di una query fuori tema. Gli output non validi o rifiuti inattesi degli altri 330 casi non vengono eseguiti, ma restano fallimenti nei rispettivi denominatori.
L'esattezza SQL normalizzata confronta la struttura testuale canonica, senza pretendere di risolvere ogni equivalenza. Una permutazione delle due condizioni AND può fallire questo exact match pur avendo gli stessi risultati. L'esattezza di esecuzione richiede le righe attese, nell'ordine atteso, sui cinque inventari. I risultati attesi sono calcolati anche da un interprete Python indipendente da SQLite. COUNT su una popolazione vuota vale zero; SUM/MIN/MAX vuoti valgono null, come verificano i test.
Vengono pubblicati separatamente: SQL validi, troncati, rifiuti, rifiuti corretti su 30, SQL esatti su 330, risultati corretti sui cinque inventari su 330, esattezza del compito su 360, esecuzioni realmente osservate e timeout. Una query costante che riesce per caso su una base non basta. Riuscire sulle cinque fixture non è comunque una prova generale di equivalenza SQL né una garanzia su dati aziendali.
Esercizio e limiti
L'esercizio addestra nuovi r16 con alpha 32, dunque lo stesso alpha/r=2 di r8/alpha 16. Ordine delle richieste, numero di aggiornamenti, risposte supervisionate e ottimizzatore restano identici. Raddoppiare il rango raddoppia il numero di parametri dell'adattatore, non quello dei parametri della base. Questo non elimina automaticamente un errore di quantizzazione; qualità, dimensione, memoria e durata vengono confrontate senza imporre un guadagno. Gli r8 e i risultati principali sono protetti da SHA-256.
Queste richieste, schemi e inventari sono sintetici e piccoli. Le formulazioni e i tipi di query sono condivisi; un solo seme di apprendimento non misura tutta la variabilità. Il rifiuto tramite grammatica riduce lo spazio SQL ammissibile: una query SQLite legale ma fuori grammatica fallisce volutamente. Il modello non avvia alcun SQL cliente, rete di database, strumento esterno, server o azione aziendale.
Creare richieste e inventari riservati
Validità SQL e semantica
Errore frequente: parafrasi dello stesso SQL vengono separate casualmente; raggruppate le famiglie prima degli split.
Invariante: famiglie disgiunte, query oracle valide sui rispettivi snapshot.
File 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
Mostrare e verificare il segnale SFT
Solo risposta e controllo del confine
Errore frequente: il prompt fornisce accidentalmente la query target; sottoponete ad audit i campi effettivamente formattati.
Invariante: i rami ricevono gli stessi ID e label.
File 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(),
},
)
Codificare LoRA poi caricare le due basi
Basso rango e quantizzazione
Errore frequente: il caricamento k-bit congela anche gli adattatori aggiunti troppo presto; verificare l'ordine delle operazioni e requires_grad.
Invariante: controllo del livello riuscito e misure delle due basi senza addestramento registrate.
File 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()),
},
)
Addestrare i due adattamenti comparabili
Stessi dati, budget e ottimizzatore
Errore frequente: durante la ripresa viene caricata una vecchia base dalla cache; rifiutare se lo SHA non corrisponde.
Invariante: i due rami hanno consumato il budget previsto e i pesi congelati sono invariati.
File 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",
},
)
Eseguire le SELECT in un controllo indipendente
Interpretazione sicura e memoria misurata
Errore frequente: una stringa con due statement supera un semplice startswith SELECT; usate un vero parser/autorizzatore.
Invariante: ogni caso ha cinque risultati oppure un rifiuto/fallimento esplicito; nessuna scrittura possibile.
File 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,
},
)
Testare la capacità di rango senza cambiare la scala
Ablazione di un fattore
Errore frequente: confrontare un r16 addestrato più a lungo falsifica l'effetto del rango; conservate i token supervisionati.
Invariante: alpha/r identico e nuova dimensione dell'adattatore calcolata.
File 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,
},
)
Fonti
LoRA, QLoRA, Qwen2.5-1.5B-Instruct, preparazione k-bit PEFT 0.17.1, livelli LoRA PEFT 0.17.1 e livelli bitsandbytes 0.47.0. Le scelte locali, compresa l'assenza di ottimizzatore paginato e di gradient checkpointing, sono esplicitate in questo corso.
Ciò che è stato osservato durante lo sviluppo
Il percorso completo ha superato i suoi 20 test pubblici. L'audit indipendente ha ricalcolato le decisioni e i risultati SQL sui cinque inventari, i denominatori, le perdite sulle risposte, le identità degli adattatori e i controlli di ripresa. Questa validazione tecnica non impone un guadagno di qualità.
Le sei condizioni ricevono le stesse 360 richieste test: 330 nello schema e 30 che chiedono un campo assente. Una query viene conteggiata corretta solo se i suoi risultati concordano con il riferimento sui cinque inventari. I cinque inventari non costituiscono una prova di equivalenza su ogni possibile base.
| Condizione | SQL corretto sulle cinque basi | Rifiuto corretto | Compito corretto | Risposte troncate |
|---|---|---|---|---|
| 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 |
Il punteggio è quello del contratto completo « solo SQL oppure REFUSE ». Un blocco Markdown, una spiegazione o un filtro fuori grammatica viene rifiutato; il punteggio non isola quindi una conoscenza astratta di SQL. Per la prima richiesta S255-0, le risposte esatte osservate sono conservate qui sotto.
{
"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 perdita seguente riguarda le stesse risposte di riferimento, con il loro EOS, e non soltanto le risposte generate che hanno avuto successo. La sua unità è il nat per token risposta. Una perdita più bassa non garantisce una query corretta in generazione libera.
| Condizione | NLL per token risposta |
|---|---|
| bf16-r0 | 0.2835442012 |
| nf4-r0 | 0.2712743014 |
| bf16-r8 | 0.003191479554 |
| nf4-r8 | 0.002342658512 |
| bf16-r16 | 0.005727622692 |
| nf4-r16 | 0.0006154038 |
I tempi qui sotto appartengono ai percorsi di addestramento continui di questa esecuzione. Ogni condizione ha 420 aggiornamenti. Anche percorsi interrotti e ripresi, acquisizioni, validazioni, inferenze ed esportazioni hanno un costo, conservato nei registri; questa tabella non è il costo totale del percorso.
| Condizione | Parametri addestrati | Ottimizzazione (s) | Ottimizzazione e validazione (s) | Picco PyTorch allocated (byte) | Picco reserved (byte) |
|---|---|---|---|---|---|
| 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 e reserved sono contatori PyTorch; non designano tutta la VRAM del processo osservata dal driver. La figura della memoria della lezione misura le generazioni test, quindi una fase diversa da questa tabella. La sua cache di allocazione può includere il preriscaldamento dichiarato. Un solo seme, un corpus procedurale e una sola macchina non consentono di promettere questi tempi né questa qualità su altre richieste.
Le uscite complete permettono di esaminare gli errori senza correggere a posteriori il parser, il set test o gli iperparametri. L'esercizio al rango 16 conserva i risultati del rango 8 e rifà le due precisioni. Nessun vantaggio sistematico del rango, di NF4 o di LoRA viene dedotto da un semplice PASS.
Osservazioni di sviluppo
Valori ricalcolati dalle uscite grezze di questo percorso. Non sono soglie di miglioramento; il percorso dallo ZIP li ricalcola sulle stesse popolazioni.
{
"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"
}
