Fine-Tuning Data Preparation

Prepare, validate, and optimize your training data for fine-tuning.

Overview

Quality training data is the most important factor in fine-tuning success. This guide covers data formats, validation, token counting, and cost estimation.

┌─────────────────────────────────────────────────────────────────────┐
│                    DATA PREPARATION PIPELINE                         │
├─────────────────────────────────────────────────────────────────────┤
│                                                                      │
│  1. COLLECT          2. FORMAT          3. VALIDATE      4. UPLOAD  │
│  ┌─────────┐        ┌─────────┐        ┌─────────┐      ┌─────────┐│
│  │Raw Data │───────►│ JSONL   │───────►│ Check   │─────►│ OpenAI  ││
│  │Examples │        │ Format  │        │ Quality │      │  Files  ││
│  └─────────┘        └─────────┘        └─────────┘      └─────────┘│
│                                                                      │
│  DATA REQUIREMENTS:                                                 │
│  ├─► Minimum: 10 examples (50-100+ recommended)                     │
│  ├─► Format: JSONL with messages array                              │
│  ├─► Quality: Diverse, representative, error-free                   │
│  └─► Size: <512MB per file                                          │
│                                                                      │
└─────────────────────────────────────────────────────────────────────┘

Data Format

Chat Format (Required)

# Each line in JSONL file must have this structure
{
    "messages": [
        {"role": "system", "content": "System prompt here"},
        {"role": "user", "content": "User input here"},
        {"role": "assistant", "content": "Ideal assistant response here"}
    ]
}

Multi-Turn Conversations

# Multi-turn example
{
    "messages": [
        {"role": "system", "content": "You are a coding assistant."},
        {"role": "user", "content": "How do I read a file in Python?"},
        {"role": "assistant", "content": "You can use open():\n```python\nwith open('file.txt', 'r') as f:\n    content = f.read()\n```"},
        {"role": "user", "content": "What if I want to read line by line?"},
        {"role": "assistant", "content": "Use readlines() or iterate:\n```python\nwith open('file.txt', 'r') as f:\n    for line in f:\n        print(line.strip())\n```"}
    ]
}

With Function Calling

# Training with tools
{
    "messages": [
        {"role": "system", "content": "You help users with weather information."},
        {"role": "user", "content": "What's the weather in Paris?"},
        {
            "role": "assistant",
            "content": null,
            "tool_calls": [
                {
                    "id": "call_abc123",
                    "type": "function",
                    "function": {
                        "name": "get_weather",
                        "arguments": "{\"location\": \"Paris\", \"unit\": \"celsius\"}"
                    }
                }
            ]
        },
        {
            "role": "tool",
            "tool_call_id": "call_abc123",
            "content": "{\"temperature\": 18, \"condition\": \"cloudy\"}"
        },
        {"role": "assistant", "content": "The weather in Paris is currently 18°C and cloudy."}
    ],
    "tools": [
        {
            "type": "function",
            "function": {
                "name": "get_weather",
                "description": "Get current weather",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "location": {"type": "string"},
                        "unit": {"type": "string", "enum": ["celsius", "fahrenheit"]}
                    },
                    "required": ["location"]
                }
            }
        }
    ]
}

Data Validation

Comprehensive Validator

import json
import tiktoken
from typing import Optional
from dataclasses import dataclass, field

@dataclass
class ValidationResult:
    is_valid: bool
    total_examples: int
    valid_examples: int
    errors: list[str] = field(default_factory=list)
    warnings: list[str] = field(default_factory=list)
    stats: dict = field(default_factory=dict)

def validate_training_file(filepath: str, model: str = "gpt-4o-mini") -> ValidationResult:
    """Comprehensive validation of training data."""

    result = ValidationResult(
        is_valid=True,
        total_examples=0,
        valid_examples=0,
        stats={
            "token_counts": [],
            "message_counts": [],
            "roles": {"system": 0, "user": 0, "assistant": 0}
        }
    )

    enc = tiktoken.encoding_for_model(model)

    with open(filepath, encoding="utf-8") as f:
        for line_num, line in enumerate(f, 1):
            result.total_examples += 1
            line = line.strip()

            if not line:
                result.warnings.append(f"Line {line_num}: Empty line")
                continue

            # Parse JSON
            try:
                entry = json.loads(line)
            except json.JSONDecodeError as e:
                result.errors.append(f"Line {line_num}: Invalid JSON - {e}")
                result.is_valid = False
                continue

            # Validate structure
            if "messages" not in entry:
                result.errors.append(f"Line {line_num}: Missing 'messages' field")
                result.is_valid = False
                continue

            messages = entry["messages"]

            if not isinstance(messages, list):
                result.errors.append(f"Line {line_num}: 'messages' must be a list")
                result.is_valid = False
                continue

            if len(messages) < 2:
                result.errors.append(f"Line {line_num}: Need at least 2 messages")
                result.is_valid = False
                continue

            # Validate each message
            valid_entry = True
            total_tokens = 0
            has_assistant = False

            for msg_num, msg in enumerate(messages):
                if not isinstance(msg, dict):
                    result.errors.append(f"Line {line_num}, Msg {msg_num}: Not a dict")
                    valid_entry = False
                    continue

                if "role" not in msg:
                    result.errors.append(f"Line {line_num}, Msg {msg_num}: Missing 'role'")
                    valid_entry = False
                    continue

                role = msg["role"]
                if role not in ["system", "user", "assistant", "tool"]:
                    result.errors.append(f"Line {line_num}, Msg {msg_num}: Invalid role '{role}'")
                    valid_entry = False
                    continue

                result.stats["roles"][role] = result.stats["roles"].get(role, 0) + 1

                if role == "assistant":
                    has_assistant = True

                # Count tokens
                content = msg.get("content", "")
                if content:
                    total_tokens += len(enc.encode(content))

            if not has_assistant:
                result.errors.append(f"Line {line_num}: No assistant message")
                valid_entry = False

            if valid_entry:
                result.valid_examples += 1
                result.stats["token_counts"].append(total_tokens)
                result.stats["message_counts"].append(len(messages))

            if not valid_entry:
                result.is_valid = False

    # Calculate summary stats
    if result.stats["token_counts"]:
        result.stats["avg_tokens"] = sum(result.stats["token_counts"]) / len(result.stats["token_counts"])
        result.stats["max_tokens"] = max(result.stats["token_counts"])
        result.stats["min_tokens"] = min(result.stats["token_counts"])
        result.stats["total_tokens"] = sum(result.stats["token_counts"])

    return result

# Usage
result = validate_training_file("training.jsonl")
print(f"Valid: {result.is_valid}")
print(f"Examples: {result.valid_examples}/{result.total_examples}")
print(f"Total tokens: {result.stats.get('total_tokens', 0)}")

if result.errors:
    print("\nErrors:")
    for error in result.errors[:10]:
        print(f"  - {error}")

Quick Validation Script

def quick_validate(filepath: str) -> bool:
    """Quick validation check."""

    errors = []

    with open(filepath) as f:
        for i, line in enumerate(f, 1):
            try:
                data = json.loads(line)
                assert "messages" in data, "Missing messages"
                assert len(data["messages"]) >= 2, "Need 2+ messages"
                assert any(m.get("role") == "assistant" for m in data["messages"]), "No assistant"
            except Exception as e:
                errors.append(f"Line {i}: {e}")
                if len(errors) >= 5:
                    break

    if errors:
        print("Validation failed:")
        for e in errors:
            print(f"  {e}")
        return False

    print("✅ Validation passed")
    return True

Token Counting and Cost Estimation

import tiktoken

def estimate_training_cost(
    filepath: str,
    model: str = "gpt-4o-mini",
    n_epochs: int = 3
) -> dict:
    """Estimate fine-tuning cost."""

    # Token pricing (as of 2024)
    TRAINING_COST_PER_1M = {
        "gpt-4o-mini": 3.00,
        "gpt-4o": 25.00,
    }

    enc = tiktoken.encoding_for_model(model)
    total_tokens = 0
    num_examples = 0

    with open(filepath) as f:
        for line in f:
            if line.strip():
                data = json.loads(line)
                for msg in data.get("messages", []):
                    content = msg.get("content", "")
                    if content:
                        total_tokens += len(enc.encode(content))
                num_examples += 1

    # Training uses each example n_epochs times
    training_tokens = total_tokens * n_epochs

    # Calculate cost
    cost_per_1m = TRAINING_COST_PER_1M.get(model, 3.00)
    estimated_cost = (training_tokens / 1_000_000) * cost_per_1m

    return {
        "num_examples": num_examples,
        "total_tokens": total_tokens,
        "tokens_per_epoch": total_tokens,
        "n_epochs": n_epochs,
        "training_tokens": training_tokens,
        "estimated_cost_usd": round(estimated_cost, 2),
        "model": model
    }

# Usage
cost = estimate_training_cost("training.jsonl", n_epochs=3)
print(f"Examples: {cost['num_examples']}")
print(f"Total tokens: {cost['total_tokens']:,}")
print(f"Training tokens ({cost['n_epochs']} epochs): {cost['training_tokens']:,}")
print(f"Estimated cost: ${cost['estimated_cost_usd']}")

Data Conversion Utilities

From CSV

import csv

def csv_to_training_jsonl(
    csv_path: str,
    output_path: str,
    input_col: str,
    output_col: str,
    system_prompt: str = ""
):
    """Convert CSV to training JSONL."""

    with open(csv_path) as csv_file, open(output_path, "w") as out_file:
        reader = csv.DictReader(csv_file)

        for row in reader:
            messages = []

            if system_prompt:
                messages.append({"role": "system", "content": system_prompt})

            messages.append({"role": "user", "content": row[input_col]})
            messages.append({"role": "assistant", "content": row[output_col]})

            out_file.write(json.dumps({"messages": messages}) + "\n")

# Usage
csv_to_training_jsonl(
    "data.csv",
    "training.jsonl",
    input_col="question",
    output_col="answer",
    system_prompt="You are a helpful assistant."
)

From DataFrame

import pandas as pd

def df_to_training_jsonl(
    df: pd.DataFrame,
    output_path: str,
    input_col: str,
    output_col: str,
    system_col: str = None
):
    """Convert DataFrame to training JSONL."""

    with open(output_path, "w") as f:
        for _, row in df.iterrows():
            messages = []

            if system_col and pd.notna(row.get(system_col)):
                messages.append({"role": "system", "content": str(row[system_col])})

            messages.append({"role": "user", "content": str(row[input_col])})
            messages.append({"role": "assistant", "content": str(row[output_col])})

            f.write(json.dumps({"messages": messages}) + "\n")

From Existing Conversations

def conversations_to_training(
    conversations: list[list[dict]],
    output_path: str,
    system_prompt: str = ""
):
    """Convert conversation logs to training format."""

    with open(output_path, "w") as f:
        for conv in conversations:
            messages = []

            if system_prompt:
                messages.append({"role": "system", "content": system_prompt})

            for turn in conv:
                messages.append({
                    "role": turn["role"],
                    "content": turn["content"]
                })

            f.write(json.dumps({"messages": messages}) + "\n")

Data Quality Guidelines

Diversity Checklist

def analyze_data_diversity(filepath: str) -> dict:
    """Analyze training data diversity."""

    analysis = {
        "unique_system_prompts": set(),
        "input_lengths": [],
        "output_lengths": [],
        "topics": [],  # Would need NLP to extract
    }

    with open(filepath) as f:
        for line in f:
            data = json.loads(line)
            messages = data.get("messages", [])

            for msg in messages:
                content = msg.get("content", "")
                if msg["role"] == "system":
                    analysis["unique_system_prompts"].add(content[:100])
                elif msg["role"] == "user":
                    analysis["input_lengths"].append(len(content.split()))
                elif msg["role"] == "assistant":
                    analysis["output_lengths"].append(len(content.split()))

    return {
        "num_unique_system_prompts": len(analysis["unique_system_prompts"]),
        "avg_input_words": sum(analysis["input_lengths"]) / max(len(analysis["input_lengths"]), 1),
        "avg_output_words": sum(analysis["output_lengths"]) / max(len(analysis["output_lengths"]), 1),
        "input_word_range": (min(analysis["input_lengths"], default=0), max(analysis["input_lengths"], default=0)),
        "output_word_range": (min(analysis["output_lengths"], default=0), max(analysis["output_lengths"], default=0)),
    }

Quality Recommendations

Aspect	Recommendation
Minimum examples	50-100 for basic fine-tuning
Diversity	Cover all expected use cases
Balance	Avoid overrepresenting any category
Length	Match expected production lengths
Quality	Each example should be production-ready
Consistency	Use consistent formatting throughout

Train/Validation Split

import random

def split_training_data(
    filepath: str,
    train_path: str,
    val_path: str,
    val_ratio: float = 0.1
):
    """Split data into training and validation sets."""

    with open(filepath) as f:
        lines = f.readlines()

    random.shuffle(lines)

    split_idx = int(len(lines) * (1 - val_ratio))
    train_lines = lines[:split_idx]
    val_lines = lines[split_idx:]

    with open(train_path, "w") as f:
        f.writelines(train_lines)

    with open(val_path, "w") as f:
        f.writelines(val_lines)

    print(f"Training: {len(train_lines)} examples")
    print(f"Validation: {len(val_lines)} examples")

# Usage
split_training_data(
    "all_data.jsonl",
    "train.jsonl",
    "val.jsonl",
    val_ratio=0.1
)

Upload to OpenAI

from openai import OpenAI

client = OpenAI()

def upload_training_file(filepath: str) -> str:
    """Upload training file to OpenAI."""

    # Validate first
    result = validate_training_file(filepath)
    if not result.is_valid:
        raise ValueError(f"Validation failed: {result.errors[:5]}")

    # Upload
    with open(filepath, "rb") as f:
        file = client.files.create(file=f, purpose="fine-tune")

    print(f"Uploaded: {file.id}")
    print(f"Status: {file.status}")
    print(f"Size: {file.bytes} bytes")

    return file.id

# Usage
file_id = upload_training_file("training.jsonl")

See Also

• Fine-Tuning Techniques
• Evaluation Flywheel
• SOP-002: Dataset Preparation