from llmfuse import FuseClient
from llmfuse.providers import GroqProvider, GeminiProvider

client = FuseClient(providers=[GroqProvider(model="..."), GeminiProvider(model="...")])
response = client.complete("Explain retrieval-augmented generation in one sentence.")

If Groq is slow, rate-limited or down, llmfuse retries it sensibly, stops calling it once it's clearly broken, and answers from Gemini instead. Your application just gets a response.

Table of Contents

• Why llmfuse?
• Features
• Installation
• Quickstart
• How It Works
• Guides
  • Configure failover
  • Tune retries
  • Circuit breakers
  • Stay under rate limits
  • Handle errors
  • Use local or other OpenAI-compatible models
  • Write a custom provider
  • Use retries on their own
  • Test your application without API keys
• Supported Providers
• API Reference
• Error Reference
• Troubleshooting
• Design Principles
• Known Limitations
• Roadmap
• Contributing
• Security
• Acknowledgements
• License

Why llmfuse?

LLM APIs fail in ordinary, predictable ways: rate limits (HTTP 429), overloaded servers (503), timeouts and full outages. Handling all of that correctly takes more than a try/except around one call.

Features

Installation

Requirements: Python 3.10 or newer.

# pip
pip install llmfuse

# uv
uv add llmfuse

To try the latest unreleased code instead: pip install "git+https://github.com/AaryanBairagi/llmfuse".

Verify the installation:

python -c "import llmfuse; print(llmfuse.__version__)"

Quickstart

Step 1: Get API keys

You need a key for at least one provider. Two or more are needed to see failover in action.

Step 2: Configure your environment

Create a .env file in your project (and add it to .gitignore):

GROQ_API_KEY=your-groq-key
GEMINI_API_KEY=your-gemini-key

# Model IDs change often. Copy current ones from each provider's docs or console.
GROQ_MODEL=your-groq-model-id
GEMINI_MODEL=your-gemini-model-id

Step 3: Make your first call

# main.py
import os

from llmfuse import AllProvidersFailedError, FuseClient
from llmfuse.providers import GeminiProvider, GroqProvider

client = FuseClient(
    providers=[
        GroqProvider(model=os.environ["GROQ_MODEL"]),      # tried first
        GeminiProvider(model=os.environ["GEMINI_MODEL"]),  # fallback
    ],
)

try:
    response = client.complete("Explain retrieval-augmented generation in one sentence.")
    print(f"[{response.provider}] {response.text}")
except AllProvidersFailedError as error:
    for provider, reason in error.errors.items():
        print(f"{provider}: {reason}")

Run it with the environment loaded:

uv run --env-file .env python main.py

Example output:

[groq] Retrieval-augmented generation (RAG) is a technique where a model retrieves relevant documents and uses them to ground its answer.

response.provider tells you which provider actually answered, so you can see failover happen.

How It Works

Every call to complete() passes through a stack of small layers. Each layer answers exactly one question.

flowchart TD
    A["Your application"] -->|"complete(prompt)"| B["FuseClient<br/><i>failover</i>"]
    subgraph P ["For each provider, in priority order"]
        C["CircuitBreaker<br/><i>healthy?</i>"] --> D["retry_call<br/><i>backoff + jitter</i>"]
        D --> E["TokenBucket<br/><i>within rate limit?</i>"]
        E --> F["Provider adapter<br/><i>request, parse, classify errors</i>"]
    end
    B --> C
    F --> G[("LLM API")]

A request during an outage

Groq is returning HTTP 503; Gemini is healthy.

sequenceDiagram
    autonumber
    participant App as Your app
    participant FC as FuseClient
    participant G as Groq
    participant M as Gemini
    App->>FC: complete("Explain RAG")
    FC->>G: attempt 1
    G-->>FC: 503 Service Unavailable
    Note over FC: retryable, so wait (backoff + jitter)
    FC->>G: attempt 2
    G-->>FC: 503
    FC->>G: attempt 3
    G-->>FC: 503
    Note over FC: retries exhausted, record a failure on Groq's breaker
    FC->>M: attempt 1
    M-->>FC: 200 OK
    FC-->>App: Response(text, provider="gemini")

Circuit breaker states

stateDiagram-v2
    [*] --> Closed
    Closed --> Open: failure_threshold consecutive failed requests
    Open --> HalfOpen: reset_timeout has passed (checked on the next request)
    HalfOpen --> Closed: trial request succeeds
    HalfOpen --> Open: trial request fails

Guides

1. Configure failover

Providers are tried in the order you list them. Order is your preference: put the fastest or cheapest first and the most reliable last.

from llmfuse import FuseClient
from llmfuse.providers import AnthropicProvider, GeminiProvider, GroqProvider

client = FuseClient(
    providers=[
        GroqProvider(model="..."),       # 1st: fast and cheap
        GeminiProvider(model="..."),     # 2nd
        AnthropicProvider(model="..."),  # 3rd: last resort
    ],
)

Failover is per request: the next call starts again from the first provider. Only an open circuit breaker makes a provider skipped across many requests.

Each provider needs a unique name. To use the same provider twice (for example, two API keys or two models), give each one its own name:

GroqProvider(model="model-a", name="groq-fast")
GroqProvider(model="model-b", name="groq-large")

2. Tune retries

from llmfuse import FuseClient, RetryPolicy

client = FuseClient(
    providers=[...],
    retry=RetryPolicy(
        max_attempts=3,   # total tries per provider, including the first
        base_delay=0.5,   # first backoff ceiling, in seconds
        multiplier=2.0,   # backoff grows 0.5s → 1s → 2s → ...
        max_delay=10.0,   # never wait longer than this between attempts
        jitter=True,      # randomise each wait in [0, backoff] (recommended)
    ),
)

backoff = min(max_delay, base_delay × multiplier^(n-1))
wait    = random(0, backoff)   if jitter else backoff

3. Circuit breakers

Every provider gets its own breaker automatically. Tune it on the client:

client = FuseClient(
    providers=[...],
    failure_threshold=5,   # consecutive failed requests before the circuit opens
    reset_timeout=30.0,    # seconds to wait before letting a trial request through
)

Inspect a provider's health, for example for a dashboard or a health check:

from llmfuse import CircuitState

if client.circuit_state("groq") is CircuitState.OPEN:
    print("Groq is currently being skipped")

4. Stay under rate limits

Rate limiting is off by default. Turn it on with requests_per_minute:

client = FuseClient(
    providers=[...],
    requests_per_minute=30,  # long-run average, per provider
    burst=5,                 # up to 5 requests may go out back-to-back
    max_wait=1.0,            # wait up to 1s for capacity, otherwise fail over
)

Every attempt counts against the budget, including retries, because providers count every HTTP call. Being throttled by your own limiter never counts as a provider failure for the circuit breaker.

Per-provider limits

Providers usually have different limits. Give each provider its own requests_per_minute; the client's value becomes the default for providers that don't set one:

client = FuseClient(
    providers=[
        GroqProvider(model="...", requests_per_minute=30),    # Groq's limit
        GeminiProvider(model="...", requests_per_minute=15),  # Gemini's limit
        AnthropicProvider(model="..."),                       # uses the default below
    ],
    requests_per_minute=50,  # default for providers without their own limit
)

5. Handle errors

If no provider can answer, complete() raises AllProvidersFailedError. Its .errors dict explains what happened with each provider:

from llmfuse import AllProvidersFailedError, CircuitOpenError, ThrottledError

try:
    response = client.complete(prompt)
except AllProvidersFailedError as error:
    for provider, reason in error.errors.items():
        if isinstance(reason, CircuitOpenError):
            print(f"{provider}: skipped, circuit open")
        elif isinstance(reason, ThrottledError):
            print(f"{provider}: skipped, local rate limit ({reason.wait:.1f}s until free)")
        else:
            print(f"{provider}: {reason}")

To catch anything raised by llmfuse, catch the base class LLMFuseError.

6. Use local or other OpenAI-compatible models

Any API that implements the OpenAI chat-completions format works through ChatCompatibleProvider. For example, a local Ollama server:

from llmfuse.providers import ChatCompatibleProvider

local = ChatCompatibleProvider(
    name="ollama",
    base_url="http://localhost:11434/v1",
    model="your-local-model",
)

No API key is needed for local servers. For hosted OpenAI-compatible services, pass api_key=....

To make a reusable preset, subclass it and set three class attributes:

class MyHostProvider(ChatCompatibleProvider):
    default_name = "myhost"
    default_base_url = "https://api.myhost.example/v1"
    api_key_env = "MYHOST_API_KEY"

7. Write a custom provider

FuseClient accepts any object with a name attribute and a complete(prompt) -> str method. No base class is required (structural typing via typing.Protocol).

from llmfuse import ProviderError

class MyModelProvider:
    name = "my-model"

    def complete(self, prompt: str) -> str:
        try:
            return call_my_model(prompt)
        except MyTimeout as error:
            # retryable=True  → llmfuse retries with backoff
            raise ProviderError("my-model timed out", provider=self.name, retryable=True) from error
        except MyAuthError as error:
            # retryable=False → llmfuse fails over immediately
            raise ProviderError("my-model auth failed", provider=self.name, retryable=False) from error

8. Use retries on their own

The retry engine is usable without FuseClient, for any flaky call:

from llmfuse import RetryPolicy, retry_call

result = retry_call(
    lambda: fetch_embeddings(texts),
    RetryPolicy(max_attempts=5, base_delay=0.5),
    on_retry=lambda attempt, error, delay: print(f"retry {attempt} in {delay:.2f}s: {error}"),
)

By default only ProviderError(retryable=True), TimeoutError and ConnectionError are retried. Pass should_retry= to customise that.

9. Test your application without API keys

llmfuse.testing ships the same fakes llmfuse uses for its own test suite:

from llmfuse import FuseClient, ProviderError, RetryPolicy
from llmfuse.testing import FakeProvider

def test_my_feature_survives_an_outage() -> None:
    down = ProviderError("503", status_code=503)
    client = FuseClient(
        providers=[
            FakeProvider("primary", errors=[down, down, down]),
            FakeProvider("backup", reply="hello from backup"),
        ],
        retry=RetryPolicy(max_attempts=3, jitter=False),
        sleep=lambda seconds: None,  # don't actually wait
    )
    assert client.complete("hi").provider == "backup"

Supported Providers

All built-in providers are imported from llmfuse.providers. OpenAIProvider sends max_completion_tokens (required by OpenAI's reasoning models); the other chat-completions providers send max_tokens.

API Reference

All public names are importable from llmfuse, except providers (llmfuse.providers) and test utilities (llmfuse.testing).

FuseClient

FuseClient(
    *,
    providers: Sequence[Provider],
    retry: RetryPolicy | None = None,
    failure_threshold: int = 5,
    reset_timeout: float = 30.0,
    requests_per_minute: float | None = None,
    burst: int = 5,
    max_wait: float = 1.0,
    sleep: Callable[[float], None] = time.sleep,
    clock: Callable[[], float] = time.monotonic,
)

Response

A frozen dataclass.

RetryPolicy

A frozen dataclass. Invalid values raise ValueError at construction.

retry_call

retry_call(
    fn: Callable[[], T],
    policy: RetryPolicy | None = None,
    *,
    should_retry: Callable[[BaseException], bool] = is_retryable,
    on_retry: Callable[[int, BaseException, float], None] | None = None,
    sleep: Callable[[float], None] = time.sleep,
) -> T

Calls fn() until it succeeds or the policy is exhausted. Non-retryable errors are re-raised immediately; exhaustion raises RetryExhaustedError (chained to the last error).

Providers

ChatCompatibleProvider(
    *,
    model: str,
    api_key: str | None = None,
    base_url: str | None = None,
    name: str | None = None,
    max_tokens: int = 1024,
    timeout: float = 30.0,
    requests_per_minute: float | None = None,
)

AnthropicProvider(
    *,
    model: str,
    api_key: str | None = None,       # default: ANTHROPIC_API_KEY
    name: str = "anthropic",
    max_tokens: int = 1024,
    timeout: float = 60.0,
    requests_per_minute: float | None = None,
)

Building blocks

Error Reference

LLMFuseError                     base class for everything llmfuse raises
├── ProviderError                a provider call failed     .provider  .status_code  .retryable
│   └── RateLimitError           HTTP 429                   .retry_after
├── RetryExhaustedError          all attempts failed        .attempts  .last_error
├── CircuitOpenError             provider skipped: circuit is open
├── ThrottledError               provider skipped: no rate-limit capacity within max_wait   .wait
└── AllProvidersFailedError      no provider answered       .errors  (provider name → error)

How responses are classified:

flowchart LR
    R["Provider response"] --> Q{"What happened?"}
    Q -->|"200 + valid answer"| OK["Return text"]
    Q -->|"429"| RL["RateLimitError<br/>retry after Retry-After"]
    Q -->|"5xx, 408, timeout, connection error"| RT["ProviderError retryable<br/>retry with backoff"]
    Q -->|"400, 401, 403, 404"| NR["ProviderError not retryable<br/>fail over now"]
    Q -->|"200 + unexpected body"| NR

Troubleshooting

uv run --env-file .env python main.py

Design Principles

Known Limitations

In the spirit of honest engineering, here is what llmfuse does not do yet:

Roadmap

• Retries with exponential backoff, full jitter and Retry-After
• Multi-provider failover
• Per-provider circuit breakers
• Per-provider token-bucket rate limiting
• Groq, Gemini, OpenAI and generic OpenAI-compatible adapters
• Anthropic Messages API adapter
• Per-provider rate limits declared by each provider
• Continuous integration across Python 3.10–3.13
• First PyPI release
• Benchmarks under simulated outages
• Overall time budget (deadline) across retries and providers
• Async client
• System prompts and multi-turn conversations
• Logging and metrics hooks

Have an idea? Open a feature request.

Contributing

Contributions are welcome. To set up a development environment:

git clone https://github.com/AaryanBairagi/llmfuse
cd llmfuse
uv sync                     # creates .venv and installs dev tools from uv.lock

Before opening a pull request:

• New behaviour is covered by tests (use the fakes in llmfuse.testing, not real network calls)
• pytest, ruff check and mypy all pass
• Public API changes are reflected in this README

Live tests are skipped automatically unless the relevant *_API_KEY and *_MODEL variables are set, so the default test run is free, fast and needs no network.

Security

• API keys are read from arguments or environment variables and are never included in error messages.
• Keep keys in .env (git-ignored) or a secrets manager, never in source code.
• To report a security issue, please contact the maintainer privately through GitHub rather than opening a public issue.

Acknowledgements

• Exponential Backoff And Jitter, AWS Architecture Blog: the full-jitter strategy.
• CircuitBreaker, Martin Fowler: the circuit breaker pattern.

License

Released under the MIT License. © 2026 Aaryan Bairagi