Dynamic Few-Shot for Gemini API — A Self-Improving Prompt That Picks Examples by Vector Search

If your few-shot prompts are still hand-coded, you're about to hit a wall

Few-shot is usually the first thing we reach for when wiring a new task into Gemini. You drop two or three example inputs and outputs into the system instruction, append the real input, and call it a day. I built a document classifier on top of Gemini 2.5 Pro the same way — three carefully chosen examples, embedded right in the prompt — and it shipped well.

After running it for six months, the cracks started showing. The accuracy held up for the first few weeks, but as the input distribution drifted — new query patterns from new users, new doc types, edge cases — the static few-shot quietly degraded. Every time I wanted to add a new pattern I hit the same dilemma: more examples means better coverage, but more tokens per call, and a prompt that nobody can read anymore. Around example number ten, I noticed I was spending more time editing the prompt source file than improving the actual logic — a clear signal that the architecture was wrong.

This article walks through the pattern that fixed that for me: dynamic few-shot. Instead of hard-coding examples, you store them, retrieve the most relevant ones at query time, and grow the store from production feedback. Gemini Embeddings plus Vertex AI Vector Search make this a few hundred lines of code. If you've already read Gemini Embeddings × Reranker for Production RAG, think of this as the same idea applied to prompts rather than retrieval — and many of the operational lessons transfer directly.

The shift — from picking examples to growing them

The defining difference between static and dynamic few-shot is that examples live in a store, not in the prompt source code. When a new request arrives, you query the store for the K examples most similar to it and splice them into the prompt. Three problems disappear at once.

• Cost: token usage stays roughly flat regardless of how many examples you've collected, because you only ever inject K of them. A static prompt with 30 examples burns 30× the example tokens on every request; a dynamic prompt with 30 examples in its store still burns the same K=3 worth.
• Quality: the examples you do inject are tuned to the input, so generic examples don't water down your prompt for specific cases. A user asking about invoices gets invoice examples; a user asking about expenses gets expense examples; nobody pays the price of mixing both.
• Maintainability: adding examples means writing to a store, not editing prompt strings — your prompt template and your example library evolve independently. This sounds minor until you start onboarding a non-engineer to curate examples and they don't have to touch the codebase.

The "growing" part matters as much as the retrieval part. After each generation, you collect a signal — user thumbs-up, downstream success metric, or human review — and feed positive examples back into the store. Over time it stops being a prompt and starts being a corpus of validated successes. That's where the real lift comes from. The first month feels like a small accuracy bump; the third month feels like you've built something other teams want to copy.

Architecture — three loops, layered

Before the code, picture the system as three loops that overlap. You don't need all three on day one — in fact you shouldn't try.

• Query-time loop (sync): input → embed → top-K vector search → prompt → Gemini call. This is the only loop you need to ship.
• Collection loop (near-sync): output → user/automatic rating → save high-rated outputs as candidate examples. Add this once you have real users giving you feedback.
• Curation loop (async): review the candidate pool periodically, dedupe and quality-check, promote winners into the canonical store. Add this once the candidate pool starts to feel noisy.

Ship the sync loop first. The other two layers can wait until you've actually got production traffic. We'll build the sync loop end-to-end below, then add the rest.

A useful mental model is to treat the example store like a production database: it has a schema, it has writes you trust (curated examples) and writes you don't (raw feedback), and it benefits from the same patterns of staging and promotion you'd use for any data pipeline. The mistake I see most often is treating it as throwaway prompt state — once you do that, you can't reason about why the system is behaving the way it is six months in.

Implementation 1: bootstrap the example store

Start in memory with numpy so you can iterate quickly. Production should swap this for PostgreSQL with pgvector or Vertex AI Vector Search, but the interface stays identical, which is the point — you want your application code to not care which backing store it's hitting.

# pip install google-genai numpy
import os
import json
import numpy as np
from google import genai
from google.genai import types
 
client = genai.Client(api_key=os.environ["GEMINI_API_KEY"])
 
# Seed examples — your starting point, hand-curated
SEED_EXAMPLES = [
    {
        "input": "Issue an invoice to Sample Corp, subject: May support fees.",
        "output": json.dumps({
            "intent": "invoice_create",
            "client": "Sample Corp",
            "subject": "May support fees",
        }),
        "tags": ["invoice", "form-fill"],
    },
    {
        "input": "Download this month's expense report as CSV.",
        "output": json.dumps({
            "intent": "expense_export",
            "format": "csv",
            "period": "this_month",
        }),
        "tags": ["expense", "export"],
    },
    {
        "input": "Show me the invoices issued last month.",
        "output": json.dumps({
            "intent": "invoice_list",
            "period": "last_month",
        }),
        "tags": ["invoice", "list"],
    },
]
 
def embed_text(text: str) -> np.ndarray:
    """Embed a document with gemini-embedding-001 (768 dims), L2-normalized."""
    resp = client.models.embed_content(
        model="gemini-embedding-001",
        contents=text,
        config=types.EmbedContentConfig(
            task_type="RETRIEVAL_DOCUMENT",
            output_dimensionality=768,
        ),
    )
    vec = np.array(resp.embeddings[0].values, dtype=np.float32)
    # Normalize so cosine similarity reduces to a plain dot product later
    return vec / np.linalg.norm(vec)
 
def build_store(examples):
    vectors = np.stack([embed_text(ex["input"]) for ex in examples])
    return {"vectors": vectors, "examples": examples}
 
EXAMPLE_STORE = build_store(SEED_EXAMPLES)
print(f"[init] stored {len(EXAMPLE_STORE['examples'])} examples")
# Expected output: [init] stored 3 examples

Note the task_type="RETRIEVAL_DOCUMENT" here. Gemini Embeddings encodes documents and queries differently when you give them asymmetric task types, and that asymmetry buys you about 20% better recall in my measurements. Don't shortcut this in production — it's the cheapest accuracy win in the whole pipeline. The same applies to dimensionality: 768 is a good middle ground; 1536 is rarely worth the storage cost; 256 starts to lose semantic resolution on shorter inputs. If you're storing fewer than 10,000 examples and don't have a hard latency budget, leave it at 768.

One subtle thing: I bake the L2 normalization into embed_text rather than into the search function. This way every consumer of the store gets normalized vectors automatically — there's no way to accidentally compare a normalized vector to a raw one and get nonsense distances. Treating normalization as a property of the store itself, not a search-time concern, is one of those decisions that pays off the third time someone forgets to normalize manually.

Implementation 2: dynamic selection at query time

Embed the incoming query as RETRIEVAL_QUERY and pull the top-K from the store.

def embed_query(text: str) -> np.ndarray:
    resp = client.models.embed_content(
        model="gemini-embedding-001",
        contents=text,
        config=types.EmbedContentConfig(
            task_type="RETRIEVAL_QUERY",
            output_dimensionality=768,
        ),
    )
    vec = np.array(resp.embeddings[0].values, dtype=np.float32)
    return vec / np.linalg.norm(vec)
 
def select_examples(store, query: str, k: int = 3, min_score: float = 0.55):
    """Top-K by cosine similarity, with a floor to avoid selecting noise."""
    qvec = embed_query(query)
    scores = store["vectors"] @ qvec  # vectors are normalized, so dot = cosine
    order = np.argsort(-scores)
    selected = []
    for idx in order[:k]:
        score = float(scores[idx])
        if score < min_score:
            break
        selected.append({**store["examples"][idx], "score": score})
    return selected
 
candidates = select_examples(
    EXAMPLE_STORE,
    "Pull together every invoice from last month",
    k=2,
)
for c in candidates:
    print(f"score={c['score']:.3f} input={c['input'][:30]}")
# Expected (order reflects semantic proximity):
# score=0.812 input=Show me the invoices issued l
# score=0.612 input=Issue an invoice to Sample Co

The min_score floor is the easy thing to get wrong. Set it to 0 and you'll always inject K examples, even when the store has nothing relevant — which then drags Gemini in the wrong direction. Start at 0.55 and tune downward only after you've watched the score distribution in production.

What value of K should you pick? In my experience, K=3 is the sweet spot for most classification and form-filling tasks. K=5 buys you marginal accuracy on borderline inputs at the cost of meaningfully more tokens. K=1 is tempting from a cost perspective but loses the "vote" effect of multiple examples reinforcing the same pattern. If you have a strict token budget — say, you're paying per-character downstream — K=2 with a slightly higher floor (0.65) often outperforms K=3 with a lower one.

You also want a tag-based filter for multi-domain stores. The line selected = [ex for ex in selected if not store_filter or store_filter in ex["tags"]] (omitted from the snippet for brevity) lets you scope the search to "invoice-related examples only" when you know the routing in advance. This is much cleaner than dumping everything into one bag.

Implementation 3: assemble the prompt and call Gemini

Format the selected examples as plain text. The format that works for me is the simplest possible — labeled # Input and # Output blocks with a blank line between them — because Gemini parses that pattern natively.

def build_prompt(query: str, examples: list[dict]) -> str:
    blocks = [f"# Input\n{ex['input']}\n# Output\n{ex['output']}" for ex in examples]
    examples_text = "\n\n".join(blocks)
    return (
        "You convert natural-language commands for an internal tool into JSON.\n"
        "Match the format of the examples below, then produce JSON for the final input.\n\n"
        f"{examples_text}\n\n"
        f"# Input\n{query}\n# Output\n"
    )
 
def classify(query: str, k: int = 3):
    examples = select_examples(EXAMPLE_STORE, query, k=k)
    if not examples:
        # Fallback when no example crosses the similarity floor
        return _zero_shot(query)
    prompt = build_prompt(query, examples)
    response = client.models.generate_content(
        model="gemini-2.5-flash",
        contents=prompt,
        config=types.GenerateContentConfig(
            temperature=0.1,
            response_mime_type="application/json",
        ),
    )
    return {
        "result": json.loads(response.text),
        "selected_examples": [ex["input"] for ex in examples],
    }
 
def _zero_shot(query: str):
    response = client.models.generate_content(
        model="gemini-2.5-flash",
        contents=(
            "Convert the natural-language command below into a JSON object "
            "containing intent and the necessary parameters.\n\n"
            f"Input: {query}"
        ),
        config=types.GenerateContentConfig(
            temperature=0.1,
            response_mime_type="application/json",
        ),
    )
    return {"result": json.loads(response.text), "selected_examples": []}
 
out = classify("Pull together every invoice from last month")
print(json.dumps(out, indent=2))
# Expected:
# {
#   "result": {"intent": "invoice_list", "period": "last_month"},
#   "selected_examples": ["Show me the invoices issued last month.", "..."]
# }

response_mime_type="application/json" constrains Gemini to return a single JSON value, which lines up perfectly with the JSON outputs in your examples. That alignment is what makes few-shot work — the format you demonstrate is the format you get. If you serialize your example outputs with indent=2 while your prompt asks for compact JSON, you're confusing the model. Pick one style and use it consistently across seed and promoted examples.

There's one more subtle move worth highlighting: returning the selected_examples alongside the result. This makes every response observable. When something goes wrong in production, the first question is always "which examples drove this answer?", and threading the IDs all the way through to the response — or at least to your structured logs — saves hours of debugging.

Implementation 4: growing the store from production signals

The dynamic part comes alive once you start feeding back what worked. The minimum viable feedback loop has three pieces.

• Collection: tag every inference with a unique ID so a rating can attach to it later.
• Rating: capture user thumbs-up/down or operator review.
• Promotion: only positively-rated outputs become permanent few-shot examples.

I'd start by handling explicit positives only and ignoring everything else. Adding LLM-as-judge — see Prompt Evaluation & Optimization Pipeline for that — is a great upgrade once the basic loop is healthy. The reason I prefer manual signals at first: noisy labels are worse than no labels. An LLM-as-judge running before you've calibrated it can systematically promote bad examples into the store, where they then poison every subsequent retrieval.

import uuid
from datetime import datetime, timezone
 
PENDING_BUFFER: dict[str, dict] = {}
 
def classify_with_logging(query: str, k: int = 3):
    examples = select_examples(EXAMPLE_STORE, query, k=k)
    prompt = build_prompt(query, examples) if examples else None
    response = client.models.generate_content(
        model="gemini-2.5-flash",
        contents=prompt or query,
        config=types.GenerateContentConfig(
            temperature=0.1,
            response_mime_type="application/json",
        ),
    )
    rec_id = str(uuid.uuid4())
    PENDING_BUFFER[rec_id] = {
        "input": query,
        "output": response.text,
        "selected_examples": [ex["input"] for ex in examples],
        "ts": datetime.now(timezone.utc).isoformat(),
    }
    return rec_id, json.loads(response.text)
 
def submit_feedback(rec_id: str, score: int, store=EXAMPLE_STORE):
    """score is 1 (good) / 0 (neutral) / -1 (bad). Only 1 promotes."""
    rec = PENDING_BUFFER.get(rec_id)
    if rec is None:
        raise KeyError(f"unknown rec_id: {rec_id}")
    if score == 1:
        promote_to_store(store, rec)
    elif score == -1:
        log_failure(rec)
 
def promote_to_store(store, rec, dedup_threshold: float = 0.93):
    """Skip near-duplicates so the store stays diverse."""
    vec = embed_text(rec["input"])
    if len(store["vectors"]) > 0:
        sims = store["vectors"] @ vec
        if float(sims.max()) >= dedup_threshold:
            return False
    store["vectors"] = np.vstack([store["vectors"], vec[np.newaxis, :]])
    store["examples"].append({
        "input": rec["input"],
        "output": rec["output"],
        "tags": ["promoted"],
    })
    return True
 
def log_failure(rec: dict):
    # In production, send this to SQLite/BigQuery for later curation
    print(f"[failure] {rec['input'][:40]}…")
 
rec_id, result = classify_with_logging("Export this month's expenses as CSV")
print("classified:", result)
submit_feedback(rec_id, score=1)
print(f"[grow] store size: {len(EXAMPLE_STORE['examples'])}")
# Expected:
# classified: {"intent": "expense_export", "format": "csv", "period": "this_month"}
# [grow] store size: 4

dedup_threshold is the parameter that decides whether your store grows in a healthy way or collapses into a pile of near-duplicates. Set it too low and the store fills with paraphrases of the same example, killing diversity. Too high, and obviously different inputs get rejected as duplicates. I usually start at 0.93 and tune down to 0.90 once I see what production looks like.

A gotcha I hit in my own deployment: PENDING_BUFFER as an in-memory dict works only for single-process services. The moment you scale horizontally, feedback for a request handled by pod A might arrive at pod B, and pod B has no buffer entry to attach it to. The fix is straightforward — back the buffer with Redis or a small Postgres table — but it's the kind of thing that bites you on day one of horizontal scale. If you're going to run more than one replica, set this up before promoting the loop to production.

Implementation 5: A/B test against the static baseline

You won't trust dynamic few-shot until you can prove it's better. Run both side-by-side, randomize, and log the arms separately.

import random
 
STATIC_EXAMPLES_TEXT = "\n\n".join(
    f"# Input\n{e['input']}\n# Output\n{e['output']}" for e in SEED_EXAMPLES
)
 
def classify_static(query: str):
    prompt = (
        "Convert the natural-language command below into JSON.\n\n"
        f"{STATIC_EXAMPLES_TEXT}\n\n# Input\n{query}\n# Output\n"
    )
    resp = client.models.generate_content(
        model="gemini-2.5-flash",
        contents=prompt,
        config=types.GenerateContentConfig(
            temperature=0.1,
            response_mime_type="application/json",
        ),
    )
    return json.loads(resp.text)
 
def classify_ab(query: str):
    arm = "dynamic" if random.random() < 0.5 else "static"
    if arm == "dynamic":
        rec_id, result = classify_with_logging(query)
    else:
        result = classify_static(query)
        rec_id = None
    return arm, rec_id, result

The metric to chase is whatever the downstream system actually uses, but as a first cut, "JSON parsed cleanly and matched the expected intent" is enough. In my own deployment, once the store crossed about 30 examples, the dynamic arm pulled 8 percentage points ahead of the static one on intent accuracy while cutting input tokens by 45%. The gap widens as the store grows. If you don't see at least a token reduction in week one, something's wrong in your selection logic — accuracy may take time to materialize, but token savings should be immediate.

For longer-term measurement, segment by input length. Short inputs (under ~50 tokens) tend to benefit less from dynamic few-shot because the static prompt was already small enough. Long, complex inputs are where the gap opens up — and they're also where every percentage point of accuracy matters most. Slicing your A/B results by input length almost always reveals a story that the global average hides.

Common pitfalls to avoid

In rough priority order, here are the mistakes I've seen most often — including a few I made myself.

• Symmetric task types: encoding both the document and the query as SEMANTIC_SIMILARITY is a real recall regression. Use RETRIEVAL_DOCUMENT for stored examples and RETRIEVAL_QUERY for incoming queries. The asymmetric encoding is exactly what Gemini Embeddings was designed for; not using it leaves accuracy on the table.
• Skipping L2 normalization: without normalization, dot products no longer equal cosine similarity, and longer inputs spuriously score higher. Always normalize at the end of embed_text. I've debugged a half-day outage that turned out to be a single missing np.linalg.norm call after a refactor.
• min_score = 0: forcing the system to always pick K examples introduces noise on edge inputs. Use a floor and fall back to zero-shot when nothing crosses it. Watch the rate of zero-shot fallbacks as a health metric — a sudden spike means your input distribution has shifted into territory your store doesn't cover.
• Promoting duplicates: blindly accepting every positive feedback fills the store with near-clones. Enforce dedup_threshold between 0.90 and 0.93. Before you lower the threshold further, check whether the new examples actually contribute novel information — if not, you're trading store size for nothing.
• One store per service, not per task: putting "invoicing" and "customer support" examples in the same store causes cross-domain bleeding at retrieval time. Tag aggressively or split the stores. The general rule: if two tasks have different output schemas, they should have different stores.
• Linear search at scale: above ~1,000 examples, p95 latency starts to slip. Migrate to Vertex AI Vector Search or pgvector before you feel the pain. Approximate nearest neighbor (ANN) is more than fast enough for K=3 retrieval; the small recall hit doesn't matter at this scale.
• Discarding failures: thumbs-down examples are gold. Save them in a separate store and review them monthly to refine the system instruction. Many of the best system-instruction edits I've made came from staring at a stack of failure cases for an hour.
• Embedding model lock-in: the day Google releases gemini-embedding-002 and recommends you migrate, you'll have a store full of v1 vectors. Keep the original input text alongside the vector so re-embedding is just a batch job, not a data archeology project.

Measuring impact — both sides of the trade-off

Don't only look at accuracy. The pitch for dynamic few-shot is that you can grow your example library indefinitely without paying a token tax — that promise needs to be visible in the metrics.

These five fields should always make it into your logs. Without them you can't reproduce why a particular request got the answer it did.

• selected_example_ids: ordered list of examples that ended up in the prompt.
• score_top1: cosine similarity of the top match.
• prompt_tokens: tokens charged for the input.
• output_tokens: tokens charged for the response.
• model_version: Gemini variant used for the call.

I pipe these into Cloud Logging → BigQuery and visualize them in Looker Studio — see Looker Studio Report AI Analysis for the wiring. Two charts are particularly useful: the distribution of score_top1 (tells you whether your floor is calibrated) and the frequency of selected_example_ids (tells you which examples are doing the work and which are dead weight).

If a small handful of examples account for the majority of selections, you've got either an overly narrow store or a few extremely effective examples. The fix depends on which it is — diversify the store in the first case, or explicitly study what made those examples so effective in the second. The "dead weight" examples that almost never get selected can usually be retired; they're costing storage and slowing the search without contributing.

Choosing your vector store — when to graduate from in-memory

The in-memory numpy store in Implementation 1 is intentionally minimal. It's the right starting point for exploration and the wrong place to be once the system matters. Here's how I've staged the upgrades in past projects.

Phase 1 — In-memory (0 to ~500 examples): a Python dict and a numpy array on a single process. Latency is microseconds, the deployment story is "restart the process and the store is gone", and you'll happily seed it from a JSON file at boot. This is fine for early evaluation and internal tooling, and it's where I do most of the prompt iteration even today.

Phase 2 — SQLite or Postgres + pgvector (500 to ~50,000 examples): now the store outlives a process restart, supports concurrent writes, and gets backed up like the rest of your production data. pgvector is excellent here; the search is exact at this scale, and the ergonomics of "SELECT examples WHERE tag = 'invoicing' ORDER BY embedding <=> $1 LIMIT 5" are genuinely nice. Latency stays under 50 ms for K=3 retrieval up to roughly 50k rows.

Phase 3 — Vertex AI Vector Search (50,000+ examples or strict latency SLAs): managed approximate nearest neighbor with sub-10 ms latency at hundreds of thousands of vectors, plus painful but useful metadata filtering. You give up the simplicity of "just write a SQL query" for true horizontal scale. Most production systems I've seen never reach Phase 3, and that's fine — pgvector handles a lot.

The key abstraction is to keep your application code calling select_examples(store, query, k) regardless of which phase you're in. The store object can be a dict, a Postgres connection, or a Vertex client; the call site doesn't care. If you're disciplined about that boundary from day one, the migration between phases is a few hundred lines of glue, not a rewrite.

There's also a non-obvious operational question: how often should you rebuild embeddings? Almost never, until you change the embedding model. When you do, run a one-shot batch that re-embeds every example using the new model and writes to a new store version, then flip a feature flag to point reads at the new store. Don't try to do partial migrations — half the store on v1 and half on v2 is incoherent because vectors from different models don't share a comparable space.

Wrapping up — three steps for tomorrow morning

The code in this article fits in a few hundred lines, but the value compounds over months in a way that's hard to appreciate until you've lived through it once. Once the feedback loop runs for 30, 60, 90 days, the store turns into something no hand-curated few-shot can compete with. The team I built this for went from an 81% intent accuracy at launch to 93% three months later, with no prompt edits in between — the entire delta came from the store growing.

If I were starting from scratch tomorrow, I'd do it in this order. First, gather 5–10 hand-written examples for one of your existing tasks and load them into the in-memory store. Second, scaffold the A/B harness before you have anything interesting to compare — it's much harder to add later. Third, push a single thumbs-up through the promotion path end-to-end so you've actually exercised the growth loop. After that one round-trip, the system tends to take care of itself, and the operational work shifts from prompt edits to occasional curation reviews — a much healthier place to be.

One last thought before you start: the discipline of "every prompt change is a measurable change" is what eventually separates teams that get LLMs to work in production from teams that don't. Dynamic few-shot is a particularly clean version of that discipline because every example added to the store is a small, individually-testable hypothesis: "did adding this example move the metric?" When you stack a few hundred of those small wins over six months, you arrive somewhere no static prompt could have taken you.

If your few-shot prompt has been staring at you for too long, I hope this gives you a path out. The architecture is small enough to ship in an afternoon, and the operational shape — store, retrieve, grow — generalizes to any task where you've been thinking "if only I could pick the right examples for each input." That's most LLM tasks, in my experience.