Tracking which prompt version served each request

Shipping a prompt change to production without tracking prompt version per production request is like deploying code without git history. When a regression surfaces in user-facing latency or output quality, you need to know exactly which prompt text served that request to reproduce and rollback. This guide lays out a concrete pipeline for versioning prompts, propagating that version through your stack, and verifying the trail end to end.

Step 1: Version Prompts as Immutable Artifacts in Git

Treat prompts like source code, not configuration you tweak in a UI. Store each prompt as a JSON file under a prompts/ directory, tagged with semantic version in the filename and inside the file. Never mutate a released version; cut a new file.

// prompts/summarize/v1.0.0.json
{
  "version": "1.0.0",
  "model": "gpt-4o-mini",
  "temperature": 0.2,
  "messages": [
    {"role": "system", "content": "Summarize the user text in two sentences."}
  ]
}

// prompts/summarize/v1.1.0.json
{
  "version": "1.1.0",
  "model": "gpt-4o-mini",
  "temperature": 0.1,
  "messages": [
    {"role": "system", "content": "You are a strict editor. Summarize in <=30 words."}
  ]
}

Commit these in a separate prompts repo or a subdirectory. Tag releases with git tag -a prompt-summarize-1.1.0 -m "tighter summary". This gives you a diffable history and a unambiguous identifier for tracking prompt version per production request later.

Step 2: Resolve the Active Version at Request Time

Do not hardcode versions in handlers. Load a small routing config that maps task → version, with support for canaries. At request time, resolve the version before you build the LLM call.

import json
from pathlib import Path
from functools import lru_cache

PROMPT_DIR = Path("prompts")

ROUTING = {
    "summarize": {"default": "1.1.0", "canary": {"version": "1.2.0-rc1", "pct": 5}},
}

@lru_cache(maxsize=128)
def load_prompt(task: str, version: str) -> dict:
    path = PROMPT_DIR / task / f"v{version}.json"
    return json.loads(path.read_text())

def resolve_version(task: str, request_id: str) -> str:
    cfg = ROUTING.get(task, {})
    # deterministic canary by hash of request_id
    if cfg.get("canary") and (hash(request_id) % 100) < cfg["canary"]["pct"]:
        return cfg["canary"]["version"]
    return cfg.get("default", "latest")

This keeps the version decision centralized and testable. The resolved string is the single source of truth you will propagate.

Step 3: Propagate Prompt Version Through Request Context

The version must travel with the request from the edge to the model call. Use contextvars in Python (or AsyncLocalStorage in Node) so it survives across async hops and is available to logging middleware.

import contextvars
from dataclasses import dataclass

prompt_version_ctx = contextvars.ContextVar("prompt_version")
request_id_ctx = contextvars.ContextVar("request_id")

@dataclass
class Ctx:
    request_id: str
    prompt_version: str

def bind_ctx(req_id: str, task: str):
    version = resolve_version(task, req_id)
    request_id_ctx.set(req_id)
    prompt_version_ctx.set(version)
    return Ctx(req_id, version)

When you call the model, attach the version to the request. OpenAI’s user field is limited to 36 chars but can encode a short version; otherwise pass it as a header your gateway forwards.

import openai

def complete(task: str, user_input: str):
    version = prompt_version_ctx.get()
    req_id = request_id_ctx.get()
    prompt = load_prompt(task, version)
    return openai.chat.completions.create(
        model=prompt["model"],
        temperature=prompt.get("temperature", 0.2),
        messages=prompt["messages"] + [{"role": "user", "content": user_input}],
        user=f"{req_id[:8]}:{version}"[:36],
    )

Tracking prompt version per production request becomes trivial when the version rides in the same context object as the request id and is emitted on every log line.

Step 4: Emit Structured Logs With Version and Token Counts

Standardize on JSON logs. Every LLM completion should emit one line containing the version, request id, model, and token usage. This is the backbone of your observability.

import logging, json, time

logger = logging.getLogger("llm")

def log_completion(ctx: Ctx, model: str, usage):
    logger.info(json.dumps({
        "request_id": ctx.request_id,
        "prompt_version": ctx.prompt_version,
        "model": model,
        "prompt_tokens": usage.prompt_tokens,
        "completion_tokens": usage.completion_tokens,
        "ts": time.time(),
    }))

If you use a gateway, ensure its per-token metering aligns; correlate by request id. Avoid logging the full prompt text in production—it bloats storage and risks leaking PII. The version string is enough to reconstruct the exact prompt from git.

Step 5: Persist the Mapping in a Queryable Store

Logs are for tailing; a database is for auditing. Write the version→request mapping to Postgres (or BigQuery) in the same transaction as your application’s business record if possible.

CREATE TABLE prompt_serving_log (
    request_id      TEXT PRIMARY KEY,
    prompt_version  TEXT NOT NULL,
    model           TEXT,
    created_at      TIMESTAMPTZ DEFAULT now()
);

import psycopg2

def persist(ctx: Ctx, model: str):
    with psycopg2.connect("postgres://app@localhost/prod") as conn:
        conn.execute(
            "INSERT INTO prompt_serving_log VALUES (%s,%s,%s) "
            "ON CONFLICT (request_id) DO NOTHING",
            (ctx.request_id, ctx.prompt_version, model),
        )

An inference gateway that honors client routing directives (e.g., n4n.ai) can carry your request id through to the provider, letting you reconcile tracking prompt version per production request with billed tokens by joining on that id. This removes guesswork when a provider invoice spikes.

Step 6: Correlate With Provider or Gateway Exports

Provider usage files or gateway exports typically include the user field or a request header you controlled. Pull those into the same warehouse.

# example: export from gateway, then load
gateway-cli export-usage --start 2024-05-01 --end 2024-05-02 > usage.jsonl
psql prod -c "COPY gateway_usage FROM 'usage.jsonl' CSV"

Now you can answer questions like: “Which prompt version consumed the most tokens yesterday?” or “Did v1.2.0-rc1 regress completion length?”

SELECT prompt_version, sum(completion_tokens) AS toks
FROM gateway_usage u
JOIN prompt_serving_log p ON p.request_id = u.request_id
GROUP BY 1 ORDER BY 2 DESC;

Step 7: Verify Success With a Synthetic Request

A pipeline is only real if a test proves it. Stand up your service locally, fire a request with a forced version, and assert the trail exists.

curl -X POST localhost:8080/summarize \
  -H "content-type: application/json" \
  -d '{"text":"Test input","force_version":"1.0.0"}'

In the handler, respect force_version for internal tests only. Then check the log line and the table:

grep '"prompt_version":"1.0.0"' logs/app.log | head -1

SELECT request_id, prompt_version, model
FROM prompt_serving_log
WHERE prompt_version = '1.0.0'
ORDER BY created_at DESC LIMIT 1;

If the row exists with the same request_id your curl returned, the chain works. Add this as a CI smoke test against a stub LLM endpoint so regressions in context propagation fail the build.

Operational Caveats

• Immutability: Never git push --force a prompt file that already served traffic. Treat versions as append-only.
• Canary safety: Resolve canary by hashed request id, not random, so a single user gets consistent behavior across retries.
• Cache busting: If you use provider prompt caching, changing the version string in the user field or system message will naturally bust the cache. That’s correct—you want separate cache keys per version.
• Retention: Keep prompt_serving_log for as long as you keep model outputs; you’ll need it for incident reviews.

Following these steps gives you a defensible audit trail. When someone asks why production output changed on Tuesday, you query one table, pull the git blob for that version, and reproduce exactly what ran.