readme_content = """
A production-grade Python framework that talks to 19 different AI providers through a single, uniform interface — with automatic failover, retries, health monitoring, circuit breaking, and statistics — so your application never has to know or care which provider actually served a given request.
💡 A note on LangChain / LangGraph: Most of the 19 providers here (SambaNova, Cloudflare Workers AI, Novita, Chutes, LLM7, BigModel, Baseten, Eden AI, AssemblyAI, Fal.ai, ...) have no maintained LangChain integration. Wrapping heterogeneous chat / speech / image APIs behind
Runnable/LangGraphnodes would add a heavy dependency layer without adding real failover control. This project implements theProviderManagerdirectly on top ofrequests, keeping it dependency-light and fully in control of retry/failover semantics.
# Clone & setup
git clone <this-repo>
cd chatbot
python3 -m venv .venv
source .venv/bin/activate # Windows: .venv\\Scripts\\activate
pip install -r requirements.txt
cp .env.example .env🔑 Edit
.envand fill in API keys for whichever providers you want to use. Any provider whose key is missing is automatically skipped — you don't need all 19 keys!
┌─────────────────────────────────────────────────────────────┐
│ 🎯 YOUR APPLICATION │
│ (calls provider_manager.invoke()) │
└────────────────────┬────────────────────────────────────────┘
│
┌────────────────────▼────────────────────────────────────────┐
│ 🔧 ProviderManager (Failover Engine) │
│ ┌─────────┐ ┌─────────┐ ┌─────────┐ ┌─────────┐ │
│ │ Retry │ │ Health │ │ Circuit │ │ Stats │ │
│ │ Engine │ │ Monitor │ │ Breaker │ │ Registry│ │
│ └────┬────┘ └────┬────┘ └────┬────┘ └────┬────┘ │
│ └─────────────┴─────────────┴─────────────┘ │
│ │ │
│ ┌────────────────┼────────────────┐ │
│ ▼ ▼ ▼ │
│ ┌─────────┐ ┌─────────┐ ┌─────────┐ │
│ │Google │ │OpenAI │ │Mistral │ ... 19 total │
│ │Gemini │ │Compat. │ │ │ │
│ └─────────┘ └─────────┘ └─────────┘ │
│ │ │ │ │
│ └────────────────┼────────────────┘ │
│ ▼ │
│ 🛡️ Automatic Failover Chain │
└─────────────────────────────────────────────────────────────┘
chatbot/
│
├── providers/
│ ├── base_provider.py # 🧩 Abstract interface
│ ├── openai_compatible_base.py # 🔌 Shared OpenAI-schema base (DRY)
│ ├── google.py # Gemini (native API)
│ ├── openrouter.py
│ ├── cohere.py # Native Chat v2 API
│ ├── mistral.py
│ ├── fireworks.py
│ ├── sambanova.py
│ ├── cloudflare.py # Workers AI
│ ├── nvidia.py
│ ├── deepinfra.py
│ ├── replicate.py # 🖼️ Image (async)
│ ├── baseten.py
│ ├── novita.py
│ ├── chutes.py
│ ├── edenai.py # 🔀 Gateway proxy
│ ├── assemblyai.py # 🎙️ Speech-to-text
│ ├── stability.py # 🖼️ Image
│ ├── fal.py # 🖼️ Image
│ ├── llm7.py
│ └── bigmodel.py
│
├── provider_manager.py # ❤️ The heart of the framework
├── retry.py # ⏱️ Exponential backoff engine
├── health_check.py # 🩺 Background monitor + circuit breaker
├── logger.py # 📝 Central logging
├── config.py # ⚙️ Env-driven settings
├── exceptions.py # 🚨 Custom exception hierarchy
├── utils.py # 🛠️ Shared helpers
├── statistics.py # 📊 Per-provider stats
├── main.py # 🖥️ CLI entry point
│
├── requirements.txt
├── .env.example
├── README.md
└── logs/
└── chatbot.log
All configuration lives in .env (loaded via python-dotenv).
| Variable | Purpose |
|---|---|
GOOGLE_API_KEY, OPENROUTER_API_KEY, ... |
🔑 Per-provider credentials |
CLOUDFLARE_ACCOUNT_ID |
☁️ Required alongside API key for Cloudflare |
MAX_RETRIES |
🔄 Retry attempts per provider before failing over |
BACKOFF_BASE_SECONDS / BACKOFF_MAX_SECONDS / BACKOFF_JITTER |
⏱️ Exponential backoff tuning |
REQUEST_TIMEOUT_SECONDS |
⏳ Per-request HTTP timeout |
FAILURE_THRESHOLD |
💥 Consecutive failures before circuit breaker trips |
COOLDOWN_SECONDS |
🧊 How long a tripped provider stays disabled |
HEALTH_CHECK_INTERVAL_SECONDS |
🩺 Background probe frequency |
CHAT_PROVIDER_PRIORITY |
📋 Comma-separated provider order (first = tried first) |
LOG_LEVEL / LOG_FILE |
📝 Logging configuration |
*_MODEL |
🎯 Per-provider default model override |
🚫 Nothing is ever hardcoded — every provider reads its key and model from
config.settings/os.getenv.
# 💬 Interactive chat loop
python main.py
# 🌊 Interactive chat with streaming
python main.py --stream
# 📝 One-shot message
python main.py --message "Explain the CAP theorem in one sentence."
# 🌊 One-shot, streamed
python main.py --message "Write a haiku about rate limits." --stream
# 📊 Print provider statistics / health report
python main.py --statsfrom provider_manager import build_default_manager
provider_manager = build_default_manager()
messages = [{"role": "user", "content": "Hello!"}]
response = provider_manager.invoke(messages)
print(response)
# 🌊 Streaming
for chunk in provider_manager.stream(messages):
print(chunk, end="", flush=True)
provider_manager.shutdown()✨ Your code never references a specific provider — it only talks to
provider_manager.
Image (Stability AI, Fal.ai, Replicate) and speech (AssemblyAI) providers implement the same BaseProvider lifecycle so they share statistics and health monitoring, but they are not part of the text invoke()/stream() failover chain.
manager = build_default_manager()
# 🎨 Generate images
stability = manager.get_provider("stability")
image_bytes = stability.generate_image("a lighthouse at sunset, oil painting")
fal = manager.get_provider("fal")
image_url = fal.generate_image("a lighthouse at sunset, oil painting")
replicate = manager.get_provider("replicate")
image_url = replicate.generate_image("a lighthouse at sunset, oil painting")
# 🎙️ Transcribe audio
assemblyai = manager.get_provider("assemblyai")
transcript = assemblyai.transcribe("https://example.com/audio.mp3")┌──────────────────────────────────────────────────────────────┐
│ FAILOVER FLOW │
├──────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────┐ │
│ │ 1. Is │──NO──▶ Skip provider, go to next │
│ │ configured? │ │
│ └──────┬──────┘ │
│ │ YES │
│ ▼ │
│ ┌─────────────┐ │
│ │ 2. Is │──YES──▶ Skip provider, go to next │
│ │ in cooldown?│ │
│ └──────┬──────┘ │
│ │ NO │
│ ▼ │
│ ┌─────────────┐ FAIL ┌─────────────┐ │
│ │ 3. Call via │────────────▶│ Record fail │ │
│ │ retry.py │ │ + health │ │
│ └──────┬──────┘ │ monitor │ │
│ │ SUCCESS └──────┬──────┘ │
│ ▼ │ │
│ ┌─────────────┐ ▼ │
│ │ Return │ ┌─────────────┐ │
│ │ response │◀─────────│ Next provider│ │
│ └─────────────┘ └─────────────┘ │
│ │
│ If ALL fail ──▶ Raise AllProvidersFailedError │
│ │
└──────────────────────────────────────────────────────────────┘
Every chat-capable, configured provider is tried in priority order:
- Skip if
is_configured()isFalse(missing API key) - Skip if health monitor says it's currently in cooldown
- Call through
retry.py— may retry the same provider for transient errors - On failure → record failure, report to health monitor (may trip circuit breaker), move to next provider
- First success wins — result returned immediately
- All fail →
AllProvidersFailedErrorraised with detailed error dictionary (exc.errors)
🌊 Streaming note: If a provider fails after streaming some tokens, failover does not silently restart on another provider (would duplicate/garble output). It raises
AllProvidersFailedErrorso the caller can decide. Failover only happens transparently when a provider fails before producing any output.
| Exception Type | Behavior |
|---|---|
AuthenticationError, ModelNotFoundError |
🚫 Non-retryable — raised immediately (dead API key) |
ProviderTimeoutError, NetworkError, RateLimitError, ServerError, InvalidResponseError |
🔄 Retryable — up to settings.max_retries times |
Backoff formula: base * 2^(attempt-1), capped at max, with ± jitter randomization to avoid thundering-herd retries.
Every attempt is recorded in a RetryResult for observability/debugging.
┌─────────────────────────────────────────────────────────────┐
│ CIRCUIT BREAKER STATE MACHINE │
├─────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────┐ failures >= threshold ┌─────────┐ │
│ │ CLOSED │ ──────────────────────────▶ │ OPEN │ │
│ │ (Normal)│ │(Cooldown│ │
│ └────┬────┘ │ Blocked)│ │
│ │ └────┬────┘ │
│ │ success │ │
│ │ (fast recovery) │ cooldown │
│ │ expires │ │
│ │ ▼ │
│ │ ┌─────────┐ │
│ └─────────────────────────────────│ HALF- │ │
│ health_check() success │ OPEN │ │
│ │ (Probing│ │
│ └────┬────┘ │
│ │ │
│ ▼ │
│ ┌─────────┐ │
│ │ CLOSED │◀──────┘
│ │(Normal) │
│ └─────────┘
└─────────────────────────────────────────────────────────────┘
- Consecutive failures reach
FAILURE_THRESHOLD→ provider marked unhealthy, enters cooldown - While unhealthy →
ProviderManagerskips entirely (no wasted requests) - Background daemon thread probes every
HEALTH_CHECK_INTERVAL_SECONDS - First successful probe → flips back to healthy (fully automatic recovery!)
- Normal successful
invoke()→ immediately clears cooldown (fast recovery path) manager.health_report()→ live table with requests, successes, failures, success rate, latency, health state
Thread-safe StatisticsRegistry tracks per provider:
| Metric | Description |
|---|---|
requests |
Total requests sent |
successes / failures |
Outcome counts |
success_rate |
Computed percentage |
total_latency_ms / average_latency_ms |
Performance metrics |
last_used |
Timestamp of last request |
last_error |
Stringified last error |
healthy |
Current health flag |
cooldown_until / in_cooldown() |
Circuit breaker state |
consecutive_failures |
Drives circuit breaker |
Access via:
provider_manager.statistics() # dict
provider_manager.health_report() # formatted stringutils.validate_response() rejects:
- ❌
None - ❌ Non-string objects
- ❌ Empty / whitespace-only strings
Raises InvalidResponseError → retryable → triggers failover exactly like any other provider error.
Every chat provider implements both:
def invoke(self, messages: list[dict], **kwargs) -> str: ...
def stream(self, messages: list[dict], **kwargs) -> Generator[str, None, None]: ...OpenAICompatibleProvider.stream()→ parses standard SSEdata: {...}chunksgoogle.py&cohere.py→ native streaming wire formatsedenai.py→ no uniform token stream, degrades gracefully to one complete chunk
Exactly 2 steps:
from providers.openai_compatible_base import OpenAICompatibleProvider
from config import settings
import os
class MyNewProvider(OpenAICompatibleProvider):
def __init__(self):
super().__init__(
name="mynewprovider",
base_url="https://api.mynewprovider.com/v1",
api_key=os.getenv("MYNEWPROVIDER_API_KEY"),
default_model=os.getenv("MYNEWPROVIDER_MODEL", "some-default-model"),
)🎯 If the vendor has an OpenAI-compatible
/chat/completionsendpoint (most do), just subclassOpenAICompatibleProvider— all 10 OpenAI-schema providers in this repo do this in ~10 lines!
from providers.mynewprovider import MyNewProvider
# In build_default_manager():
all_providers.append(MyNewProvider())✅ That's it! No changes to ProviderManager, retry.py, health_check.py, statistics.py, or main.py required.
from langchain_core.runnables import RunnableLambda
from provider_manager import build_default_manager
manager = build_default_manager()
def _invoke(input_messages: list[dict]) -> str:
return manager.invoke(input_messages)
chatbot_runnable = RunnableLambda(_invoke)
# Now usable anywhere a LangChain Runnable / LangGraph node is expected:
# graph.add_node("chat", chatbot_runnable)🛡️ Keeps battle-tested failover/retry/health logic in
ProviderManagerwhile giving you a LangGraph-compatible node.
| Symptom | Likely Cause | Fix |
|---|---|---|
AllProvidersFailedError: No chat providers are configured |
No API keys set in .env |
Add at least one valid key |
| Provider always skipped | Missing key, or is_configured() returns False |
Check .env (e.g., Cloudflare needs both API_KEY and ACCOUNT_ID) |
| Provider unhealthy, never recovers | health_check() keeps failing |
Check logs/chatbot.log for probe failures; fix credential/model issue |
| High latency / frequent timeouts | REQUEST_TIMEOUT_SECONDS too low |
Increase timeout; check average_latency_ms in --stats |
| Rate-limited constantly | Too high in priority for your quota | Reorder CHAT_PROVIDER_PRIORITY, or lower FAILURE_THRESHOLD |
| Want to force a specific provider | Call directly: manager.get_provider("mistral").invoke(messages) |
Bypasses failover entirely |
| Logs not appearing | logs/ directory creation failed |
Check filesystem permissions |
InvalidResponseError on every request |
Vendor changed response schema | Check provider's _extract_text / parsing logic against current API docs |
| Principle | Application |
|---|---|
| SOLID | BaseProvider interface (ISP/DIP); ProviderManager depends only on abstraction |
| Open/Closed | Adding providers never requires modifying ProviderManager |
| DRY | OpenAICompatibleProvider eliminates 10x duplication across schema-compatible vendors |
| Single Responsibility | Retries, health, stats, logging each isolated in own module |
| Type Safety | Type hints + dataclasses throughout config.py, statistics.py, retry.py |
| Abstract Base Classes | BaseProvider(ABC) enforces required methods at import time |
⭐ Star this repo if it helps you build resilient AI systems!
output_path = "/mnt/agents/output/README.md" with open(output_path, "w", encoding="utf-8") as f: f.write(readme_content)
print(f"README saved to: {output_path}") print(f"File size: {len(readme_content)} characters")