Turn a deliberately messy support Q&A corpus (220 pairs seeded with exact duplicates, embedding-detectable near-duplicates, PII contamination, junk answers, and leakage probes) into a training-ready dataset through a six-stage factory: exact dedup → embedding near-dedup → quality gate → PII scrub → chat-format validation → leakage-checked train/val/test split.
Every stage records its counts into a data-quality datasheet, and you finish by generating a synthetic evaluation set in RAGAS-compatible format from the same corpus. The capstone move is defense-in-depth verification: you will deliberately disable the near-dedup stage and watch the leakage check catch 19 contaminated eval rows that would otherwise have inflated your fine-tune's measured quality — the exact failure mode that makes teams believe a model works until production disagrees. Fine-tuning lives or dies on this pipeline; the training run itself is the easy part.
~120 minutes hands-on · Advanced tier (maps to module M9).
Laptop-only, $0 in dry-run mode. The quality-scoring stage has a local heuristic scorer (default) and a Claude-based scorer you can enable with an API key; the synthetic eval-set generator likewise has a dry-run template mode and an API mode. Everything else — dedup, scrub, validation, splits, leakage checks — is fully local always.
mkdir -p ~/labs/de-lab-15 && cd ~/labs/de-lab-15
python3 -m venv .venv && source .venv/bin/activate
pip install sentence-transformers numpy pydantic faker anthropic
Create make_messy_corpus.py — a deterministic generator with tagged seeds so you can audit which stage removed what:
"""Generate a messy support Q&A corpus (220 pairs) with seeded problems."""
import json, random, pathlib
from faker import Faker
random.seed(15); Faker.seed(15)
fake = Faker("en_IN")
SYSTEMS = ["hydraulic press", "conveyor belt", "industrial chiller", "packaging robot",
"steam boiler", "air compressor", "CNC lathe", "paint booth ventilator",
"forklift charger", "wastewater pump"]
PROBLEMS = [
("overheating during long runs", "check the coolant loop first, flush and refill with the approved glycol mixture, then verify the fan clutch engages at temperature"),
("high-pitched whine under load", "inspect the bearing assembly for wear, repack with high-temperature grease, and confirm shaft alignment with a dial indicator"),
("breaker trips at startup", "measure inrush current, inspect the contactor for pitting, and verify the motor windings are not shorted to ground"),
("pressure drops below spec", "recalibrate the pressure transducer against the master gauge and pressure-test the manifold for slow leaks"),
("excessive vibration at speed", "check foundation bolts to torque spec, rebalance the rotating assembly, and inspect coupling alignment"),
("fluid leak at the main seal", "replace the O-ring kit, inspect the shaft sleeve for scoring, and pressure-test before returning to service"),
("won't reach operating temperature", "verify the heating element resistance, check the thermostat calibration, and confirm the controller setpoint profile"),
("error code halts mid-cycle", "pull the controller fault log, update firmware to the current release, and restore factory PID defaults before re-tuning"),
]
pairs = []
pid = 0
def add(q, a, tag):
global pid
pairs.append({"pair_id": f"qa-{pid:04d}", "question": q, "answer": a, "seed_tag": tag})
pid += 1
# 80 unique clean pairs: one per (system, problem) combination
clean_pool = []
for system in SYSTEMS:
for problem, fix in PROBLEMS:
q = f"What should I do when the {system} shows {problem}?"
a = f"For a {system} with {problem}: {fix}. Log the work order before closing."
add(q, a, "clean")
clean_pool.append(pairs[-1])
# 25 exact duplicates (copy-pasted tickets)
for p in random.sample(clean_pool, 25):
add(p["question"], p["answer"], "exact_dup")
# 40 near-duplicates (light paraphrase of question, same answer)
REPHRASE = [("What should I do when", "How do I handle it when"),
("What should I do when", "Recommended steps when"),
("shows", "is showing"), ("shows", "keeps showing")]
for p in random.sample(clean_pool, 40):
q = p["question"]
for old, new in random.sample(REPHRASE, 2):
q = q.replace(old, new)
add(q, p["answer"], "near_dup")
# 30 PII-contaminated pairs (emails, phones, reporter names)
for i in range(30):
system = random.choice(SYSTEMS)
problem, fix = random.choice(PROBLEMS)
name = fake.name(); email = fake.email(); phone = fake.phone_number()
q = f"{name} reported the {system} has {problem}, reachable at {email} or {phone} — what next?"
a = f"Tell {name} to power down the unit. Then {fix}. Escalate to {fake.email()} if unresolved."
add(q, a, "pii")
# 35 low-quality pairs
JUNK_A = ["idk try turning it off and on", "see manual", "ask your supervisor", "n/a", " "]
for i in range(35):
system = random.choice(SYSTEMS)
problem, _ = random.choice(PROBLEMS)
if i % 2 == 0:
add(f"What should I do when the {system} shows {problem}?", random.choice(JUNK_A), "low_quality")
else:
add(f"{system}???", "urgent pls help", "low_quality")
# 10 leak probes: near-identical to clean pairs; near-dedup SHOULD remove these —
# the leakage-checked split is the safety net if that stage regresses
for p in random.sample(clean_pool, 10):
add(p["question"] + " Please advise.", p["answer"], "leak_probe")
random.shuffle(pairs)
pathlib.Path("raw_pairs.jsonl").write_text("\n".join(json.dumps(p) for p in pairs))
from collections import Counter
print(len(pairs), "pairs:", dict(Counter(p["seed_tag"] for p in pairs)))
Run it: expect 220 pairs: {'clean': 80, 'pii': 30, 'exact_dup': 25, 'near_dup': 40, 'low_quality': 35, 'leak_probe': 10}.
Stages 1–2: exact dedup, then embedding near-dedup. Create factory.py. Exact dedup hashes normalized (question, answer); near-dedup embeds question + answer with all-MiniLM-L6-v2 and greedily drops any pair whose cosine similarity to an already-kept pair is ≥ 0.92 (first occurrence wins).
import hashlib, json, pathlib, random, re, sys
import numpy as np
from pydantic import BaseModel, Field, ValidationError
from sentence_transformers import SentenceTransformer
random.seed(15)
pairs = [json.loads(l) for l in pathlib.Path("raw_pairs.jsonl").read_text().splitlines()]
counts = {"raw": len(pairs)}
# Stage 1: exact dedup on normalized (question, answer)
seen, deduped = set(), []
for p in pairs:
key = hashlib.sha256((p["question"].strip().lower() + "\x1f"
+ p["answer"].strip().lower()).encode()).hexdigest()
if key not in seen:
seen.add(key); deduped.append(p)
counts["after_exact_dedup"] = len(deduped)
# Stage 2: near-dup removal via embeddings (cosine >= 0.92)
model = SentenceTransformer("sentence-transformers/all-MiniLM-L6-v2")
texts = [p["question"] + " " + p["answer"] for p in deduped]
emb = model.encode(texts, normalize_embeddings=True, show_progress_bar=False)
sim = emb @ emb.T
keep, kept_idx = [], []
SKIP_NEAR = "--skip-near-dedup" in sys.argv
for i in range(len(deduped)):
if (not SKIP_NEAR) and any(sim[i, j] >= 0.92 for j in kept_idx):
continue
kept_idx.append(i); keep.append(deduped[i])
counts["after_near_dedup"] = len(keep)
The 0.92 threshold is not folklore — it was chosen by probing this corpus: seeded near-duplicates sit at similarity ≥ 0.95 to their twin while the closest distinct clean pairs (same problem, different machine) top out at ~0.90. Measure before you set a threshold; on your corpus the gap will sit elsewhere.
Stage 3: quality gate. The dry-run scorer is a transparent heuristic; the LLM scorer is what production uses. Keep the interface identical — score 0–10, keep if ≥ 6:
def quality_score_dry(q, a):
score = 10
if len(a.split()) < 8: score -= 6
if len(q.split()) < 5: score -= 3
if not any(w in a.lower() for w in ["check", "inspect", "replace", "verify", "measure",
"recalibrate", "flush", "update", "pull", "tell"]):
score -= 3
return max(score, 0)
def quality_score_llm(q, a):
import anthropic
from pydantic import BaseModel
class Score(BaseModel):
score: int = Field(ge=0, le=10)
reason: str
client = anthropic.Anthropic()
response = client.messages.parse(
model="claude-opus-4-8", max_tokens=200, thinking={"type": "adaptive"},
messages=[{"role": "user", "content":
"Rate this Q&A pair 0-10 as fine-tuning data for a maintenance assistant. "
"Penalize vague answers, missing steps, and questions with no context.\n\n"
f"Q: {q}\nA: {a}"}],
output_format=Score)
return response.parsed_output.score
scorer = quality_score_dry if "--llm-score" not in sys.argv else quality_score_llm
quality = [p for p in keep if scorer(p["question"], p["answer"]) >= 6]
counts["after_quality_gate"] = len(quality)
Stage 4: PII scrub. Regex out emails and phone numbers; because reporter names travel with contact details in this corpus and name-detection by regex is unreliable, the policy here is drop-the-row when contact PII is found rather than pretend a partial scrub is clean. Write that policy decision down — scrub-and-keep vs drop is a judgment call you should be able to defend per corpus.
EMAIL = re.compile(r"[\w.+-]+@[\w-]+\.[\w.]+")
PHONE = re.compile(r"(\+?\d[\d\s-]{7,}\d)")
scrubbed = []
for p in quality:
q, a = EMAIL.sub("[EMAIL]", p["question"]), EMAIL.sub("[EMAIL]", p["answer"])
q, a = PHONE.sub("[PHONE]", q), PHONE.sub("[PHONE]", a)
if "[EMAIL]" in q + a or "[PHONE]" in q + a:
continue # names travel with contact details here; drop the whole row
scrubbed.append({**p, "question": q, "answer": a})
counts["after_pii_scrub"] = len(scrubbed)
Stage 5: chat-format validation. Emit provider-standard chat JSONL and validate every row with Pydantic before it can ship. On this corpus the gate catches zero rows — earlier stages already removed the junk — and that is fine: format validation is the belt-and-braces stage that catches scrub artifacts and future generator changes, and it costs nothing.
class ChatMessage(BaseModel):
role: str = Field(pattern="^(user|assistant)$")
content: str = Field(min_length=1)
class TrainingRow(BaseModel):
messages: list[ChatMessage] = Field(min_length=2, max_length=2)
valid_rows = []
for p in scrubbed:
row = {"messages": [{"role": "user", "content": p["question"]},
{"role": "assistant", "content": p["answer"]}]}
try:
TrainingRow.model_validate(row); valid_rows.append((p, row))
except ValidationError:
pass
counts["after_format_validation"] = len(valid_rows)
Stage 6: leakage-checked split. Split 80/10/10, then embed the questions and drop any val/test row whose question is ≥ 0.95 similar to any train question. This is the safety net behind the dedup stages: if anything semantically identical survived into both train and eval, the eval is measuring memorization.
random.shuffle(valid_rows)
n = len(valid_rows)
train = valid_rows[:int(0.8 * n)]
val = valid_rows[int(0.8 * n):int(0.9 * n)]
test = valid_rows[int(0.9 * n):]
tr_emb = model.encode([p["question"] for p, _ in train],
normalize_embeddings=True, show_progress_bar=False)
leaked = 0
def strip_leaks(split, name):
global leaked
clean = []
q_emb = model.encode([p["question"] for p, _ in split],
normalize_embeddings=True, show_progress_bar=False)
sims = q_emb @ tr_emb.T
for k, (p, row) in enumerate(split):
if sims[k].max() >= 0.95: leaked += 1
else: clean.append((p, row))
print(f"{name}: {len(split)} -> {len(clean)} after leakage check")
return clean
val = strip_leaks(val, "val"); test = strip_leaks(test, "test")
counts["train"], counts["val"], counts["test"] = len(train), len(val), len(test)
counts["leaked_eval_rows_dropped"] = leaked
for name, split in [("train", train), ("val", val), ("test", test)]:
pathlib.Path(f"{name}.jsonl").write_text("\n".join(json.dumps(r) for _, r in split))
print(json.dumps(counts, indent=2))
Run the healthy pipeline and read the funnel. python factory.py. Expected counts: raw 220 → exact dedup 186 → near-dedup 133 → quality 110 → PII 80 → format 80 → train 64 / val 8 / test 8, leaked 0. Sanity-check the funnel against the seeds: exact dedup removed 34 (25 seeded exact dups plus 9 collisions among the templated junk pairs — duplicates exist in junk too); near-dedup collapsed the paraphrase clusters and leak probes; quality removed the surviving junk; PII dropped exactly 30. Note one subtlety for your datasheet: dedup keeps the first occurrence in shuffled order, so a surviving row may carry a near_dup or exact_dup tag while its clean twin was dropped — clusters collapse to one representative, and 80 semantic survivors is the number that matters.
Break the pipeline to prove the safety net. python factory.py --skip-near-dedup. Expected: 186 rows sail through to the split, and the leakage check now fires — val: 13 -> 3, test: 13 -> 4, leaked_eval_rows_dropped: 19. Put both runs side by side in the datasheet. This is the money exhibit: without the leakage check, those 19 near-identical eval rows would have scored your fine-tune against its own training data.
Generate the RAGAS-compatible synthetic eval set. Create make_evalset.py: for each of 20 sampled clean training pairs, produce a RAGAS-schema row — {"question", "ground_truth", "contexts": [answer_source_text]} — where the question is a fresh paraphrase (dry-run: template rewording, e.g. "Recommended procedure when the {system} has {problem}?"; API mode: client.messages.parse with a Pydantic model asking for a reworded question a real technician might type). Validate the output rows with Pydantic and run the same ≥ 0.95 leakage check against train.jsonl questions — synthetic eval sets leak too, and generated paraphrases sit close to their sources by construction, so expect to drop some and report how many. Note in the datasheet that the production-scale equivalent is the RAGAS TestsetGenerator (and distilabel for instruction data); the schema you emitted is what those tools consume and produce.
Write the datasheet. datasheet.md is the artifact reviewers actually read: the stage-by-stage funnel table for both runs (healthy and --skip-near-dedup), the threshold justifications (0.92 near-dup, 0.95 leakage, quality floor 6, PII drop-not-scrub policy), a 5-row hand audit of train.jsonl (read them; do they teach what you want taught?), and the synthetic eval-set counts with its leakage-check result.
Leakage checks compare eval rows against train — never the reverse direction, and never before splitting. A dedup pass before the split reduces leakage; only the post-split check proves it, and only if it runs on the semantic (embedding) level rather than exact match.
leaked_eval_rows_dropped: 19 with val and test visibly shrunk.train.jsonl rows all parse as two-message chat format; no row contains an @ email or a phone-shaped digit run.question, ground_truth, and non-empty contexts, and its leakage-drop count is reported.factory.py produces byte-identical splits (determinism: seeds fixed, first-occurrence dedup stable).A de-lab-15/ folder containing make_messy_corpus.py, factory.py, make_evalset.py, the three split files, evalset.jsonl, and datasheet.md with the dual funnel table, threshold justifications, hand-audit notes, and synthetic-set report. Self-grading rubric (pass = 8/10):
question + " " + answer.min_length validation never fires: correct on this corpus (see step 4's note); prove the gate works by injecting one empty-content row manually.random call or set iteration — everything random must flow from random.seed(15).Seeded composition: 80 unique clean pairs, 25 exact duplicates, 40 near-duplicate paraphrases (question reworded, answer identical; similarity to twin ≥ 0.95), 30 PII rows (Faker names, emails, phones in both question and answer), 35 junk rows (answers under 8 words or content-free, some questions of the form "steam boiler???"), 10 leak probes (clean question + " Please advise.", identical answer — similarity ≈ 0.99). Healthy pipeline: exact dedup −34 (25 seeded + 9 junk-template collisions), near-dedup −53 (paraphrases, probes, junk clusters), quality −23 (remaining junk), PII −30, format −0, split 64/8/8 with 0 leaks. Broken (--skip-near-dedup): the 40+10 near-dup/probe rows survive to the split, and the question-level 0.95 check catches every val/test row whose twin landed in train — 19 rows in this seed. If your broken run catches ~0, the leakage check is comparing the wrong field or running pre-split.
The factory is entirely local in dry-run mode — corpus, embeddings, scrubbing, and splits never leave the machine — which is precisely the posture fine-tuning data prep must have on client engagements: this pipeline is the PII boundary, so it runs inside the client's environment before any external service sees a byte. The stage order encodes the compliance rule — scrub and drop PII before any LLM-based quality scoring, so contaminated rows are never transmitted even to an approved model endpoint. Under DPDP or GDPR, the datasheet doubles as processing documentation: what was removed, why, at what threshold, and with counts — exactly what a client DPO asks for when you propose fine-tuning on their support logs.