Quantized Export

from unsloth import FastLanguageModel

model, tokenizer = FastLanguageModel.from_pretrained(
    model_name=CHECKPOINT_DIR,
    max_seq_length=4096,
    load_in_4bit=False,  # load full precision before merge
)

# fp16/bf16 merged export — safe default, no quant loss at this step
model.save_pretrained_merged(
    MERGED_DIR,
    tokenizer,
    save_method="merged_16bit",
)

# 4-bit merged export — only if the serving stack
# consumes bitsandbytes 4-bit directly (rare; most
# deployments quantize downstream instead — see the
# AWQ and GGUF sections below)
model.save_pretrained_merged(
    MERGED_DIR + "-4bit",
    tokenizer,
    save_method="merged_4bit",
)

model.save_pretrained_gguf(
    GGUF_DIR,
    tokenizer,
    quantization_method=["q4_k_m", "q8_0"],
)

cmake -B build
cmake --build build --config Release -j"$(nproc)"

pip install ./gguf-py
pip install sentencepiece

# 1. Convert HF safetensors to GGUF (f16, no quant yet)
python convert_hf_to_gguf.py \
    "$MERGED_DIR" \
    --outfile "$GGUF_DIR/model-f16.gguf" \
    --outtype f16

# 2. Generate the importance matrix from a calibration
#    corpus — domain-representative text, several
#    hundred KB minimum; a generic corpus (e.g. the
#    llama.cpp wikitext sample) works if no
#    domain corpus is available
./llama-imatrix \
    -m "$GGUF_DIR/model-f16.gguf" \
    -f calibration-corpus.txt \
    -o "$GGUF_DIR/imatrix.dat" \
    --chunks 200

# 3. Quantize using the imatrix — Q4_K_M is the
#    edge/llama.cpp default from the Format Map
./llama-quantize \
    --imatrix "$GGUF_DIR/imatrix.dat" \
    "$GGUF_DIR/model-f16.gguf" \
    "$GGUF_DIR/model-Q4_K_M.gguf" \
    Q4_K_M

./llama-completion \
    -m "$GGUF_DIR/model-Q4_K_M.gguf" \
    -p "$(cat prompt.txt)" \
    -n 512 --temp 0 --seed 0

from awq import AutoAWQForCausalLM
from transformers import AutoTokenizer

quant_config = {
    "zero_point": True,
    "q_group_size": 128,
    "w_bit": 4,
    "version": "GEMM",
}

model = AutoAWQForCausalLM.from_pretrained(MERGED_DIR)
tokenizer = AutoTokenizer.from_pretrained(MERGED_DIR)

# calibration data: a few hundred domain-representative
# samples; reuses the same calibration-corpus concept
# as the llama.cpp imatrix step above
model.quantize(tokenizer, quant_config=quant_config)
model.save_quantized(MERGED_DIR + "-awq")
tokenizer.save_pretrained(MERGED_DIR + "-awq")

vllm serve "$MERGED_DIR" \
    --quantization fp8 \
    --served-model-name checkpoint-fp8 \
    --port 8000 > vllm-fp8-load.log 2>&1 &
VLLM_PID=$!

# Bounded wait for readiness — up to 60s, not a blind sleep.
ready=0
for _ in $(seq 1 30); do
    if curl -sf http://localhost:8000/v1/models | grep -q checkpoint-fp8; then
        ready=1
        break
    fi
    sleep 2
done

if [ "$ready" -ne 1 ]; then
    echo "FAILED: endpoint did not come up within 60s"
    kill "$VLLM_PID" 2>/dev/null
    exit 1
fi

# Endpoint liveness alone does not prove FP8 loaded — vLLM can
# silently fall back to bf16. Confirm the actual dtype from the
# startup log before trusting the deployment.
if grep -qi 'fp8' vllm-fp8-load.log; then
    echo "FP8 endpoint up and confirmed FP8 in startup log"
else
    echo "FAILED: endpoint up but startup log does not confirm FP8 (possible silent bf16 fallback)"
    kill "$VLLM_PID" 2>/dev/null
    exit 1
fi

import json
import sys

# Deterministic decoding, persisted and reused for both the
# pre-export run (that produced pre-export-outputs.jsonl) and the
# post-export run below — greedy (temperature 0) with a fixed seed.
# Any drift in these settings between the two runs can flip an
# otherwise-valid export into a spurious mismatch.
SMOKE_TEST_GENERATION_KWARGS = {"temperature": 0, "seed": 0, "max_new_tokens": 512}

def load_pre_export_outputs(path: str) -> dict:
    """goldens.jsonl-keyed pre-export generations,
    produced from the promoted checkpoint before any
    quantization/export step, using
    SMOKE_TEST_GENERATION_KWARGS."""
    with open(path) as f:
        return {row["task_id"]: row["output"] for row in map(json.loads, f)}

def load_goldens(path: str, n: int = 5) -> list:
    with open(path) as f:
        rows = [json.loads(line) for line in f]
    return rows[:n]

def load_exported_model(export_path: str):
    """Load in the ACTUAL target runtime — vLLM,
    llama.cpp, or the AWQ loader. Never substitute
    a different framework than production here."""
    raise NotImplementedError("wire to target runtime")

def generate(model, prompt: str, **generation_kwargs) -> str:
    raise NotImplementedError("wire to target runtime")

def grade(task_id: str, output: str) -> bool:
    """Run the golden's actual task grader (from
    eval-harness-first) against a single output.
    Wire to the real grader module — never stub this
    with a byte-match; that defeats the point of the
    lossy-export path below."""
    raise NotImplementedError("wire to eval-harness-first's grader for this golden")

def diff_report(golden_id: str, pre: str, post: str, *, lossless: bool) -> dict:
    """lossless=True: gate on byte/string match.
    lossless=False (any quantized format): gate on
    grader verdict agreement — byte match is expected
    to fail for a healthy lossy export, so it is
    recorded for triage only, never as `match`."""
    byte_match = pre.strip() == post.strip()
    if lossless:
        match = byte_match
    else:
        match = grade(golden_id, pre) == grade(golden_id, post)
    return {
        "task_id": golden_id,
        "match": match,
        "byte_match": byte_match,
        "pre_export": pre,
        "post_export": post,
    }

def main(export_path: str, goldens_path: str, pre_export_path: str, *, lossless: bool):
    goldens = load_goldens(goldens_path, n=5)
    pre_outputs = load_pre_export_outputs(pre_export_path)
    model = load_exported_model(export_path)

    reports = []
    for row in goldens:
        post = generate(model, row["prompt"], **SMOKE_TEST_GENERATION_KWARGS)
        pre = pre_outputs[row["task_id"]]
        reports.append(diff_report(row["task_id"], pre, post, lossless=lossless))

    failures = [r for r in reports if not r["match"]]
    print(json.dumps({"total": len(reports), "failures": len(failures),
                       "mode": "byte-match" if lossless else "graded-verdict"}, indent=2))
    for r in failures:
        print(f"MISMATCH {r['task_id']} (byte_match={r['byte_match']}):")
        print(f"  pre : {r['pre_export'][:200]}")
        print(f"  post: {r['post_export'][:200]}")

    # non-zero exit on any mismatch — this script
    # gates the export, it does not just report on it
    sys.exit(1 if failures else 0)

if __name__ == "__main__":
    # lossless=True only for an unquantized fp16/bf16 merge;
    # every quantized format (Q4_K_M, AWQ, FP8, ...) is lossless=False
    main(*sys.argv[1:4], lossless=False)