Few-Shot vs Zero-Shot Prompting: Complete Production Guide

import json
from openai import AsyncOpenAI
from typing import Any

client = AsyncOpenAI()
ZERO_SHOT_PROMPT = """Classify the sentiment of this text as POSITIVE, NEGATIVE, or NEUTRAL.

Text: {text}

Return JSON: {{"sentiment": "POSITIVE|NEGATIVE|NEUTRAL", "confidence": 0.0-1.0}}"""

# Few-shot example
FEW_SHOT_PROMPT = """Classify the sentiment of this text as POSITIVE, NEGATIVE, or NEUTRAL.

Examples:

Text: "This product exceeded my expectations! Great quality."
{{"sentiment": "POSITIVE", "confidence": 0.95}}

Text: "Terrible customer service. Would not recommend."
{{"sentiment": "NEGATIVE", "confidence": 0.92}}

Text: "The item arrived on time. It's exactly as described."
{{"sentiment": "NEUTRAL", "confidence": 0.88}}

Text: {text}

Return JSON: {{"sentiment": "POSITIVE|NEGATIVE|NEUTRAL", "confidence": 0.0-1.0}}"""

async def zero_shot_classify(text: str) -> dict[str, Any]:
    response = await client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{
            "role": "user",
            "content": ZERO_SHOT_PROMPT.format(text=text),
        }],
        response_format={"type": "json_object"},
    )
    return json.loads(response.choices[0].message.content)

async def few_shot_classify(text: str) -> dict[str, Any]:
    response = await client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{
            "role": "user",
            "content": FEW_SHOT_PROMPT.format(text=text),
        }],
        response_format={"type": "json_object"},
    )
    return json.loads(response.choices[0].message.content)

# Benchmark
test_cases = [
    "Love it! Best purchase this year.",
    "Completely broke after two weeks.",
    "Standard quality for the price.",
]

print("Zero-shot results:")
for text in test_cases:
    result = await zero_shot_classify(text)
    print(f"  {text[:30]}... → {result}")

print("\nFew-shot results:")
for text in test_cases:
    result = await few_shot_classify(text)
    print(f"  {text[:30]}... → {result}")

# Good zero-shot use case: Simple translation
ZERO_SHOT_TRANSLATION = """Translate the following English text to Spanish:

{text}

Return only the translation."""

# Good zero-shot use case: Basic summarization
ZERO_SHOT_SUMMARY = """Summarize this article in 2-3 sentences:

{article}"""

# Good few-shot use case: Legal document classification
FEW_SHOT_LEGAL = """Classify this legal document by type.

Examples:

Document: "Pursuant to the agreement dated January 15, 2024..."
{{"type": "CONTRACT", "subtype": "SERVICE_AGREEMENT"}}

Document: "Plaintiff hereby files this complaint against..."
{{"type": "LITIGATION", "subtype": "CIVIL_COMPLAINT"}}

Document: "Last Will and Testament of John Doe..."
{{"type": "ESTATE", "subtype": "WILL"}}

Document: {document_text}
"""

# Good few-shot use case: Code snippet classification
FEW_SHOT_CODE = """Identify the programming pattern in this code.

Examples:

Code: ```
def get_or_create(session, model, **kwargs):
    instance = session.query(model).filter_by(**kwargs).first()
    if not instance:
        instance = model(**kwargs)
        session.add(instance)
    return instance

Pair few-shot with [prompt versioning](/blog/prompt-versioning-production-management) to A/B test example effectiveness.

### Decision Table

| Situation | Recommended Strategy | Why |
|-----------|---------------------|-----|
| Generic task, frontier model, clear instructions | Zero-shot | Instruction tuning already covers it; examples add cost, little accuracy |
| Strict custom output format | One-shot | A single example fixes format compliance more reliably than prose |
| Domain classification with 5+ labels | Few-shot (3-7), balanced | Examples define label space and boundaries between labels |
| Heterogeneous inputs (many sub-types) | Dynamic few-shot | Fixed examples rarely match the current input; retrieval does |
| Multi-step reasoning | Few-shot with chain-of-thought | Demonstrates the reasoning path, not just the answer |
| Small model, cost-sensitive, high volume | Few-shot + prompt caching | Examples recover accuracy; caching removes most of their cost |
| Accuracy still short after 10 examples | Fine-tune | You have hit the ceiling of in-context learning for this model |

---

## Few-Shot Failure Modes {#failure-modes}

Few-shot prompts fail in ways that zero-shot prompts do not. Each of these has bitten a production system we have reviewed.

### Order Sensitivity

The same examples in a different order can swing accuracy by double digits on smaller models. [Lu et al. (2022)](https://arxiv.org/abs/2104.08786) documented this, and the underlying cause is **recency bias**: the last example exerts outsized influence on the prediction. A prompt ending with two `NEGATIVE` examples over-predicts `NEGATIVE`.

**Fix:** Evaluate at least three orderings on your held-out set and lock the best one; or interleave labels so no class appears last twice. Dynamic retrieval should sort by similarity ascending so the most relevant example sits closest to the query.

### Majority Label and Format Copying

If examples skew toward one label, the model inherits that skew. If examples share an incidental feature (all invoices from the same vendor, all code in Python), the model may treat the feature as a rule. [Zhao et al. (2021)](https://arxiv.org/abs/2102.09690) showed that calibrating against a content-free input ("N/A") can correct much of this bias without changing the prompt.

**Fix:** Balance labels, vary incidental attributes deliberately, and check predictions on a content-free or neutral input to see the prompt's baseline bias.

### Example Leakage Into Evaluation

The most common silent error: examples in the prompt are drawn from the same pool as the evaluation set. The model "scores" 95% because it saw the answers. Keep example pools and eval sets disjoint, and when using dynamic retrieval, exclude the current record from the candidate pool.

### Token Budget Erosion

Ten examples at 150 tokens each is 1,500 tokens on every request. On a long-context task the examples compete with the actual content for the model's attention, and on a high-volume endpoint they dominate spend. Use [token budget management](/blog/token-budget-management) to cap examples as a share of context, and use **prompt caching** so a stable example block is billed at a discount — Anthropic's [prompt caching](https://docs.anthropic.com/en/docs/build-with-claude/prompt-caching) and OpenAI's automatic caching both apply to a fixed prefix, which is exactly what a static few-shot block is.

### Stale Examples

Examples written at launch describe last year's inputs. When the input distribution drifts (new product lines, new ticket categories), fixed examples quietly become misleading. Treat the example pool as data with an owner, and refresh it from recent correctly-labeled production traffic on a schedule.

---

## Example Selection Strategies: Quality Over Quantity {#example-selection}

**Not all examples are equal.** Strategic selection dramatically impacts accuracy.

### Strategy 1: Diversity-Based Selection

```python
from typing import List, Dict
import numpy as np

class ExampleSelector:
    """Select diverse examples for few-shot prompting."""
    
    def __init__(self, embedding_model):
        self.embedding_model = embedding_model
        self.examples: List[Dict[str, Any]] = []
        self.embeddings: np.ndarray = None
    
    def add_example(self, input_text: str, output: str) -> None:
        """Add example to pool."""
        self.examples.append({
            "input": input_text,
            "output": output,
        })
    
    async def select_diverse(self, k: int = 5) -> List[Dict[str, Any]]:
        """Select k diverse examples using maximal marginal relevance."""
        if len(self.examples) <= k:
            return self.examples
        
        # Compute embeddings
        texts = [ex["input"] for ex in self.examples]
        embeddings = await self._embed_batch(texts)
        
        # Greedy diversity selection
        selected_indices = [0]  # Start with first example
        
        for _ in range(k - 1):
            # Find example most dissimilar to already selected
            max_min_sim = -1
            best_idx = -1
            
            for i, emb in enumerate(embeddings):
                if i in selected_indices:
                    continue
                
                # Minimum similarity to selected examples
                min_sim = min(
                    self._cosine_similarity(emb, embeddings[j])
                    for j in selected_indices
                )
                
                if min_sim > max_min_sim:
                    max_min_sim = min_sim
                    best_idx = i
            
            selected_indices.append(best_idx)
        
        return [self.examples[i] for i in selected_indices]
    
    @staticmethod
    def _cosine_similarity(a: np.ndarray, b: np.ndarray) -> float:
        return float(np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b)))
    
    async def _embed_batch(self, texts: List[str]) -> np.ndarray:
        response = await client.embeddings.create(
            model="text-embedding-3-small",
            input=texts,
        )
        return np.array([e.embedding for e in response.data])

# Usage
selector = ExampleSelector(embedding_model=client)
selector.add_example("Great product!", "POSITIVE")
selector.add_example("Terrible quality", "NEGATIVE")
selector.add_example("It's okay", "NEUTRAL")
# ... add more examples

diverse_examples = await selector.select_diverse(k=5)

class RelevantExampleSelector(ExampleSelector):
    """Select examples most similar to current query."""
    
    async def select_relevant(self, query: str, k: int = 5) -> List[Dict[str, Any]]:
        """Select k examples most similar to query."""
        query_emb = await self._embed_batch([query])
        query_emb = query_emb[0]
        
        example_texts = [ex["input"] for ex in self.examples]
        example_embs = await self._embed_batch(example_texts)
        
        # Compute similarities
        similarities = [
            self._cosine_similarity(query_emb, ex_emb)
            for ex_emb in example_embs
        ]
        
        # Get top-k
        top_indices = np.argsort(similarities)[-k:][::-1]
        return [self.examples[i] for i in top_indices]

# Usage
relevant_selector = RelevantExampleSelector(embedding_model=client)
# ... add examples

query = "This exceeded all my expectations"
relevant_examples = await relevant_selector.select_relevant(query, k=5)

from dataclasses import dataclass
from typing import List

@dataclass
class PerformancePoint:
    num_examples: int
    accuracy: float
    avg_input_tokens: int
    avg_cost_usd: float

async def benchmark_few_shot_scaling(
    test_cases: List[Dict[str, Any]],
    example_pool: List[Dict[str, Any]],
    max_examples: int = 10,
) -> List[PerformancePoint]:
    """Find optimal number of examples by measuring accuracy vs cost."""
    results = []
    
    for k in range(0, max_examples + 1):
        examples = example_pool[:k] if k > 0 else []
        
        correct = 0
        total_tokens = 0
        
        for test_case in test_cases:
            prompt = build_few_shot_prompt(test_case["input"], examples)
            
            response = await client.chat.completions.create(
                model="gpt-4o-mini",
                messages=[{"role": "user", "content": prompt}],
            )
            
            predicted = response.choices[0].message.content
            if predicted == test_case["expected"]:
                correct += 1
            
            total_tokens += response.usage.prompt_tokens
        
        accuracy = correct / len(test_cases)
        avg_tokens = total_tokens / len(test_cases)
        
        # Cost calculation (GPT-4o-mini: $0.15 per 1M input tokens)
        avg_cost = (avg_tokens / 1_000_000) * 0.15
        
        results.append(PerformancePoint(
            num_examples=k,
            accuracy=accuracy,
            avg_input_tokens=avg_tokens,
            avg_cost_usd=avg_cost,
        ))
    
    return results

def build_few_shot_prompt(query: str, examples: List[Dict[str, Any]]) -> str:
    if not examples:
        return f"Classify: {query}"
    
    example_text = "\n\n".join([
        f"Input: {ex['input']}\nOutput: {ex['output']}"
        for ex in examples
    ])
    
    return f"{example_text}\n\nInput: {query}\nOutput:"

# Find optimal configuration
performance = await benchmark_few_shot_scaling(
    test_cases=eval_dataset,
    example_pool=training_examples,
    max_examples=10,
)

# Analysis
for point in performance:
    print(f"{point.num_examples} examples: "
          f"{point.accuracy:.2%} accuracy, "
          f"{point.avg_input_tokens} tokens, "
          f"${point.avg_cost_usd:.6f} per request")

# Typical result: accuracy plateaus around 5-7 examples
# Example output:
# 0 examples: 68% accuracy, 120 tokens, $0.000018
# 1 examples: 78% accuracy, 220 tokens, $0.000033
# 3 examples: 86% accuracy, 380 tokens, $0.000057
# 5 examples: 89% accuracy, 520 tokens, $0.000078
# 7 examples: 90% accuracy, 660 tokens, $0.000099
# 10 examples: 90% accuracy, 880 tokens, $0.000132

class DynamicFewShotPrompt:
    """Generate few-shot prompts with query-relevant examples."""
    
    def __init__(
        self,
        example_store,
        num_examples: int = 5,
    ):
        self.example_store = example_store
        self.num_examples = num_examples
        self.selector = RelevantExampleSelector(embedding_model=client)
        
        # Load examples into selector
        for example in example_store.all():
            self.selector.add_example(example["input"], example["output"])
    
    async def build_prompt(self, query: str, template: str) -> str:
        """Build prompt with dynamically selected examples."""
        # Retrieve relevant examples
        examples = await self.selector.select_relevant(query, self.num_examples)
        
        # Format examples
        examples_text = "\n\n".join([
            f"Input: {ex['input']}\nOutput: {ex['output']}"
            for ex in examples
        ])
        
        # Construct full prompt
        return template.format(
            examples=examples_text,
            query=query,
        )

# Example usage
DYNAMIC_TEMPLATE = """Classify the sentiment of product reviews.

Examples:
{examples}

Now classify:
Input: {query}
Output:"""

dynamic_prompt = DynamicFewShotPrompt(
    example_store=example_database,
    num_examples=5,
)

query = "Best purchase I've made this year!"
prompt = await dynamic_prompt.build_prompt(query, DYNAMIC_TEMPLATE)

response = await client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[{"role": "user", "content": prompt}],
)

# Zero-shot
ZERO_SHOT_MATH = """Solve this problem: {problem}"""

# Few-shot with chain-of-thought
FEW_SHOT_COT = """Solve math word problems step by step.

Example 1:
Problem: A restaurant had 23 customers at lunch. 17 more came for dinner. How many customers total?
Reasoning:
- Started with 23 customers at lunch
- Added 17 customers for dinner
- Total = 23 + 17 = 40
Answer: 40 customers

Example 2:
Problem: Sarah has 48 apples. She gives 12 to her friend and sells half of the remaining. How many does she have left?
Reasoning:
- Started with 48 apples
- After giving away 12: 48 - 12 = 36
- Sells half of 36: 36 / 2 = 18 (sold)
- Remaining: 36 - 18 = 18
Answer: 18 apples

Problem: {problem}
Reasoning:"""

# Benchmark: CoT vs standard few-shot
async def compare_cot_impact():
    test_problems = [
        "A store had 156 items. They sold 67 and received 43 new items. How many total?",
        "Tom has $450. He spends 1/3 on rent and $80 on food. How much left?",
    ]
    
    print("Few-shot (no CoT):")
    for prob in test_problems:
        result = await zero_shot_classify(prob)
        print(f"  {result}")
    
    print("\nFew-shot with CoT:")
    for prob in test_problems:
        result = await few_shot_classify(prob)  # Uses CoT examples
        print(f"  {result}")

# Typical result: CoT improves accuracy 15-25% on reasoning tasks

# One-shot is often sufficient for format specification
ONE_SHOT_FORMAT = """Extract key information from invoices.

Example:
Invoice text: "Invoice #INV-2024-001 dated March 15, 2024 from Acme Corp for $1,250.00"
Output: {{"invoice_number": "INV-2024-001", "date": "2024-03-15", "vendor": "Acme Corp", "amount": 1250.00}}

Invoice text: {text}
Output:"""

# One-shot for style transfer
ONE_SHOT_STYLE = """Rewrite in professional business tone.

Example:
Casual: "Hey, the meeting got pushed to next week. My bad!"
Professional: "Please note that the meeting has been rescheduled to next week. I apologize for any inconvenience."

Casual: {text}
Professional:"""

from enum import Enum

class PromptStrategy(str, Enum):
    ZERO_SHOT = "zero_shot"
    ONE_SHOT = "one_shot"
    FEW_SHOT = "few_shot"
    DYNAMIC = "dynamic"

class AdaptiveFewShotSystem:
    """Production system that adapts prompting strategy based on performance."""
    
    def __init__(
        self,
        example_store,
        accuracy_threshold: float = 0.90,
        token_budget: int = 1000,
    ):
        self.example_store = example_store
        self.accuracy_threshold = accuracy_threshold
        self.token_budget = token_budget
        self.performance_history = {}
    
    async def execute(self, query: str, task_type: str) -> tuple[str, PromptStrategy]:
        """Execute with optimal strategy for task type."""
        # Check historical performance
        if task_type in self.performance_history:
            perf = self.performance_history[task_type]
            if perf["zero_shot_accuracy"] >= self.accuracy_threshold:
                strategy = PromptStrategy.ZERO_SHOT
            elif perf["one_shot_accuracy"] >= self.accuracy_threshold:
                strategy = PromptStrategy.ONE_SHOT
            else:
                strategy = PromptStrategy.FEW_SHOT
        else:
            # Default to few-shot for new task types
            strategy = PromptStrategy.FEW_SHOT
        
        # Build prompt based on strategy
        if strategy == PromptStrategy.ZERO_SHOT:
            prompt = self._build_zero_shot(query, task_type)
        elif strategy == PromptStrategy.ONE_SHOT:
            examples = await self._get_examples(task_type, k=1)
            prompt = self._build_few_shot(query, examples)
        else:
            examples = await self._get_examples(task_type, k=5)
            prompt = self._build_few_shot(query, examples)
        
        # Execute
        response = await client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": prompt}],
        )
        
        return response.choices[0].message.content, strategy
    
    async def _get_examples(self, task_type: str, k: int) -> List[Dict[str, Any]]:
        """Retrieve relevant examples for task type."""
        return await self.example_store.get_by_task(task_type, limit=k)
    
    def _build_zero_shot(self, query: str, task_type: str) -> str:
        """Build zero-shot prompt."""
        templates = {
            "sentiment": "Classify sentiment: {query}",
            "extraction": "Extract key fields: {query}",
        }
        return templates.get(task_type, "").format(query=query)
    
    def _build_few_shot(self, query: str, examples: List[Dict[str, Any]]) -> str:
        """Build few-shot prompt with examples."""
        example_text = "\n\n".join([
            f"Input: {ex['input']}\nOutput: {ex['output']}"
            for ex in examples
        ])
        return f"{example_text}\n\nInput: {query}\nOutput:"
    
    async def update_performance(
        self,
        task_type: str,
        strategy: PromptStrategy,
        was_correct: bool,
    ) -> None:
        """Track performance to optimize future strategy selection."""
        if task_type not in self.performance_history:
            self.performance_history[task_type] = {
                "zero_shot_accuracy": 0.0,
                "one_shot_accuracy": 0.0,
                "few_shot_accuracy": 0.0,
            }
        
        key = f"{strategy.value}_accuracy"
        # Simple moving average update
        current = self.performance_history[task_type][key]
        self.performance_history[task_type][key] = (current * 0.9) + (1.0 if was_correct else 0.0) * 0.1

# Usage in production
adaptive_system = AdaptiveFewShotSystem(
    example_store=example_db,
    accuracy_threshold=0.90,
    token_budget=1000,
)

result, strategy = await adaptive_system.execute(
    "I love this product!",
    task_type="sentiment",
)
print(f"Used strategy: {strategy}, Result: {result}")

# Update performance tracking
await adaptive_system.update_performance(
    task_type="sentiment",
    strategy=strategy,
    was_correct=True,  # Verify against ground truth
)