From Prompt to Production: Designing a Reliable AI Workflow

A workflow that calls a model, waits, and returns the answer works fine right up until the call times out. Retry it, and one of two things happens: either the retry is safe, or the email it already sent, the record it already wrote, the charge it already made, happens a second time. That's not a hypothetical edge case, it's the default behavior of a retry on anything that isn't explicitly built to survive one. Reliability in an AI workflow isn't about the model being right more often. It's about the system around it correctly handling the model being slow, wrong, or interrupted, three things that are guaranteed to happen eventually at real volume, not exceptions to plan for later.

From Prototype to Production names reliability as one of the problems that stays invisible until the scale stage, when a rare failure stops being rare. This is the hands-on toolkit for actually building it in: retries that know what's safe to repeat, idempotency that makes retries safe in the first place, structured output validation, model fallbacks, and checkpointing that survives a crash without starting over.

The Architecture

Retries: Not Every Failure Deserves the Same Response

The Anthropic SDK's own exception hierarchy already encodes the distinction that matters: which failures are worth retrying, and which ones will just fail identically a second time.

import time
import random
import anthropic
 
client = anthropic.Anthropic()
 
def call_with_retry(prompt: str, max_attempts: int = 4) -> str:
    for attempt in range(1, max_attempts + 1):
        try:
            response = client.messages.create(
                model="claude-sonnet-5",
                max_tokens=1024,
                messages=[{"role": "user", "content": prompt}],
            )
            return response.content[0].text
        except anthropic.RateLimitError:
            pass  # retryable, back off and try again
        except anthropic.APIConnectionError:
            pass  # retryable, likely transient
        except anthropic.BadRequestError:
            raise  # the request itself is malformed, retrying won't fix it
        except anthropic.APIStatusError as e:
            if e.status_code < 500:
                raise  # a 4xx that isn't rate limiting is not a retry candidate
        if attempt == max_attempts:
            raise RuntimeError("Exhausted retries")
        backoff = (2 ** attempt) + random.uniform(0, 1)
        time.sleep(backoff)

This is the same retryable-versus-not distinction covered under Failure Handling in Multi-Agent AI Systems, applied here at the level of a single call instead of a whole worker. If the workflow also needs volume limits and a breaker that stops calling a struggling dependency altogether, the rate limiter and circuit breaker built in How to Give AI Agents Access to APIs apply directly on top of this.

Idempotency: What Actually Makes a Retry Safe

Retrying a read is free. Retrying a write, send this email, charge this card, create this ticket, is only safe if the system can tell "this already happened" from "this hasn't happened yet."

processed_keys: set[str] = set()  # a real system uses a database, not memory
 
def send_confirmation_email(idempotency_key: str, to: str, body: str) -> dict:
    if idempotency_key in processed_keys:
        return {"status": "already_sent", "idempotency_key": idempotency_key}
 
    result = email_client.send(to=to, body=body)
    processed_keys.add(idempotency_key)
    return {"status": "sent", "idempotency_key": idempotency_key, "result": result}

Not every step needs this. A step that only reads data, or one that's naturally idempotent already, setting a status field to "complete" twice has the same effect as setting it once, doesn't need a key at all. The steps that need it are exactly the ones with a real-world side effect that isn't naturally safe to repeat.

Structured Output Validation: A Malformed Answer Isn't an Exception

A model that returns text where JSON was expected doesn't throw an error, it returns a 200 with a response that quietly fails the next step downstream. That failure needs to be caught explicitly, with a retry that tells the model what was wrong, not just what was asked for again.

from pydantic import BaseModel, ValidationError
import json
 
class OrderSummary(BaseModel):
    order_id: str
    total: float
    line_items: int
 
def get_structured_output(prompt: str, max_attempts: int = 3) -> OrderSummary:
    feedback = ""
    for attempt in range(1, max_attempts + 1):
        full_prompt = prompt + (f"\n\nYour previous response was invalid: {feedback}" if feedback else "")
        response = client.messages.create(
            model="claude-sonnet-5",
            max_tokens=512,
            messages=[{"role": "user", "content": full_prompt}],
        )
        raw = response.content[0].text
        try:
            return OrderSummary.model_validate(json.loads(raw))
        except (json.JSONDecodeError, ValidationError) as e:
            feedback = str(e)
            if attempt == max_attempts:
                raise ValueError(f"Model never produced valid output: {feedback}")

Fallback Models: What Runs When the First Choice Won't

Retries handle a transient failure on the same model. A fallback model handles the case where the primary model is degraded long enough that retrying it isn't going to help within the time the workflow has.

def call_with_fallback(prompt: str) -> tuple[str, str]:
    for model in ["claude-sonnet-5", "claude-haiku-4-5"]:
        try:
            response = client.messages.create(
                model=model,
                max_tokens=1024,
                messages=[{"role": "user", "content": prompt}],
            )
            return response.content[0].text, model
        except (anthropic.RateLimitError, anthropic.APIConnectionError, anthropic.APIStatusError):
            continue
    raise RuntimeError("All models in the fallback chain failed")

Checkpointing: Surviving a Crash Without Starting Over

A five-step workflow that crashes on step four, with no record of steps one through three having already run, doesn't resume, it restarts, redoing work, and re-triggering every side effect those first three steps already caused. Checkpointing after each step is what turns a crash into a pause instead of a loss.

import json
from pathlib import Path
 
def save_checkpoint(workflow_id: str, step_index: int, state: dict) -> None:
    path = Path(f"checkpoints/{workflow_id}.json")
    path.write_text(json.dumps({"step_index": step_index, "state": state}))
 
def load_checkpoint(workflow_id: str) -> dict | None:
    path = Path(f"checkpoints/{workflow_id}.json")
    return json.loads(path.read_text()) if path.exists() else None

def run_workflow(workflow_id: str, steps: list) -> dict:
    checkpoint = load_checkpoint(workflow_id)
    start_index = checkpoint["step_index"] + 1 if checkpoint else 0
    state = checkpoint["state"] if checkpoint else {}
 
    for index in range(start_index, len(steps)):
        state = steps[index](state)
        save_checkpoint(workflow_id, index, state)
 
    return state

Timeouts Everywhere, and Graceful Degradation

When a non-critical step in an otherwise-working workflow genuinely can't complete in time, the honest options are to fail the whole thing or return a partial result with that fact clearly flagged, not to silently drop the missing piece and hand back something that looks complete. A report missing one section because a slow data source timed out, clearly marked as missing, is more useful than either a total failure or a report that looks whole and isn't.

Gate Deploys on Evaluation, Not on Vibes

A prompt change, a model swap, or a new fallback tier is itself a reliability risk if nothing checks that it didn't quietly make things worse. The measurement discipline for this is covered in full in LLM Evaluation; the reliability-specific point is that the eval suite belongs in the deploy path, not run manually and occasionally:

def deploy_if_eval_passes(new_prompt: str, eval_suite, min_pass_rate: float = 0.95) -> bool:
    results = eval_suite.run(new_prompt)
    pass_rate = results.passed / results.total
    if pass_rate < min_pass_rate:
        raise RuntimeError(f"Eval pass rate {pass_rate:.2%} below {min_pass_rate:.2%}, blocking deploy")
    return True

A prompt change that passes a human's quick read-through but drops the eval pass rate from 97% to 91% is exactly the kind of regression that looks fine in a demo and shows up as a real incident once it's serving real traffic.

A Worked Trace

A report-generation workflow: gather data, summarize it with the model, validate the structured output, send the finished report by email.

When Simpler Is Fine

The Bottom Line

None of these mechanisms exist to make the model more accurate. They exist because a model call is one part of a system that also has to survive timeouts, crashes, malformed output, and retries, and a workflow that only works when none of those things happen isn't reliable, it's untested. Retries that respect what's actually safe to repeat, idempotency that makes repeating safe in the first place, validated output, a fallback tier, checkpointed progress, and an evaluation gate on every change, together, are what close the actual gap between a workflow that worked in a demo and one that survives real production traffic.

A demo never has to survive a retry, a crash, or a malformed answer, because nobody's testing it hard enough for any of those to happen. Production always does, eventually, and the workflow's job is to already have an answer ready when it does.