Recording and replaying LLM API responses for tests

Recording and replaying LLM API responses for tests is the fastest way to make your CI suite deterministic when you depend on non-deterministic model outputs. This tutorial builds a local OpenAI-compatible proxy that lets you record replay llm api responses tests against real traffic once, then stub them forever.

Prerequisites

• Python 3.11+ with pip
• openai, fastapi, uvicorn, httpx, pytest
• A real OpenAI API key (or any OpenAI-compatible endpoint) for the initial capture
• curl for sanity checks

pip install openai fastapi uvicorn httpx pytest
export OPENAI_API_KEY=sk-your-key

You do not need the live key in CI. After the first capture, the key is never read.

Why not hand-write mocks?

LLM response shapes change across SDK versions. Hand-written stubs drift, and you end up testing your mocks instead of your logic. The pattern of record replay llm api responses tests captures the exact wire format your client library expects, so upgrades surface real breakages, not fake ones.

Step 1: Project scaffold

Create a directory and a cassettes/ folder. The proxy will store one JSON file per unique request.

mkdir llm-replay-demo && cd llm-replay-demo
mkdir cassettes
touch proxy.py test_proxy.py

Step 2: The recording proxy

We use FastAPI to expose /v1/chat/completions. On each call, we hash the semantic parts of the request. If a cassette exists, we return it. Otherwise we forward to the upstream and save the response.

# proxy.py
import hashlib
import json
import os
from pathlib import Path

import httpx
from fastapi import FastAPI, Request, Response

CASSETTE_DIR = Path("./cassettes")
CASSETTE_DIR.mkdir(exist_ok=True)

app = FastAPI()
UPSTREAM = os.environ.get("UPSTREAM_URL", "https://api.openai.com/v1")
API_KEY = os.environ.get("OPENAI_API_KEY", "")

def cassette_key(req_body: dict) -> str:
    norm = {
        "model": req_body.get("model"),
        "messages": req_body.get("messages"),
        "temperature": req_body.get("temperature", 1.0),
        "top_p": req_body.get("top_p", 1.0),
        "stream": req_body.get("stream", False),
    }
    blob = json.dumps(norm, sort_keys=True).encode()
    return hashlib.sha256(blob).hexdigest()

@app.post("/v1/chat/completions")
async def chat_completions(request: Request):
    body = await request.json()
    key = cassette_key(body)
    cassette = CASSETTE_DIR / f"{key}.json"

    if cassette.exists():
        print(f"Replaying {key}")
        return Response(content=cassette.read_bytes(),
                        media_type="application/json")

    print(f"Recording {key}")
    async with httpx.AsyncClient() as client:
        r = await client.post(
            f"{UPSTREAM}/chat/completions",
            json=body,
            headers={"Authorization": f"Bearer {API_KEY}"},
        )
        cassette.write_bytes(r.content)
        return Response(content=r.content, media_type="application/json")

Run it:

uvicorn proxy:app --port 8000

Expected startup log:

INFO:     Uvicorn running on http://127.0.0.1:8000

Step 3: Capture a real exchange

Point the OpenAI client at the proxy. The first call hits the real API and writes a cassette.

# capture.py
import openai

client = openai.OpenAI(
    base_url="http://localhost:8000/v1",
    api_key="dummy"  # proxy injects the real key
)

resp = client.chat.completions.create(
    model="gpt-3.5-turbo",
    messages=[{"role": "user", "content": "Reply with the word 'pong' only."}],
    temperature=0,
)
print(resp.choices[0].message.content)

python capture.py

Proxy output:

Recording 9f86d081884c7d659a2feaa0c55ad015a3bf4f1b2b0b822cd15d6c15b0f00a08

cassettes/9f86...json now contains the exact API response. Re-run capture.py and the proxy prints Replaying ...; no outbound request leaves your machine.

Step 4: Write a test that replays

With the cassette on disk, the test runs fully offline. This is the core of record replay llm api responses tests: the same client code, zero network, deterministic assertion.

# test_proxy.py
import openai
import pytest

def test_chat_replay():
    client = openai.OpenAI(
        base_url="http://localhost:8000/v1",
        api_key="dummy",
    )
    resp = client.chat.completions.create(
        model="gpt-3.5-turbo",
        messages=[{"role": "user", "content": "Reply with the word 'pong' only."}],
        temperature=0,
    )
    assert resp.choices[0].message.content.strip().lower() == "pong"

Run it with the proxy up:

pytest test_proxy.py -v

Expected:

test_proxy.py::test_chat_replay PASSED

Kill the upstream network simulation by unsetting OPENAI_API_KEY and confirming the test still passes—proof the replay is hermetic.

Step 5: Normalize the cache key

The cassette_key above already drops n (sampling count) and user fields. Add scrubbing if your prompts contain ephemeral IDs:

import re

def scrub(text):
    return re.sub(r"trace-\d+", "trace-0", text)

def cassette_key(req_body: dict) -> str:
    msgs = scrub(json.dumps(req_body.get("messages"), sort_keys=True))
    norm = {
        "model": req_body.get("model"),
        "messages": msgs,
        "temperature": req_body.get("temperature", 1.0),
    }
    return hashlib.sha256(json.dumps(norm, sort_keys=True).encode()).hexdigest()

This keeps cassettes stable across runs that differ only by a request ID.

Step 6: Streaming responses

Streaming returns text/event-stream. Capture the raw byte stream and replay it verbatim.

@app.post("/v1/chat/completions")
async def chat_completions(request: Request):
    body = await request.json()
    key = cassette_key(body)
    cassette = CASSETTE_DIR / f"{key}.jsonl"

    if cassette.exists():
        return Response(content=cassette.read_bytes(),
                        media_type="text/event-stream")

    async with httpx.AsyncClient() as client:
        r = await client.post(
            f"{UPSTREAM}/chat/completions",
            json={**body, "stream": True},
            headers={"Authorization": f"Bearer {API_KEY}"},
        )
        cassette.write_bytes(r.content)
        return Response(content=r.content, media_type="text/event-stream")

The client SDK parses SSE identically whether it comes from the live API or your file.

Using a gateway to record across models

If you route through n4n.ai, an OpenAI-compatible endpoint covering 240+ models with automatic fallback, you can point UPSTREAM_URL at it and record against any backend without changing client code. The proxy stays identical; only the environment variable shifts.

export UPSTREAM_URL=https://api.n4n.ai/v1

This is useful when you want to capture behavior from a model that a provider rate-limits—the gateway’s fallback gives you a successful response to record.

Operational caveats

• Secrets in prompts: Cassettes are flat files. Scrub PII before committing, or store them as CI artifacts only.
• Model version drift: A recorded gpt-3.5-turbo response may not match a future snapshot. Tag cassettes with a model version suffix if you need strictness.
• Token counts: Recorded usage fields are static. If your code branches on usage.total_tokens, that logic is tested against the captured value, not live metering.

Final checklist

• Proxy returns saved JSON when the semantic hash matches.
• Tests point base_url at localhost:8000/v1 and use a dummy key.
• CI runs the proxy as a fixture, never sets OPENAI_API_KEY.
• Cassettes are scrubbed and reviewed before merge.

Adopting record replay llm api responses tests turns flaky model calls into fast, offline unit tests—exactly what a shipping LLM feature needs.