How to Build an AI Agent Circuit Breaker Pattern That Automatically Isolates Failing Foundation Model Endpoints Before Cascading Failures Corrupt Downstream Multi-Agent Workflow State in H2 2026

Multi-agent systems running on top of foundation model endpoints are the backbone of production AI workloads in 2026. Orchestrators fan out tasks to specialized sub-agents, each of which calls one or more model endpoints, writes intermediate results to shared state stores, and hands off context to the next agent in the chain. When everything works, this is elegant. When a single model endpoint starts returning garbage, timing out intermittently, or degrading silently, the elegance collapses fast.

The insidious part is not the hard failure. A hard failure is easy to catch. The real danger is the partial failure: a foundation model endpoint that is still technically alive but returning truncated outputs, hallucinated tool calls, or malformed JSON. Downstream agents ingest that corrupted output as ground truth, mutate shared workflow state, and propagate the corruption further down the chain. By the time you see an error, the blast radius spans multiple agents, multiple state mutations, and potentially external side effects that cannot be rolled back.

This tutorial shows you exactly how to build a production-grade AI Agent Circuit Breaker that detects both hard and soft failures in foundation model endpoints, isolates them automatically, routes around them, and protects your multi-agent workflow state from corruption. Every code sample is in Python, and the architecture is framework-agnostic so it works whether your orchestration layer is LangGraph, CrewAI, a bespoke system, or anything in between.

Why Standard Circuit Breakers Are Not Enough for Foundation Models

The classic circuit breaker pattern, popularized by Michael Nygard and later formalized in Netflix's Hystrix, operates on a simple premise: count failures, trip the breaker when failures exceed a threshold, wait, probe, and reset. This works beautifully for microservices returning HTTP 500s. Foundation model endpoints are a different beast entirely.

Here is what makes LLM endpoints uniquely dangerous in a multi-agent context:

• Silent semantic degradation: The endpoint returns HTTP 200 with a well-formed JSON body, but the content is wrong. A standard circuit breaker never trips because no exception was raised.
• Latency spikes masking correctness issues: A model endpoint under load may return correct outputs slowly, or incorrect outputs quickly. Latency alone is not a reliable health signal.
• Non-deterministic output variance: Some output variance is expected and healthy. Distinguishing healthy variance from degradation requires semantic awareness, not just structural checks.
• Stateful downstream contamination: Unlike a stateless API, a bad LLM response that gets written into a vector store, a task queue, or a shared scratchpad persists. The damage outlives the bad call.
• Multi-endpoint fan-out: A single agent step may call several model endpoints (a router model, an embedder, a generator). Failure attribution is non-trivial.

To handle all of this, your circuit breaker needs to be semantics-aware, state-protective, and multi-signal. Let's build it layer by layer.

The Architecture at a Glance

Before diving into code, here is the high-level component map:

• HealthProbe: Collects raw signals per endpoint (latency, error rate, output schema violations, semantic confidence scores).
• BreakerStateMachine: Maintains per-endpoint state (CLOSED, OPEN, HALF-OPEN) and transitions based on aggregated health signals.
• SemanticValidator: Runs lightweight checks on model outputs to detect soft failures before they enter workflow state.
• StateGuard: Wraps all workflow state writes with a transactional boundary that rolls back on breaker trip.
• FallbackRouter: Redirects traffic to a secondary endpoint or a cached safe response when the primary breaker is OPEN.
• AgentCircuitBreakerMiddleware: The thin wrapper that composes all of the above and plugs into your agent execution loop.

Step 1: Define Your Health Signal Schema

Start by defining what "unhealthy" means for a foundation model endpoint. This is the most important design decision you will make, because the wrong signals produce false trips or, worse, missed detections.

from dataclasses import dataclass, field
from typing import Optional
import time

@dataclass
class EndpointHealthSignal:
    endpoint_id: str
    timestamp: float = field(default_factory=time.time)

    # Hard failure signals
    http_error: bool = False
    timeout: bool = False
    connection_refused: bool = False

    # Soft failure signals
    output_schema_valid: bool = True
    required_fields_present: bool = True
    output_length_ratio: float = 1.0   # observed / expected; 1.0 is nominal
    latency_ms: float = 0.0

    # Semantic signals (populated by SemanticValidator)
    semantic_confidence: float = 1.0   # 0.0 to 1.0
    tool_call_parseable: bool = True
    json_parseable: bool = True

    @property
    def is_hard_failure(self) -> bool:
        return self.http_error or self.timeout or self.connection_refused

    @property
    def is_soft_failure(self) -> bool:
        return (
            not self.output_schema_valid
            or not self.required_fields_present
            or self.output_length_ratio < 0.2   # severely truncated
            or not self.tool_call_parseable
            or not self.json_parseable
            or self.semantic_confidence < 0.35
        )

    @property
    def is_failure(self) -> bool:
        return self.is_hard_failure or self.is_soft_failure

The output_length_ratio and semantic_confidence fields are the keys that separate this from a naive circuit breaker. A ratio below 0.2 means the model returned less than 20% of the expected output length, which is a strong signal of truncation or context window exhaustion. Semantic confidence is populated by the SemanticValidator in Step 3.

Step 2: Build the Breaker State Machine

The state machine is the heart of the circuit breaker. It tracks a rolling window of health signals per endpoint and transitions between states accordingly.

from collections import deque
from enum import Enum
from threading import Lock
import time

class BreakerState(Enum):
    CLOSED = "closed"       # Normal operation; traffic flows
    OPEN = "open"           # Endpoint isolated; traffic blocked
    HALF_OPEN = "half_open" # Probe mode; limited traffic allowed

@dataclass
class BreakerConfig:
    window_size: int = 20               # Rolling window of signal samples
    failure_threshold: float = 0.4     # Trip if >40% of window is failures
    soft_failure_weight: float = 0.6   # Soft failures count as 0.6 of a failure
    open_duration_seconds: float = 30.0
    half_open_probe_count: int = 3
    recovery_threshold: float = 0.85   # % success needed to close from half-open

class BreakerStateMachine:
    def __init__(self, endpoint_id: str, config: BreakerConfig):
        self.endpoint_id = endpoint_id
        self.config = config
        self.state = BreakerState.CLOSED
        self._window: deque = deque(maxlen=config.window_size)
        self._lock = Lock()
        self._opened_at: Optional[float] = None
        self._half_open_successes: int = 0
        self._half_open_attempts: int = 0

    def record(self, signal: EndpointHealthSignal) -> BreakerState:
        with self._lock:
            weight = self._signal_weight(signal)
            self._window.append(weight)
            self._evaluate_transition()
            return self.state

    def _signal_weight(self, signal: EndpointHealthSignal) -> float:
        if signal.is_hard_failure:
            return 1.0
        if signal.is_soft_failure:
            return self.config.soft_failure_weight
        return 0.0  # success

    def _failure_rate(self) -> float:
        if not self._window:
            return 0.0
        return sum(self._window) / len(self._window)

    def _evaluate_transition(self):
        now = time.time()

        if self.state == BreakerState.CLOSED:
            if (len(self._window) >= self.config.window_size // 2
                    and self._failure_rate() >= self.config.failure_threshold):
                self._trip(now)

        elif self.state == BreakerState.OPEN:
            if now - self._opened_at >= self.config.open_duration_seconds:
                self._enter_half_open()

        elif self.state == BreakerState.HALF_OPEN:
            if self._half_open_attempts >= self.config.half_open_probe_count:
                success_rate = self._half_open_successes / self._half_open_attempts
                if success_rate >= self.config.recovery_threshold:
                    self._close()
                else:
                    self._trip(now)

    def _trip(self, now: float):
        self.state = BreakerState.OPEN
        self._opened_at = now
        self._window.clear()
        print(f"[BREAKER] Endpoint '{self.endpoint_id}' TRIPPED. State: OPEN.")

    def _enter_half_open(self):
        self.state = BreakerState.HALF_OPEN
        self._half_open_successes = 0
        self._half_open_attempts = 0
        print(f"[BREAKER] Endpoint '{self.endpoint_id}' entering HALF-OPEN probe mode.")

    def _close(self):
        self.state = BreakerState.CLOSED
        self._window.clear()
        print(f"[BREAKER] Endpoint '{self.endpoint_id}' RECOVERED. State: CLOSED.")

    def allow_request(self) -> bool:
        with self._lock:
            if self.state == BreakerState.CLOSED:
                return True
            if self.state == BreakerState.OPEN:
                return False
            if self.state == BreakerState.HALF_OPEN:
                self._half_open_attempts += 1
                return True
        return False

    def record_half_open_success(self):
        with self._lock:
            if self.state == BreakerState.HALF_OPEN:
                self._half_open_successes += 1

Notice the soft_failure_weight parameter. Rather than treating a soft failure as a binary event, we assign it a fractional weight. This prevents a single ambiguous response from tripping the breaker while still accumulating pressure as soft failures pile up. Tune this value based on your tolerance for false positives versus missed detections.

Step 3: Build the Semantic Validator

This is the component that makes your circuit breaker genuinely AI-aware. The semantic validator runs a fast, lightweight check on each model output before it is accepted into the workflow. It does not call another LLM (that would be circular and expensive). Instead, it uses a combination of structural heuristics, confidence score extraction, and optional embedding-based drift detection.

import json
import re
from typing import Any, Dict, List, Optional

class SemanticValidator:
    def __init__(
        self,
        required_keys: Optional[List[str]] = None,
        expected_output_schema: Optional[Dict] = None,
        min_token_estimate: int = 10,
        tool_call_pattern: Optional[str] = None,
    ):
        self.required_keys = required_keys or []
        self.expected_schema = expected_output_schema
        self.min_token_estimate = min_token_estimate
        self.tool_call_pattern = re.compile(tool_call_pattern) if tool_call_pattern else None

    def validate(self, raw_output: str, expected_length_estimate: int = 200) -> EndpointHealthSignal:
        signal = EndpointHealthSignal(endpoint_id="__validator__")

        # Length ratio check
        observed_tokens = len(raw_output.split())
        signal.output_length_ratio = observed_tokens / max(expected_length_estimate, 1)

        # JSON parseability
        parsed = None
        try:
            parsed = json.loads(raw_output)
            signal.json_parseable = True
        except json.JSONDecodeError:
            # Not all outputs are JSON; only flag if JSON was expected
            signal.json_parseable = not self.required_keys  # if no keys expected, JSON is optional

        # Required key presence
        if self.required_keys and parsed is not None:
            signal.required_fields_present = all(k in parsed for k in self.required_keys)
        elif self.required_keys and parsed is None:
            signal.required_fields_present = False

        # Tool call parseability
        if self.tool_call_pattern:
            signal.tool_call_parseable = bool(self.tool_call_pattern.search(raw_output))

        # Semantic confidence heuristic
        signal.semantic_confidence = self._estimate_confidence(raw_output, parsed)

        return signal

    def _estimate_confidence(self, raw: str, parsed: Optional[Any]) -> float:
        score = 1.0

        # Penalize very short outputs
        word_count = len(raw.split())
        if word_count < self.min_token_estimate:
            score -= 0.5

        # Penalize outputs that are repetitive (a common degradation pattern)
        words = raw.lower().split()
        if len(words) > 10:
            unique_ratio = len(set(words)) / len(words)
            if unique_ratio < 0.3:
                score -= 0.4

        # Penalize outputs containing known error markers
        error_markers = [
            "i cannot", "i'm unable", "error occurred", "null", "undefined",
            "context length exceeded", "rate limit", "sorry, i"
        ]
        lower_raw = raw.lower()
        for marker in error_markers:
            if marker in lower_raw:
                score -= 0.25
                break

        # Penalize if required keys are missing from a parsed response
        if self.required_keys and parsed and isinstance(parsed, dict):
            missing = [k for k in self.required_keys if k not in parsed]
            score -= 0.15 * len(missing)

        return max(0.0, min(1.0, score))

In production, you can extend _estimate_confidence with embedding-based drift detection. Keep a rolling embedding centroid of known-good outputs for each endpoint using a lightweight model like a local sentence transformer, then flag outputs whose cosine distance from the centroid exceeds a threshold. This catches semantic drift that heuristics alone will miss.

Step 4: Build the StateGuard (Transactional State Protection)

This is the piece most tutorials skip, and it is arguably the most important. Even if your circuit breaker trips correctly, any state mutations that happened before the trip are already in your workflow store. The StateGuard wraps every state write in a context manager that can be rolled back if a breaker trips mid-workflow.

from contextlib import contextmanager
from copy import deepcopy
from typing import Any, Dict

class WorkflowStateStore:
    """A simple in-memory workflow state store with snapshot/rollback support."""

    def __init__(self):
        self._state: Dict[str, Any] = {}
        self._snapshots: list = []

    def get(self, key: str) -> Any:
        return self._state.get(key)

    def set(self, key: str, value: Any):
        self._state[key] = value

    def snapshot(self) -> int:
        """Take a snapshot and return its ID."""
        self._snapshots.append(deepcopy(self._state))
        return len(self._snapshots) - 1

    def rollback(self, snapshot_id: int):
        """Restore state to a previous snapshot."""
        if 0 <= snapshot_id < len(self._snapshots):
            self._state = self._snapshots[snapshot_id]
            self._snapshots = self._snapshots[:snapshot_id]
            print(f"[STATE_GUARD] Rolled back to snapshot {snapshot_id}.")
        else:
            raise ValueError(f"Invalid snapshot ID: {snapshot_id}")

    def commit(self, snapshot_id: int):
        """Discard snapshots up to and including this ID (commit the changes)."""
        self._snapshots = self._snapshots[snapshot_id + 1:]


class StateGuard:
    def __init__(self, store: WorkflowStateStore):
        self.store = store

    @contextmanager
    def protected_write(self):
        snapshot_id = self.store.snapshot()
        try:
            yield self.store
            self.store.commit(snapshot_id)
        except BreakerTripException as e:
            self.store.rollback(snapshot_id)
            raise
        except Exception as e:
            self.store.rollback(snapshot_id)
            raise


class BreakerTripException(Exception):
    """Raised when a circuit breaker trips during a protected write block."""
    def __init__(self, endpoint_id: str, state: BreakerState):
        self.endpoint_id = endpoint_id
        self.breaker_state = state
        super().__init__(
            f"Circuit breaker OPEN for endpoint '{endpoint_id}'. "
            f"State write rolled back."
        )

The key insight here is that protected_write takes a snapshot before any agent writes happen, then commits only if the entire block succeeds. If a BreakerTripException is raised at any point during the block, the state is rolled back to the pre-block snapshot. This gives you exactly-once semantics for state mutations in the happy path, and zero-corruption semantics when the breaker trips.

Step 5: Build the Fallback Router

When a breaker is OPEN, traffic must go somewhere. The FallbackRouter handles this with a priority-ordered list of fallback options per endpoint.

from typing import Callable, List, Optional, Tuple
import hashlib

@dataclass
class FallbackOption:
    endpoint_id: str
    priority: int          # Lower is higher priority
    is_cached: bool = False
    cache_key_fn: Optional[Callable] = None

class FallbackRouter:
    def __init__(self):
        self._routes: Dict[str, List[FallbackOption]] = {}
        self._cache: Dict[str, str] = {}

    def register_fallback(self, primary_endpoint: str, fallback: FallbackOption):
        if primary_endpoint not in self._routes:
            self._routes[primary_endpoint] = []
        self._routes[primary_endpoint].append(fallback)
        self._routes[primary_endpoint].sort(key=lambda f: f.priority)

    def cache_response(self, endpoint_id: str, prompt_hash: str, response: str):
        cache_key = f"{endpoint_id}:{prompt_hash}"
        self._cache[cache_key] = response

    def get_fallback(
        self,
        primary_endpoint: str,
        breakers: Dict[str, "BreakerStateMachine"],
        prompt: Optional[str] = None,
    ) -> Tuple[Optional[str], Optional[str]]:
        """
        Returns (endpoint_id, cached_response).
        If a live fallback is found, returns (endpoint_id, None).
        If a cached response is found, returns (None, cached_response).
        If nothing is available, returns (None, None).
        """
        options = self._routes.get(primary_endpoint, [])
        prompt_hash = hashlib.md5(prompt.encode()).hexdigest() if prompt else ""

        for option in options:
            if option.is_cached:
                cache_key = f"{primary_endpoint}:{prompt_hash}"
                cached = self._cache.get(cache_key)
                if cached:
                    print(f"[FALLBACK] Serving cached response for '{primary_endpoint}'.")
                    return None, cached
                continue

            breaker = breakers.get(option.endpoint_id)
            if breaker and breaker.allow_request():
                print(f"[FALLBACK] Routing '{primary_endpoint}' traffic to '{option.endpoint_id}'.")
                return option.endpoint_id, None

        print(f"[FALLBACK] No viable fallback for '{primary_endpoint}'. Failing gracefully.")
        return None, None

Step 6: Compose the AgentCircuitBreakerMiddleware

Now we wire everything together into a single middleware class that your agent execution loop calls for every model invocation.

import asyncio
from typing import Awaitable

class AgentCircuitBreakerMiddleware:
    def __init__(
        self,
        state_store: WorkflowStateStore,
        breaker_config: Optional[BreakerConfig] = None,
    ):
        self.state_store = state_store
        self.state_guard = StateGuard(state_store)
        self.breaker_config = breaker_config or BreakerConfig()
        self.breakers: Dict[str, BreakerStateMachine] = {}
        self.fallback_router = FallbackRouter()
        self.validators: Dict[str, SemanticValidator] = {}

    def register_endpoint(
        self,
        endpoint_id: str,
        validator: Optional[SemanticValidator] = None,
        fallbacks: Optional[List[FallbackOption]] = None,
    ):
        self.breakers[endpoint_id] = BreakerStateMachine(endpoint_id, self.breaker_config)
        if validator:
            self.validators[endpoint_id] = validator
        for fb in (fallbacks or []):
            self.fallback_router.register_fallback(endpoint_id, fb)

    async def call(
        self,
        endpoint_id: str,
        prompt: str,
        model_fn: Callable[[str, str], Awaitable[str]],
        expected_length: int = 200,
        write_to_state: Optional[Callable[[str, WorkflowStateStore], None]] = None,
    ) -> Optional[str]:
        breaker = self.breakers.get(endpoint_id)
        if not breaker:
            raise ValueError(f"Endpoint '{endpoint_id}' not registered.")

        # Check if breaker allows the request
        if not breaker.allow_request():
            fallback_ep, cached = self.fallback_router.get_fallback(
                endpoint_id, self.breakers, prompt
            )
            if cached:
                return cached
            if fallback_ep:
                return await self.call(fallback_ep, prompt, model_fn, expected_length, write_to_state)
            return None  # Graceful degradation: return None, let caller decide

        # Attempt the model call
        raw_output = None
        signal = EndpointHealthSignal(endpoint_id=endpoint_id)

        try:
            start = asyncio.get_event_loop().time()
            raw_output = await model_fn(endpoint_id, prompt)
            signal.latency_ms = (asyncio.get_event_loop().time() - start) * 1000

        except TimeoutError:
            signal.timeout = True
        except ConnectionRefusedError:
            signal.connection_refused = True
        except Exception:
            signal.http_error = True

        # Run semantic validation if we got a response
        if raw_output is not None:
            validator = self.validators.get(endpoint_id)
            if validator:
                validation_signal = validator.validate(raw_output, expected_length)
                signal.output_schema_valid = validation_signal.output_schema_valid
                signal.required_fields_present = validation_signal.required_fields_present
                signal.output_length_ratio = validation_signal.output_length_ratio
                signal.tool_call_parseable = validation_signal.tool_call_parseable
                signal.json_parseable = validation_signal.json_parseable
                signal.semantic_confidence = validation_signal.semantic_confidence

        # Record the signal and get the new breaker state
        new_state = breaker.record(signal)

        # Cache good responses for future fallback use
        if raw_output and not signal.is_failure:
            import hashlib
            prompt_hash = hashlib.md5(prompt.encode()).hexdigest()
            self.fallback_router.cache_response(endpoint_id, prompt_hash, raw_output)
            if new_state == BreakerState.HALF_OPEN:
                breaker.record_half_open_success()

        # If the signal was a failure and the breaker just tripped, raise to trigger rollback
        if signal.is_failure and new_state == BreakerState.OPEN:
            raise BreakerTripException(endpoint_id, new_state)

        # If we have a good response, write to state inside the guard
        if raw_output and not signal.is_failure and write_to_state:
            with self.state_guard.protected_write():
                write_to_state(raw_output, self.state_store)

        return raw_output if not signal.is_failure else None

Step 7: Integrate with Your Multi-Agent Workflow

Here is a complete example showing how the middleware slots into a two-agent workflow where Agent A feeds Agent B through shared state.

import asyncio

# --- Mock model function (replace with your real SDK calls) ---
async def mock_model_fn(endpoint_id: str, prompt: str) -> str:
    # Simulate a degraded endpoint returning truncated output
    if endpoint_id == "gpt-primary" and "corrupt" in prompt:
        return '{"result": null}'   # Soft failure: null result
    return f'{{"result": "Processed: {prompt[:30]}...", "confidence": 0.92}}'

# --- Setup ---
store = WorkflowStateStore()
middleware = AgentCircuitBreakerMiddleware(
    state_store=store,
    breaker_config=BreakerConfig(
        window_size=10,
        failure_threshold=0.4,
        open_duration_seconds=15.0,
    )
)

middleware.register_endpoint(
    endpoint_id="gpt-primary",
    validator=SemanticValidator(
        required_keys=["result", "confidence"],
        min_token_estimate=5,
    ),
    fallbacks=[
        FallbackOption(endpoint_id="gpt-secondary", priority=1),
        FallbackOption(endpoint_id="cached", priority=2, is_cached=True),
    ]
)
middleware.register_endpoint(endpoint_id="gpt-secondary")

# --- Agent A: Generates a plan ---
async def agent_a(prompt: str):
    def write_plan(output: str, s: WorkflowStateStore):
        s.set("agent_a_plan", output)

    result = await middleware.call(
        endpoint_id="gpt-primary",
        prompt=prompt,
        model_fn=mock_model_fn,
        expected_length=50,
        write_to_state=write_plan,
    )
    print(f"[Agent A] Result: {result}")
    return result

# --- Agent B: Executes the plan from shared state ---
async def agent_b():
    plan = store.get("agent_a_plan")
    if plan is None:
        print("[Agent B] No plan in state. Skipping execution (safe degradation).")
        return

    def write_execution(output: str, s: WorkflowStateStore):
        s.set("agent_b_execution", output)

    result = await middleware.call(
        endpoint_id="gpt-primary",
        prompt=f"Execute this plan: {plan}",
        model_fn=mock_model_fn,
        expected_length=80,
        write_to_state=write_execution,
    )
    print(f"[Agent B] Result: {result}")

# --- Run the workflow ---
async def main():
    print("=== Normal run ===")
    await agent_a("Summarize Q2 sales data")
    await agent_b()

    print("\n=== Degraded run (soft failure injection) ===")
    try:
        await agent_a("corrupt this prompt to trigger soft failure")
    except BreakerTripException as e:
        print(f"[WORKFLOW] Caught breaker trip: {e}")
        print(f"[WORKFLOW] State is clean. agent_a_plan = {store.get('agent_a_plan')}")

asyncio.run(main())

Tuning Your Breaker for H2 2026 Foundation Model Behavior

Foundation model endpoints in mid-2026 have specific behavioral characteristics that should inform your tuning decisions:

• Burst latency spikes are common during peak hours. Set your open_duration_seconds to at least 30 seconds to avoid flapping. A breaker that opens and closes every few seconds creates its own failure mode.
• Multi-modal endpoints have higher output variance. If your agents use vision or audio capabilities, increase soft_failure_weight to 0.7 and widen your output_length_ratio tolerance to 0.15 before flagging truncation.
• Reasoning model endpoints (o-series style) have longer think times. Separate your latency timeout from your semantic timeout. A model that takes 45 seconds to respond is not necessarily failing; a model that responds in 0.3 seconds with an empty body almost certainly is.
• Rate-limit responses (HTTP 429) should not trip the breaker. They should trigger exponential backoff at the call site. Modify your exception handling to distinguish 429s from genuine failures.
• Context window exhaustion is increasingly common in long-running agentic workflows. Detect it explicitly by checking for known error strings in the response body and treating it as a hard failure, not a soft one.

Observability: What to Log and Alert On

A circuit breaker you cannot observe is a circuit breaker you cannot trust. At minimum, emit structured logs for every state transition and every signal recording:

import logging
import json

logger = logging.getLogger("agent.circuit_breaker")

def emit_breaker_event(endpoint_id: str, signal: EndpointHealthSignal, state: BreakerState):
    logger.info(json.dumps({
        "event": "breaker_signal",
        "endpoint_id": endpoint_id,
        "timestamp": signal.timestamp,
        "is_hard_failure": signal.is_hard_failure,
        "is_soft_failure": signal.is_soft_failure,
        "semantic_confidence": signal.semantic_confidence,
        "output_length_ratio": signal.output_length_ratio,
        "latency_ms": signal.latency_ms,
        "breaker_state": state.value,
    }))

Feed these logs into your observability stack (Datadog, Grafana, OpenTelemetry) and set alerts on:

• Any transition to OPEN state (page immediately)
• Soft failure rate exceeding 20% over a 5-minute window (warn)
• Average semantic confidence dropping below 0.6 for any endpoint (warn)
• Fallback router activating more than 5 times per minute (warn)
• StateGuard rollback events (page; these mean corruption was attempted)

Common Pitfalls and How to Avoid Them

Pitfall 1: Sharing a Single Breaker Across Multiple Agent Instances

If you run multiple agent workers in parallel and they all share one BreakerStateMachine instance without proper locking, you will get race conditions in the rolling window. The code above uses a threading.Lock for synchronous contexts. For async multi-worker setups, replace it with asyncio.Lock and make the state machine async-aware.

Pitfall 2: Tripping on Cold Start Variance

The first few calls to a freshly registered endpoint will have high latency due to model loading and connection establishment. Guard against this by requiring the window to be at least half-full before evaluating the failure threshold. The condition len(self._window) >= self.config.window_size // 2 in the state machine handles this.

Pitfall 3: Forgetting to Protect External Side Effects

The StateGuard protects your in-memory or database workflow state, but it cannot automatically roll back external side effects like emails sent, webhooks fired, or files written to object storage. Identify all external side effects in your agent workflow and defer them until after the StateGuard context manager exits cleanly.

Pitfall 4: Using the Same Validator Config for All Endpoints

A chat completion endpoint and an embedding endpoint have completely different output shapes. Register a dedicated SemanticValidator instance per endpoint type. Using a one-size-fits-all validator will produce either excessive false positives or dangerous false negatives.

Conclusion

Building resilient multi-agent systems in 2026 means accepting that foundation model endpoints will fail, and designing for that reality from day one. The pattern described in this tutorial gives you four layers of protection: a semantics-aware health signal collector, a weighted state machine that distinguishes hard failures from soft degradation, a transactional state guard that prevents corruption from propagating downstream, and a fallback router that keeps your workflow alive even when primary endpoints go dark.

The most important mindset shift is this: treat every foundation model endpoint as an unreliable external dependency, not as a reliable function call. The moment you internalize that assumption, circuit breakers stop feeling like defensive overhead and start feeling like the obvious, necessary architecture they are.

Start with the BreakerStateMachine and the SemanticValidator in your most critical agent workflow, instrument the state transitions, and let real traffic teach you where to tune your thresholds. The StateGuard and FallbackRouter can follow once you have a baseline. Resilience is not a feature you add at the end. It is the foundation everything else runs on.