4. Advanced Few-Shot: Example Engineering

Capsule overview

Example engineering goes beyond picking examples by hand: it includes generating synthetic examples with LLMs when you have no labeled data, dynamic selection with embeddings (dynamic few-shot), per-domain example banks, and optimizing the cost/quality balance. These techniques scale few-shot to real production scenarios.

In this capsule you'll learn when to invest in each technique, how to implement them in Python, and how to measure whether the extra investment (the cost of embeddings, the time spent curating synthetics) justifies the accuracy gain. The result is a toolkit you can take straight into the final project's Few-Shot Classification System.

Why it matters: In production, you rarely have exactly the right examples available. Knowing how to generate valid synthetic ones can bootstrap a classification system in hours instead of days spent waiting for real labeled data.

Synthetic Examples Generated by an LLM

Why generate them

When you start a new classification domain, you rarely have labeled examples ready. Writing them by hand is slow (you need domain experts). LLMs can generate realistic examples for many tasks in minutes.

Use case: Build an initial example bank for bootstrapping, then progressively replace it with real examples.

Basic generator

from openai import OpenAI
import json
from dotenv import load_dotenv

load_dotenv()
client = OpenAI()

def generate_synthetic_examples(
    task_description: str,
    categories: list[str],
    n_per_category: int = 3,
    temperature: float = 0.7
) -> list[tuple[str, str]]:
    """
    Generates synthetic classification examples using an LLM.
    
    Args:
        task_description: What kind of text gets classified (e.g. "support queries")
        categories: List of categories with their description
        n_per_category: How many examples to generate per category
        temperature: Higher = more variety, lower = more predictable
    
    Returns:
        A list of (input_text, category)
    """
    cats_str = "\n".join([f"- {c}" for c in categories])
    
    prompt = f"""
Generate {n_per_category} REALISTIC examples of "{task_description}" for each category.
Each example must be short (1-2 sentences), natural, and clearly belong to that category.

Categories:
{cats_str}

Response format (JSON):
{{
  "examples": [
    {{"input": "the example's text", "category": "CATEGORY_NAME"}},
    ...
  ]
}}

Generate exactly {n_per_category * len(categories)} examples in total.
"""
    
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}],
        temperature=temperature,
        response_format={"type": "json_object"}
    )
    
    data = json.loads(response.choices[0].message.content)
    return [(ex["input"], ex["category"]) for ex in data.get("examples", [])]

# Usage
categories = [
    "TECHNICAL: software errors, broken functionality",
    "BILLING: charges, invoices, plans, payments",
    "ACCOUNT: profile data, cancellation, password",
    "INFORMATION: general questions, usage doubts"
]

examples = generate_synthetic_examples(
    task_description="SaaS technical support queries",
    categories=categories,
    n_per_category=3
)

print(f"Generated {len(examples)} examples:\n")
for inp, cat in examples:
    print(f"  [{cat}] {inp}")

Typical output:

Generated 12 examples:

  [TECHNICAL] The app hasn't loaded the dashboard since yesterday's update
  [TECHNICAL] Error 403 when trying to access the Q3 reports
  [TECHNICAL] The export module freezes when I select more than 100 records
  [BILLING] Why am I being charged twice for the Enterprise plan this month?
  [BILLING] I need an invoice with tax details to give to my accountant
  [BILLING] I want to switch from monthly to annual payment to get the discount
  [ACCOUNT] How do I change the email tied to my account without losing my data
  [ACCOUNT] I want to close my account and export all my data first
  [ACCOUNT] I need to add two more admins to the team
  [INFORMATION] Do you have a native Slack integration?
  [INFORMATION] How long does it take for a new active user to show up in the reports?
  [INFORMATION] Does the basic plan include API access?

Validate and filter the synthetics

Not every generated example is good. Validate before adding to the bank:

def validate_example(input_text: str, category: str, valid_categories: set[str]) -> tuple[bool, str]:
    """
    Validates that an example is usable for few-shot.
    Returns: (is_valid, reason_if_invalid)
    """
    # Check that the category is valid
    if category not in valid_categories:
        return False, f"Category '{category}' is not valid"
    
    # Check the minimum length
    word_count = len(input_text.split())
    if word_count < 4:
        return False, f"Too short ({word_count} words)"
    
    # Check the maximum length (very long prompts can backfire)
    if word_count > 80:
        return False, f"Too long ({word_count} words) for a few-shot example"
    
    # Check that it contains no instructions (don't confuse the model)
    problem_words = ["classify", "categorize", "respond", "ignore", "forget"]
    lower = input_text.lower()
    for word in problem_words:
        if word in lower:
            return False, f"Contains an instruction: '{word}'"
    
    return True, ""

def filter_valid_examples(
    examples: list[tuple[str, str]],
    valid_categories: set[str]
) -> list[tuple[str, str]]:
    """Filters out invalid examples and prints a report."""
    valid = []
    rejected = 0
    
    for inp, cat in examples:
        is_valid, reason = validate_example(inp, cat, valid_categories)
        if is_valid:
            valid.append((inp, cat))
        else:
            rejected += 1
            print(f"  ❌ Rejected: '{inp[:40]}...' — {reason}")
    
    print(f"\nResult: {len(valid)}/{len(examples)} valid ({rejected} rejected)")
    return valid

# Test
valid_cats = {"TECHNICAL", "BILLING", "ACCOUNT", "INFORMATION"}
filtered_examples = filter_valid_examples(examples, valid_cats)

The complete flow: bootstrap → validate → use

def bootstrap_example_bank(
    description: str,
    categories_with_desc: list[str],
    n_initial: int = 3,
    validate: bool = True
) -> list[tuple[str, str]]:
    """
    The complete flow: generate → validate → return the initial bank.
    """
    # Pull the category names out of "NAME: description"
    cat_names = {c.split(":")[0].strip() for c in categories_with_desc}
    
    # Generate examples
    print(f"Generating {n_initial * len(categories_with_desc)} synthetic examples...")
    examples = generate_synthetic_examples(description, categories_with_desc, n_initial)
    
    if validate:
        print("Validating examples...")
        examples = filter_valid_examples(examples, cat_names)
    
    print(f"\nInitial bank ready: {len(examples)} examples")
    return examples

Dynamic Few-Shot: Selection by Embeddings

Why dynamic selection matters

With a fixed bank of 5 examples, you use the same ones for every input. With dynamic selection, for each new input you pick the K most semantically similar examples. This improves accuracy especially when the bank is large (20+ examples) and diverse.

Input: "The app crashes on Android 14"
  
  Fixed bank (5 examples):
    → Always the same 5, even if 2 of them are irrelevant
    
  Dynamic few-shot (embeddings):
    → Picks the 3 most semantically similar: 
       "App closes on mobile" (0.85 sim)
       "Error on the new version" (0.79 sim)
       "Problem when updating" (0.72 sim)
    → Ignores the BILLING examples that don't help here

Implementation with OpenAI Embeddings

from openai import OpenAI
import numpy as np
from typing import Optional

client = OpenAI()

def get_embedding(text: str, model: str = "text-embedding-3-small") -> list[float]:
    """Gets an embedding from OpenAI. Cost: ~$0.00002 per 1K tokens."""
    response = client.embeddings.create(
        model=model,
        input=text
    )
    return response.data[0].embedding

def cosine_similarity(a: list[float], b: list[float]) -> float:
    """Cosine similarity between two vectors."""
    va = np.array(a)
    vb = np.array(b)
    return float(np.dot(va, vb) / (np.linalg.norm(va) * np.linalg.norm(vb) + 1e-8))

class DynamicExampleBank:
    """
    Example bank with dynamic selection by semantic similarity.
    Pre-computes embeddings for efficiency in production.
    """
    
    def __init__(self, embedding_model: str = "text-embedding-3-small"):
        self.embedding_model = embedding_model
        self.examples: list[dict] = []  # {input, output, embedding}
    
    def add(self, input_text: str, output: str) -> None:
        """Adds an example and pre-computes its embedding."""
        embedding = get_embedding(input_text, self.embedding_model)
        self.examples.append({
            "input": input_text,
            "output": output,
            "embedding": embedding
        })
        print(f"  Added: [{output}] '{input_text[:50]}'")
    
    def add_batch(self, examples: list[tuple[str, str]]) -> None:
        """Adds multiple examples (one embeddings API call per example)."""
        for inp, out in examples:
            self.add(inp, out)
    
    def select(self, new_input: str, k: int = 3) -> list[tuple[str, str]]:
        """
        Selects the K most semantically similar examples.
        Only computes the embedding of new_input at runtime.
        """
        if not self.examples:
            return []
        
        # Embedding of the current input (the only API call at runtime)
        new_emb = get_embedding(new_input, self.embedding_model)
        
        # Compare against the pre-computed embeddings
        scored = []
        for ex in self.examples:
            sim = cosine_similarity(new_emb, ex["embedding"])
            scored.append((sim, ex["input"], ex["output"]))
        
        # Sort by similarity, descending
        scored.sort(key=lambda x: x[0], reverse=True)
        
        return [(inp, out) for _, inp, out in scored[:k]]
    
    def build_prompt(self, task_description: str, new_input: str, k: int = 3) -> str:
        """Builds a prompt with the K most similar examples."""
        examples = self.select(new_input, k)
        
        lines = [task_description, ""]
        for inp, out in examples:
            lines.append(f"Input: {inp}")
            lines.append(f"Output: {out}")
            lines.append("")
        lines.append(f"Input: {new_input}")
        lines.append("Output:")
        
        return "\n".join(lines)

# Usage
print("Building the dynamic bank...")
bank = DynamicExampleBank()

bank.add_batch([
    ("Error 500 when saving the form data", "TECHNICAL"),
    ("The app freezes when uploading large files", "TECHNICAL"),
    ("I haven't been able to log in since Tuesday", "TECHNICAL"),
    ("My October invoice has a double charge", "BILLING"),
    ("Can I switch from monthly to annual payment?", "BILLING"),
    ("I want to cancel my account and request a refund", "ACCOUNT"),
    ("Does the basic plan include API access?", "INFORMATION"),
])

# Dynamic selection in action
queries = [
    "The app crashes when opening PDF documents",
    "Incorrect charge on my credit card",
    "Do you have a discount for startups?"
]

print("\n=== Dynamic Few-Shot Selection ===\n")
for query in queries:
    selected_examples = bank.select(query, k=3)
    print(f"Input: '{query}'")
    print("Selected examples:")
    for inp, out in selected_examples:
        print(f"  [{out}] '{inp[:50]}'")
    print()

Typical output:

Input: 'The app crashes when opening PDF documents'
Selected examples:
  [TECHNICAL] 'The app freezes when uploading large files'
  [TECHNICAL] 'Error 500 when saving the form data'
  [TECHNICAL] 'I haven't been able to log in since Tuesday'

Input: 'Incorrect charge on my credit card'
Selected examples:
  [BILLING] 'My October invoice has a double charge'
  [BILLING] 'Can I switch from monthly to annual payment?'
  [ACCOUNT] 'I want to cancel my account and request a refund'

Input: 'Do you have a discount for startups?'
Selected examples:
  [INFORMATION] 'Does the basic plan include API access?'
  [BILLING] 'Can I switch from monthly to annual payment?'
  [TECHNICAL] 'I haven't been able to log in since Tuesday'

The selected examples are semantically relevant to each input.

K-Nearest Examples: The Alternative Without Embeddings

When the cost of embeddings is prohibitive or you're working offline:

def jaccard_similarity(a: str, b: str) -> float:
    """Similarity based on shared words. No API calls."""
    wa = set(a.lower().split())
    wb = set(b.lower().split())
    if not wa or not wb:
        return 0.0
    return len(wa & wb) / len(wa | wb)

def tfidf_like_similarity(a: str, b: str, vocabulary: Optional[set] = None) -> float:
    """Improved similarity, weighted by how rare the words are."""
    # Common English stopwords
    STOPWORDS = {"the", "a", "an", "of", "to", "and", "in", "on", "for", "with", 
                 "is", "it", "at", "my", "i", "this", "that"}
    
    def words(text: str) -> set:
        return set(w for w in text.lower().split() if w not in STOPWORDS)
    
    wa = words(a)
    wb = words(b)
    
    if not wa or not wb:
        return 0.0
    
    return len(wa & wb) / len(wa | wb)

def k_nearest_static(
    bank: list[tuple[str, str]],
    new_input: str,
    k: int = 3,
    sim_fn = jaccard_similarity
) -> list[tuple[str, str]]:
    """The K most similar examples, without embeddings."""
    scored = [(sim_fn(inp, new_input), inp, out) for inp, out in bank]
    scored.sort(key=lambda x: x[0], reverse=True)
    return [(inp, out) for _, inp, out in scored[:k]]

# Comparing similarity methods
test_bank = [
    ("Error logging in", "TECHNICAL"),
    ("The app closes on its own", "TECHNICAL"),
    ("Invoice with a double charge", "BILLING"),
    ("Change payment method", "BILLING"),
    ("Cancel subscription", "ACCOUNT"),
]

new_query = "I can't get in with my username and password"

print("Jaccard similarity:")
for inp, out in k_nearest_static(test_bank, new_query, k=3, sim_fn=jaccard_similarity):
    print(f"  [{out}] '{inp}'")

print("\nTFIDF-like similarity:")
for inp, out in k_nearest_static(test_bank, new_query, k=3, sim_fn=tfidf_like_similarity):
    print(f"  [{out}] '{inp}'")

Domain-Specific Example Banks

For systems that handle multiple domains, keep separate banks and pick one based on the context:

from typing import Optional

class MultiDomainExampleBank:
    """
    Example bank per domain.
    Selects the right bank based on signals in the context.
    """
    
    def __init__(self):
        # One bank per domain
        self.banks: dict[str, list[tuple[str, str]]] = {}
        # Keywords per domain, for routing
        self.keywords: dict[str, set[str]] = {}
    
    def add_domain(self, domain: str, keywords: set[str]) -> None:
        """Registers a domain with its routing keywords."""
        self.banks[domain] = []
        self.keywords[domain] = keywords
    
    def add_example(self, domain: str, input_text: str, output: str) -> None:
        """Adds an example to a specific domain."""
        if domain not in self.banks:
            raise ValueError(f"Domain '{domain}' is not registered")
        self.banks[domain].append((input_text, output))
    
    def detect_domain(self, text: str) -> str:
        """
        Detects the most likely domain based on keywords.
        Falls back to 'default' if there's no match.
        """
        text_lower = text.lower()
        scores = {}
        
        for domain, keywords in self.keywords.items():
            score = sum(1 for kw in keywords if kw in text_lower)
            scores[domain] = score
        
        if not scores or max(scores.values()) == 0:
            return "default"
        
        return max(scores.items(), key=lambda x: x[1])[0]
    
    def get_examples(self, text: str, k: int = 3) -> list[tuple[str, str]]:
        """Gets examples from the most relevant domain."""
        domain = self.detect_domain(text)
        bank = self.banks.get(domain, self.banks.get("default", []))
        return bank[:k]

# Configure the banks per domain
multi_bank = MultiDomainExampleBank()

multi_bank.add_domain("support", {"error", "doesn't work", "fails", "problem", "bug"})
multi_bank.add_domain("sales", {"price", "cost", "plan", "discount", "buy", "upgrade"})
multi_bank.add_domain("legal", {"contract", "terms", "privacy", "gdpr", "personal data"})

multi_bank.add_example("support", "Error 500 when loading", "TECHNICAL")
multi_bank.add_example("support", "The app won't open", "TECHNICAL")
multi_bank.add_example("sales", "How much does the Pro plan cost?", "PRICE")
multi_bank.add_example("sales", "Do you have an annual discount?", "PROMO")
multi_bank.add_example("legal", "Where are your servers located?", "PRIVACY")

# Test
queries = [
    "Error loading the reports module",
    "Can I negotiate the price of the Enterprise plan?",
    "Are you GDPR compliant?"
]

for c in queries:
    domain = multi_bank.detect_domain(c)
    examples = multi_bank.get_examples(c, k=2)
    print(f"'{c[:50]}' → Domain: {domain}, Examples: {[out for _, out in examples]}")

Cost Optimization: Pre-Compute the Embeddings

In production, never recompute the embeddings of fixed examples on every request:

import numpy as np
import json
import os
from pathlib import Path

def precompute_and_cache_embeddings(
    examples: list[tuple[str, str]],
    cache_path: str = "embeddings_cache.json"
) -> list[dict]:
    """
    Pre-computes the embeddings and caches them on disk.
    At runtime, you only need to compute the embedding of the new input.
    """
    if Path(cache_path).exists():
        print(f"Loading embeddings from the cache: {cache_path}")
        with open(cache_path) as f:
            return json.load(f)
    
    print(f"Computing {len(examples)} embeddings (this happens only once)...")
    bank_with_embeddings = []
    
    for inp, out in examples:
        emb = get_embedding(inp)
        bank_with_embeddings.append({
            "input": inp,
            "output": out,
            "embedding": emb  # A list of floats, serializable to JSON
        })
    
    with open(cache_path, "w") as f:
        json.dump(bank_with_embeddings, f)
    
    print(f"Embeddings saved to {cache_path}")
    return bank_with_embeddings

def select_from_cached(
    cached_bank: list[dict],
    new_input: str,
    k: int = 3
) -> list[tuple[str, str]]:
    """Selects the k most similar using the pre-computed embeddings."""
    new_emb = np.array(get_embedding(new_input))
    
    scored = []
    for ex in cached_bank:
        emb = np.array(ex["embedding"])
        sim = cosine_similarity(new_emb.tolist(), emb.tolist())
        scored.append((sim, ex["input"], ex["output"]))
    
    scored.sort(reverse=True)
    return [(inp, out) for _, inp, out in scored[:k]]

# Usage in production
# The first time: compute and cache
cached_bank = precompute_and_cache_embeddings(filtered_examples, "my_bank.json")

# On every request: only embed the new input (1 API call)
result = select_from_cached(cached_bank, "The export module is failing")

Cost impact: With a bank of 20 examples, without pre-computing = 21 embeddings calls/request. With pre-computing = 1 call/request. A 95% reduction.

Connection to the Project

In the Few-Shot Classification System (capsule 08) you'll use:

• DynamicExampleBank for embedding-based selection in advanced few-shot mode
• generate_synthetic_examples() to bootstrap the bank if there's no labeled data
• precompute_and_cache_embeddings() to optimize cost in the comparative evaluator
• MultiDomainExampleBank as the basis for the optional multi-domain mode

Troubleshooting

Problem 1: The synthetic examples aren't realistic

Cause: The generation prompt is vague, or the temperature is too high.

Fix:

# Be more specific in the generation prompt
improved_prompt = f"""
Generate {n} REALISTIC queries for a B2B SaaS company's support system.
The users are engineers or managers, not end consumers.
Each query should sound like a real support email (1-2 sentences).
Avoid generic phrases like "It doesn't work" or "I have a problem".
"""

Problem 2: Dynamic few-shot picks examples from the wrong category

Cause: Semantic similarity can pick examples from a different category if the input is ambiguous.

Fix: Add a category diversity filter to the selection:

def select_balanced(cached_bank, new_input, k=3, max_per_cat=2):
    """Selects examples balanced across categories."""
    all_examples = select_from_cached(cached_bank, new_input, k=k*2)  # Ask for more
    
    selected = []
    cat_count = {}
    
    for inp, out in all_examples:
        if cat_count.get(out, 0) < max_per_cat:
            selected.append((inp, out))
            cat_count[out] = cat_count.get(out, 0) + 1
        if len(selected) >= k:
            break
    
    return selected

Problem 3: The cost of embeddings is too high

Cause: You're computing the bank's embeddings on every request.

Fix: Pre-compute and cache the bank's embeddings. Only compute the new input's embedding at runtime. The cost goes from O(bank) to O(1) per request.

Problem 4: The synthetic bank is biased toward certain examples

Cause: The generating LLM has its own biases and can over-represent certain kinds of inputs.

Fix: Generate 10+ per category and pick the most diverse ones with measure_diversity(). Add real examples progressively to replace the synthetic ones.

Exercises

Exercise 1: Bootstrap with synthetics (Easy)

Generate a bank of 12 examples (3 per category) for an intent classifier of marketing emails: PURCHASE, INQUIRY, COMPLAINT, UNSUBSCRIBE.

cats = [
    "PURCHASE: the user wants to buy or already bought something",
    "INQUIRY: the user has a question about the product/service",
    "COMPLAINT: the user is unhappy or has a problem with a purchase",
    "UNSUBSCRIBE: the user wants to unsubscribe or cancel"
]

examples = generate_synthetic_examples(
    task_description="emails from customers of an online store",
    categories=cats,
    n_per_category=3
)

# Validate
valid_cats = {"PURCHASE", "INQUIRY", "COMPLAINT", "UNSUBSCRIBE"}
ok_examples = filter_valid_examples(examples, valid_cats)
print(f"Bank ready: {len(ok_examples)} valid examples")

Exercise 2: Dynamic few-shot with a small bank (Medium)

Given a bank of 8 email examples (URGENT/NORMAL/LOW_PRIORITY), implement dynamic selection using Jaccard (no embeddings). Verify that for "SYSTEM DOWN URGENT" it picks URGENT examples.

bank = [
    ("System completely down, customers affected", "URGENT"),
    ("Critical error in production", "URGENT"),
    ("When will feature X be ready?", "NORMAL"),
    ("Following up on a previous ticket", "NORMAL"),
    ("Information about the next release", "LOW_PRIORITY"),
    ("Could you add this functionality?", "LOW_PRIORITY"),
    ("Timeout on 2 out of 100 requests", "NORMAL"),
    ("Intermittent error when exporting", "NORMAL"),
]

new_query = "SYSTEM DOWN URGENT customers can't get in"
selected = k_nearest_static(bank, new_query, k=3, sim_fn=jaccard_similarity)
print("Selected for the system-down input:")
for inp, out in selected:
    print(f"  [{out}] {inp}")

# Verify that URGENT shows up
assert any(out == "URGENT" for _, out in selected), "It should select URGENT examples"
print("✅ Correctly selects URGENT")

Exercise 3: Pre-compute and use the cache (Medium)

Implement the complete flow: generate 9 synthetic examples → pre-compute the embeddings → save to JSON → load from JSON → use for dynamic selection.

CACHE = "bank_cache.json"

# 1. Generate examples
examples = generate_synthetic_examples(
    "technical support",
    ["TECHNICAL: software errors", "BILLING: charges and payments", "ACCOUNT: user profile"],
    n_per_category=3
)

# 2. Pre-compute and cache
cached_bank = precompute_and_cache_embeddings(examples, CACHE)
print(f"Cache created with {len(cached_bank)} examples")

# 3. Verify that the second load comes from the cache
cached_bank_2 = precompute_and_cache_embeddings(examples, CACHE)
assert len(cached_bank) == len(cached_bank_2)
print("✅ Second load from the cache (faster, no API calls)")

# 4. Use it to classify
new_query = "Error syncing with Salesforce"
selected = select_from_cached(cached_bank, new_query, k=3)
print(f"\nFor '{new_query}':")
for inp, out in selected:
    print(f"  [{out}] {inp[:50]}")

Summary

In this capsule you learned:

• Synthetic examples: Generate them with an LLM when you have no labeled data. Validate before using (length, valid category, no embedded instructions)
• Dynamic few-shot: Pick examples by similarity to the current input. It improves accuracy when the bank is large and diverse
• Embeddings vs Jaccard: Embeddings = better semantic quality, at an API cost. Jaccard = free, and enough for short texts in the same language
• Pre-compute the embeddings: Compute once, save to JSON. At runtime you only embed the new input. Cuts cost by 90-95%
• Multi-domain banks: Separate banks per domain + keyword routing. Scales to many domains without mixing examples

Next capsule: Output formatting and parsing — how to guarantee the LLM's output is always parseable by code.

Further resources

1. OpenAI Embeddings API — Official documentation for text-embedding-3-small and text-embedding-3-large, with prices and use cases
2. What Makes Good In-Context Examples? — An analysis of which characteristics make an example good for few-shot
3. Self-Generated In-Context Learning (SG-ICL) — A paper on automatically generating examples for few-shot
4. Sentence Transformers — An open-source alternative for semantic embeddings with no API cost (for offline environments)
5. FAISS (Facebook AI Similarity Search) — For K-nearest search in very large example banks (100K+)
6. NumPy Dot Product — Reference for the cosine similarity computation used in the example