DEV Community

kongkong
kongkong

Posted on

Add AI Features Behind a Transcript Store, Not a Live Call

The demo froze at 4:11 p.m., right when I thought the incident summary feature was finally done. Someone had clicked Generate on a long report, waited, and watched a spinner that never resolved. Was the model merely slow, or had our handler already forgotten what it asked? I opened the API logs, found a 502 from the provider, and discovered nothing durable remained.

That empty log is why I now take a stubborn position on AI feature design. You should not put a live model call on the request path of a user action. You should store a transcript first, then serve the feature from records you can replay. A chat window is a debugging toy, and a transcript table is the actual product surface.

Live calls feel fast because they hide every layer that later fails during production traffic. Auth, persistence, retries, and the UI empty state all get postponed until the demo looks clever. Then the provider blinks, and you cannot answer a simple question from an impatient teammate. What did we send, what came back, and which user is staring at a blank panel?

Think of the model as a warehouse forklift, not as the building that holds the inventory. You would not let customers walk into the aisle and grab pallets during a live shift. You would scan goods into inventory, then pick from records you can audit after midnight. Why do we still let a user request reach the forklift before we have a stock ledger?

I used to wire the provider SDK straight into the POST handler, because every tutorial did that. The first outage taught me that a missing transcript is worse than a wrong summary. You can rewrite a summary, but you cannot reconstruct a prompt that never hit disk. After that, I stopped calling the approach a prototype and started calling it a liability.

The contract I want is boring, and that boredom is the entire point of the design. Every model interaction becomes a row with status, input hash, output, and an actor id. The HTTP handler writes a queued transcript, and a worker may fill the output later. The UI reads the row, not the socket that happens to be talking to a model today.

Here is a schema I actually keep in the migration folder, because arguments die when SQL exists. Copy it before you import another vendor client, and you will thank yourself during the first incident review. Does your current backend even have a durable place to put a failed generation?

CREATE TABLE model_transcripts (
  id              UUID PRIMARY KEY,
  actor_id        UUID NOT NULL,
  feature         TEXT NOT NULL,
  input_hash      CHAR(64) NOT NULL,
  input_payload   JSONB NOT NULL,
  output_payload  JSONB,
  status          TEXT NOT NULL CHECK (
    status IN ('queued', 'running', 'succeeded', 'failed', 'rejected')
  ),
  provider_name   TEXT,
  error_code      TEXT,
  created_at      TIMESTAMPTZ NOT NULL DEFAULT now(),
  updated_at      TIMESTAMPTZ NOT NULL DEFAULT now()
);

CREATE INDEX model_transcripts_actor_created
  ON model_transcripts (actor_id, created_at DESC);
Enter fullscreen mode Exit fullscreen mode

Notice what is missing from that table: there is no column named temperature, and there is no column named vibe. Provider details are optional metadata, because the feature must survive a vendor swap without a rewrite. If your product meaning lives inside an SDK client, do you even have a product yet?

The API then becomes a plan and apply workflow that never waits on a remote token stream. The following handlers are a compact working sketch, not a dump from a named production system. I want the browser to create a transcript, poll it, and render whatever status the row currently holds. That sounds slower in a hallway conversation, and it is faster when the provider starts returning 429s after lunch.

# transcript_api.py
from fastapi import APIRouter, Depends, HTTPException
from pydantic import BaseModel, Field
import hashlib, json, uuid

router = APIRouter()

class SummaryPlan(BaseModel):
    incident_id: str
    notes: str = Field(min_length=1, max_length=8000)

def sha256_payload(data: dict) -> str:
    blob = json.dumps(data, sort_keys=True, separators=(",", ":")).encode()
    return hashlib.sha256(blob).hexdigest()

@router.post("/features/incident-summary/transcripts")
def plan_summary(body: SummaryPlan, actor=Depends(current_user), db=Depends(get_db)):
    if not actor.can("incident.summarize"):
        raise HTTPException(403, "missing incident.summarize")
    incident = db.incidents.get(body.incident_id, actor_id=actor.id)
    if incident is None:
        raise HTTPException(404, "incident not found")
    payload = {
        "incident_id": incident.id,
        "notes": body.notes,
        "title": incident.title,
        "severity": incident.severity,
    }
    row = db.transcripts.insert(
        id=str(uuid.uuid4()),
        actor_id=actor.id,
        feature="incident_summary",
        input_hash=sha256_payload(payload),
        input_payload=payload,
        status="queued",
    )
    db.outbox.enqueue("transcript.fill", row.id)
    return {"transcript_id": row.id, "status": row.status}

@router.get("/features/incident-summary/transcripts/{transcript_id}")
def read_summary(transcript_id: str, actor=Depends(current_user), db=Depends(get_db)):
    row = db.transcripts.get(transcript_id)
    if row is None or row.actor_id != actor.id:
        raise HTTPException(404, "transcript not found")
    return row.as_public_dict()
Enter fullscreen mode Exit fullscreen mode

Would I still let the worker call a model after the transcript row exists in storage? I would, but only after that row exists, and only through an idempotent fill. The worker is allowed to fail, because failure is just another status the UI already knows how to render. If you cannot draw queued, running, failed, and rejected, you are not shipping an AI feature.

# fill_transcript.py
def fill_transcript(transcript_id: str, db, provider, limiter) -> None:
    row = db.transcripts.lock(transcript_id)
    if row.status in {"succeeded", "rejected"}:
        return
    if not limiter.allow(row.actor_id, row.feature):
        db.transcripts.mark(row.id, status="failed", error_code="rate_limited")
        return
    db.transcripts.mark(row.id, status="running")
    try:
        output = provider.complete(feature=row.feature, payload=row.input_payload)
        if not output.get("summary"):
            db.transcripts.mark(row.id, status="rejected", error_code="empty_summary")
            return
        db.transcripts.mark(
            row.id,
            status="succeeded",
            output_payload=output,
            provider_name=provider.name,
        )
    except ProviderError as exc:
        db.transcripts.mark(row.id, status="failed", error_code=exc.code)
Enter fullscreen mode Exit fullscreen mode

This is the part most demos skip, and it is the part that later saves the release. Empty model output is not a success that happens to carry a blank string. It is a rejected transcript, which is a first-class outcome your support tools can filter. Have you ever tried to grep a chat UI for empty successes at two in the morning?

I bother with input hashes because replay has to work when the original vendor is asleep. If two requests share an input hash, I can serve the stored output instead of spending another call. More importantly, I can replay a failed row against a stub provider in continuous integration. I do not need the original vendor to be awake for that assertion to matter.

# replay_transcript.py
import argparse, json
from pathlib import Path

def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("--fixture", required=True)
    parser.add_argument("--expect-status", default="succeeded")
    args = parser.parse_args()
    fixture = json.loads(Path(args.fixture).read_text())
    provider = StubProvider(script=fixture["script"])
    db = FakeDb.seed(fixture["transcript"])
    fill_transcript(
        fixture["transcript"]["id"],
        db=db,
        provider=provider,
        limiter=AllowAll(),
    )
    row = db.transcripts.get(fixture["transcript"]["id"])
    assert row.status == args.expect_status, row.error_code
    print(json.dumps({"status": row.status, "error_code": row.error_code}))

if __name__ == "__main__":
    main()
Enter fullscreen mode Exit fullscreen mode

A fixture file is the original artifact I want in the pull request, not a screenshot of a clever answer. Check this into git, and the argument about model quality becomes an argument about a JSON document. That kind of argument is one a team can actually finish before standup begins.

{
  "transcript": {
    "id": "11111111-1111-1111-1111-111111111111",
    "actor_id": "22222222-2222-2222-2222-222222222222",
    "feature": "incident_summary",
    "input_hash": "replace-me",
    "input_payload": {
      "incident_id": "inc-9",
      "notes": "checkout timeouts after deploy",
      "title": "payments p95 spike",
      "severity": "sev2"
    },
    "status": "queued"
  },
  "script": {
    "summary": "Payments latency rose after the checkout deploy.",
    "confidence": "low"
  }
}
Enter fullscreen mode Exit fullscreen mode

Where does a free generation lane fit, if the live path is no longer sacred? I use it to mint fixtures and fill a disposable server with ugly transcripts before users arrive. Disclosure: This article was prepared as part of MonkeyCode's product outreach. MonkeyCode offers free model access and a free server option for rehearsal, not for the application itself.

That distinction matters much more than whatever brand name happens to sit on the GPU. If the rehearsal host vanished tomorrow, the transcript table and the replay command would still describe the product. If your architecture collapses when a free endpoint sleeps, you did not build a feature so much as rent a demo.

I keep the rehearsal worker pointed at a throwaway base URL, and I never let that URL leak into the production environment file. The production worker reads transcripts and writes status, and it can stay stubbed until the rejection UI is honest. Can your staging deploy survive with the provider key unset, or does the boot sequence panic like a toddler?

export TRANSCRIPT_PROVIDER_BASE_URL="https://rehearsal.example.invalid"
python -m fill_worker --once
python replay_transcript.py --fixture tests/fixtures/empty_summary.json --expect-status rejected
python replay_transcript.py --fixture tests/fixtures/happy_summary.json --expect-status succeeded
Enter fullscreen mode Exit fullscreen mode

Production caveats are not decorations, and I will not soften them to make the opinion nicer. Transcripts contain prompts, and prompts often contain customer language you should not copy into a random host. Hash the input, encrypt the payload at rest, and expire rows that never reached a terminal status. A free server is the wrong place for production secrets, passport scans, or anything your counsel would call regulated.

This approach is also the wrong default for a synchronous product that must speak in under a second, like an in-editor complete. If the user is typing, a queued row will feel like molasses, and they will disable your feature before you finish the migration. In that case, keep the live call, but still write a transcript asynchronously after the stream ends. Please do not confuse a hard latency constraint with a convenient excuse to own nothing.

Which teams should refuse this as a hard rule across every product surface they own? Teams that cannot insert a queue between click and answer, and teams that cannot store prompts at all. Also skip the free rehearsal host if your data cannot leave the building, because a complimentary endpoint is still someone else's computer. I am not arguing for delay; I am arguing that durability is the feature and the model is a plugin.

I still see pull requests that add a provider client, a sparkly button, and zero status codes for the empty output case. Those patches always look finished in review, and they always fail at the first 502. If you want a test of seriousness, ask whether a rejected transcript can be rendered without a redeploy. If the answer is a shrug from the author, the model is not the real problem.

So here is the working path I want a reader to copy, without turning it into a shrine. Create the table, protect the POST with a real permission, and enqueue the fill job. Render four statuses, reject empty output, and replay fixtures in CI before you attach a paid key. Only then point a rehearsal worker at free model access, on a server you can afford to delete.

If you try this and the first failure is not a 502, I want to hear the exact handoff that broke. Was it auth returning 403 after a successful fill, or the UI treating queued as a hard error? Tell me the status code and the layer, because that seam is where AI products actually die.

Top comments (0)