Fine-tuning open source models

My opinion is that fine tuning is only worth it when you can name the win you want, the risk you accept, and the test you will run before anyone depends on it.

Fine-tuning is not always the answer. Let me be direct about when it makes sense.

Data quality matters more than quantity. A small, high-quality dataset beats a large, noisy one.

"""
Dataset Preparation for Fine-Tuning
====================================
How to structure training data for instruction fine-tuning.
"""

import json
from typing import List, Dict


def create_training_example(
    instruction: str,
    input_text: str,
    output: str,
    system_prompt: str = ""
) -> Dict:
    """
    Create a single training example in the standard format.
    
    Args:
        instruction: What you want the model to do
        input_text: Context or input data (can be empty)
        output: The desired response
        system_prompt: Optional system context
        
    Returns:
        Dictionary in training format
    """
    example = {
        "instruction": instruction,
        "input": input_text,
        "output": output,
    }
    
    if system_prompt:
        example["system"] = system_prompt
    
    return example


# Example: Creating a dataset for customer support fine-tuning

support_examples = [
    create_training_example(
        instruction="Respond to this customer query professionally.",
        input_text="I have not received my order after 2 weeks.",
        output="""I apologise for the delay with your order. Let me look into this for you straightaway.

Could you please provide your order number? I will check the shipping status and arrange a resolution. If the order has been lost, I will ensure we send a replacement or process a full refund.

Thank you for your patience.""",
        system_prompt="You are a helpful customer support agent. Be empathetic and solution-focused."
    ),
    create_training_example(
        instruction="Respond to this customer query professionally.",
        input_text="Your product stopped working after one day.",
        output="""I am sorry to hear your product is not working as expected. That is certainly not the experience we want you to have.

For a product that failed within 24 hours, you are entitled to a full refund or immediate replacement. Which would you prefer?

If you would like a replacement, I can arrange express delivery at no extra cost. Please let me know which option works best for you.""",
        system_prompt="You are a helpful customer support agent. Be empathetic and solution-focused."
    ),
]


def save_dataset(examples: List[Dict], filepath: str):
    """Save dataset in JSONL format (one JSON object per line)."""
    with open(filepath, "w") as f:
        for example in examples:
            f.write(json.dumps(example) + "\n")


def validate_dataset(filepath: str) -> Dict:
    """
    Validate a training dataset.
    
    Returns statistics and any issues found.
    """
    stats = {
        "total_examples": 0,
        "avg_instruction_length": 0,
        "avg_output_length": 0,
        "issues": []
    }
    
    instruction_lengths = []
    output_lengths = []
    
    with open(filepath, "r") as f:
        for line_num, line in enumerate(f, 1):
            try:
                example = json.loads(line)
                stats["total_examples"] += 1
                
                # Check required fields
                if "instruction" not in example:
                    stats["issues"].append(f"Line {line_num}: Missing instruction")
                if "output" not in example:
                    stats["issues"].append(f"Line {line_num}: Missing output")
                
                # Track lengths
                instruction_lengths.append(len(example.get("instruction", "")))
                output_lengths.append(len(example.get("output", "")))
                
                # Check for very short outputs (likely low quality)
                if len(example.get("output", "")) < 50:
                    stats["issues"].append(f"Line {line_num}: Very short output")
                    
            except json.JSONDecodeError:
                stats["issues"].append(f"Line {line_num}: Invalid JSON")
    
    if instruction_lengths:
        stats["avg_instruction_length"] = sum(instruction_lengths) / len(instruction_lengths)
    if output_lengths:
        stats["avg_output_length"] = sum(output_lengths) / len(output_lengths)
    
    return stats