↓ Skip to main content
Ishan

Agent Circuit Breaker

·313 words·2 mins

Tags

Circuit-Breaker Agents Llm Python Resilience

Technologies

Python asyncio LLM agents
View on GitHub

Agent Circuit Breaker
#

Circuit breaker pattern for agent/LLM calls: monitors failures and temporarily disables expensive operations when a threshold is exceeded.

  • Lightweight — zero dependencies for the core library.
  • Sync and async — decorator, context manager, and call() / call_async().
  • Configurable — consecutive or sliding-window failure counting, custom predicate, fallback, excluded exceptions.

Install
#

pip install -e .

Or add to your project and use the package agent_circuit_breaker.

Quick start
#

Decorator
#

from agent_circuit_breaker import circuit_breaker

@circuit_breaker(failure_threshold=5, recovery_timeout=60)
def call_llm(prompt: str) -> str:
    # your LLM/agent call
    return response

@circuit_breaker(failure_threshold=3, recovery_timeout=30)
async def async_call_llm(prompt: str) -> str:
    return await some_async_client(prompt)

Context manager
#

from agent_circuit_breaker import CircuitBreaker

breaker = CircuitBreaker(failure_threshold=5, recovery_timeout=60)

with breaker:
    result = agent.run(task)

# async
async with breaker:
    result = await agent.run_async(task)

Class-based
#

from agent_circuit_breaker import CircuitBreaker, CircuitBreakerOpenError

breaker = CircuitBreaker(failure_threshold=5, recovery_timeout=60)

try:
    result = breaker.call(agent_function, arg1, arg2)
except CircuitBreakerOpenError:
    result = "Service unavailable"

Configuration
#

ParameterDefaultDescription
failure_threshold5Number of failures that open the circuit.
recovery_timeout60Seconds the circuit stays open before a trial (half-open).
failure_windowNoneIf set, use a sliding time window (seconds) instead of consecutive failures.
failure_predicateNoneCallable (exc) -> bool: only count exception as failure when it returns True.
fallbackNoneCallable to run when the circuit is open instead of raising.
excluded_exceptionsNoneTuple of exception types that never count as failures (still re-raised).

Exception handling order: first check excluded_exceptions; if the exception is in that tuple, do not count it. Otherwise use failure_predicate if set, else treat as failure.

States
#

  • CLOSED — Calls allowed; failures are counted.
  • OPEN — Calls blocked; CircuitBreakerOpenError (or fallback) until recovery_timeout has passed.
  • HALF_OPEN — One trial call allowed; success closes the circuit, failure reopens it.

Monitoring
#

Set callbacks on the breaker:

  • on_state_change(old_state, new_state)
  • on_failure() / on_success()
  • on_open() / on_close() / on_half_open()

Requirements
#

  • Python 3.9+

License
#

MIT