Self-Instruct › Complete Self-Instruct Implementation

import random


class MockLanguageModel:
    """
    Simulates LLM responses for demonstration.
    In practice, replace with API calls to GPT-3, LLaMA, etc.
    """

    def __init__(self):
        self.instruction_templates = [
            "Rewrite the given sentence in passive voice.",
            "Identify the main theme of the passage.",
            "List the pros and cons of the given topic.",
            "Explain the concept to a five-year-old.",
            "Write a professional email about the topic.",
            "Compare and contrast the two given items.",
            "Provide three examples of the given concept.",
            "Summarize the key takeaways from the text.",
            "Correct the grammatical errors in the sentence.",
            "Generate a creative title for the story.",
            "Classify the text as formal or informal.",
            "Extract the named entities from the sentence.",
            "Paraphrase the following paragraph.",
            "Write a brief description of the product.",
            "Determine the tone of the message.",
        ]

    def generate(self, prompt: str, max_tokens: int = 200) -> str:
        """Generate response based on prompt type."""
        if "Come up with a series of tasks" in prompt:
            return self._generate_instructions()
        elif "Input:" in prompt:
            return self._generate_instance()
        elif "determine whether" in prompt.lower():
            return random.choice(["Classification", "Generation"])
        else:
            return "Sample output for the given input."

    def _generate_instructions(self) -> str:
        selected = random.sample(self.instruction_templates, 5)
        return "\n".join(f"{i + 4}. {inst}" for i, inst in enumerate(selected))

    def _generate_instance(self) -> str:
        inputs = [
            "The cat sat on the mat.",
            "Climate change affects biodiversity.",
            "Machine learning transforms industries.",
        ]
        outputs = [
            "The mat was sat on by the cat.",
            "Climate change has significant impacts on biodiversity.",
            "Industries are being transformed by machine learning.",
        ]
        idx = random.randint(0, len(inputs) - 1)
        # Include Input: prefix so parser can find it
        return f"Input: {inputs[idx]}\nOutput: {outputs[idx]}"

import random
from typing import Dict, List


class SelfInstructGenerator:
    """
    Complete Self-Instruct implementation.

    Generates new instruction-tuning examples by leveraging
    an existing language model to create diverse tasks.
    """

    def __init__(
        self, seed_tasks: List[Dict], model, similarity_threshold: float = 0.7
    ):
        self.task_pool = list(seed_tasks)
        self.model = model
        self.similarity_threshold = similarity_threshold
        self.generated_instructions = set()

        # Track existing instructions for similarity checking
        for task in seed_tasks:
            self.generated_instructions.add(task["instruction"].lower())

    def generate_batch(self, batch_size: int = 10) -> List[Dict]:
        """Generate a batch of new task instances."""
        new_tasks = []
        attempts = 0
        max_attempts = batch_size * 3  # Allow some failures

        while len(new_tasks) < batch_size and attempts < max_attempts:
            attempts += 1

            # Step 1: Generate new instructions
            sampled = random.sample(self.task_pool, min(3, len(self.task_pool)))
            prompt = create_instruction_generation_prompt(sampled)
            raw_output = self.model.generate(prompt)
            instructions = parse_generated_instructions(raw_output)

            for instruction in instructions:
                # Step 2: Apply filters
                if not self._passes_filters(instruction):
                    continue

                # Step 3: Classify task type
                is_classification = self._classify_task(instruction)

                # Step 4: Generate instance
                instance = self._generate_instance(
                    instruction, is_classification
                )

                if instance and self._validate_instance(instance):
                    new_tasks.append(instance)
                    self.generated_instructions.add(instruction.lower())

                    if len(new_tasks) >= batch_size:
                        break

        return new_tasks

    def _passes_filters(self, instruction: str) -> bool:
        """Check if instruction passes all filters."""
        # Similarity filter
        if not filter_by_similarity(
            instruction,
            list(self.generated_instructions),
            self.similarity_threshold,
        ):
            return False

        # Keyword filter
        passed, _ = filter_by_keywords(instruction)
        if not passed:
            return False

        # Length check
        if len(instruction.split()) < 3 or len(instruction.split()) > 150:
            return False

        return True

    def _classify_task(self, instruction: str) -> bool:
        """Determine if task is classification."""
        prompt = create_classification_prompt(instruction)
        response = self.model.generate(prompt, max_tokens=20)
        return "classification" in response.lower()

    def _generate_instance(
        self, instruction: str, is_classification: bool
    ) -> Dict:
        """Generate input-output pair for instruction."""
        # Use demonstrations from pool
        demos = random.sample(self.task_pool, min(2, len(self.task_pool)))

        if is_classification:
            # Output-first for classification
            prompt = create_output_first_prompt(
                instruction,
                ["Yes", "No"],  # Simplified for demo
                demos,
            )
        else:
            # Input-first for generation
            prompt = create_input_first_prompt(instruction, demos)

        response = self.model.generate(prompt)

        # Parse response to extract input and output
        return self._parse_instance(instruction, response)

    def _parse_instance(self, instruction: str, response: str) -> Dict:
        """Parse model response into structured instance."""
        # Simple parsing - production code would be more reliable
        input_text = ""
        output_text = ""

        if "Input:" in response and "Output:" in response:
            parts = response.split("Output:")
            input_part = parts[0]
            output_text = parts[1].strip() if len(parts) > 1 else ""

            if "Input:" in input_part:
                input_text = input_part.split("Input:")[1].strip()
        else:
            output_text = response.strip()

        return {
            "instruction": instruction,
            "input": input_text,
            "output": output_text,
        }

    def _validate_instance(self, instance: Dict) -> bool:
        """Final validation of complete instance."""
        length_ok, _ = apply_length_filters(
            instance["instruction"], instance["input"], instance["output"]
        )
        format_ok, _ = apply_format_filters(
            instance["instruction"], instance["input"], instance["output"]
        )
        return length_ok and format_ok

# Run Self-Instruct generation
model = MockLanguageModel()
generator = SelfInstructGenerator(seed_tasks, model, similarity_threshold=0.7)

# Generate a small batch
generated_batch = generator.generate_batch(batch_size=5)