Instruction Following › Code Implementation

from dataclasses import dataclass
from typing import Optional


@dataclass
class InstructionExample:
    """Represents a single instruction-tuning example."""

    instruction: str
    input_text: Optional[str]
    output: str

    def format_prompt(self, template: str = "default") -> str:
        """Format the example according to a template."""
        if template == "default":
            if self.input_text:
                return f"### Instruction:\n{self.instruction}\n\n### Input:\n{self.input_text}\n\n### Response:\n"
            else:
                return (
                    f"### Instruction:\n{self.instruction}\n\n### Response:\n"
                )
        elif template == "alpaca":
            if self.input_text:
                return f"Below is an instruction that describes a task, paired with an input that provides further context. Write a response that appropriately completes the request.\n\n### Instruction:\n{self.instruction}\n\n### Input:\n{self.input_text}\n\n### Response:\n"
            else:
                return f"Below is an instruction that describes a task. Write a response that appropriately completes the request.\n\n### Instruction:\n{self.instruction}\n\n### Response:\n"
        else:
            raise ValueError(f"Unknown template: {template}")

    def format_full(self, template: str = "default") -> str:
        """Format the complete example including the response."""
        return self.format_prompt(template) + self.output

# Create diverse instruction examples
examples = [
    # Translation task
    InstructionExample(
        instruction="Translate the following English text to French.",
        input_text="The weather is beautiful today.",
        output="Le temps est magnifique aujourd'hui.",
    ),
    # Summarization task
    InstructionExample(
        instruction="Summarize this paragraph in one sentence.",
        input_text="Machine learning is a subset of artificial intelligence that enables systems to learn and improve from experience without being explicitly programmed. It focuses on developing computer programs that can access data and use it to learn for themselves. The process begins with observations or data, such as examples, direct experience, or instruction, to look for patterns in data and make better decisions in the future.",
        output="Machine learning enables computer systems to automatically learn and improve from data and experience rather than requiring explicit programming.",
    ),
    # Generation task (no input)
    InstructionExample(
        instruction="Write a haiku about programming.",
        input_text=None,
        output="Code flows like water\nBugs emerge from hidden depths\nDebug, compile, run",
    ),
    # Classification task
    InstructionExample(
        instruction="Classify the sentiment of the following review as positive, negative, or neutral.",
        input_text="The product arrived on time but the quality was disappointing. I expected better for the price.",
        output="Negative",
    ),
    # Question answering task
    InstructionExample(
        instruction="Answer the following question based on the given context.",
        input_text="Context: The Great Wall of China was built over many centuries, with construction beginning in the 7th century BC. The most well-known sections were built during the Ming Dynasty (1368-1644 AD).\n\nQuestion: When were the most famous parts of the Great Wall built?",
        output="The most famous sections of the Great Wall of China were built during the Ming Dynasty, which lasted from 1368 to 1644 AD.",
    ),
    # Code generation task
    InstructionExample(
        instruction="Write a Python function that calculates the factorial of a number.",
        input_text=None,
        output="def factorial(n):\n    if n < 0:\n        raise ValueError('Factorial not defined for negative numbers')\n    if n == 0 or n == 1:\n        return 1\n    return n * factorial(n - 1)",
    ),
]

# Generate formatted outputs for the first example
example = examples[0]
default_format = example.format_full("default")
alpaca_format = example.format_full("alpaca")

class InstructionValidator:
    """Validates instruction examples for quality issues."""

    def __init__(
        self,
        min_instruction_length: int = 5,
        min_output_length: int = 1,
        max_instruction_length: int = 1000,
    ):
        self.min_instruction_length = min_instruction_length
        self.min_output_length = min_output_length
        self.max_instruction_length = max_instruction_length

    def validate(self, example: InstructionExample) -> dict:
        """
        Validate an instruction example.
        Returns dict with 'valid' bool and 'issues' list.
        """
        issues = []

        # Check instruction length
        if len(example.instruction) < self.min_instruction_length:
            issues.append(
                f"Instruction too short ({len(example.instruction)} chars)"
            )
        if len(example.instruction) > self.max_instruction_length:
            issues.append(
                f"Instruction too long ({len(example.instruction)} chars)"
            )

        # Check output exists and has content
        if (
            not example.output
            or len(example.output.strip()) < self.min_output_length
        ):
            issues.append("Output is empty or too short")

        # Check for placeholder text
        placeholder_patterns = [
            "[your response here]",
            "[insert",
            "TODO",
            "...",
        ]
        for pattern in placeholder_patterns:
            if pattern.lower() in example.output.lower():
                issues.append(f"Output contains placeholder text: {pattern}")

        # Check instruction ends appropriately (not mid-sentence)
        if example.instruction and example.instruction[-1] not in ".?!:":
            # This is a soft warning, not necessarily an error
            issues.append("Instruction doesn't end with punctuation (minor)")

        return {
            "valid": len([i for i in issues if "minor" not in i]) == 0,
            "issues": issues,
        }

# Test the validator
validator = InstructionValidator()

# Create some examples with quality issues
test_examples = [
    InstructionExample(
        instruction="Hi",  # Too short
        input_text=None,
        output="Hello!",
    ),
    InstructionExample(
        instruction="Complete this task",  # Missing punctuation
        input_text="Some input",
        output="",  # Empty output
    ),
    InstructionExample(
        instruction="Summarize this article.",
        input_text="Article content here.",
        output="[Your response here]",  # Placeholder
    ),
    InstructionExample(
        instruction="Translate to Spanish.",
        input_text="Good morning",
        output="Buenos días",  # Valid example
    ),
]

validation_results = [validator.validate(ex) for ex in test_examples]

import re
from collections import Counter


def analyze_diversity(examples: list[InstructionExample]) -> dict:
    """
    Analyze the diversity of an instruction dataset.
    Returns statistics about task types, instruction patterns, and lengths.
    """

    # Extract first verb from instructions as a rough task indicator
    def extract_task_verb(instruction: str) -> str:
        words = instruction.lower().split()
        task_verbs = [
            "write",
            "translate",
            "summarize",
            "classify",
            "answer",
            "explain",
            "list",
            "describe",
            "compare",
            "generate",
            "create",
            "analyze",
            "extract",
            "rewrite",
            "convert",
        ]
        for word in words[:5]:  # Check first 5 words
            clean_word = re.sub(r"[^a-z]", "", word)
            if clean_word in task_verbs:
                return clean_word
        return "other"

    task_verbs = [extract_task_verb(ex.instruction) for ex in examples]

    # Analyze instruction lengths
    instruction_lengths = [len(ex.instruction) for ex in examples]
    output_lengths = [len(ex.output) for ex in examples]

    # Check for input presence
    has_input = sum(1 for ex in examples if ex.input_text)

    return {
        "total_examples": len(examples),
        "task_distribution": dict(Counter(task_verbs)),
        "instruction_length": {
            "min": min(instruction_lengths),
            "max": max(instruction_lengths),
            "mean": sum(instruction_lengths) / len(instruction_lengths),
        },
        "output_length": {
            "min": min(output_lengths),
            "max": max(output_lengths),
            "mean": sum(output_lengths) / len(output_lengths),
        },
        "examples_with_input": has_input,
        "examples_without_input": len(examples) - has_input,
    }

diversity_stats = analyze_diversity(examples)

def prepare_for_training(
    examples: list[InstructionExample],
    template: str = "default",
    max_length: int = 512,
) -> list[dict]:
    """
    Prepare instruction examples for training.
    Returns list of dicts with 'prompt' and 'completion' keys.
    """
    prepared = []

    for example in examples:
        prompt = example.format_prompt(template)
        completion = example.output
        full_text = prompt + completion

        # Check length (rough estimate using character count)
        # In practice, you'd use actual tokenization
        if len(full_text) > max_length * 4:  # Rough char-to-token ratio
            continue  # Skip examples that are too long

        prepared.append(
            {"prompt": prompt, "completion": completion, "full_text": full_text}
        )

    return prepared

training_data = prepare_for_training(examples, template="alpaca")