48 小時實作筆記：以 Python 建立模型 API 回放工具

At hour three of a prompt-refactor session, I asked myself a question that should have had an obvious answer. Did the new system prompt actually improve the parser output, or did it just change it in a way that felt better? I could not answer, because I had no frozen baseline and every comparison cost another live call.

Worse, the "identical" request I sent ten seconds later came back with different wording. Sampling settings explain part of that, batching and server-side routing explain the rest, and none of it matters when you are trying to tell improvement from noise. So I stopped tuning prompts and spent the rest of the window building a cassette recorder instead.

What follows is the 48-hour log: the three requirements, the four failures, the code that survived, and the cases where this whole approach is the wrong tool.

The three requirements I wrote down before touching code

I forced myself to state the contract first, because cache layers rot when their contract stays implicit.

1. A replay run must be free. I want to run the same suite forty times while fixing a regex in the parser, without spending a single live call.
2. A miss must be loud. If a recorded response is missing, the harness exits with code 2 and prints the fingerprint. It must never silently fall through to the network.
3. The key must be explainable. I want to print the fingerprint next to the request and see exactly which field moved.

Requirement 2 is the one people skip, and it is the one that turns a cache into a test harness. A cache that quietly calls the network on a miss hides the exact regression you built it to catch.

The artifact: a dependency-free cassette

Everything below is standard-library Python, and it works against any HTTP model endpoint because it wraps your call function instead of a vendor SDK. The snippet is a working proposal from this session, tested only against the endpoints I had access to.

# cassette.py — stdlib only, Python 3.10+
import hashlib, json, os, pathlib, time

CASSETTE_DIR = pathlib.Path(os.environ.get("CASSETTE_DIR", "./cassettes"))
CASSETTE_DIR.mkdir(parents=True, exist_ok=True)
LOG = CASSETTE_DIR / "calls.jsonl"

class CassetteMiss(RuntimeError):
    def __init__(self, key, payload):
        super().__init__(f"cassette miss: {key}")
        self.key, self.payload = key, payload

def key_for(payload: dict) -> str:
    blob = json.dumps(payload, sort_keys=True, separators=(",", ":"), ensure_ascii=False)
    return hashlib.sha256(blob.encode("utf-8")).hexdigest()

The store is append-only JSONL, which survives a hard stop better than a rewritten JSON file and is trivial to grep and diff.

def load_index() -> dict[str, dict]:
    if not LOG.exists():
        return {}
    index = {}
    for line in LOG.read_text(encoding="utf-8").splitlines():
        if not line.strip():
            continue
        rec = json.loads(line)
        index[rec["key"]] = rec   # later wins; earlier records stay for audit
    return index

def call(payload: dict, live_fn, mode: str | None = None) -> dict:
    mode = mode or os.environ.get("CASSETTE_MODE", "replay")
    key = key_for(payload)
    index = load_index()
    if mode == "replay":
        if key not in index:
            raise CassetteMiss(key, payload)
        return index[key]["response"]
    response = live_fn(payload)
    with LOG.open("a", encoding="utf-8") as fh:
        fh.write(json.dumps({"key": key, "ts": time.time(),
                             "payload": payload, "response": response}) + "\n")
    return response

Then a tiny runner that maps the miss onto exit code 2:

# run.py
import json, sys
from cassette import call, CassetteMiss

try:
    for case in suite():
        out = call(case.payload, live_fn=case.live)
        case.assert_output(out)
except CassetteMiss as exc:
    print(f"MISS {exc.key}", file=sys.stderr)
    print(json.dumps(exc.payload, indent=2, sort_keys=True), file=sys.stderr)
    sys.exit(2)

Two commands now define the whole loop, and the second one never touches the network:

CASSETTE_MODE=record python run.py --suite smoke   # costs live calls, once
CASSETTE_MODE=replay python run.py --suite smoke   # free, repeatable, CI-safe

Hour 6: my key matched nothing, then matched the wrong thing

The first fingerprint I wrote hashed the raw request dictionary, and it missed on every replay for three boring reasons.

• Dictionary order. Python preserves insertion order, so two logically equal payloads hashed differently. Fix: sort_keys=True with compact separators.
• Whitespace drift. Trailing newlines inside message content changed the hash but never changed the meaning. Fix: strip trailing spaces per line and collapse repeated blank lines.
• Implicit defaults. {"temperature": 1.0} and {} are the same request to the server and different keys to me. Fix: merge defaults before hashing.

DEFAULTS = {"temperature": 1.0, "top_p": 1.0, "max_tokens": 512, "stream": False}

def normalize_params(params: dict) -> dict:
    merged = {**DEFAULTS, **params}
    return {k: merged[k] for k in sorted(merged)}

def normalize_text(text: str) -> str:
    lines = [line.rstrip() for line in text.strip().splitlines()]
    return "\n".join(line for line in lines if line)

A cache key is an API contract with your past self, and I had been writing it by accident. Write the normalization function first, pin it with a version field, and bump that field on the day its meaning changes.

Then my key was technically correct and still wrong, which is a different problem entirely.

Hour 27: the system prompt that was never in the request I hashed

Somewhere around hour 27 I changed the system prompt, replayed, and got the old answer back with a fully green test run. The fingerprint looked fine. The bug lived in my harness: the system message was injected by a thin client wrapper reading an environment variable, so the dictionary I hashed contained only the user turn.

Two different prompts produced the same key, and replay confidently returned the first answer. I fixed it with two guards I should have written on day one.

import hashlib, pathlib

def template_sha(path: str) -> str:
    return hashlib.sha256(pathlib.Path(path).read_bytes()).hexdigest()[:16]

def fingerprint(model, messages, params, template_path):
    assert messages and messages[0]["role"] == "system", "system prompt must be explicit"
    return {
        "v": 2,                       # bump when key semantics change
        "model": model,
        "template": template_sha(template_path),
        "params": normalize_params(params),
        "messages": [{"role": m["role"], "content": normalize_text(m["content"])}
                     for m in messages],
    }

The lesson generalizes past this project: never let a library assemble the request you are hashing. Build the message list yourself, assert the shape you expect, and put a hash of the prompt file bytes into the key.

Hour 14: streaming broke the cassette, and I made it worse first

My first instinct was to persist chunk boundaries so that a replayed stream looked byte-identical to the original. That was a mistake, because chunking is a transport artifact and providers are free to split differently tomorrow.

What the parser actually needs is the assembled text plus the finish reason, delivered through the same iterator interface:

def replay_stream(text: str, finish_reason: str = "stop", chunk_size: int = 24):
    for i in range(0, len(text), chunk_size):
        yield {"delta": text[i:i + chunk_size], "finish_reason": None}
    yield {"delta": "", "finish_reason": finish_reason}

Now the streaming path is exercised on every replay, the parser sees real partial chunks, and I am not pretending to reproduce a provider's internal buffering. If your parser depends on exact chunk shapes, that is a finding about your parser, not something to hide inside a fixture.

Hour 38: running the suite on a free server

I moved the second half of this work onto MonkeyCode's free server option and used its free model access for the record pass, so what follows is the workflow I exercised rather than a plan. Disclosure: This article was prepared as part of MonkeyCode's product outreach. I did not measure quotas, hardware, or availability windows, and I am deliberately not quoting numbers I cannot reproduce — treat those details as operator-supplied and verify them on the day you read this.

What I can report is the shape of the workflow, and it fits a shared free box for one specific reason. On such a machine the scarce resources are wall-clock time and live-call budget, not replay CPU. The first pass records; every pass after that is local, offline, and independent of the network.

export CASSETTE_DIR=/tmp/cassette-smoke
CASSETTE_MODE=record /usr/bin/time -f "record %e s" python run.py --suite smoke
CASSETTE_MODE=replay /usr/bin/time -f "replay %e s" python run.py --suite smoke
wc -l "$CASSETTE_DIR/calls.jsonl"

Two practical notes from doing this on a remote box instead of my laptop. Keep the cassette directory outside the repo unless the recorded responses are safe to publish, and commit only a small, scrubbed seed cassette for the smoke suite. Keep the suite small enough that a record pass stays comfortable, because a big suite turns the "free" part into a scheduling problem.

When to record, and when to call live

This table is what I actually use now, and it is the part I would copy into another project before copying the code.

The distinction that matters: replay answers "did my code change?" and never answers "is the model good today?" Confusing those two questions is how teams end up trusting a stale fixture for months.

What I would repeat, and what I would skip

Repeat the boring parts: writing the normalization function before the cache, the exit-code-2 miss behavior, and hashing the prompt file instead of trusting the client wrapper. Skip persisting stream chunk boundaries, and skip my first attempt at "just hash the request dictionary".

I would also add one thing much earlier — a stability probe that sends the same payload five times in record mode and reports how many distinct answers come back. That single number tells you how much of your supposed improvement is noise, and I only built it on day two.

Limitations and who should not use this

A cassette is a frozen observation, so it cannot validate quality and it will pass happily after a provider swaps the model behind an alias. Do not store responses containing user data unless you have a scrubbing step you trust, because an append-only log is extremely good at keeping secrets forever. Teams doing continuous evaluation against live traffic should use a real eval pipeline rather than a local cassette.

Skip this if you make one call per day, or if your only recurring question is whether a model is currently reachable. The harness pays for itself when you have a dozen cases and you run them dozens of times.

What are you re-running against a live model right now just to check that your own code still works? If the answer is "everything", the cassette is roughly two hours of work — and if you want a free place to run it, MonkeyCode is worth trying; tell me what breaks on your second pass.