Agent Ops: orchestrate, monitor, scale and guard AI agents

A demo agent works on a laptop. A production agent copes with crashed pods, rate limits, poisoned inputs, $40k token bills and a 2 a.m. page. This track covers what changes between the two, with runnable code and the interview questions that test it.

Orchestration

Orchestration is deciding who controls the next step: your code, one LLM, or several LLMs coordinating. Most production systems mix all three.

Patterns at a glance

Deterministic shell, agentic core

The most reliable production shape: plain code owns the workflow (load data, enforce policy, persist results) and the LLM only makes decisions inside one bounded step. You get the flexibility of an agent where you need it and the predictability of code everywhere else.

"""Deterministic shell, agentic core: code owns the workflow, the LLM only decides inside one step."""
from typing import TypedDict

from langchain.agents import create_agent
from langchain_core.messages import AIMessage
from langchain_core.tools import tool
from langgraph.graph import END, START, StateGraph

from _fake import call, script


@tool
def lookup_order(order_id: str) -> str:
    """Look up an order by id."""
    return f"{order_id}: delivered 3 days ago"


# The agentic core: a standard tool-calling agent (a compiled subgraph).
agent = create_agent(
    script(call("lookup_order", {"order_id": "A1001"}), AIMessage("Your order A1001 was delivered.")),
    tools=[lookup_order],
    system_prompt="You are a support agent. Use tools; never guess.",
)


class Ticket(TypedDict):
    customer_id: str
    question: str
    tier: str
    reply: str
    escalate: bool


def load_customer(t: Ticket):                       # deterministic: plain code, no LLM
    return {"tier": "gold" if t["customer_id"].startswith("G") else "standard"}


def run_agent(t: Ticket):                           # agentic: the LLM picks tools and order
    prompt = f"[customer tier: {t['tier']}] {t['question']}"
    out = agent.invoke({"messages": [("user", prompt)]})
    return {"reply": out["messages"][-1].text}


def policy_check(t: Ticket):                        # deterministic gate before anything is sent
    risky = any(w in t["reply"].lower() for w in ("refund approved", "guarantee"))
    return {"escalate": risky}


g = StateGraph(Ticket)
g.add_node("load_customer", load_customer)
g.add_node("agent", run_agent)
g.add_node("policy_check", policy_check)
g.add_edge(START, "load_customer")
g.add_edge("load_customer", "agent")
g.add_edge("agent", "policy_check")
g.add_edge("policy_check", END)
workflow = g.compile()

if __name__ == "__main__":
    out = workflow.invoke({"customer_id": "G-42", "question": "Where is A1001?",
                           "tier": "", "reply": "", "escalate": False})
    print(out["tier"], "|", out["reply"], "| escalate:", out["escalate"])
    # gold | Your order A1001 was delivered. | escalate: False

Choosing a pattern

1. Can you write the steps down in advance? → workflow. Stop here.
2. One domain, ≤ ~15 tools? → single agent inside a deterministic shell.
3. Distinct domains with different tools or permissions? → router to specialised agents.
4. Tasks need several specialists to collaborate on one output? → supervisor.
5. Takes minutes or hours, or is triggered by events? → run it in the background (queue + workers) whatever the pattern.

Durable execution

Pods get rescheduled, deploys roll, providers time out. With a checkpointer, LangGraph saves state after every step, so a failed run resumes where it stopped instead of starting over. That matters for side effects: you must not charge a card twice because the email step after it failed.

"""Crash in the middle of a run, then resume from the last checkpoint without redoing finished steps."""
from typing import TypedDict

from langgraph.checkpoint.memory import InMemorySaver   # PostgresSaver in production
from langgraph.graph import END, START, StateGraph

calls = {"charge_card": 0, "send_receipt": 0}
receipt_service_up = False


class Order(TypedDict):
    order_id: str
    charged: bool
    receipt_sent: bool


def charge_card(s: Order):
    calls["charge_card"] += 1
    return {"charged": True}


def send_receipt(s: Order):
    calls["send_receipt"] += 1
    if not receipt_service_up:
        raise ConnectionError("email service down")      # simulated outage / pod crash
    return {"receipt_sent": True}


g = StateGraph(Order)
g.add_node("charge_card", charge_card)
g.add_node("send_receipt", send_receipt)
g.add_edge(START, "charge_card")
g.add_edge("charge_card", "send_receipt")
g.add_edge("send_receipt", END)
app = g.compile(checkpointer=InMemorySaver())

if __name__ == "__main__":
    cfg = {"configurable": {"thread_id": "order-7"}}
    try:
        # durability="sync": each checkpoint is written before the next step starts
        app.invoke({"order_id": "A7", "charged": False, "receipt_sent": False}, cfg, durability="sync")
    except ConnectionError as e:
        print("run failed:", e, "| next step:", app.get_state(cfg).next)   # ('send_receipt',)

    receipt_service_up = True
    out = app.invoke(None, cfg)          # None = resume this thread from its last checkpoint
    print(out, calls)
    # {'order_id': 'A7', 'charged': True, 'receipt_sent': True} {'charge_card': 1, 'send_receipt': 2}
    # charge_card ran ONCE: the customer was not charged twice.

• durability="sync" writes each checkpoint before the next step starts (safest); "async" writes in the background (faster); "exit" writes only at the end (fastest, no mid-run recovery).
• Resuming re-runs the failed node from its start, so side effects inside a node still need idempotency keys.

State & monitoring

Designing agent state

• Keep it small. State is checkpointed on every step. Store IDs and references (doc_ids, an S3 key), not 5 MB tool outputs.
• Be explicit. Typed keys like intent, pending_refund, attempts are easier to monitor, test and route on than parsing the message list.
• Separate private state. Sub-agents get their own schema (subgraphs); only the summary flows back to the parent.
• Version it. Threads can live for weeks. When the schema changes, give new keys defaults so old checkpoints still load, and keep a migration path.
• Plan retention. Checkpoints grow with every step. Delete or archive old threads on a schedule, and treat them as personal data (they contain user messages).

Inspecting and fixing live threads

Because every step is checkpointed, an operator can see exactly where a conversation is, why it's waiting, and repair it — without a redeploy.

"""Operate on live threads: inspect, read history, fix bad state, resume."""
from typing import TypedDict

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, StateGraph
from langgraph.types import Command, interrupt


class Refund(TypedDict):
    order_id: str
    amount: float
    status: str


def propose(s: Refund):
    return {"amount": 1200.0, "status": "proposed"}          # bug: the agent proposed too much


def approve(s: Refund):
    ok = interrupt({"order_id": s["order_id"], "amount": s["amount"]})
    return {"status": "refunded" if ok else "denied"}


g = StateGraph(Refund)
g.add_node("propose", propose)
g.add_node("approve", approve)
g.add_edge(START, "propose")
g.add_edge("propose", "approve")
g.add_edge("approve", END)
app = g.compile(checkpointer=InMemorySaver())

if __name__ == "__main__":
    cfg = {"configurable": {"thread_id": "refund-99"}}
    app.invoke({"order_id": "A9", "amount": 0, "status": "new"}, cfg)

    snap = app.get_state(cfg)
    print("waiting at:", snap.next, "| interrupt:", snap.interrupts[0].value)
    print("values:", snap.values)
    print("history:", [(h.metadata["step"], h.next) for h in app.get_state_history(cfg)])

    # An operator corrects the amount, as if the 'propose' node had written it
    app.update_state(cfg, {"amount": 120.0}, as_node="propose")
    out = app.invoke(Command(resume=True), cfg)
    print("final:", out)   # {'order_id': 'A9', 'amount': 120.0, 'status': 'refunded'}

Build a small internal admin page on these three calls (get_state, get_state_history, update_state). It turns "the agent is stuck for customer X" from a code investigation into a two-minute fix.

What to monitor

Instrumentation

Tracing tools (LangSmith, Langfuse, Arize Phoenix, OpenTelemetry) give you the full run tree. You still want structured logs and metrics in your own stack for alerting. A small wrapper around each node is enough:

"""Structured per-node telemetry: one JSON log line per node with thread, duration and outcome."""
import json
import logging
import time
from typing import TypedDict

from langchain_core.runnables import RunnableConfig
from langgraph.graph import END, START, StateGraph

logging.basicConfig(level=logging.INFO, format="%(message)s")
log = logging.getLogger("agent")


def instrumented(name, fn):
    # Not functools.wraps: LangGraph reads the signature to decide whether to inject config
    def wrapper(state, config: RunnableConfig):
        start = time.perf_counter()
        status = "ok"
        try:
            return fn(state)
        except Exception as e:
            status = f"error:{type(e).__name__}"
            raise
        finally:
            log.info(json.dumps({
                "event": "node", "node": name, "status": status,
                "thread_id": config["configurable"].get("thread_id"),
                "duration_ms": round((time.perf_counter() - start) * 1000, 1),
            }))
    return wrapper


class S(TypedDict):
    q: str
    a: str


def retrieve(s: S): time.sleep(0.02); return {"a": "3 docs"}
def answer(s: S):   return {"a": s["a"] + " -> answer"}


g = StateGraph(S)
for name, fn in [("retrieve", retrieve), ("answer", answer)]:
    g.add_node(name, instrumented(name, fn))
g.add_edge(START, "retrieve")
g.add_edge("retrieve", "answer")
g.add_edge("answer", END)
app = g.compile()

if __name__ == "__main__":
    app.invoke({"q": "refund policy?", "a": ""}, {"configurable": {"thread_id": "t-1"}})
    # {"event": "node", "node": "retrieve", "status": "ok", "thread_id": "t-1", "duration_ms": 20.3}
    # {"event": "node", "node": "answer", "status": "ok", "thread_id": "t-1", "duration_ms": 0.0}

# Token usage for cost dashboards (works with any LangChain chat model):
#   from langchain_core.callbacks import get_usage_metadata_callback
#   with get_usage_metadata_callback() as cb:
#       graph.invoke(inputs, cfg)
#   print(cb.usage_metadata)   # {'claude-sonnet-5-5': {'input_tokens': ..., 'output_tokens': ...}}

Alerts & dashboards

• Page someone for: provider error rate > 5% for 5 min, p95 latency doubling, cost per hour above 3× baseline, queue depth growing for 15 min.
• Ticket (no page) for: success rate down week over week, guardrail block rate change, approval queue older than SLA.
• Always tag traces and logs with thread_id, user_id/tenant, prompt version and model id, so you can slice any regression to a release.
• Sample and review ~1–2% of conversations by hand every week. Dashboards miss "polite but wrong".

Scaling

Where agents actually bottleneck

Agent servers spend most of their time waiting on the LLM. CPU is rarely the limit. The real constraints, in order:

1. Provider rate limits: requests per minute and tokens per minute per model, often per region.
2. Cost: every extra loop iteration multiplies tokens.
3. Database connections: every replica × every concurrent run wants a checkpoint connection.
4. Long-lived connections: SSE streams hold a request open for the whole run, so load balancer idle timeouts and per-instance concurrency limits matter.

Capacity math (worked example)

Practise this one; it comes up in system-design rounds.

• Requests: 100 × 5 = 500 LLM requests/min.
• Tokens: 500 × 4,000 = 2M input tokens/min and 500 × 400 = 200k output tokens/min. Check these against your provider quota first; this is usually the binding limit. Prompt caching makes 75% of those input tokens much cheaper and faster.
• Concurrency (Little's law: arrival rate × duration): 100/60 × 20 s ≈ 34 runs in flight. At 40 concurrent runs per replica, one replica handles it. Run 2–3 for availability.
• Database: ~10 checkpoints per run × 100/min ≈ 17 writes/s, which is trivial for Postgres. A pool of ~10 connections per replica is plenty.

Conclusion: compute scaling is easy. The plan is about quota, cost and back-pressure.

Scale-out setup

Stateless replicas behind a load balancer, a shared Postgres checkpointer through a connection pool, a client-side rate limiter so replicas don't stampede the provider, and a cap on concurrent runs per replica. This exact setup was tested with 100 concurrent conversations through a 10-connection pool.

"""Scale-out setup: shared connection pool, client-side rate limiting, bounded concurrency."""
import asyncio
import os

from langchain.chat_models import init_chat_model
from langchain_core.rate_limiters import InMemoryRateLimiter
from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver
from psycopg_pool import AsyncConnectionPool

# 1) Rate-limit LLM calls per replica (keep replicas x rate under your provider quota)
limiter = InMemoryRateLimiter(requests_per_second=8, max_bucket_size=16)
llm = init_chat_model("anthropic:claude-sonnet-5-5", rate_limiter=limiter, timeout=60)

# 2) Bound concurrent runs per replica so a spike queues instead of failing
RUN_SLOTS = asyncio.Semaphore(int(os.getenv("MAX_CONCURRENT_RUNS", "40")))


async def main(build_graph):
    # 3) One pool per process, shared by all requests (not a connection per request)
    async with AsyncConnectionPool(
        os.environ["DATABASE_URL"], min_size=2, max_size=10,
        kwargs={"autocommit": True, "prepare_threshold": 0},   # required by the Postgres saver
    ) as pool:
        saver = AsyncPostgresSaver(pool)
        await saver.setup()           # better: run once from a migration job
        graph = build_graph(saver)

        async def handle(thread_id: str, message: str):
            async with RUN_SLOTS:
                cfg = {"configurable": {"thread_id": thread_id}, "recursion_limit": 25}
                return await graph.ainvoke({"messages": [("user", message)]}, cfg)

        return await asyncio.gather(*(handle(f"t{i}", "hi") for i in range(100)))

• Any replica can serve any thread, so no sticky sessions are needed.
• Scale on concurrent requests or queue depth, not CPU.
• For sustained high volume, buy provisioned throughput or spread across regions/providers, and route easy steps to smaller models.

Long-running work: queues & workers

Anything that takes longer than a comfortable HTTP request (research reports, bulk document processing, multi-hour approvals) should be enqueued. The API returns a run id immediately; workers execute graphs; clients poll or get a webhook. Using the run id as thread_id means a crashed worker's run can be resumed from its checkpoint.

"""Long-running agents: API enqueues, workers run graphs, clients poll status."""
import asyncio
import uuid
from typing import TypedDict

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, StateGraph


class Job(TypedDict):
    topic: str
    report: str


async def research(s: Job):
    await asyncio.sleep(0.05)                      # stands in for minutes of tool calls
    return {"report": f"report on {s['topic']}"}

g = StateGraph(Job)
g.add_node("research", research)
g.add_edge(START, "research")
g.add_edge("research", END)
graph = g.compile(checkpointer=InMemorySaver())

queue: asyncio.Queue = asyncio.Queue()             # SQS / Pub/Sub / Service Bus in production
status: dict[str, str] = {}                         # a DB table in production


async def submit(topic: str) -> str:               # what POST /runs does
    run_id = str(uuid.uuid4())[:8]
    status[run_id] = "queued"
    await queue.put((run_id, topic))
    return run_id


async def worker(n: int):
    while True:
        run_id, topic = await queue.get()
        status[run_id] = "running"
        try:
            cfg = {"configurable": {"thread_id": run_id}}   # thread = run: resumable after a crash
            await graph.ainvoke({"topic": topic, "report": ""}, cfg)
            status[run_id] = "done"
        except Exception:
            status[run_id] = "failed"                       # alert + dead-letter for a human
        finally:
            queue.task_done()


async def main():
    workers = [asyncio.create_task(worker(i)) for i in range(3)]   # scale workers on queue depth
    ids = [await submit(t) for t in ("pricing", "churn", "nps", "support load")]
    await queue.join()
    print({i: status[i] for i in ids})
    for w in workers:
        w.cancel()

if __name__ == "__main__":
    asyncio.run(main())   # {'1a2b3c4d': 'done', ...}

Failures & recovery

Failure taxonomy

Retries vs. fallbacks

Retry what might succeed next time (rate limits, timeouts). Route around what never will (404, validation errors). Retrying a permanent error just burns time and money.

"""Transient vs. permanent failures: retry the first, route the second to a fallback."""
from typing import Literal, TypedDict

from langgraph.graph import END, START, StateGraph
from langgraph.types import Command, RetryPolicy

attempts = {"crm": 0}


class TransientError(Exception): ...   # 429, 503, timeouts
class PermanentError(Exception): ...   # 400, 404, validation errors


class S(TypedDict):
    customer_id: str
    profile: str
    reply: str


def fetch_profile(s: S) -> Command[Literal["answer", "fallback"]]:
    attempts["crm"] += 1
    if attempts["crm"] < 3:
        raise TransientError("CRM returned 503")            # retried by RetryPolicy
    if s["customer_id"] == "unknown":
        # don't retry what can't succeed: route around it instead
        return Command(goto="fallback", update={"profile": ""})
    return Command(goto="answer", update={"profile": "gold, 3 open orders"})


def answer(s: S):   return {"reply": f"Personalised reply using: {s['profile']}"}
def fallback(s: S): return {"reply": "Generic reply + ticket created for follow-up"}


g = StateGraph(S)
g.add_node("fetch_profile", fetch_profile,
           retry_policy=RetryPolicy(max_attempts=4, initial_interval=0.05, retry_on=TransientError))
g.add_node("answer", answer)
g.add_node("fallback", fallback)
g.add_edge(START, "fetch_profile")
g.add_edge("answer", END)
g.add_edge("fallback", END)
app = g.compile()

if __name__ == "__main__":
    print(app.invoke({"customer_id": "C1", "profile": "", "reply": ""})["reply"], "| attempts:", attempts["crm"])
    # Personalised reply using: gold, 3 open orders | attempts: 3

Model calls get the same treatment: llm.with_retry(...) for transient errors and llm.with_fallbacks([backup]) or ModelFallbackMiddleware to switch provider.

Side effects & compensation

When one run makes several real-world changes, a failure halfway through leaves things inconsistent. The saga pattern pairs every action with an undo and runs the undos in reverse on failure. Combine it with idempotency keys so retries and resumes are safe.

"""Saga pattern: multi-step side effects that undo themselves if a later step fails."""
log: list[str] = []


def reserve_stock(order):  log.append("stock reserved")
def release_stock(order):  log.append("stock released")
def charge_card(order):    log.append("card charged")
def refund_card(order):    log.append("card refunded")
def book_courier(order):
    raise RuntimeError("courier API rejected the address")


STEPS = [                                   # (action, compensation)
    (reserve_stock, release_stock),
    (charge_card, refund_card),
    (book_courier, None),
]


def run_saga(order) -> bool:
    done = []
    for action, undo in STEPS:
        try:
            action(order)
            done.append(undo)
        except Exception as e:
            log.append(f"FAILED: {e}")
            for compensate in reversed(done):          # undo in reverse order
                if compensate:
                    compensate(order)
            return False
    return True


if __name__ == "__main__":
    print(run_saga({"id": "A1"}), log)
    # False ['stock reserved', 'card charged', 'FAILED: courier API rejected the address',
    #        'card refunded', 'stock released']

Incident playbook

1. Contain: feature-flag the agent off or route to the fallback (ticket / human queue). Have this switch before you need it.
2. Find the step: open failing traces; is it retrieval, routing, a tool, or the model?
3. Check what changed: prompt version, model version, tool API, data source, traffic mix.
4. Repair threads: resume, update or close affected threads with the state APIs.
5. Prevent: add the failing cases to the eval set, fix the root cause, add a monitor that would have caught it sooner.

Guardrails

Layered defence

Built-in middleware

LangChain 1.x agents accept middleware that wraps model and tool calls, with ready-made guardrails. This demo redacts emails, blocks card numbers, caps model calls, allows at most one refund per run, and pauses refunds for approval:

"""Layered guardrails on a LangChain 1.x agent using built-in middleware."""
from langchain.agents import create_agent
from langchain.agents.middleware import (
    HumanInTheLoopMiddleware,
    ModelCallLimitMiddleware,
    PIIMiddleware,
    ToolCallLimitMiddleware,
)
from langchain_core.messages import AIMessage
from langchain_core.tools import tool
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import Command

from _fake import call, script


@tool
def lookup_order(order_id: str) -> str:
    """Look up an order."""
    return f"{order_id}: delivered"


@tool
def issue_refund(order_id: str, amount: float) -> str:
    """Refund an order."""
    return f"refunded {amount} for {order_id}"


agent = create_agent(
    script(call("issue_refund", {"order_id": "A1", "amount": 80.0}), AIMessage("Refund done.")),
    tools=[lookup_order, issue_refund],
    middleware=[
        PIIMiddleware("email", strategy="redact"),                   # input: mask emails
        PIIMiddleware("credit_card", strategy="block"),              # input: refuse card numbers
        ModelCallLimitMiddleware(run_limit=10),                      # budget: stop runaway loops
        ToolCallLimitMiddleware(tool_name="issue_refund", run_limit=1),  # at most one refund per run
        HumanInTheLoopMiddleware(interrupt_on={"issue_refund": True, "lookup_order": False}),
    ],
    checkpointer=InMemorySaver(),       # needed to pause for approval
)

if __name__ == "__main__":
    cfg = {"configurable": {"thread_id": "g-1"}}
    out = agent.invoke({"messages": [("user", "Refund A1, my email is priya@example.com")]}, cfg)
    print("user message stored as:", out["messages"][0].content)
    print("paused for approval:", out["__interrupt__"][0].value["action_requests"][0]["name"])

    out = agent.invoke(Command(resume={"decisions": [{"type": "approve"}]}), cfg)
    print("after approval:", out["messages"][-2].content, "|", out["messages"][-1].content)

Running it prints the stored message as Refund A1, my email is [REDACTED_EMAIL], pauses on issue_refund, and only executes the refund after the approve decision. Other built-ins worth knowing: ModelRetryMiddleware, ToolRetryMiddleware, ModelFallbackMiddleware, SummarizationMiddleware and LLMToolSelectorMiddleware.

Tool policy as data

Hard-coding permissions inside each tool doesn't scale. Keep one default-deny policy table that security can review, and enforce it where tool calls are routed (allow → run, approve → interrupt(), deny → error message back to the model).

"""Policy as data: which roles may use which tools, and when a human must approve."""
from dataclasses import dataclass
from typing import Callable


@dataclass(frozen=True)
class Rule:
    roles: frozenset[str]
    approve_if: Callable[[dict], bool] | None = None    # predicate over the tool args


POLICY = {
    "lookup_order":  Rule(frozenset({"customer", "agent", "supervisor"})),
    "issue_refund":  Rule(frozenset({"agent", "supervisor"}), approve_if=lambda a: a["amount"] > 100),
    "delete_account": Rule(frozenset({"supervisor"}), approve_if=lambda a: True),
}


def decide(tool: str, args: dict, role: str) -> str:
    """Return 'allow', 'approve' or 'deny'. Unknown tools are denied (default-deny)."""
    rule = POLICY.get(tool)
    if rule is None or role not in rule.roles:
        return "deny"
    if rule.approve_if and rule.approve_if(args):
        return "approve"
    return "allow"


if __name__ == "__main__":
    for t, a, r in [("lookup_order", {"order_id": "A1"}, "customer"),
                    ("issue_refund", {"amount": 40}, "agent"),
                    ("issue_refund", {"amount": 400}, "agent"),
                    ("issue_refund", {"amount": 40}, "customer"),
                    ("run_shell", {}, "supervisor")]:
        print(f"{t:14} {str(a):18} {r:10} -> {decide(t, a, r)}")
    # The router sends 'approve' calls to an interrupt() node and turns 'deny' into a ToolMessage error.

Output checks

Check the reply before the user sees it. Cheap deterministic checks catch a lot; add an LLM judge for groundedness where stakes are high. Always bound the repair loop and have a safe fallback.

"""Check the agent's reply BEFORE it reaches the user; repair or fall back if it fails."""
import re
from typing import Literal, TypedDict

from langgraph.graph import END, START, StateGraph
from langgraph.types import Command

EMAIL = re.compile(r"[\w.+-]+@[\w-]+\.[\w.]+")
PROMISES = ("guarantee", "we will refund", "legal advice")


class S(TypedDict):
    draft: str
    sources: list[str]
    reply: str
    attempts: int


def draft(s: S):
    # stands in for the agent; first draft leaks an email and has no citation
    if s["attempts"] == 0:
        return {"draft": "Contact lee@corp.com, we guarantee a refund.", "attempts": 1}
    return {"draft": "Refunds are possible within 30 days [refund-policy.md].", "attempts": s["attempts"] + 1}


def check(s: S) -> Command[Literal["draft", "__end__"]]:
    problems = []
    if EMAIL.search(s["draft"]):
        problems.append("contains an email address")
    if any(p in s["draft"].lower() for p in PROMISES):
        problems.append("makes a promise policy forbids")
    if not any(f"[{src}]" in s["draft"] for src in s["sources"]):
        problems.append("no citation")
    if not problems:
        return Command(goto=END, update={"reply": s["draft"]})
    if s["attempts"] >= 2:                               # bounded repair, then safe fallback
        return Command(goto=END, update={"reply": "Let me connect you with a teammate."})
    return Command(goto="draft")                          # in real code: pass problems back to the LLM


g = StateGraph(S)
g.add_node("draft", draft)
g.add_node("check", check)
g.add_edge(START, "draft")
g.add_edge("draft", "check")
app = g.compile()

if __name__ == "__main__":
    print(app.invoke({"draft": "", "sources": ["refund-policy.md"], "reply": "", "attempts": 0})["reply"])
    # Refunds are possible within 30 days [refund-policy.md].

Testing guardrails

• Keep an adversarial eval set: injection attempts (direct and inside documents/tool results), PII, jailbreaks, out-of-scope requests, over-limit refunds.
• Measure both sides: attack success rate (should be ~0) and false-block rate on normal traffic (too many blocks frustrate users).
• Run it in CI on every prompt, model or policy change, next to the quality evals.
• Assume a guardrail will fail eventually; design so the blast radius is small (least privilege, approvals, audit).

Real-world scenarios

Composite scenarios based on common production patterns. Each one is also a strong interview story.

Customer-support agent at scale

• Setup: router → orders / refunds / FAQ agents; Postgres checkpoints; refunds over a threshold go to a supervisor queue via interrupt().
• What went wrong: a holiday traffic spike hit the provider's tokens-per-minute limit; retries from every replica amplified it into an outage.
• Fix: per-replica rate limiter sized to the quota, a semaphore on concurrent runs, prompt caching for the long system prompt, a small model for triage, and a fallback that creates a ticket with "we'll email you" when the queue is too deep.
• Monitoring that would have caught it: 429-rate alert and "tokens per minute vs quota" dashboard.

IT-ops runbook agent

• Setup: on-call engineers ask the agent to diagnose alerts; tools read logs and metrics and can restart services.
• Risk: log lines are attacker-controllable, so a crafted log message could instruct the agent to run a destructive command (prompt injection through tool output).
• Fix: split into a read-only diagnosis agent and a separate action step that only executes runbook-approved commands from an allow-list, with dry-run output shown and human approval for every write. All tool calls go to an audit log.
• Lesson: separate reading untrusted data from taking privileged actions.

Document-processing pipeline

• Setup: 50k invoices a night are extracted into structured records by a queue of workers, each running a small graph: classify → extract (structured output) → validate → write.
• What went wrong: a worker crash mid-batch re-processed documents and created duplicate payments.
• Fix: thread_id = document id with a durable checkpointer (resume, don't restart); idempotent writes keyed by document hash; a dead-letter queue for documents that fail validation twice, reviewed by a human.
• Lesson: mostly a workflow, with the LLM only in the extract step. That is exactly why it scales.

Interview corner

Questions to practise

How to talk about production experience

• Name the failure modes unprompted: crashes, rate limits, loops, injection, cost. It signals you've run agents, not just built demos.
• Pair every feature with how you'd measure it ("…and I'd track approval latency against a 4-hour SLA").
• Prefer boring reliability: deterministic shells, idempotency, queues, feature flags.
• Quantify: rough capacity maths and cost per run beat vague "it scales".
• Drill more in the Interview Q&A bank (filter by Deployment & Ops and Safety & Security).