Data Preparation for Fine-Tuning

# Alpaca format: instruction, input (optional), output
alpaca_examples = [
 {
 "instruction": "Summarize the following research paper abstract.",
 "input": "We present a novel approach to protein structure prediction "
 "using graph neural networks. Our method achieves state-of-the-art "
 "results on CASP14 benchmarks, improving GDT-TS scores by 12% over "
 "existing methods. We demonstrate that incorporating evolutionary "
 "coupling information as edge features significantly enhances "
 "prediction accuracy for proteins with low sequence identity.",
 "output": "The paper introduces a graph neural network approach for protein "
 "structure prediction that achieves a 12% improvement in GDT-TS "
 "scores on CASP14 benchmarks. The key innovation is using "
 "evolutionary coupling information as edge features, which "
 "particularly helps with low-sequence-identity proteins."
 },
 {
 "instruction": "Classify the sentiment of this product review.",
 "input": "The battery life is incredible but the screen is too dim outdoors.",
 "output": "Mixed sentiment. Positive: battery life. Negative: screen brightness."
 },
 {
 # No input field: instruction-only format
 "instruction": "What are three common causes of overfitting in neural networks?",
 "output": "1. Training for too many epochs without early stopping.\n"
 "2. Using a model with excessive capacity relative to the dataset size.\n"
 "3. Insufficient regularization (no dropout, weight decay, or data augmentation)."
 }
]

# Save as JSONL (one JSON object per line)
import json

with open("train_alpaca.jsonl", "w") as f:
 for example in alpaca_examples:
 f.write(json.dumps(example) + "\n")

# ChatML / Messages format (OpenAI-compatible)
chatml_examples = [
 {
 "messages": [
 {
 "role": "system",
 "content": "You are a medical coding assistant. Assign ICD-10 codes "
 "to clinical descriptions. Always provide the code and a "
 "brief justification."
 },
 {
 "role": "user",
 "content": "Patient presents with acute lower back pain radiating to "
 "the left leg, worse with sitting. MRI shows L4-L5 disc "
 "herniation with nerve root compression."
 },
 {
 "role": "assistant",
 "content": "ICD-10 Code: M51.16\n\n"
 "Justification: M51.16 covers intervertebral disc "
 "degeneration with radiculopathy in the lumbar region. The "
 "radiating leg pain indicates radiculopathy, and the MRI "
 "confirms L4-L5 disc herniation with nerve compression, "
 "which falls under this code."
 }
 ]
 }
]

# Validate the format
def validate_chatml(examples):
 """Validate ChatML format for common issues."""
 issues = []
 for i, ex in enumerate(examples):
 messages = ex.get("messages", [])
 if not messages:
 issues.append(f"Example {i}: empty messages")
 continue

 # Check role sequence
 roles = [m["role"] for m in messages]
 if roles[-1] != "assistant":
 issues.append(f"Example {i}: last message must be from assistant")

 # Check for empty content
 for j, msg in enumerate(messages):
 if not msg.get("content", "").strip():
 issues.append(f"Example {i}, message {j}: empty content")

 return issues

issues = validate_chatml(chatml_examples)
print(f"Validation: {'PASSED' if not issues else issues}")

# Load a pre-trained tokenizer and configure padding/truncation
# Consistent tokenization is critical for reproducible fine-tuning
from transformers import AutoTokenizer

# Load a tokenizer with its built-in chat template
tokenizer = AutoTokenizer.from_pretrained("meta-llama/Llama-3.1-8B-Instruct")

messages = [
 {"role": "system", "content": "You are a helpful coding assistant."},
 {"role": "user", "content": "Write a Python function to reverse a string."},
 {"role": "assistant", "content": "def reverse_string(s):\n return s[::-1]"}
]

# Apply the chat template to see the raw text
formatted = tokenizer.apply_chat_template(
 messages,
 tokenize=False, # Return string, not token IDs
 add_generation_prompt=False # Training: no trailing prompt
)
print(formatted)

# For training, we typically want tokenized output with labels
tokenized = tokenizer.apply_chat_template(
 messages,
 tokenize=True,
 return_dict=True,
 return_tensors="pt"
)
print(f"\nToken count: {tokenized['input_ids'].shape[1]}")

# Load and preprocess a dataset from Hugging Face Hub
# Map the tokenization function across all splits for training
from datasets import Dataset, concatenate_datasets
import random

def mix_datasets(
 datasets: dict, # {"task_name": Dataset}
 target_ratios: dict, # {"task_name": 0.3}
 total_size: int = None, # Target total size
 seed: int = 42
) -> Dataset:
 """Mix multiple datasets with specified ratios."""
 random.seed(seed)

 if total_size is None:
 # Default: use the smallest dataset scaled by its ratio
 min_effective = min(
 len(ds) / ratio
 for ds, ratio in zip(datasets.values(), target_ratios.values())
 )
 total_size = int(min_effective)

 mixed_parts = []
 for task_name, dataset in datasets.items():
 ratio = target_ratios[task_name]
 n_samples = int(total_size * ratio)

 if n_samples <= len(dataset):
 # Subsample
 indices = random.sample(range(len(dataset)), n_samples)
 sampled = dataset.select(indices)
 else:
 # Oversample (repeat with shuffling)
 repeats = n_samples // len(dataset) + 1
 indices = list(range(len(dataset))) * repeats
 random.shuffle(indices)
 sampled = dataset.select(indices[:n_samples])

 mixed_parts.append(sampled)
 print(f" {task_name}: {len(dataset)} -> {n_samples} samples ({ratio:.0%})")

 # Concatenate and shuffle
 combined = concatenate_datasets(mixed_parts)
 combined = combined.shuffle(seed=seed)

 print(f"\nTotal mixed dataset: {len(combined)} examples")
 return combined

# pip install datasets transformers
from datasets import load_dataset
from transformers import AutoTokenizer

tokenizer = AutoTokenizer.from_pretrained("meta-llama/Llama-3.1-8B-Instruct")

# Load a popular instruction dataset from the Hub
dataset = load_dataset("HuggingFaceH4/no_robots", split="train")
print(f"Loaded {len(dataset)} examples with columns: {dataset.column_names}")

# Filter: keep only examples where the response exceeds 20 tokens
def is_long_enough(example):
 return len(tokenizer.encode(example["messages"][-1]["content"])) > 20

dataset = dataset.filter(is_long_enough)
print(f"After filtering: {len(dataset)} examples")

# Map: apply the chat template to each example
def apply_template(example):
 example["text"] = tokenizer.apply_chat_template(
 example["messages"], tokenize=False, add_generation_prompt=False
 )
 return example

dataset = dataset.map(apply_template, remove_columns=["messages"])
print(f"First example (truncated): {dataset[0]['text'][:200]}...")

# Define typed configuration for the fine-tuning data pipeline
# Centralize hyperparameters for dataset preparation and formatting
from typing import List, Dict
import numpy as np

def pack_sequences(
 examples: List[Dict],
 tokenizer,
 max_seq_length: int = 2048,
 pad_token_id: int = None
) -> List[Dict]:
 """Pack multiple examples into fixed-length sequences."""
 if pad_token_id is None:
 pad_token_id = tokenizer.pad_token_id or tokenizer.eos_token_id

 packed = []
 current_input_ids = []
 current_attention_mask = []
 current_labels = []

 for example in examples:
 tokens = tokenizer(
 example["text"],
 truncation=True,
 max_length=max_seq_length,
 add_special_tokens=True
 )
 example_ids = tokens["input_ids"]

 # Check if adding this example would exceed max length
 if len(current_input_ids) + len(example_ids) > max_seq_length:
 # Pad the current sequence and save it
 pad_length = max_seq_length - len(current_input_ids)
 current_input_ids.extend([pad_token_id] * pad_length)
 current_attention_mask.extend([0] * pad_length)
 current_labels.extend([-100] * pad_length)

 packed.append({
 "input_ids": current_input_ids,
 "attention_mask": current_attention_mask,
 "labels": current_labels
 })

 # Start a new packed sequence
 current_input_ids = []
 current_attention_mask = []
 current_labels = []

 # Add this example to the current sequence
 current_input_ids.extend(example_ids)
 current_attention_mask.extend([1] * len(example_ids))
 current_labels.extend(example_ids) # Causal LM: labels = input_ids

 # Save the last sequence if non-empty
 if current_input_ids:
 pad_length = max_seq_length - len(current_input_ids)
 current_input_ids.extend([pad_token_id] * pad_length)
 current_attention_mask.extend([0] * pad_length)
 current_labels.extend([-100] * pad_length)
 packed.append({
 "input_ids": current_input_ids,
 "attention_mask": current_attention_mask,
 "labels": current_labels
 })

 return packed

# Calculate efficiency improvement
def packing_efficiency(lengths: List[int], max_length: int) -> dict:
 """Compare padding waste vs. packing efficiency."""
 # Without packing: pad each to max_length
 padded_tokens = len(lengths) * max_length
 useful_tokens_padded = sum(lengths)
 pad_efficiency = useful_tokens_padded / padded_tokens

 # With packing: fit multiple examples per sequence
 packed_sequences = 0
 current_length = 0
 for length in sorted(lengths):
 if current_length + length > max_length:
 packed_sequences += 1
 current_length = 0
 current_length += length
 if current_length > 0:
 packed_sequences += 1

 packed_tokens = packed_sequences * max_length
 pack_efficiency = useful_tokens_padded / packed_tokens

 return {
 "sequences_without_packing": len(lengths),
 "sequences_with_packing": packed_sequences,
 "efficiency_without_packing": f"{pad_efficiency:.1%}",
 "efficiency_with_packing": f"{pack_efficiency:.1%}",
 "speedup": f"{len(lengths) / packed_sequences:.1f}x"
 }

# Example with realistic distribution
np.random.seed(42)
lengths = np.random.lognormal(mean=5.5, sigma=0.8, size=10000).astype(int)
lengths = np.clip(lengths, 50, 2048)
result = packing_efficiency(lengths.tolist(), max_length=2048)
for k, v in result.items():
 print(f" {k}: {v}")

# Define the data_quality_audit function
# This handles the core processing logic
def data_quality_audit(dataset, tokenizer, max_seq_length=2048):
 """Run a comprehensive data quality audit before training."""
 report = {
 "total_examples": len(dataset),
 "issues": [],
 "warnings": [],
 "stats": {}
 }

 lengths = []
 empty_count = 0
 duplicate_count = 0
 seen_hashes = set()

 for i, example in enumerate(dataset):
 messages = example.get("messages", [])

 # Check for empty messages
 for msg in messages:
 if not msg.get("content", "").strip():
 empty_count += 1

 # Check for duplicates (hash-based)
 content_hash = hash(str(messages))
 if content_hash in seen_hashes:
 duplicate_count += 1
 seen_hashes.add(content_hash)

 # Tokenize and check length
 text = tokenizer.apply_chat_template(messages, tokenize=False)
 tokens = tokenizer(text)["input_ids"]
 lengths.append(len(tokens))

 # Check for truncation
 if len(tokens) > max_seq_length:
 report["warnings"].append(
 f"Example {i}: {len(tokens)} tokens (will be truncated)"
 )

 # Summary statistics
 import numpy as np
 lengths = np.array(lengths)
 report["stats"] = {
 "mean_length": f"{lengths.mean():.0f}",
 "median_length": f"{np.median(lengths):.0f}",
 "p95_length": f"{np.percentile(lengths, 95):.0f}",
 "max_length": f"{lengths.max():.0f}",
 "truncated_pct": f"{(lengths > max_seq_length).mean():.1%}",
 "empty_messages": empty_count,
 "duplicates": duplicate_count,
 }

 # Flag issues
 if empty_count > 0:
 report["issues"].append(f"{empty_count} empty messages found")
 if duplicate_count > len(dataset) * 0.05:
 report["issues"].append(f"{duplicate_count} duplicates ({duplicate_count/len(dataset):.1%})")
 if (lengths > max_seq_length).mean() > 0.1:
 report["issues"].append(f"{(lengths > max_seq_length).mean():.1%} examples will be truncated")

 return report