By Deepak Bagada, CEO at SaaSNext & Principal AI Architect.

Build an Ultrafast Incident-Response Agent with LangGraph

On August 13, 2026, OpenAI previewed Ultrafast — a service tier that runs GPT-5.6 Sol up to 14 times faster than the Standard tier while sustaining up to 750 output tokens per second. The engine underneath is Cerebras wafer-scale silicon: 900,000 cores and 44GB of SRAM holding the entire model on-chip, which eliminates the weight-transfer bottleneck that dominates GPU inference latency. Cerebras' own benchmarks make the delta concrete — GPT-5.6 Sol Ultrafast runs roughly 11 times faster than Claude Fable 5 and 5 times faster than Claude Opus 4.8 Fast mode. A 2,500-question Humanity's Last Exam session finished in 11h11m where Fable 5 needed 78h27m, and GDP-Val showed a 5.6x end-to-end speedup. OpenAI listed four launch use cases — production outage response, cybersecurity rapid detection, agent workflows, and financial analysis — and the tier arrives first in the OpenAI API as a limited preview with no pricing disclosed.

Latency is not a convenience metric for those workloads. In incident response, latency is the product. Every unresolved minute is measured in failed checkouts, angry customers, and burnt SRE attention — and in India's busiest shopping hours, a slow payment API means lost conversions measured in lakhs of rupees. This dispatch builds fastres, a LangGraph workflow that treats inference speed as a first-class constraint. It detects an incident, ingests logs and traces, and routes every task down one of two paths: a latency-critical fast path on Ultrafast-class inference for S1/S2 events, and a reasoning-heavy deep-dive path for the rare cases that need slow deliberation. Along the way it caches hot diagnostics so known failure signatures skip the model entirely, guards the premium fast-tier budget, and degrades to standard inference instead of dying when the fast tier hiccups. The rest of the AI workflows library plugs into the same skeleton.

Why latency is the incident bottleneck

An on-call loop has three phases, and each one is latency-bound. Detection: how many seconds before the pager fires. Diagnosis: how long before a root-cause hypothesis exists. Remediation: how long before the fix ships. Conventional LLM agents compress the first but leave diagnosis slow, because a reasoning model burns 30–60 seconds per hop and an incident can need several hops. Ultrafast-class inference compresses the second phase — 750 tokens per second means a root-cause analysis that used to take a minute now lands in a few seconds, which is the difference between catching a cascading failure at the first 5xx spike and discovering it at the fifth alarm.

But raw speed is useless if it is ungoverned. The fast tier is premium — pricing is undisclosed, but assume a multi-dollar-per-million-tokens premium over standard — so a naive agent would happily burn the whole budget on a trivial 403 configuration error. And a single noisy alert can trigger a stampede: ten agents all asking the fast tier the same question. That is why this workflow routes before it thinks. A cheap signature cache answers the questions that were already answered, the budget guard answers the questions that got too expensive, and only genuinely novel S1/S2 events reach the fast tier. Treat fast inference as a scarce, expensive resource and you get the speed without the bill.

Architecture

flowchart TD
    A[Detect incident: alert rules + trace signals] --> B[Ingest logs and traces]
    B --> C[Classify severity + tier routing]
    C --> D{Cache hit?}
    D -- yes --> E[Hot diagnostic cache: instant answer]
    D -- no --> F{Budget allowed + tier up?}
    F -- yes --> G[Fast analysis: GPT-5.6 Sol Ultrafast]
    F -- no --> H[Standard analysis: deep dive]
    E --> I[Verify root-cause hypothesis]
    G --> I
    H --> I
    I -- low confidence --> B
    I -- confirmed --> J[Draft runbook fix]
    J --> K{Human approval gate}
    K -- approved --> L[Apply runbook + verify remediation]
    K -- rejected --> M[Escalate to on-call engineer]
    L --> N[Post-incident review]
    M --> N

Project setup

mkdir fastres && cd fastres
python -m venv .venv && source .venv/bin/activate
pip install langgraph langchain-openai pydantic redis

# .env
OPENAI_API_KEY=sk-...
FAST_MODEL=gpt-5.6-sol-ultrafast
STANDARD_MODEL=gpt-5.6-sol-standard
DAILY_FAST_BUDGET_USD=40.00
FAST_TOKEN_PRICE_USD_PER_M=6.00
STANDARD_TOKEN_PRICE_USD_PER_M=1.50
HOT_CACHE_TTL_S=900
FAST_TIER_STATUS=up
TRACE_ENDPOINT=http://otel-collector:4318
SLACK_WEBHOOK_URL=https://hooks.slack.com/services/T00/B00/xx

schemas.py

import uuid
from enum import Enum
from pydantic import BaseModel, Field

class Tier(str, Enum):
    FAST = "fast"
    STANDARD = "standard"

class Severity(str, Enum):
    S1 = "s1"
    S2 = "s2"
    S3 = "s3"

class Incident(BaseModel):
    incident_id: str = Field(default_factory=lambda: uuid.uuid4().hex[:12])
    severity: Severity = Severity.S3
    title: str
    trace_ids: list[str] = Field(default_factory=list)
    first_seen_at: str = ""

class Diagnostic(BaseModel):
    signature: str = Field(..., description="normalized error signature")
    summary: str
    ttl_until: str = ""

class RootCause(BaseModel):
    incident_id: str
    hypothesis: str
    confidence: float = Field(ge=0, le=1)
    evidence: list[str] = Field(default_factory=list)
    tier_used: Tier = Tier.STANDARD

class RunbookDraft(BaseModel):
    incident_id: str
    cause: RootCause
    steps: list[str] = Field(default_factory=list)
    rollback: list[str] = Field(default_factory=list)

class BudgetState(BaseModel):
    daily_spend_usd: float = 0.0
    cap_usd: float = 40.0
    over: bool = False

tools.py

import os, time, json, hashlib
import urllib.request
from schemas import Tier

def retry(fn, attempts=3, backoff=(0.25, 0.5, 1.0)):
    last = None
    for i, wait in enumerate(backoff[:attempts]):
        try:
            return fn()
        except Exception as e:
            last = e
            if i < attempts - 1:
                time.sleep(wait)
    raise last

def call_openai(model, prompt, temperature=0.2) -> str:
    body = json.dumps({"model": model, "messages": [
        {"role": "user", "content": prompt}],
        "temperature": temperature}).encode()
    req = urllib.request.Request(
        "https://api.openai.com/v1/chat/completions", data=body,
        headers={"Authorization": f"Bearer {os.getenv('OPENAI_API_KEY')}",
                 "Content-Type": "application/json"})
    with urllib.request.urlopen(req, timeout=60) as resp:
        return json.load(resp)["choices"][0]["message"]["content"]

def fast_infer(prompt):
    return retry(lambda: call_openai(os.getenv("FAST_MODEL"), prompt))

def standard_infer(prompt):
    return call_openai(os.getenv("STANDARD_MODEL"), prompt, temperature=0.0)

def ingest_logs(trace_ids):
    # In production this pulls from OTel/Tempo or a Kafka topic.
    return [{"trace_id": t, "level": "error",
             "msg": f"5xx spike detected on {t}"} for t in trace_ids]

def make_signature(entries):
    raw = "|".join(e.get("msg", "") for e in entries[:5])
    return hashlib.sha256(raw.encode()).hexdigest()[:16]

class HotCache:
    def __init__(self, ttl_s=900):
        self._store = {}
        self.ttl_s = ttl_s
    def get(self, sig):
        item = self._store.get(sig)
        if item and time.time() < float(item.ttl_until):
            return item
        return None
    def set(self, diag):
        diag.ttl_until = str(time.time() + self.ttl_s)
        self._store[diag.signature] = diag

class BudgetGuard:
    def __init__(self, cap_usd=40.0):
        self.cap_usd = cap_usd
        self.spend = 0.0
    def charge(self, tokens, tier):
        price = float(os.getenv("FAST_TOKEN_PRICE_USD_PER_M", "6.0")
                      if tier == Tier.FAST else
                      os.getenv("STANDARD_TOKEN_PRICE_USD_PER_M", "1.5"))
        self.spend += tokens / 1_000_000 * price
    def allow_fast(self):
        return self.spend < self.cap_usd

def fast_tier_available():
    return os.getenv("FAST_TIER_STATUS", "up") == "up"

def post_slack(msg):
    try:
        urllib.request.urlopen(urllib.request.Request(
            os.getenv("SLACK_WEBHOOK_URL"),
            data=json.dumps({"text": msg}).encode(),
            headers={"Content-Type": "application/json"}), timeout=5)
    except Exception:
        pass

graph.py

import os
from typing import TypedDict, Literal
from langgraph.graph import StateGraph, END
from schemas import (Tier, Severity, Diagnostic, RootCause,
                     RunbookDraft, BudgetState)
from tools import (ingest_logs, make_signature, HotCache, BudgetGuard,
                   fast_infer, standard_infer, fast_tier_available,
                   post_slack)

class FastResState(TypedDict):
    incident: Incident | None
    logs: list[dict]
    sig: str
    cause: RootCause | None
    draft: RunbookDraft | None
    budget: BudgetState
    approved: bool
    status: Literal["resolved", "escalated", "in_progress"]

cache = HotCache(int(os.getenv("HOT_CACHE_TTL_S", "900")))
guard = BudgetGuard(float(os.getenv("DAILY_FAST_BUDGET_USD", "40")))

def detect_node(state):
    return {**state}  # alert rules populate state["incident"] upstream

def ingest_node(state):
    logs = ingest_logs(state["incident"].trace_ids)
    return {**state, "logs": logs, "sig": make_signature(logs)}

def route_tier(state) -> str:
    if cache.get(state["sig"]):
        return "cache"
    sev = state["incident"].severity
    if sev in (Severity.S1, Severity.S2) and guard.allow_fast()             and fast_tier_available():
        return "fast"
    return "standard"

def cache_node(state):
    diag = cache.get(state["sig"])
    cause = RootCause(incident_id=state["incident"].incident_id,
                      hypothesis=diag.summary, confidence=0.95,
                      evidence=["hot cache hit"], tier_used=Tier.STANDARD)
    return {**state, "cause": cause}

def fast_node(state):
    prompt = f"Root-cause this S{state['incident'].severity.value} incident.
{state['logs']}"
    answer = fast_infer(prompt)
    guard.charge(len(answer) // 4, Tier.FAST)
    cause = RootCause(incident_id=state["incident"].incident_id,
                      hypothesis=answer, confidence=0.80,
                      evidence=[l.get("msg", "") for l in state["logs"]],
                      tier_used=Tier.FAST)
    cache.set(Diagnostic(signature=state["sig"], summary=answer))
    return {**state, "cause": cause}

def standard_node(state):
    prompt = (f"Reason carefully about {state['incident'].model_dump_json()}. "
              f"Logs:
{state['logs']}
Produce one verified hypothesis.")
    answer = standard_infer(prompt)
    guard.charge(len(answer) // 4, Tier.STANDARD)
    cause = RootCause(incident_id=state["incident"].incident_id,
                      hypothesis=answer, confidence=0.85,
                      evidence=[l.get("msg", "") for l in state["logs"]],
                      tier_used=Tier.STANDARD)
    return {**state, "cause": cause}

def verify_node(state):
    return {**state}  # cross-check hypothesis against live traces here

def route_verify(state) -> str:
    if state["cause"] and state["cause"].confidence >= 0.7:
        return "draft"
    return "ingest"  # low confidence: re-ingest with more traces

def draft_node(state):
    draft = RunbookDraft(
        incident_id=state["incident"].incident_id, cause=state["cause"],
        steps=["Roll back deploy tag a86b1c4",
               "Restart payment-worker replicas",
               "Re-run ingest smoke test, confirm P99 under 200ms"],
        rollback=["git revert a86b1c4 --no-commit"])
    post_slack(f"Runbook ready for {state['incident'].incident_id}")
    return {**state, "draft": draft}

def approve_node(state):
    return {**state, "approved": True}  # real gate waits on a Slack action

def route_approval(state) -> str:
    return "apply" if state["approved"] else "escalate"

def apply_node(state):
    for step in state["draft"].steps:
        print(f"[apply] {step}")  # real terraform/kubectl calls here
    return {**state, "status": "resolved"}

def escalate_node(state):
    post_slack(f"PAGE {state['incident'].incident_id}: approval rejected")
    return {**state, "status": "escalated"}

def pir_node(state):
    post_slack(f"PIR: {state['incident'].incident_id} "
               f"tier={state['cause'].tier_used.value} "
               f"spend=${round(guard.spend, 2)} status={state['status']}")
    return {**state}

def build_graph():
    g = StateGraph(FastResState)
    for name, fn in [("detect", detect_node), ("ingest", ingest_node),
                     ("cache", cache_node), ("fast", fast_node),
                     ("standard", standard_node), ("verify", verify_node),
                     ("draft", draft_node), ("approve", approve_node),
                     ("apply", apply_node), ("escalate", escalate_node),
                     ("pir", pir_node)]:
        g.add_node(name, fn)
    g.set_entry_point("detect")
    g.add_edge("detect", "ingest")
    g.add_conditional_edges("ingest", route_tier,
        {"cache": "cache", "fast": "fast", "standard": "standard"})
    for n in ("cache", "fast", "standard"):
        g.add_edge(n, "verify")
    g.add_conditional_edges("verify", route_verify,
        {"draft": "draft", "ingest": "ingest"})
    g.add_edge("draft", "approve")
    g.add_conditional_edges("approve", route_approval,
        {"apply": "apply", "escalate": "escalate"})
    g.add_edge("apply", "pir")
    g.add_edge("escalate", "pir")
    g.add_edge("pir", END)
    return g.compile()

main.py

import asyncio, json
from graph import build_graph
from schemas import Incident, Severity, BudgetState

async def main():
    graph = build_graph()
    incident = Incident(
        severity=Severity.S1,
        title="Payment API 5xx spike",
        trace_ids=["tr_9f2a", "tr_9f2b", "tr_9f2c"],
    )
    result = await graph.ainvoke({
        "incident": incident, "logs": [], "sig": "",
        "cause": None, "draft": None,
        "budget": BudgetState(), "approved": True, "status": "in_progress",
    })
    print(json.dumps({
        "incident": result["incident"].incident_id,
        "hypothesis": result["cause"].hypothesis[:120],
        "tier_used": result["cause"].tier_used.value,
        "status": result["status"],
    }, indent=2))

if __name__ == "__main__":
    asyncio.run(main())

How the speed-tier routing works

The graph is a straight shot with three deliberate shortcuts. ingest pulls logs and traces, then computes a signature over the first five messages. The route_tier conditional edge then makes one cheap decision, not a model call: is this signature already in the hot cache? If yes, cache answers instantly — a known signature like "disk full on db-04" returns the cached hypothesis with no inference and no cost. If no, the router checks the guard: is the incident S1 or S2, is the daily budget under the cap, and is the fast tier reporting healthy? Only if all three hold does the event go to fast, which runs GPT-5.6 Sol Ultrafast and writes the result back into the cache so the next identical alarm is free. Everything else — S3 noise, budget exhaustion, a rate-limited preview tier — falls through to standard, which runs the full reasoning model and is allowed to be slow because the incident is not actually on fire.

The verify node is where the workflow refuses to be fooled. A hypothesis with confidence below 0.7 routes back to ingest, pulling more traces and re-signaturing, instead of letting a half-baked theory reach a runbook. Confirmed hypotheses flow to draft, which writes the fix steps and a rollback, then stops at the approve gate — a real deployment waits on a human Slack/PagerDuty action. A rejected approval escalates to the on-call engineer with the runbook attached. Whatever the outcome, pir posts a post-incident review that records the tier used and the money spent, so the cost of speed is audited, not discovered at month-end. For teams hardening this loop further, the containment and tooling patterns in the MCP directory bolt directly onto the same nodes.

Retry Rules & Error Handling

Every tier failure has a designed response. The table below is the contract the graph runs under; the retry helper implements the backoff columns.

Cost & decision matrix

The fast tier is a premium resource, so the workflow spends it like one. Prices are illustrative at 2026 frontier rates (fast plausibly lands around ₹40–60 per million tokens, standard around ₹10–15); replace with your real ledger before production.

Testing the workflow

Test four scenarios. First, a cache hit: run the same incident twice; the second run must skip both model tiers and land in cache. Second, a budget cap: set DAILY_FAST_BUDGET_USD=0.01 and confirm an S1 routes to standard with an over-budget log line. Third, a fast-tier outage: set FAST_TIER_STATUS=down and confirm graceful degradation plus the Slack alert after three failures. Fourth, an approval rejection: set approved: false and confirm the workflow escalates instead of applying. The budget test is the one to watch — if the guard ever lets an S3 through to the fast tier, your routing node is broken and will cost you the day's entire inference budget in one noisy hour.

Closing thoughts

OpenAI's Ultrafast tier is the first genuinely new inference economy since the token pricing wars began: 14x the speed, 750 tokens a second, on silicon most teams will never own. fastres treats that resource the way good SRE treats a hot spare — it is always ready, never wasted, and wired into a decision path that knows when speed matters and when it does not. Cache the known, fast-path the novel, deep-dive the noise, and audit every token. Track the wider agent-routing wave on latest AI news, and grab the full pattern library from AI workflows.