### Installation Commands

Source: https://github.com/simranmultani197/agentcircuit/blob/main/README.md

Install the core package or specific extras for LLM providers and framework adapters.

```bash
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
```

--------------------------------

### Install AgentCircuit

Source: https://github.com/simranmultani197/agentcircuit/blob/main/README.md

Install the core package via pip.

```bash
pip install agentcircuit
```

--------------------------------

### Run Tests

Source: https://github.com/simranmultani197/agentcircuit/blob/main/README.md

Install development dependencies and execute the test suite.

```bash
pip install agentcircuit[dev]
pytest
```

--------------------------------

### Installation Options

Source: https://github.com/simranmultani197/agentcircuit/blob/main/README.md

Various methods to install the AgentCircuit library, including core functionality and optional dependencies for different LLM providers and adapters.

```APIDOC
## Installation Options

```bash
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
```
```

--------------------------------

### Configure and Use Storage Backends

Source: https://context7.com/simranmultani197/agentcircuit/llms.txt

Demonstrates setting up and using different storage backends for traces, including in-memory, SQLite, and PostgreSQL. Shows how to query traces, get statistics, and prune old data.

```python
from agentcircuit import (
    reliable,
    set_default_storage,
    get_default_storage,
    InMemoryStorage,
    Storage
)
from agentcircuit.storage import create_storage, PostgresStorage

# Default: In-memory storage (lost when process exits)
storage = get_default_storage()  # Returns InMemoryStorage

# Switch to SQLite for persistence
sqlite_storage = Storage(db_path=".agentcircuit/traces.db")
set_default_storage(sqlite_storage)

# All @reliable decorators now use SQLite storage
@reliable(fuse_limit=3)
def my_node(state):
    return {"result": process(state)}

# Per-decorator storage override
custom_db = Storage(db_path="custom_traces.db")

@reliable(storage=custom_db)
def isolated_node(state):
    return {"isolated": True}

# Query stored traces
traces = sqlite_storage.get_traces(limit=10, status="repaired")
for trace in traces:
    print(f"Node: {trace['node_id']}, Status: {trace['status']}")

# Get statistics
stats = sqlite_storage.get_stats()
print(f"Total traces: {stats['total_traces']}")
print(f"Total cost: ${stats['total_cost']:.4f}")
print(f"Total saved: ${stats['total_saved']:.4f}")

# PostgreSQL for teams (requires psycopg2-binary)
pg_storage = create_storage(
    "postgresql",
    connection_string="postgresql://user:pass@localhost/agentcircuit"
)
set_default_storage(pg_storage)

# Prune old traces
deleted = sqlite_storage.prune_old_traces(days=30)
print(f"Deleted {deleted} old traces")
```

--------------------------------

### Install Auto-Repair Dependencies

Source: https://github.com/simranmultani197/agentcircuit/blob/main/README.md

Install optional dependencies to enable LLM-based auto-repair.

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

--------------------------------

### Fuse - Loop Detection Example

Source: https://context7.com/simranmultani197/agentcircuit/llms.txt

Instantiate Fuse with a limit to detect loops. The check method takes a history of state hashes and the current state to identify repeated states.

```python
from agentcircuit import Fuse, LoopError

# Create a fuse that trips after 3 identical states
fuse = Fuse(limit=3)

# History of previous state hashes
history = []
current_state = {"step": 1, "data": "same input"}

# Check if we're in a loop
try:
    fuse.check(
        history=[fuse._hash_state(s) for s in history],
        current_state=current_state
    )
    print("No loop detected, safe to proceed")
except LoopError as e:
    print(f"Loop detected: {e}")
    # Handle the loop - maybe break out or take corrective action

# Simulating a loop scenario
same_state = {"action": "retry", "count": 5}
history = [fuse._hash_state(same_state) for _ in range(3)]

try:
    fuse.check(history=history, current_state=same_state)
except LoopError as e:
    print(f"Caught: {e}")  # "Fuse Tripped: Loop detected. State repeated 3 times."
```

--------------------------------

### Sentinel - Output Validation Example

Source: https://context7.com/simranmultani197/agentcircuit/llms.txt

Use Sentinel to validate outputs against Pydantic schemas. It raises SentinelError for invalid outputs and returns a Pydantic model instance for valid ones. A schema of None allows any data.

```python
from agentcircuit import Sentinel, SentinelError
from pydantic import BaseModel, Field

class AgentResponse(BaseModel):
    answer: str = Field(min_length=1)
    confidence: float = Field(ge=0.0, le=1.0)
    sources: list[str] = Field(default_factory=list)

sentinel = Sentinel(schema=AgentResponse)

# Valid output - passes validation
valid_output = {
    "answer": "The capital of France is Paris",
    "confidence": 0.95,
    "sources": ["wikipedia.org", "britannica.com"]
}

validated = sentinel.validate(valid_output)
print(f"Valid: {validated}")  # Returns AgentResponse instance

# Invalid output - raises SentinelError
invalid_output = {
    "answer": "",  # Too short
    "confidence": 1.5,  # Out of range
}

try:
    sentinel.validate(invalid_output)
except SentinelError as e:
    print(f"Validation failed: {e}")
    # "Sentinel Alert: Output validation failed: ..."

# No schema - passes through unchanged
no_schema_sentinel = Sentinel(schema=None)
result = no_schema_sentinel.validate({"any": "data"})
print(result)  # {"any": "data"}
```

--------------------------------

### Custom LLM Repair Strategy

Source: https://context7.com/simranmultani197/agentcircuit/llms.txt

Implement a custom LLM repair strategy with a specific prompt to guide the LLM in fixing agent outputs. This is useful for complex or domain-specific error correction.

```python
# Custom LLM repair with custom prompt
custom_llm_repair = LLMRepairStrategy(
    custom_prompt="""Fix this output for an AI agent.

ERROR: {error}
INPUT: {input}
FAILED OUTPUT: {output}
SCHEMA: {schema}
HINT: {hint}

Return only valid JSON."""
)
```

--------------------------------

### Running Tests

Source: https://github.com/simranmultani197/agentcircuit/blob/main/README.md

Instructions on how to set up the development environment and run tests for the AgentCircuit library.

```APIDOC
## Running Tests

```bash
pip install agentcircuit[dev]
pytest
```
```

--------------------------------

### Integration with LangChain, CrewAI, and AutoGen

Source: https://github.com/simranmultani197/agentcircuit/blob/main/README.md

Demonstrates how to wrap external framework components using AgentCircuit adapters.

```python
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)
```

--------------------------------

### List Available Models and Pricing

Source: https://context7.com/simranmultani197/agentcircuit/llms.txt

Prints a sorted list of supported models and their pricing per million tokens for input and output. Requires MODEL_PRICING to be defined.

```python
print("Supported models:")
for model in sorted(MODEL_PRICING.keys()):
    p = MODEL_PRICING[model]
    print(f"  {model}: ${p.input_per_token * 1_000_000:.2f}/${p.output_per_token * 1_000_000:.2f} per 1M tokens")
```

--------------------------------

### Import Budget Components

Source: https://github.com/simranmultani197/agentcircuit/blob/main/README.md

Import budget and timeout management utilities for circuit breaking.

```python
from agentcircuit import BudgetFuse, TimeoutFuse, GlobalBudget
```

--------------------------------

### LangGraph Integration with AgentCircuit

Source: https://context7.com/simranmultani197/agentcircuit/llms.txt

Demonstrates integrating AgentCircuit's reliable decorators with LangGraph state graphs. Shows how to define agent states, budgets, and nodes, then compile and run the graph.

```python
from langgraph.graph import StateGraph, END
from agentcircuit import reliable, GlobalBudget
from pydantic import BaseModel
from typing import TypedDict, Annotated
import operator

class AgentState(TypedDict):
    messages: Annotated[list[str], operator.add]
    result: str

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

class ProcessOutput(BaseModel):
    messages: list[str]
    result: str

@reliable(
    sentinel_schema=ProcessOutput,
    fuse_limit=3,
    budget=budget,
    model="gpt-4o"
)
def process_node(state: AgentState) -> dict:
    # Your processing logic
    response = call_llm(state["messages"][-1])
    return {
        "messages": [response],
        "result": response
    }

@reliable(
    sentinel_schema=ProcessOutput,
    fuse_limit=3,
    budget=budget,
    model="gpt-4o-mini"
)
def summarize_node(state: AgentState) -> dict:
    summary = summarize_messages(state["messages"])
    return {
        "messages": [summary],
        "result": summary
    }

def should_continue(state: AgentState) -> str:
    if len(state["messages"]) > 5:
        return "summarize"
    return "process"

# Build the graph
graph = StateGraph(AgentState)
graph.add_node("process", process_node)
graph.add_node("summarize", summarize_node)
graph.add_conditional_edges("process", should_continue, {
    "process": "process",
    "summarize": "summarize"
})
graph.add_edge("summarize", END)
graph.set_entry_point("process")

# Compile and run
app = graph.compile()
result = app.invoke({"messages": ["Hello, start processing"], "result": ""})

print(f"Final result: {result['result']}")
print(f"Total spent: ${budget.total_spent:.4f}")
```

--------------------------------

### Catching Budget and Timeout Errors

Source: https://github.com/simranmultani197/agentcircuit/blob/main/README.md

Demonstrates how to handle budget and timeout exceptions during node execution.

```python
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")
```

--------------------------------

### Integrate with LangChain, CrewAI, and AutoGen

Source: https://context7.com/simranmultani197/agentcircuit/llms.txt

Shows how to use AgentCircuit adapters to integrate with LangChain, CrewAI, and AutoGen. Includes wrapping LangChain chains and tools, and agents/functions for CrewAI and AutoGen.

```python
from agentcircuit import get_adapter
from pydantic import BaseModel

class OutputSchema(BaseModel):
    result: str
    confidence: float

# LangChain Integration
langchain_adapter = get_adapter("langchain", fuse_limit=5)

# Wrap a LangChain chain
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI

prompt = ChatPromptTemplate.from_template("Summarize: {text}")
llm = ChatOpenAI(model="gpt-4o-mini")
chain = prompt | llm

wrapped_chain = langchain_adapter.wrap_chain(chain, schema=OutputSchema)
result = wrapped_chain.invoke({"text": "Long document..."})

# Wrap a LangChain tool
from langchain_core.tools import tool

@tool
def search_tool(query: str) -> str:
    """Search for information."""
    return f"Results for: {query}"

wrapped_tool = langchain_adapter.wrap_tool(search_tool)

# CrewAI Integration
crewai_adapter = get_adapter("crewai", fuse_limit=3)
wrapped_agent = crewai_adapter.wrap_agent(my_crew_agent)

# AutoGen Integration
autogen_adapter = get_adapter("autogen")
wrapped_function = autogen_adapter.wrap_function(my_autogen_tool)
```

--------------------------------

### Persistence Configuration

Source: https://github.com/simranmultani197/agentcircuit/blob/main/README.md

Configures persistent storage for traces using SQLite or custom storage backends.

```python
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
```

```python
from agentcircuit.storage import Storage

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

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

--------------------------------

### Import Core Components

Source: https://github.com/simranmultani197/agentcircuit/blob/main/README.md

Import individual components for loop detection, error recovery, and schema validation.

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

--------------------------------

### Import Pricing Components

Source: https://github.com/simranmultani197/agentcircuit/blob/main/README.md

Import utilities for calculating and retrieving model cost information.

```python
from agentcircuit import CostCalculator, get_model_pricing, MODEL_PRICING
```

--------------------------------

### Using CostCalculator Directly

Source: https://github.com/simranmultani197/agentcircuit/blob/main/README.md

Provides utilities to check model pricing and calculate costs for specific token counts.

```python
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}")
```

--------------------------------

### Set Per-Node Budget Limits

Source: https://github.com/simranmultani197/agentcircuit/blob/main/README.md

Apply cost and time constraints to individual agent nodes.

```python
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
```

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

--------------------------------

### Manage Global Budget Across Nodes

Source: https://github.com/simranmultani197/agentcircuit/blob/main/README.md

Share a single budget instance across multiple agent nodes to track total pipeline costs.

```python
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")
```

--------------------------------

### Configure Retry Strategy Options

Source: https://context7.com/simranmultani197/agentcircuit/llms.txt

Define detailed retry configurations, including backoff strategies, delays, and jitter, to handle transient errors effectively. This allows for robust retries with controlled backoff.

```python
# Retry configuration options
 aggressive_retry = RetryConfig(
    max_attempts=5,
    strategy=RetryStrategy.JITTERED_BACKOFF,
    base_delay=0.5,
    max_delay=60.0,
    exponential_base=2.0,
    jitter_factor=0.2
)

# Calculate delay for attempt
delay = aggressive_retry.get_delay(attempt=3)
print(f"Delay for attempt 3: {delay:.2f}s")
```

--------------------------------

### License

Source: https://github.com/simranmultani197/agentcircuit/blob/main/README.md

Information about the license under which the AgentCircuit library is distributed.

```APIDOC
## License

MIT
```

--------------------------------

### Budget Components

Source: https://github.com/simranmultani197/agentcircuit/blob/main/README.md

Details on budget-related components: BudgetFuse for dollar-based circuit breaking, TimeoutFuse for time-based circuit breaking, and GlobalBudget for shared budget management.

```APIDOC
### Budget Components

```python
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
```

--------------------------------

### Integration with LangGraph

Source: https://github.com/simranmultani197/agentcircuit/blob/main/README.md

Shows how to use the reliable decorator within a LangGraph StateGraph.

```python
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)
```

--------------------------------

### Per-Node Budget and Timeout Limits

Source: https://context7.com/simranmultani197/agentcircuit/llms.txt

Implement circuit breakers for individual node cost and time limits using `reliable` decorator arguments `max_cost_usd` and `max_seconds`. Handle `BudgetExceededError` and `TimeoutExceededError` for fallback logic.

```python
from agentcircuit import reliable, BudgetExceededError, TimeoutExceededError

# Per-node dollar limit
@reliable(max_cost_usd=2.0, model="gpt-4o")
def expensive_analysis(state):
    """This node cannot spend more than $2.00"""
    return call_expensive_llm(state["prompt"])

# Per-node time limit
@reliable(max_seconds=30)
def time_sensitive_node(state):
    """This node must complete within 30 seconds"""
    return call_external_api(state["query"])

# Combined limits
@reliable(max_cost_usd=1.0, max_seconds=15, model="gpt-4o-mini")
def constrained_node(state):
    """Limited by both cost and time"""
    return process_data(state)
```

```python
# Handling budget errors
try:
    result = expensive_analysis({"prompt": "Very long analysis request..."})
except BudgetExceededError as e:
    print(f"Cost limit hit: ${e.spent:.4f} >= ${e.limit:.4f}")
    # Implement fallback logic

try:
    result = time_sensitive_node({"query": "slow endpoint"})
except TimeoutExceededError as e:
    print(f"Time limit hit: {e.elapsed:.1f}s >= {e.limit:.1f}s")
    # Implement timeout handling
```

--------------------------------

### Pricing Components

Source: https://github.com/simranmultani197/agentcircuit/blob/main/README.md

Information on pricing components, including CostCalculator for cost estimation, get_model_pricing for retrieving model costs, and MODEL_PRICING for the full pricing table.

```APIDOC
### Pricing Components

```python
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
```

--------------------------------

### Build Custom Strategy Chains for Error Repair

Source: https://context7.com/simranmultani197/agentcircuit/llms.txt

Construct custom StrategyChain instances to define a sequence of repair strategies for different error types. This allows fine-grained control over error recovery.

```python
from agentcircuit.strategies import (
    StrategyChain,
    JSONRepairStrategy,
    SchemaRepairStrategy,
    RetryWithBackoffStrategy,
    LLMRepairStrategy,
    TruncateContextStrategy,
    FailFastStrategy,
    RetryConfig,
    RetryStrategy,
    create_default_chain
)
from pydantic import BaseModel

class OutputSchema(BaseModel):
    data: str
    score: float

# Use the default strategy chain
default_chain = create_default_chain()

# Or build a custom chain
custom_chain = StrategyChain(
    strategies=[
        JSONRepairStrategy(),       # Fix malformed JSON
        SchemaRepairStrategy(),     # Fix schema mismatches
        RetryWithBackoffStrategy(   # Retry transient errors
            config=RetryConfig(
                max_attempts=3,
                strategy=RetryStrategy.EXPONENTIAL_BACKOFF,
                base_delay=1.0,
                max_delay=30.0
            )
        ),
        TruncateContextStrategy(max_chars=8000),  # Handle context length
        LLMRepairStrategy(),        # Generic LLM-based repair
    ]
)
```

--------------------------------

### Global Budget for Shared Cost Control

Source: https://context7.com/simranmultani197/agentcircuit/llms.txt

Enforce total cost and time limits across multiple decorated functions using `GlobalBudget`. Initialize with `max_cost_usd` and `max_seconds`. Reset the budget using `budget.reset()` after execution.

```python
from agentcircuit import reliable, GlobalBudget, BudgetExceededError, TimeoutExceededError

# Create a shared budget for an entire agent graph
budget = GlobalBudget(max_cost_usd=10.0, max_seconds=120)

@reliable(budget=budget, model="gpt-4o")
def analyze_node(state):
    return {"analysis": analyze_with_llm(state["data"])}

@reliable(budget=budget, model="gpt-4o-mini")
def summarize_node(state):
    return {"summary": summarize_with_llm(state["analysis"])}

@reliable(budget=budget, model="gpt-4o")
def generate_node(state):
    return {"output": generate_with_llm(state["summary"])}
```

```python
# Run your pipeline with budget protection
try:
    tasks = [{"data": f"task_{i}"} for i in range(100)]

    for task in tasks:
        result1 = analyze_node(task)
        result2 = summarize_node(result1)
        result3 = generate_node(result2)

        # Check remaining budget
        print(f"Spent: ${budget.total_spent:.4f}, Remaining: ${budget.remaining:.4f}")
        print(f"Elapsed: {budget.elapsed_seconds:.1f}s")

except BudgetExceededError as e:
    print(f"Budget exceeded! Spent ${e.spent:.2f} of ${e.limit:.2f} limit")

except TimeoutExceededError as e:
    print(f"Timeout! {e.elapsed:.1f}s of {e.limit:.1f}s limit")

finally:
    # Reset for next run
    budget.reset()
    print("Budget reset for next execution")
```

--------------------------------

### Reset Budget

Source: https://github.com/simranmultani197/agentcircuit/blob/main/README.md

Resets the budget for the next execution run.

```python
budget.reset()
```

--------------------------------

### Specify Model for Cost Tracking

Source: https://github.com/simranmultani197/agentcircuit/blob/main/README.md

Configures the model to enable automatic cost calculation based on built-in pricing tables.

```python
@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
```

--------------------------------

### Medic - Auto-Repair Initialization

Source: https://context7.com/simranmultani197/agentcircuit/llms.txt

Initialize Medic with an LLM callable, maximum recovery attempts, and cost tracking enabled. It attempts to repair failed outputs using LLM-based recovery.

```python
from agentcircuit import Medic, MedicError
from pydantic import BaseModel

class ExtractedData(BaseModel):
    name: str
    email: str
    age: int

def my_llm(prompt: str) -> str:
    # Your LLM implementation
    return '{"name": "John", "email": "john@example.com", "age": 30}'

# Initialize Medic with LLM callable
medic = Medic(
    llm_callable=my_llm,
    max_recovery_attempts=2,
    track_costs=True
)
```

--------------------------------

### Cost Calculation and Model Pricing

Source: https://context7.com/simranmultani197/agentcircuit/llms.txt

Utilize `CostCalculator` for accurate cost tracking and `get_model_pricing` to retrieve pricing for various models. Supports custom pricing and estimation from arbitrary objects.

```python
from agentcircuit import CostCalculator, get_model_pricing, MODEL_PRICING

# Check model pricing
pricing = get_model_pricing("gpt-4o")
print(f"GPT-4o Input:  ${pricing.input_per_token * 1_000_000:.2f}/1M tokens")
print(f"GPT-4o 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 for 1000 input + 500 output tokens: ${cost:.6f}")
```

```python
# Use custom pricing for fine-tuned or enterprise models
custom_calc = CostCalculator(cost_per_token=0.00003)  # $30/1M tokens
custom_cost = custom_calc.calculate(input_tokens=5000, output_tokens=2000)
print(f"Custom pricing cost: ${custom_cost:.6f}")
```

```python
# Estimate from arbitrary objects
calc = CostCalculator(model="claude-3-5-sonnet")
input_data = {"messages": ["Hello", "How are you?"], "context": "..."}
output_data = {"response": "I'm doing well, thank you!"}

tokens, estimated_cost = calc.estimate_from_objects(input_data, output_data)
print(f"Estimated {tokens} tokens, cost: ${estimated_cost:.6f}")
```

--------------------------------

### Reliable Decorators

Source: https://github.com/simranmultani197/agentcircuit/blob/main/README.md

Documentation for the `@reliable()` and `@reliable_node()` decorators, including their parameters and purpose for creating reliable agent circuits.

```APIDOC
## 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) |
```

--------------------------------

### Import Error Types

Source: https://github.com/simranmultani197/agentcircuit/blob/main/README.md

Import custom exception classes for handling circuit breaker and loop errors.

```python
from agentcircuit import BudgetExceededError, TimeoutExceededError, LoopError
```

--------------------------------

### Core Components

Source: https://github.com/simranmultani197/agentcircuit/blob/main/README.md

Explanation of the core components of AgentCircuit: Fuse for loop detection, Medic for LLM-based error recovery, and Sentinel for Pydantic schema validation.

```APIDOC
### Core Components

```python
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
```

--------------------------------

### Implement Reliable Agent Functions

Source: https://github.com/simranmultani197/agentcircuit/blob/main/README.md

Compare standard agent functions with those wrapped in the @reliable decorator for self-healing and validation.

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

```python
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
```

--------------------------------

### Configure Reliable Node with Auto-Repair

Source: https://github.com/simranmultani197/agentcircuit/blob/main/README.md

Provide an llm_callable to the @reliable decorator to enable automatic output repair.

```python
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"]}
```

--------------------------------

### Custom Cost Per Token

Source: https://github.com/simranmultani197/agentcircuit/blob/main/README.md

Overrides the default pricing with a custom cost per token for specific models.

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

--------------------------------

### Configure Minimal Reliable Node

Source: https://github.com/simranmultani197/agentcircuit/blob/main/README.md

Use the @reliable decorator for basic loop detection and schema validation without an LLM.

```python
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"]}
```

--------------------------------

### Error Types

Source: https://github.com/simranmultani197/agentcircuit/blob/main/README.md

Overview of the error types raised by AgentCircuit, including BudgetExceededError, TimeoutExceededError, and LoopError, with details on when they are raised and their attributes.

```APIDOC
### Error Types

```python
from agentcircuit import BudgetExceededError, TimeoutExceededError, LoopError
```

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

--------------------------------

### Enhanced @reliable Decorator with LLM Auto-Repair

Source: https://context7.com/simranmultani197/agentcircuit/llms.txt

Configure @reliable with LLM callable for auto-repair, set loop limits, maximum cost, and execution time. Specify the model for LLM calls.

```python
from agentcircuit import reliable
from pydantic import BaseModel
import os
from groq import Groq
import json

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

def my_llm(prompt: str) -> str:
    client = Groq(api_key=os.environ["GROQ_API_KEY"])
    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,
    fuse_limit=3,
    max_cost_usd=2.0,
    max_seconds=30,
    model="llama-3.3-70b"
)
def enhanced_search_node(state):
    result = call_llm(state["text"])
    return json.loads(result)

# Execute the node
result = enhanced_search_node({"q": "machine learning"})
print(result)  # SearchResult(query='machine learning', results=['result1', 'result2'], confidence=0.95)
```

--------------------------------

### Classify and Handle Errors with ErrorClassifier

Source: https://context7.com/simranmultani197/agentcircuit/llms.txt

Use ErrorClassifier to categorize exceptions for intelligent handling and obtain recovery hints. This is useful for debugging and implementing fallback mechanisms.

```python
from agentcircuit import (
    BudgetExceededError,
    TimeoutExceededError,
    LoopError,
    MedicError,
    reliable
)
from agentcircuit.errors import (
    ErrorClassifier,
    ErrorCategory,
    ErrorSeverity,
    ClassifiedError,
    AgentCircuitError,
    RecoveryError,
    ConfigurationError,
    ProviderError
)

# Classify an error for intelligent handling
try:
    result = json.loads("invalid json{")
except Exception as e:
    classified = ErrorClassifier.classify(e)

    print(f"Category: {classified.category.value}")  # "json_parse"
    print(f"Severity: {classified.severity.value}")  # "low"
    print(f"Recoverable: {classified.recoverable}")  # True
    print(f"Suggested strategy: {classified.suggested_strategy}")  # "json_repair"

    # Get recovery hint for LLM prompt
    hint = ErrorClassifier.get_recovery_prompt_hint(classified)
    print(f"Hint: {hint}")
```

--------------------------------

### Basic @reliable Decorator Usage

Source: https://context7.com/simranmultani197/agentcircuit/llms.txt

Use the @reliable decorator for basic schema validation and loop detection. Ensure Pydantic models are defined for schema validation.

```python
from agentcircuit import reliable
from pydantic import BaseModel

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

# Basic usage with schema validation and loop detection
@reliable(sentinel_schema=SearchResult, fuse_limit=5)
def search_node(state):
    # Your agent logic here
    return {
        "query": state["q"],
        "results": ["result1", "result2"],
        "confidence": 0.95
    }
```

--------------------------------

### Attempt Recovery from Validation Errors

Source: https://context7.com/simranmultani197/agentcircuit/llms.txt

Use `medic.attempt_recovery` to fix data extraction errors based on a schema. Requires `SentinelError` and `ExtractedData` schema definition. Handles recovery attempts and reports success or failure.

```python
try:
    from agentcircuit.sentinel import SentinelError

    # Simulate a failed extraction
    error = SentinelError("Missing required field: email")
    input_state = {"text": "Contact John, age 30"}
    raw_output = {"name": "John", "age": "thirty"}  # Invalid output

    fixed_data = medic.attempt_recovery(
        error=error,
        input_state=input_state,
        raw_output=raw_output,
        node_id="extract_node",
        recovery_attempts=1,
        schema=ExtractedData
    )

    print(f"Repaired output: {fixed_data}")
    # {"name": "John", "email": "john@example.com", "age": 30}

except SentinelError as e:
    print(f"Recovery failed after all attempts: {e}")
```

```python
# Get recovery statistics
stats = medic.get_stats()
print(f"Success rate: {stats['success_rate']:.1%}")
print(f"Total tokens used: {stats['total_tokens']}")
print(f"Total cost: ${stats['total_cost']:.4f}")
```

--------------------------------

### Comprehensive Error Handling with @reliable Decorator

Source: https://context7.com/simranmultani197/agentcircuit/llms.txt

Apply the @reliable decorator to functions for built-in error handling, including budget, timeout, loop, and medic errors. This simplifies robust function execution.

```python
# Comprehensive error handling
@reliable(
    sentinel_schema=MySchema,
    max_cost_usd=5.0,
    max_seconds=60,
    fuse_limit=3
)
def risky_node(state):
    return process(state)

try:
    result = risky_node({"data": "input"})
except BudgetExceededError as e:
    print(f"Budget: spent ${e.spent:.2f} of ${e.limit:.2f}")
    # Fallback to cheaper model or cached result
except TimeoutExceededError as e:
    print(f"Timeout: {e.elapsed:.1f}s of {e.limit:.1f}s")
    # Return partial result or retry with simpler input
except LoopError as e:
    print(f"Loop detected: {e}")
    # Break the cycle, maybe randomize input
except MedicError as e:
    print(f"Recovery failed: {e}")
    # Log for manual review, return safe default
except AgentCircuitError as e:
    print(f"AgentCircuit error: {e}")
    # Generic handling for other AgentCircuit errors
```

=== COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.