Building Reliable AI Agents: The 2026 Production Playbook

A demo agent and a production agent differ in one thing: what happens when something goes wrong at step seven. This playbook is the checklist and the code we would want before putting an agent in front of customers: choose the least agentic design that works, make every tool contract strict, make every side effect safe to repeat, gate the dangerous actions, checkpoint the work, put hard limits around it, measure it, and roll it out in stages.

Why agents fail more than the demo suggests

Reliability in a multi-step system compounds. If every step succeeds with probability p and failures are independent, a task of n steps succeeds with probability p to the power n.

Real failures are not fully independent, but the shape holds: long chains of "usually works" steps are unreliable. Every technique in this playbook does one of three things: it removes steps, raises per-step reliability, or stops one failure from erasing the progress of the others.

Rule one: use the least agentic design that works

A good pattern is a workflow with an agent step inside it: code controls the outer sequence and the dangerous actions, and the model handles the one step that needs judgment. Move down the table only when the simpler design demonstrably fails on your evaluation set.

The failure taxonomy

You cannot defend against failures you have not named. These are the ones that show up in production, with the layer of this playbook that handles each.

The reliability stack

1. Scoped tools and strict contracts. Few tools, narrow purposes, validated arguments.
2. Risk tiers. Read, write, and irreversible actions are treated differently.
3. Idempotent side effects. Doing a step twice must be safe.
4. Checkpoints. Completed work is stored so a crash resumes instead of restarting.
5. Hard limits. Turn, call, cost, and time budgets, plus loop detection.
6. Failure handling. Retries inside tools, provider failover, and a clean way to give up.
7. Human gates. Approval for what cannot be undone, and handoff when the agent is stuck.
8. Measurement. Evals before release and monitoring after.

The next sections build layers one to four in a single tested component, then cover the rest.

Layers 1 to 4: a safe tool executor

This executor sits between the model's tool requests and your real functions. It validates arguments and returns errors the model can act on, applies a risk policy, records every state-changing step in a ledger so repeats and resumes are safe, retries only what is safe to retry, enforces timeouts, and caps output size.

import hashlib, json, sqlite3, time
from concurrent.futures import ThreadPoolExecutor, TimeoutError as FutureTimeout
from dataclasses import dataclass
from typing import Any, Callable

READ, WRITE, IRREVERSIBLE = "read", "write", "irreversible"
TYPES = {"string": str, "integer": int, "number": (int, float),
         "boolean": bool, "array": list, "object": dict}


class TransientError(Exception):
    pass                                           # timeouts, 429s, 503s: safe to retry


@dataclass
class ToolSpec:
    name: str
    fn: Callable[..., Any]
    params: dict                                   # {"order_id": {"type": "string", "required": True}}
    risk: str = READ
    timeout_s: float = 20.0
    max_retries: int = 2
    max_output_chars: int = 8000
    takes_key: bool = False                        # does the tool forward an idempotency key?


def validate_args(spec: ToolSpec, args: Any) -> list:
    if not isinstance(args, dict):
        return ["Arguments must be a JSON object."]
    errors = [f"Unknown argument '{k}'. Allowed: {sorted(spec.params)}."
              for k in args if k not in spec.params]
    for k, p in spec.params.items():
        if k not in args:
            if p.get("required"):
                errors.append(f"Missing required argument '{k}' ({p['type']}).")
            continue
        v = args[k]
        wrong_bool = isinstance(v, bool) and p["type"] in ("integer", "number")
        if wrong_bool or not isinstance(v, TYPES[p["type"]]):
            errors.append(f"Argument '{k}' must be {p['type']}, got {type(v).__name__}.")
        elif "enum" in p and v not in p["enum"]:
            errors.append(f"Argument '{k}' must be one of {p['enum']}.")
    return errors


class Ledger:
    # Doubles as the checkpoint log: one row per completed state-changing step.
    def __init__(self, path: str = ":memory:"):
        self.db = sqlite3.connect(path, check_same_thread=False)
        self.db.execute("""CREATE TABLE IF NOT EXISTS steps (
            key TEXT PRIMARY KEY, run_id TEXT, idx INTEGER, tool TEXT,
            args TEXT, result TEXT, ts REAL)""")

    def get(self, key: str):
        row = self.db.execute("SELECT result FROM steps WHERE key=?", (key,)).fetchone()
        return row[0] if row else None

    def put(self, key, run_id, idx, tool, args, result) -> None:
        self.db.execute("INSERT OR REPLACE INTO steps VALUES (?,?,?,?,?,?,?)",
                        (key, run_id, idx, tool, json.dumps(args, sort_keys=True), result, time.time()))
        self.db.commit()

    def load_run(self, run_id: str) -> list:
        return self.db.execute("SELECT idx, tool, args, result FROM steps "
                               "WHERE run_id=? ORDER BY idx", (run_id,)).fetchall()


@dataclass
class ToolResult:
    ok: bool
    kind: str          # ok | replayed | invalid | denied | timeout | error
    content: str


def step_key(run_id: str, idx: int, tool: str, args: dict) -> str:
    raw = json.dumps([run_id, idx, tool, args], sort_keys=True, default=str)
    return hashlib.sha256(raw.encode()).hexdigest()[:32]


class Executor:
    def __init__(self, tools: list, ledger: Ledger, approve=None, sleep=time.sleep):
        self.tools = {t.name: t for t in tools}
        self.ledger, self.approve, self.sleep = ledger, approve, sleep

    def call(self, run_id: str, idx: int, name: str, args: dict) -> ToolResult:
        spec = self.tools.get(name)
        if spec is None:
            return ToolResult(False, "invalid", f"Unknown tool '{name}'. Available: {sorted(self.tools)}.")
        errors = validate_args(spec, args)
        if errors:
            return ToolResult(False, "invalid", "Invalid arguments: " + " ".join(errors))

        key = step_key(run_id, idx, name, args)
        if spec.risk != READ:
            done = self.ledger.get(key)
            if done is not None:                   # resumed or retried step: do not act twice
                return ToolResult(True, "replayed", done)
        if spec.risk == IRREVERSIBLE and not (self.approve and self.approve(name, args)):
            return ToolResult(False, "denied",
                              "Not executed: this action needs human approval and none was given. "
                              "Say what you intended and ask the user to confirm.")

        # A write without an idempotency key is never retried: a timeout may hide a success.
        retries = spec.max_retries if (spec.risk == READ or spec.takes_key) else 0
        kwargs = dict(args, idempotency_key=key) if spec.takes_key else dict(args)
        kind, last = "error", ""
        for attempt in range(retries + 1):
            pool = ThreadPoolExecutor(max_workers=1)
            try:
                out = pool.submit(spec.fn, **kwargs).result(timeout=spec.timeout_s)
            except FutureTimeout:
                kind, last = "timeout", f"Timed out after {spec.timeout_s}s."
            except TransientError as e:
                kind, last = "error", f"Temporary failure: {e}."
            except Exception as e:                 # permanent failure: tell the model, do not retry
                return ToolResult(False, "error", f"{type(e).__name__}: {e}. Do not retry unchanged.")
            else:
                text = str(out)
                if len(text) > spec.max_output_chars:
                    cut = len(text) - spec.max_output_chars
                    text = text[:spec.max_output_chars] + f"\n[truncated {cut} chars; narrow the request]"
                if spec.risk != READ:
                    self.ledger.put(key, run_id, idx, name, args, text)
                return ToolResult(True, "ok", text)
            finally:
                pool.shutdown(wait=False)
            if attempt < retries:
                self.sleep(0.2 * 2 ** attempt)
        return ToolResult(False, kind, last + " Stop and report if this keeps failing.")

Design decisions worth copying:

• Errors are written for the model. "Unknown argument 'order'. Allowed: ['order_id']" lets it fix the call. "Invalid request" invites an unchanged retry.
• A write is never retried without an idempotency key. After a timeout you cannot know whether the action happened, so repeating it blindly is how customers get double refunds. When the tool forwards the key to the downstream API, the retry is safe because the API deduplicates.
• The ledger is the checkpoint. Because completed writes are stored under a deterministic key, a restarted run replays recorded results instead of acting again, and load_run tells you where it stopped.
• Irreversible means approval. The default is denial, and the denial message tells the model how to proceed gracefully.
• Output is capped, with a message saying how to narrow the request, which protects the context window.

To use it, call Executor.call in place of the plain tool-calling function in the loop from our post on detecting and stopping runaway loops, and keep the loop guard around it. The two compose: the guard limits how much work the agent may attempt, and the executor makes each attempt safe.

Layer 5: Limits, loops, and giving up cleanly

Every run needs a ceiling on turns, tool calls, cost, and wall-clock time, enforced in code, not in the prompt. Add detectors for repeats, ping-pong patterns, and error streaks so a loop is caught in a handful of calls rather than at the budget. When a limit is hit, give the model one tool-free turn to summarize, and return the run as stopped with a reason, never as a success. All of this is built and tested in the loops post.

Define the ways a run can end before you write the agent: completed, needs_approval, blocked (a dependency is down or data is missing), stopped_budget, and gave_up. Structured outcomes let your product route each one correctly instead of showing "something went wrong."

Layer 6: Handling provider and tool failures

• Retry at the right level. Retry transient tool failures inside the tool wrapper, with backoff, and report one final answer to the model. Never let the model drive your retry policy.
• Fail over between model providers before the first token reaches a user, using normalized message history so a conversation can move mid-task. See the provider-agnostic client post.
• Add circuit breakers to tools. When a downstream API is failing, stop calling it for a cool-off period and return a clear "temporarily unavailable" so the agent adapts instead of hammering.
• Handle stream failures in agent turns: check for the terminal marker and treat a dropped stream as a failed turn, as covered in debugging streaming failures.
• Plan a degraded mode. If a tool is down, can the agent still answer partially, or hand the task to a person with what it has?

Context: the quiet reliability problem

Many agent failures are context failures. The model forgot an instruction, drowned in a huge tool result, or repeated an approach that already failed.

The principle: the conversation is a working memory, not a database. Anything that matters after the run, or after a crash, belongs in storage you control.

Verification beats reflection

Asking a model to "double-check your work" helps a little. Checking the work with something other than the model helps a lot.

• Run the code and the tests it should pass.
• Validate output against a schema and business rules.
• Look up the entity the agent claims exists.
• Compare totals, counts, or dates against the source.
• For high-stakes steps, use a second, independent check and require agreement.

Put these verifiers between "the agent says it is done" and "the task is marked complete."

Security: assume the data is hostile

• Tool results and retrieved pages are untrusted input. They can contain instructions. Keep a clear separation between your instructions and fetched content, and never let fetched text change permissions.
• Least privilege per tool. A tool that reads invoices should not hold credentials that can delete them. Use scoped, short-lived tokens.
• Limit egress. Restrict which domains and addresses an agent can contact or send data to.
• Keep secrets out of context. Inject credentials in the tool layer, never in prompts.
• Gate irreversible actions regardless of how confident the agent seems.

Our AI security playbook goes deeper on injection and data leakage.

Layer 7: Human gates that do not annoy people

• Gate by risk, not by default. Approval fatigue makes people approve without reading.
• Show the action, not the reasoning. "Refund $84.20 to the card ending 4417 for order A-1093" beats a paragraph of model prose.
• Make approval resumable. The run pauses with its state stored, waits for the answer for minutes or days, and continues from the checkpoint.
• Give approvals a timeout and a default. Silence should mean "not approved," never "go ahead."
• Hand off with context. When the agent gives up, the human should see what was asked, what was tried, and what is blocking progress.

Layer 8: Measure it

You need numbers before and after release. Before release, run a golden dataset with a statistical regression gate, as built in our post on evaluating LLM output quality. After release, trace every run and watch these:

Record a trace per run with a span per model call and tool call, including the run ID, step index, tool, argument hash, result kind, latency, token usage, and the model version the API returned. The observability guide covers what to capture and how to alert on it.

Test the executor before you trust it

These tests cover argument validation, replay protection, approval gates, retry rules, timeouts, truncation, and resume. They need no network and no model.

import time

ORDER = {"order_id": {"type": "string", "required": True}}


def make(tools, approve=None):
    return Executor(tools, Ledger(), approve=approve, sleep=lambda s: None)


def test_invalid_arguments_explain_themselves():
    ex = make([ToolSpec("get_order", lambda order_id: "x", ORDER)])
    r = ex.call("r1", 0, "get_order", {"order": "1"})
    assert r.kind == "invalid"
    assert "Unknown argument 'order'" in r.content and "Missing required argument 'order_id'" in r.content
    assert ex.call("r1", 1, "get_order", {"order_id": 5}).kind == "invalid"
    assert ex.call("r1", 2, "nope", {}).kind == "invalid"


def test_bool_is_not_an_integer():
    spec = ToolSpec("t", lambda n: n, {"n": {"type": "integer", "required": True}})
    assert validate_args(spec, {"n": True})


def test_a_write_step_runs_once_even_if_retried_or_resumed():
    calls = []
    refund = ToolSpec("refund", lambda order_id: calls.append(order_id) or "refunded", ORDER, risk=WRITE)
    ex = make([refund])
    assert ex.call("r1", 3, "refund", {"order_id": "A"}).kind == "ok"
    assert ex.call("r1", 3, "refund", {"order_id": "A"}).kind == "replayed"
    assert calls == ["A"]
    ex.call("r1", 4, "refund", {"order_id": "A"})          # a new step is a new decision
    assert calls == ["A", "A"]


def test_irreversible_actions_need_approval():
    calls = []
    wipe = ToolSpec("delete_account", lambda order_id: calls.append(1) or "deleted", ORDER, risk=IRREVERSIBLE)
    assert make([wipe], approve=lambda n, a: False).call("r", 0, "delete_account", {"order_id": "A"}).kind == "denied"
    assert make([wipe]).call("r", 0, "delete_account", {"order_id": "A"}).kind == "denied"
    assert calls == []
    assert make([wipe], approve=lambda n, a: True).call("r", 0, "delete_account", {"order_id": "A"}).ok
    assert calls == [1]


def test_reads_retry_transient_errors():
    n = {"c": 0}

    def flaky(order_id):
        n["c"] += 1
        if n["c"] < 3:
            raise TransientError("503")
        return "ok"
    assert make([ToolSpec("get", flaky, ORDER)]).call("r", 0, "get", {"order_id": "A"}).ok
    assert n["c"] == 3


def test_unkeyed_writes_are_never_retried():
    n = {"c": 0}

    def flaky(order_id):
        n["c"] += 1
        raise TransientError("503")
    r = make([ToolSpec("pay", flaky, ORDER, risk=WRITE)]).call("r", 0, "pay", {"order_id": "A"})
    assert n["c"] == 1 and not r.ok


def test_keyed_writes_retry_with_the_same_key():
    keys = []

    def flaky(order_id, idempotency_key):
        keys.append(idempotency_key)
        if len(keys) < 2:
            raise TransientError("503")
        return "paid"
    r = make([ToolSpec("pay", flaky, ORDER, risk=WRITE, takes_key=True)]).call("r", 0, "pay", {"order_id": "A"})
    assert r.ok and len(keys) == 2 and keys[0] == keys[1]


def test_permanent_errors_are_not_retried():
    n = {"c": 0}

    def broken(order_id):
        n["c"] += 1
        raise ValueError("bad id")
    r = make([ToolSpec("get", broken, ORDER)]).call("r", 0, "get", {"order_id": "A"})
    assert n["c"] == 1 and "Do not retry" in r.content


def test_timeout_and_truncation():
    slow = ToolSpec("slow", lambda order_id: time.sleep(0.3) or "late", ORDER, timeout_s=0.05, max_retries=0)
    assert make([slow]).call("r", 0, "slow", {"order_id": "A"}).kind == "timeout"
    big = ToolSpec("big", lambda order_id: "x" * 100, ORDER, max_output_chars=10)
    assert "truncated 90 chars" in make([big]).call("r", 0, "big", {"order_id": "A"}).content


def test_ledger_supports_resume():
    refund = ToolSpec("refund", lambda order_id: "done", ORDER, risk=WRITE)
    ex = make([refund])
    ex.call("run9", 2, "refund", {"order_id": "B"})
    ex.call("run9", 0, "refund", {"order_id": "A"})
    assert [row[0] for row in ex.ledger.load_run("run9")] == [0, 2]

Roll out on an autonomy ladder

1. Shadow mode. The agent proposes actions and people perform them. You compare proposals against what people did.
2. Approve everything. The agent acts only after a person approves each step. Measure approval rate and edits.
3. Automate low-risk actions. Reads and reversible writes run on their own. Dangerous ones still ask.
4. Release gradually. Start with a small percentage of traffic or a few customers, with a kill switch that can disable one tool, one tenant, or the whole agent without a deploy.
5. Promote only on evidence. Move up a rung when success, escalation, and cost metrics hold for long enough to trust them.

Set service levels that match the stakes, such as a minimum task success rate, a maximum escalation rate, and a cost-per-success ceiling, and treat breaching them like any other production incident.

Production readiness checklist

Pitfalls to avoid

• Reaching for multi-agent designs first. Coordination failures are harder to debug than single-agent ones.
• Trusting the prompt for safety. Anything that must never happen must be impossible in code.
• Retrying writes blindly. A timeout is not a failure. It may be a success you did not see.
• Giving the agent a huge toolbox. More tools mean more wrong choices. Scope tools to the task.
• Unlimited autonomy on day one. Earn it through the ladder.
• Letting the conversation be the only record of state. Store state where a crash cannot erase it.
• Treating a stopped run as a success. Report stops honestly, with reasons.
• No owner. An agent in production is a service. Someone has to be on call for it.

Frequently asked questions

Keep reading