AgentCircuit

Stop burning money on broken AI agents.

AgentCircuit is a circuit breaker and runtime safety layer for AI agents — loop detection, auto-repair, output validation, and budget control in a single decorator.

Two problems kept breaking my agents:

1. Loops that silently drained my API budget — $200+ before I noticed
2. Bad LLM outputs that crashed downstream code — no validation, no recovery

AgentCircuit fixes both with one decorator.

pip install agentcircuit

AgentCircuit wraps your AI agent functions with invisible safety nets:

• Fuse - Detects infinite loops and kills them before they drain your wallet
• Medic - Catches exceptions and auto-repairs outputs using an LLM
• Sentinel - Validates every output against your Pydantic schema
• Budget - Dollar and time circuit breakers that stop runaway costs
• Pricing - Accurate cost tracking with built-in pricing for 25+ models

Zero config. No server. No database. Just a decorator.

Works with LangGraph, LangChain, CrewAI, and AutoGen.

---

Before / After

Before — your agent crashes, loops forever, or returns garbage:

def extract_data(state):
    result = call_llm(state["text"])
    return json.loads(result)  # crashes on malformed JSON

After - one line change, your agent self-heals:

from agentcircuit import reliable
from pydantic import BaseModel

class Output(BaseModel):
    name: str
    age: int

@reliable(sentinel_schema=Output)
def extract_data(state):
    result = call_llm(state["text"])
    return json.loads(result)  # now validated against Output schema

What happens behind the scenes:

1. Fuse checks if this node is stuck in a loop (same input seen 3+ times)
2. Your function runs normally
3. Sentinel validates the output against Output schema
4. If validation fails, error is raised (add llm_callable to enable auto-repair)

---

Quick Start

Minimal (no LLM, just validation + loop detection)

from agentcircuit import reliable
from pydantic import BaseModel

class SearchResult(BaseModel):
    query: str
    results: list[str]

@reliable(sentinel_schema=SearchResult, fuse_limit=5)
def search_node(state):
    return {"query": state["q"], "results": ["result1", "result2"]}

With Auto-Repair (add an LLM for self-healing)

pip install agentcircuit[groq]  # or agentcircuit[openai] or agentcircuit[anthropic]

import os
from groq import Groq

client = Groq(api_key=os.environ["GROQ_API_KEY"])

def my_llm(prompt: str) -> str:
    return client.chat.completions.create(
        messages=[{"role": "user", "content": prompt}],
        model="llama-3.3-70b-versatile"
    ).choices[0].message.content

@reliable(sentinel_schema=SearchResult, llm_callable=my_llm)
def search_node(state):
    return {"query": state["q"], "results": ["result1", "result2"]}

Now if search_node throws an exception or returns invalid data, Medic will use your LLM to generate a valid output matching the schema.

---

Cost Saving

AgentCircuit provides three layers of cost protection to prevent runaway agent loops from draining your wallet.

Per-Node Dollar Limit

Stop a single node from spending too much:

from agentcircuit import reliable

@reliable(max_cost_usd=2.0, model="gpt-4o")
def expensive_node(state):
    return call_llm(state["prompt"])
# Raises BudgetExceededError if this node's cumulative cost exceeds $2.00

Per-Node Time Limit

Kill long-running nodes before they hang your pipeline:

@reliable(max_seconds=30)
def slow_node(state):
    return call_external_api(state["query"])
# Raises TimeoutExceededError if execution exceeds 30 seconds

Global Budget (Shared Across Nodes)

Set one budget for your entire agent graph — all nodes share it:

from agentcircuit import reliable, GlobalBudget

budget = GlobalBudget(max_cost_usd=10.0, max_seconds=120)

@reliable(budget=budget, model="gpt-4o")
def node_a(state):
    return call_llm_a(state)

@reliable(budget=budget, model="gpt-4o-mini")
def node_b(state):
    return call_llm_b(state)

# Run your pipeline...
result_a = node_a({"prompt": "analyze this"})
result_b = node_b({"prompt": "summarize that"})

# Check spend after execution
print(f"Total spent: ${budget.total_spent:.4f}")
print(f"Remaining:   ${budget.remaining:.4f}")
print(f"Elapsed:     {budget.elapsed_seconds:.1f}s")

# Reset for next run
budget.reset()

If the combined cost of node_a + node_b exceeds $10.00 or 120 seconds, the next node call raises an error immediately — before making another LLM call.

Catching Budget Errors

from agentcircuit import reliable, GlobalBudget, BudgetExceededError, TimeoutExceededError

budget = GlobalBudget(max_cost_usd=5.0, max_seconds=60)

@reliable(budget=budget, model="gpt-4o")
def my_node(state):
    return call_llm(state)

try:
    for task in tasks:
        my_node(task)
except BudgetExceededError as e:
    print(f"Budget hit: spent ${e.spent:.2f} of ${e.limit:.2f} limit")
except TimeoutExceededError as e:
    print(f"Timeout hit: {e.elapsed:.1f}s of {e.limit:.1f}s limit")

---

Pricing & Cost Tracking

AgentCircuit automatically tracks the cost of every node execution. It uses a priority chain to get the most accurate cost:

Priority	Source	Accuracy
1	Actual API token usage (from AgentCircuit providers)	Exact
2	User-provided cost_per_token	Explicit
3	Built-in model pricing table lookup	Good estimate
4	Default $5/1M tokens flat rate	Rough fallback

Specify Your Model (recommended)

@reliable(model="gpt-4o")
def my_node(state):
    return call_openai(state)
# Cost is calculated using GPT-4o's actual pricing:
# $2.50/1M input tokens, $10.00/1M output tokens

Custom Cost Per Token

Override with your own rate (e.g., for fine-tuned models or enterprise pricing):

@reliable(cost_per_token=0.00003)  # $30/1M tokens
def my_node(state):
    return call_custom_model(state)

Supported Models

Built-in pricing for 40+ models across major providers:

Provider	Models
OpenAI	gpt-5, gpt-4.5-preview, o3, o3-pro, o3-mini, o1, o1-mini, gpt-4o, gpt-4o-mini, gpt-4-turbo, gpt-4, gpt-3.5-turbo
Anthropic	claude-opus-4.5, claude-sonnet-4.5, claude-haiku-4.5, claude-opus-4.1, claude-opus-4, claude-sonnet-4, claude-3-5-sonnet, claude-3-5-haiku, claude-3-opus, claude-3-haiku
Google	gemini-3-pro, gemini-3-flash, gemini-2.5-pro, gemini-2.5-flash, gemini-2.5-flash-lite, gemini-2.0-flash, gemini-1.5-pro, gemini-1.5-flash
Groq	llama-3.3-70b, llama-3.1-8b, mixtral-8x7b, llama-3.1-70b, gemma2-9b

Model names are matched flexibly — "gpt-4o", "gpt-4o-2024-08-06", and "GPT-4o" all resolve to the same pricing.

Using CostCalculator Directly

from agentcircuit import CostCalculator, get_model_pricing

# Check a model's pricing
pricing = get_model_pricing("gpt-4o")
print(f"Input:  ${pricing.input_per_token * 1_000_000:.2f}/1M tokens")
print(f"Output: ${pricing.output_per_token * 1_000_000:.2f}/1M tokens")

# Calculate cost for known token counts
calc = CostCalculator(model="gpt-4o")
cost = calc.calculate(input_tokens=1000, output_tokens=500)
print(f"Cost: ${cost:.6f}")

---

With LangGraph

from langgraph.graph import StateGraph
from agentcircuit import reliable, GlobalBudget
from pydantic import BaseModel

class AgentState(BaseModel):
    messages: list[str]
    result: str = ""

budget = GlobalBudget(max_cost_usd=5.0, max_seconds=120)

@reliable(sentinel_schema=AgentState, fuse_limit=3, budget=budget, model="gpt-4o")
def process_node(state):
    # Your agent logic
    return {"messages": state["messages"], "result": "done"}

graph = StateGraph(AgentState)
graph.add_node("process", process_node)

With LangChain / CrewAI / AutoGen

from agentcircuit import get_adapter

# LangChain
adapter = get_adapter("langchain", fuse_limit=5)
wrapped_chain = adapter.wrap_chain(my_chain, schema=OutputSchema)

# CrewAI
adapter = get_adapter("crewai")
wrapped_agent = adapter.wrap_agent(my_agent)

# AutoGen
adapter = get_adapter("autogen")
wrapped_function = adapter.wrap_function(my_tool)

---

How It Works

Your Function
     |
     v
  [ Budget ] -- Over budget? --> STOP (BudgetExceededError)
     |
     v
  [ Fuse ] -- Loop detected? --> STOP (LoopError)
     |
     v
  Run your function
     |
     v
  [ Pricing ] -- Track cost (model-aware)
     |
     v
  [ Sentinel ] -- Output valid? --> Return result
     |
     v (invalid)
  [ Medic ] -- LLM fixes output --> [ Sentinel ] validates again
     |
     v (still fails after 2 attempts)
  Raise original error (no silent failures)

Storage

By default, AgentCircuit stores traces in memory (lost when process exits). For persistence:

from agentcircuit import reliable, set_default_storage
from agentcircuit.storage import Storage  # SQLite backend

set_default_storage(Storage())  # Now traces persist to .agentcircuit/traces.db

@reliable()
def my_node(state):
    return state

Or pass storage per-decorator:

from agentcircuit.storage import Storage

db = Storage(db_path="my_traces.db")

@reliable(storage=db)
def my_node(state):
    return state

Installation Options

pip install agentcircuit                 # Core only (pydantic). Validation + loop detection.
pip install agentcircuit[groq]           # + Groq for auto-repair
pip install agentcircuit[openai]         # + OpenAI for auto-repair
pip install agentcircuit[anthropic]      # + Anthropic for auto-repair
pip install agentcircuit[langchain]      # + LangChain adapter
pip install agentcircuit[crewai]         # + CrewAI adapter
pip install agentcircuit[llm]            # All LLM providers
pip install agentcircuit[all]            # Everything

API Reference

@reliable() / @reliable_node()

Parameter	Type	Default	Description
sentinel_schema	BaseModel	None	Pydantic model for output validation
llm_callable	Callable[[str], str]	None	LLM function for auto-repair
fuse_limit	int	3	Max identical states before loop detection
node_name	str	None	Override function name for traces
storage	BaseStorage	InMemoryStorage	Storage backend for traces
medic_repair	Callable	None	Custom repair callback (legacy)
max_cost_usd	float	None	Per-node dollar budget limit
max_seconds	float	None	Per-node execution time limit (seconds)
budget	GlobalBudget	None	Shared budget across multiple nodes
model	str	None	Model name for pricing table lookup
cost_per_token	float	None	Custom cost per token override (USD)

Core Components

from agentcircuit import Fuse, Medic, Sentinel  # Use individually if needed

• Fuse(limit=3) - Loop detection via state hashing
• Medic(llm_callable=...) - LLM-based error recovery
• Sentinel(schema=...) - Pydantic schema validation

Budget Components

from agentcircuit import BudgetFuse, TimeoutFuse, GlobalBudget

• BudgetFuse(max_cost_usd=1.0) - Dollar-based circuit breaker
• TimeoutFuse(max_seconds=30) - Time-based circuit breaker
• GlobalBudget(max_cost_usd=10.0, max_seconds=120) - Thread-safe shared budget

Pricing Components

from agentcircuit import CostCalculator, get_model_pricing, MODEL_PRICING

• CostCalculator(model="gpt-4o") - Calculate costs using model pricing
• CostCalculator(cost_per_token=0.00003) - Calculate costs using custom rate
• get_model_pricing("gpt-4o") - Look up a model's pricing
• MODEL_PRICING - Full pricing table dict

Error Types

from agentcircuit import BudgetExceededError, TimeoutExceededError, LoopError

Error	Raised When	Attributes
BudgetExceededError	Dollar limit exceeded	spent, limit
TimeoutExceededError	Time limit exceeded	elapsed, limit
LoopError	Infinite loop detected	—

Running Tests

pip install agentcircuit[dev]
pytest

License

MIT