Training Data Pipeline

Training Data Pipeline

When to Use This Skill

Use this skill when you need to:

• Format production logs into SFT training data (API logs, user corrections, accept/reject signals)
• Distill from frontier models using batch APIs (OpenAI, Anthropic) to label production inputs
• Bootstrap synthetic data when fewer than 1000 real examples exist
• Validate data quality before training (dedup, schema check, diversity metrics)
• Split data into train/eval sets with production data reserved for evaluation

Three Data Paths

Always prefer Path A — production data is the moat competitors can't replicate.

JSONL Chat Format

All training platforms (Tinker, Unsloth, TRL, Axolotl) accept this standard chat format:

{"messages": [{"role": "system", "content": "You are a helpful assistant."}, {"role": "user", "content": "What is 2+2?"}, {"role": "assistant", "content": "4"}]}
{"messages": [{"role": "user", "content": "Translate to French: Hello"}, {"role": "assistant", "content": "Bonjour"}]}

Format Rules

• One JSON object per line, no trailing commas
• messages array with role and content fields
• Roles: system (optional, first only), user, assistant (alternating)
• Multi-turn: alternate user/assistant pairs within a single messages array
• UTF-8 encoding, no BOM
• assistant messages are the training targets — everything else is context

Platform-Specific Notes

Tinker: Standard chat format above. Max 32K tokens per example. System message optional.

Unsloth/TRL: Same format. Also accepts {"prompt": "...", "completion": "..."} for simple pairs. Chat format preferred for multi-turn.

Axolotl: Supports multiple formats via config. Recommend chat_template type with standard JSONL.

Path A: Production Data Collection

From API Logs

If you log API requests/responses, convert them directly:

import json

def api_log_to_training(log_entry):
    """Convert an API request/response log to training format."""
    messages = []

    # Add system prompt if present
    if log_entry.get("system_prompt"):
        messages.append({
            "role": "system",
            "content": log_entry["system_prompt"]
        })

    # Add the user's input
    messages.append({
        "role": "user",
        "content": log_entry["user_input"]
    })

    # Add the response (use corrected version if available)
    response = log_entry.get("corrected_response") or log_entry["api_response"]
    messages.append({
        "role": "assistant",
        "content": response
    })

    return {"messages": messages}

# Process logs
with open("api_logs.jsonl") as f, open("training_data.jsonl", "w") as out:
    for line in f:
        log = json.loads(line)
        example = api_log_to_training(log)
        out.write(json.dumps(example) + "\n")

From User Corrections

User corrections (edits to model output) are the highest-quality training signal:

def correction_to_training(original_input, corrected_output, system_prompt=None):
    """Convert a user correction into a training example.
    The corrected output becomes the training target."""
    messages = []
    if system_prompt:
        messages.append({"role": "system", "content": system_prompt})
    messages.append({"role": "user", "content": original_input})
    messages.append({"role": "assistant", "content": corrected_output})
    return {"messages": messages}

From Accept/Reject Signals

If users accept or reject model outputs, use accepted outputs as positive examples:

def filter_accepted(logs):
    """Keep only examples where user accepted the output."""
    accepted = []
    for log in logs:
        if log.get("user_action") == "accepted":
            accepted.append({
                "messages": [
                    {"role": "user", "content": log["input"]},
                    {"role": "assistant", "content": log["output"]}
                ]
            })
    return accepted

Path B: Frontier Distillation

Use frontier models to label your production inputs. Best when you have real inputs but no gold labels.

OpenAI Batch API (50% discount)

import json

def create_batch_file(inputs, system_prompt, model="gpt-4o"):
    """Create a batch file for OpenAI Batch API."""
    requests = []
    for i, user_input in enumerate(inputs):
        requests.append({
            "custom_id": f"request-{i}",
            "method": "POST",
            "url": "/v1/chat/completions",
            "body": {
                "model": model,
                "messages": [
                    {"role": "system", "content": system_prompt},
                    {"role": "user", "content": user_input}
                ],
                "max_tokens": 4096
            }
        })

    with open("batch_input.jsonl", "w") as f:
        for req in requests:
            f.write(json.dumps(req) + "\n")
    return "batch_input.jsonl"

# Submit batch
# openai api batches create -i batch_input.jsonl -e /v1/chat/completions -c 24h

Anthropic Batch API

import anthropic

client = anthropic.Anthropic()

def create_anthropic_batch(inputs, system_prompt, model="claude-sonnet-4-5-20250929"):
    """Create batch request for Anthropic Message Batches API."""
    requests = []
    for i, user_input in enumerate(inputs):
        requests.append({
            "custom_id": f"request-{i}",
            "params": {
                "model": model,
                "max_tokens": 4096,
                "system": system_prompt,
                "messages": [
                    {"role": "user", "content": user_input}
                ]
            }
        })

    batch = client.messages.batches.create(requests=requests)
    return batch.id

Processing Batch Results

def batch_results_to_training(results_file, inputs, system_prompt=None):
    """Convert batch API results into training JSONL."""
    training = []
    with open(results_file) as f:
        for line in f:
            result = json.loads(line)
            idx = int(result["custom_id"].split("-")[1])
            messages = []
            if system_prompt:
                messages.append({"role": "system", "content": system_prompt})
            messages.append({"role": "user", "content": inputs[idx]})
            # Extract assistant response from batch result
            content = result["response"]["body"]["choices"][0]["message"]["content"]
            messages.append({"role": "assistant", "content": content})
            training.append({"messages": messages})

    with open("distilled_training.jsonl", "w") as f:
        for example in training:
            f.write(json.dumps(example) + "\n")
    return len(training)

Path C: Synthetic Bootstrapping

Generate training data from scratch when you have < 1000 real examples. Use as a starting point, then replace with production data as it accumulates.

Seed Prompt Strategy

import openai

client = openai.OpenAI()

def generate_synthetic_examples(task_description, seed_examples, n=500, model="gpt-4o"):
    """Generate diverse synthetic training examples from seed examples."""

    meta_prompt = f"""You are generating training data for an LLM that will be fine-tuned for:
{task_description}

Here are {len(seed_examples)} real examples of the desired behavior:
{json.dumps(seed_examples[:5], indent=2)}

Generate a NEW, diverse example. The input should cover a different scenario than
the seeds. The output should match the quality and style of the examples above.

Return JSON: {{"input": "...", "output": "..."}}"""

    examples = []
    for i in range(n):
        response = client.chat.completions.create(
            model=model,
            messages=[{"role": "user", "content": meta_prompt}],
            response_format={"type": "json_object"},
            temperature=0.9,  # High temp for diversity
        )
        example = json.loads(response.choices[0].message.content)
        examples.append({
            "messages": [
                {"role": "user", "content": example["input"]},
                {"role": "assistant", "content": example["output"]}
            ]
        })
    return examples

Diversity Strategies

• Vary temperature (0.7-1.0) across generation batches
• Use different frontier models (GPT-4o, Claude, Gemini) to reduce model-specific bias
• Seed with representative prompts from different categories/difficulty levels
• Include edge cases and adversarial examples explicitly in seed prompts

Data Quality Validation

Schema Validation

def validate_jsonl(filepath):
    """Validate JSONL training file format."""
    errors = []
    valid = 0
    with open(filepath) as f:
        for i, line in enumerate(f, 1):
            try:
                obj = json.loads(line)
            except json.JSONDecodeError as e:
                errors.append(f"Line {i}: Invalid JSON — {e}")
                continue

            if "messages" not in obj:
                errors.append(f"Line {i}: Missing 'messages' key")
                continue

            msgs = obj["messages"]
            if not isinstance(msgs, list) or len(msgs) < 2:
                errors.append(f"Line {i}: 'messages' must be a list with >= 2 entries")
                continue

            # Check roles
            has_user = any(m.get("role") == "user" for m in msgs)
            has_assistant = any(m.get("role") == "assistant" for m in msgs)
            if not has_user or not has_assistant:
                errors.append(f"Line {i}: Must have at least one user and one assistant message")
                continue

            for j, msg in enumerate(msgs):
                if "role" not in msg or "content" not in msg:
                    errors.append(f"Line {i}, message {j}: Missing 'role' or 'content'")
                elif msg["role"] not in ("system", "user", "assistant"):
                    errors.append(f"Line {i}, message {j}: Invalid role '{msg['role']}'")

            valid += 1

    return {"valid": valid, "errors": errors, "total": valid + len(errors)}

Deduplication (MinHash)

from datasketch import MinHash, MinHashLSH

def deduplicate_dataset(examples, threshold=0.8):
    """Remove near-duplicate examples using MinHash LSH."""
    lsh = MinHashLSH(threshold=threshold, num_perm=128)
    unique = []

    for i, ex in enumerate(examples):
        # Hash the assistant's response (training target)
        text = ex["messages"][-1]["content"]
        m = MinHash(num_perm=128)
        for word in text.lower().split():
            m.update(word.encode("utf-8"))

        key = f"doc-{i}"
        if not lsh.query(m):
            lsh.insert(key, m)
            unique.append(ex)

    removed = len(examples) - len(unique)
    print(f"Removed {removed} duplicates ({removed/len(examples)*100:.1f}%)")
    return unique

Diversity Metrics

from collections import Counter

def distinct_n(texts, n=2):
    """Calculate distinct-n metric (ratio of unique n-grams to total n-grams)."""
    total_ngrams = Counter()
    for text in texts:
        words = text.lower().split()
        ngrams = [tuple(words[i:i+n]) for i in range(len(words)-n+1)]
        total_ngrams.update(ngrams)
    if sum(total_ngrams.values()) == 0:
        return 0
    return len(total_ngrams) / sum(total_ngrams.values())

def dataset_diversity_report(examples):
    """Generate diversity metrics for a training dataset."""
    responses = [ex["messages"][-1]["content"] for ex in examples]
    inputs = [m["content"] for ex in examples for m in ex["messages"] if m["role"] == "user"]

    report = {
        "total_examples": len(examples),
        "avg_response_length": sum(len(r.split()) for r in responses) / len(responses),
        "avg_input_length": sum(len(i.split()) for i in inputs) / len(inputs),
        "distinct_1": distinct_n(responses, 1),
        "distinct_2": distinct_n(responses, 2),
        "distinct_3": distinct_n(responses, 3),
    }
    return report

Train/Eval Split

import random

def split_dataset(examples, eval_ratio=0.1, production_indices=None):
    """Split dataset into train/eval, keeping production data in eval for ground truth.

    Args:
        examples: List of training examples
        eval_ratio: Fraction of data for evaluation (default 10%)
        production_indices: Indices of real production examples (always go to eval)
    """
    production_indices = set(production_indices or [])
    synthetic = [ex for i, ex in enumerate(examples) if i not in production_indices]
    production = [ex for i, ex in enumerate(examples) if i in production_indices]

    # Production data goes to eval (ground truth)
    eval_set = list(production)

    # Fill remaining eval budget from synthetic
    remaining_eval = max(0, int(len(examples) * eval_ratio) - len(eval_set))
    random.shuffle(synthetic)
    eval_set.extend(synthetic[:remaining_eval])
    train_set = synthetic[remaining_eval:]

    print(f"Train: {len(train_set)}, Eval: {len(eval_set)} "
          f"({len(production)} production + {len(eval_set)-len(production)} synthetic)")
    return train_set, eval_set

Common Issues

Quick Start Checklist

1. Identify data source: Production logs (A), frontier distillation (B), or synthetic (C)
2. Format to JSONL: Standard chat format with messages array
3. Validate: Run validate_jsonl() on the output file
4. Deduplicate: Run MinHash dedup with 0.8 threshold
5. Check diversity: Run dataset_diversity_report(), aim for distinct-2 > 0.5
6. Split: 90/10 train/eval, production data in eval set
7. Count tokens: Verify no examples exceed model's context window
8. Proceed to training: Load tinker or unsloth skill for next step