07. Few-Shot Learning and Rapid Adaptation — New Intents with <100 Examples

graph TB
    subgraph "Support Set (5-shot, 3-way for illustration)"
        subgraph "order_status (5 examples)"
            OS1["'Where is my Berserk order?'"]
            OS2["'Track my delivery'"]
            OS3["'When will volume 2 arrive?'"]
            OS4["'Order #123 status?'"]
            OS5["'Is my package shipped?'"]
        end
        subgraph "recommendation (5 examples)"
            R1["'Manga like One Piece?'"]
            R2["'Recommend dark fantasy'"]
            R3["'What should I read next?'"]
            R4["'Similar to Naruto?'"]
            R5["'Best shōnen for adults?'"]
        end
        subgraph "preorder_anime (5 examples — NEW)"
            PA1["'Preorder Jujutsu Kaisen 0?'"]
            PA2["'When can I preorder?'"]
            PA3["'Reserve anime manga'"]
            PA4["'Upcoming anime tie-in'"]
            PA5["'New anime manga release'"]
        end
    end

    subgraph "Embedding (DistilBERT → 768d)"
        OS1 --> E_OS["Mean of 5 embeddings"]
        R1 --> E_R["Mean of 5 embeddings"]
        PA1 --> E_PA["Mean of 5 embeddings"]
    end

    E_OS --> P_OS["Prototype: p_order ∈ ℝ⁷⁶⁸"]
    E_R --> P_R["Prototype: p_reco ∈ ℝ⁷⁶⁸"]
    E_PA --> P_PA["Prototype: p_preorder ∈ ℝ⁷⁶⁸"]

    subgraph "Query Classification"
        Q["Query: 'Can I preorder the Chainsaw Man anime edition?'"]
        Q --> E_Q["Embedding ∈ ℝ⁷⁶⁸"]
    end

    E_Q --> D1["d(q, p_order) = 4.82"]
    E_Q --> D2["d(q, p_reco) = 3.15"]
    E_Q --> D3["d(q, p_preorder) = 1.24"]

    P_OS --> D1
    P_R --> D2
    P_PA --> D3

    D1 --> SM["Softmax(-distances)<br>order: 0.02<br>reco: 0.12<br>preorder: 0.86"]

    style P_PA fill:#c8e6c9
    style D3 fill:#c8e6c9

sequenceDiagram
    participant Meta as Meta-Initialization θ
    participant Task1 as Task 1: Intent (10-way 5-shot)
    participant Task2 as Task 2: Sentiment (3-way 5-shot)
    participant Task3 as Task 3: Genre (8-way 5-shot)
    participant Update as Meta-Update

    Note over Meta: θ = random init (or pretrained)

    rect rgb(255, 249, 196)
        Note over Task1: INNER LOOP — Task 1
        Meta->>Task1: Copy θ → θ₁
        Note over Task1: 3 gradient steps on support set:<br>θ₁' = θ₁ - α∇L₁(θ₁)<br>θ₁'' = θ₁' - α∇L₁(θ₁')<br>θ₁''' = θ₁'' - α∇L₁(θ₁'')
        Task1->>Update: Evaluate θ₁''' on query set → L₁(θ₁''')
    end

    rect rgb(187, 222, 251)
        Note over Task2: INNER LOOP — Task 2
        Meta->>Task2: Copy θ → θ₂
        Note over Task2: 3 gradient steps on support set:<br>θ₂' ... θ₂'''
        Task2->>Update: Evaluate θ₂''' on query set → L₂(θ₂''')
    end

    rect rgb(200, 230, 201)
        Note over Task3: INNER LOOP — Task 3
        Meta->>Task3: Copy θ → θ₃
        Note over Task3: 3 gradient steps on support set:<br>θ₃' ... θ₃'''
        Task3->>Update: Evaluate θ₃''' on query set → L₃(θ₃''')
    end

    Note over Update: OUTER LOOP<br>θ ← θ - β∇θ[L₁(θ₁''') + L₂(θ₂''') + L₃(θ₃''')]<br><br>This gradient flows THROUGH<br>the inner loop steps<br>(second-order derivative)

    Update->>Meta: Updated meta-initialization θ

graph TB
    subgraph "Input: 30 labeled examples (3 intents × 10 each)"
        D["order_status: 10<br>recommendation: 10<br>preorder_anime: 10"]
    end

    subgraph "Stage 1: Contrastive Fine-Tuning"
        D --> PAIRS["Generate contrastive pairs<br>Positive (same class): 3 × C(10,2) = 135<br>Negative (cross-class): 3 × C(10,1)² × 3 = 2700+<br>Total: ~3,915 pairs"]

        PAIRS --> ST["Sentence Transformer<br>(all-MiniLM-L6-v2, 22M params)"]
        ST --> CONT["Contrastive Loss<br>Pull same-class closer<br>Push diff-class apart<br><br>Fine-tune for ~20 epochs<br>~2 minutes on GPU"]
    end

    subgraph "Stage 2: Classification Head"
        CONT --> EMB["Embed all 30 examples<br>with fine-tuned encoder"]
        EMB --> LR["Logistic Regression<br>on 30 × 384d vectors<br>Training: <1 second"]
    end

    subgraph "Inference"
        Q["New query"] --> EMB2["Encode with<br>fine-tuned Sentence-T"]
        EMB2 --> PRED["Logistic Regression<br>→ Intent prediction"]
    end

    style CONT fill:#fff9c4
    style LR fill:#c8e6c9

graph LR
    subgraph "Before SetFit (Pre-trained Embeddings)"
        B1["order_status queries<br>scattered — overlap with<br>product_inquiry"]
        B2["recommendation queries<br>partially clustered"]
        B3["preorder_anime queries<br>NO cluster — unseen intent"]
    end

    subgraph "After SetFit (30 examples, 2 min training)"
        A1["order_status queries<br>tight cluster — separated<br>from product_inquiry"]
        A2["recommendation queries<br>compact cluster"]
        A3["preorder_anime queries<br>NEW cluster formed from<br>contrastive pairs"]
    end

    B1 -->|"Contrastive<br>fine-tuning"| A1
    B2 -->|"2 min"| A2
    B3 -->|"130× data<br>amplification"| A3

    style A1 fill:#c8e6c9
    style A2 fill:#c8e6c9
    style A3 fill:#c8e6c9

graph TD
    START["New intent emerges<br>How many labeled examples?"]

    START -->|"5-20 examples"| PROTO["Prototypical Networks<br>No training needed<br>Just compute prototypes<br>Expected: 70-78% accuracy"]

    START -->|"20-50 examples"| SETFIT["SetFit<br>2-5 min contrastive training<br>+ logistic regression<br>Expected: 78-85% accuracy"]

    START -->|"50-100 examples"| COMBINED["SetFit + FOMAML init<br>5-10 min training<br>Expected: 82-88% accuracy"]

    START -->|"100+ examples"| FULL["Standard fine-tuning<br>(with continual learning, doc 06)<br>Expected: 88-92% accuracy"]

    PROTO --> VAL{"Accuracy ≥ 80%?"}
    SETFIT --> VAL
    COMBINED --> VAL
    FULL --> DEPLOY["Deploy to production"]

    VAL -->|Yes| DEPLOY
    VAL -->|No| MORE["Collect more examples<br>or try next method"]
    MORE --> START

    style SETFIT fill:#c8e6c9
    style COMBINED fill:#c8e6c9

import torch
import torch.nn.functional as F
from transformers import AutoModel, AutoTokenizer
from collections import defaultdict
import numpy as np


class PrototypicalClassifier:
    """
    Few-shot intent classification using prototypical networks.
    No training required — just compute prototypes from support examples.
    """

    def __init__(self, model_name: str = "distilbert-base-uncased"):
        self.tokenizer = AutoTokenizer.from_pretrained(model_name)
        self.model = AutoModel.from_pretrained(model_name)
        self.model.eval()
        self.prototypes: dict[str, torch.Tensor] = {}

    @torch.no_grad()
    def encode(self, texts: list[str]) -> torch.Tensor:
        """Encode texts to [CLS] embeddings."""
        inputs = self.tokenizer(
            texts,
            padding=True,
            truncation=True,
            max_length=128,
            return_tensors="pt",
        )
        outputs = self.model(**inputs)
        # Use [CLS] token embedding
        return outputs.last_hidden_state[:, 0, :]  # (N, 768)

    def register_intent(self, intent_name: str, examples: list[str]):
        """
        Register a new intent from few-shot examples.
        Prototype = mean embedding of all examples.
        """
        embeddings = self.encode(examples)  # (K, 768)
        prototype = embeddings.mean(dim=0)  # (768,)
        # L2 normalize for stable distance computation
        prototype = F.normalize(prototype, dim=0)
        self.prototypes[intent_name] = prototype

    def classify(self, query: str, top_k: int = 3) -> list[dict]:
        """Classify a query by distance to prototypes."""
        query_emb = self.encode([query])[0]  # (768,)
        query_emb = F.normalize(query_emb, dim=0)

        distances = {}
        for intent, proto in self.prototypes.items():
            # Euclidean distance (on normalized vectors)
            distances[intent] = torch.norm(query_emb - proto).item()

        # Convert distances to probabilities
        dist_tensor = torch.tensor(list(distances.values()))
        probs = F.softmax(-dist_tensor, dim=0)

        results = []
        for (intent, dist), prob in zip(distances.items(), probs):
            results.append({
                "intent": intent,
                "distance": dist,
                "probability": prob.item(),
            })

        results.sort(key=lambda x: x["probability"], reverse=True)
        return results[:top_k]


# Usage
clf = PrototypicalClassifier()

clf.register_intent("order_status", [
    "Where is my Berserk order?",
    "Track my delivery",
    "When will volume 2 arrive?",
    "Order #123 status?",
    "Is my package shipped?",
])

clf.register_intent("preorder_anime", [
    "Preorder Jujutsu Kaisen 0?",
    "When can I preorder the anime edition?",
    "Reserve anime manga",
    "Upcoming anime tie-in preorder",
    "New anime manga release date",
])

result = clf.classify("Can I preorder the Chainsaw Man anime edition?")
# [{"intent": "preorder_anime", "probability": 0.86, "distance": 1.24}, ...]

from setfit import SetFitModel, SetFitTrainer, TrainingArguments
from datasets import Dataset
import pandas as pd


def train_setfit_intent(
    examples: list[dict],
    model_name: str = "sentence-transformers/all-MiniLM-L6-v2",
    num_epochs: int = 20,
):
    """
    Train SetFit model for few-shot intent classification.

    From 30 examples, generates ~3,900 contrastive pairs.
    Training takes 2-5 minutes on a single GPU.
    """
    # Prepare dataset
    df = pd.DataFrame(examples)
    dataset = Dataset.from_pandas(df)

    # Split if enough data
    if len(examples) > 20:
        splits = dataset.train_test_split(test_size=0.2, seed=42)
        train_dataset = splits["train"]
        eval_dataset = splits["test"]
    else:
        train_dataset = dataset
        eval_dataset = None

    # Load SetFit model
    model = SetFitModel.from_pretrained(
        model_name,
        labels=list(set(df["label"])),
    )

    args = TrainingArguments(
        batch_size=16,
        num_epochs=num_epochs,
        num_iterations=20,  # Number of text pairs to generate per class
        evaluation_strategy="epoch" if eval_dataset else "no",
    )

    trainer = SetFitTrainer(
        model=model,
        args=args,
        train_dataset=train_dataset,
        eval_dataset=eval_dataset,
    )

    trainer.train()
    metrics = trainer.evaluate() if eval_dataset else {}

    return model, metrics


# Example: 30 examples for 3 intents
examples = [
    {"text": "Where is my Berserk order?", "label": "order_status"},
    {"text": "Track my delivery", "label": "order_status"},
    # ... 8 more order_status examples
    {"text": "Manga like One Piece?", "label": "recommendation"},
    # ... 9 more recommendation examples
    {"text": "Preorder Jujutsu Kaisen 0?", "label": "preorder_anime"},
    # ... 9 more preorder_anime examples
]

model, metrics = train_setfit_intent(examples)
# Training: ~2 minutes on T4 GPU
# Generated pairs: ~3,900
# Accuracy: ~83% with 30 examples

import torch
import torch.nn as nn
from transformers import AutoModel, AutoTokenizer
from copy import deepcopy
import random


class FOMAMLIntentClassifier(nn.Module):
    """
    First-Order MAML for few-shot intent classification.
    Learns a meta-initialization that can be adapted with a few gradient steps.
    """

    def __init__(self, model_name: str = "distilbert-base-uncased", num_classes: int = 10):
        super().__init__()
        self.encoder = AutoModel.from_pretrained(model_name)
        self.classifier = nn.Linear(768, num_classes)

    def forward(self, input_ids, attention_mask):
        outputs = self.encoder(input_ids=input_ids, attention_mask=attention_mask)
        cls_emb = outputs.last_hidden_state[:, 0, :]
        return self.classifier(cls_emb)


def fomaml_train(
    model: FOMAMLIntentClassifier,
    tasks: list,
    meta_lr: float = 1e-5,
    inner_lr: float = 1e-3,
    inner_steps: int = 3,
    meta_epochs: int = 100,
):
    """
    Train FOMAML with episodic sampling.

    Each episode:
    1. Sample a task (N-way K-shot)
    2. Inner loop: adapt model on support set
    3. Outer loop: update meta-params using query set loss
    """
    meta_optimizer = torch.optim.Adam(model.parameters(), lr=meta_lr)
    tokenizer = AutoTokenizer.from_pretrained("distilbert-base-uncased")

    for epoch in range(meta_epochs):
        meta_loss = 0

        # Sample batch of tasks
        task_batch = random.sample(tasks, min(4, len(tasks)))

        for task in task_batch:
            support, query = task["support"], task["query"]

            # Clone model for inner loop
            adapted_model = deepcopy(model)
            inner_optimizer = torch.optim.SGD(
                adapted_model.parameters(), lr=inner_lr
            )

            # Inner loop: adapt on support set
            for _ in range(inner_steps):
                support_inputs = tokenizer(
                    [s["text"] for s in support],
                    padding=True, truncation=True,
                    max_length=128, return_tensors="pt",
                )
                support_labels = torch.tensor([s["label"] for s in support])

                logits = adapted_model(**support_inputs)
                loss = nn.functional.cross_entropy(logits, support_labels)

                inner_optimizer.zero_grad()
                loss.backward()
                inner_optimizer.step()

            # Outer loop: evaluate adapted model on query set
            query_inputs = tokenizer(
                [q["text"] for q in query],
                padding=True, truncation=True,
                max_length=128, return_tensors="pt",
            )
            query_labels = torch.tensor([q["label"] for q in query])

            query_logits = adapted_model(**query_inputs)
            query_loss = nn.functional.cross_entropy(query_logits, query_labels)
            meta_loss += query_loss

        # FOMAML: use gradients from adapted model directly
        # (approximates second-order gradient without Hessian)
        meta_optimizer.zero_grad()
        meta_loss /= len(task_batch)
        meta_loss.backward()
        meta_optimizer.step()

        if epoch % 10 == 0:
            print(f"Epoch {epoch}: meta_loss = {meta_loss.item():.4f}")

    return model

import time
from typing import Literal


def rapid_adapt_and_deploy(
    new_intent: str,
    examples: list[dict],
    method: Literal["prototype", "setfit", "fomaml"] = "auto",
):
    """
    Rapid adaptation pipeline: from labeled examples to deployed model.
    Automatically selects method based on example count.
    """
    n_examples = len(examples)
    start_time = time.time()

    # Auto-select method
    if method == "auto":
        if n_examples < 20:
            method = "prototype"
        elif n_examples < 100:
            method = "setfit"
        else:
            method = "fomaml"

    if method == "prototype":
        clf = PrototypicalClassifier()
        # Register all existing intents (from production model's support set)
        for intent, support in load_existing_support_sets().items():
            clf.register_intent(intent, support)
        # Register new intent
        clf.register_intent(new_intent, [e["text"] for e in examples])
        accuracy = evaluate_prototype(clf, examples)

    elif method == "setfit":
        # Combine existing intent examples with new intent
        all_examples = load_existing_few_shot_examples()
        all_examples.extend(examples)
        model, metrics = train_setfit_intent(all_examples)
        accuracy = metrics.get("accuracy", 0)

    elif method == "fomaml":
        model = FOMAMLIntentClassifier()
        model.load_state_dict(torch.load("meta_init.pt"))
        # Few inner loop steps on new data
        adapted = adapt_fomaml(model, examples, inner_steps=5)
        accuracy = evaluate_adapted(adapted, examples)

    elapsed = time.time() - start_time

    result = {
        "intent": new_intent,
        "method": method,
        "n_examples": n_examples,
        "accuracy": accuracy,
        "training_time_seconds": elapsed,
        "ready_for_production": accuracy >= 0.80,
    }

    if result["ready_for_production"]:
        deploy_to_lambda(result)

    return result