aetherroute

AetherRoute ⚡

AetherRoute is a production-grade multi-provider LLM orchestration, validation, and routing framework. It is designed to mitigate critical LLM failure modes — provider outages, cost spikes, prompt injection attacks, structured formatting errors, and token window overflows — in professional environments.

Request ──► Input Sanitizer ──► Cache Lookup (Exact/Semantic)
                                       │
  ┌────────────────────────────────────┘ (Cache Miss)
  ▼
Context Curation (Sliding Window & Summarization)
  │
  ▼
Scoring & Decision Router (Task Fit, Cost, Latency, SQLite History)
  │
  ▼
Provider Pool (OpenAI, Anthropic Claude, Mistral, Ollama)
  │
  ├──► Success ──► Pydantic Validator ──► DB Logging ──► Response
  │                    │ (Validation Error)
  │                    ▼ (Up to 3 Retries)
  │             Repair Re-prompt Loop
  │
  └──► Failure ──► Hot Failover (Try next-best provider)

---

🚀 Key Features & Solved Production Failure Modes

1. Resilient Failover & Registry

• Problem: OpenAI/Anthropic downtime or rate limiting crashes your application.
• Solution: A live health registry (healthy, degraded, down) updates via periodic pings. On failure, AetherRoute performs a hot-failover to the next-best provider seamlessly.

2. Prompt Normalization

• Problem: Providers expect different message structures (Anthropic alternating roles, OpenAI permissive).
• Solution: Unified prompt format dynamically mapped to each provider's API requirements.

3. Real-Time Cost Governor

• Problem: Recursive loops and heavy context cause runaway API bills.
• Solution: Hard per-request and per-session cost ceilings enforced against live token counts. Blocked calls raise CostLimitExceeded.

4. Sliding Context Curation

• Problem: Full chat history causes token overflows and irrelevant retrieval.
• Solution: TF-IDF cosine relevance ranking prunes old messages, and cheap models asynchronously summarize older context.

5. Output Validation & Repair Loops

• Problem: LLMs return invalid JSON or miss required fields.
• Solution: Pydantic schema validation with automatic re-prompting on failure (up to 3 retries with error detail injected into the repair prompt).

6. Security Input Sanitization

• Problem: Prompt injection scripts bypass rules and leak system prompts.
• Solution: Regex-based injection detection + RBAC permission guards raise SecurityBlockError/PermissionDeniedError.

7. Fuzzy Semantic Cache

• Problem: Semantically identical repeated queries waste money and increase latency.
• Solution: Async Redis cache (in-memory fallback) returns hits for both exact and semantically similar queries using cosine similarity.

8. CLI Observability Report

• Problem: Black-box execution blocks debugging.
• Solution: Every routing decision, latency, cost, and validation retry is persisted in SQLite and rendered as a beautiful terminal report via aetherroute-report.

---

📦 Installation

Core (no provider SDK)

pip install aetherroute

With specific providers

pip install aetherroute[openai]        # OpenAI only
pip install aetherroute[anthropic]     # Anthropic Claude only
pip install aetherroute[mistral]       # Mistral AI only
pip install aetherroute[all]           # All providers

From source

git clone https://github.com/mithunbarath/aetherroute.git
cd aetherroute
pip install -e ".[all,dev]"

---

⚙️ Configuration

Set API keys as environment variables (all optional — AetherRoute runs in mock mode without them):

# Windows CMD
set OPENAI_API_KEY=your-openai-key
set ANTHROPIC_API_KEY=your-anthropic-key
set MISTRAL_API_KEY=your-mistral-key

# Linux / macOS
export OPENAI_API_KEY=your-openai-key
export ANTHROPIC_API_KEY=your-anthropic-key
export MISTRAL_API_KEY=your-mistral-key

Or edit config.yaml directly in your project directory.

---

⚡ Quick Start

import asyncio
from aetherroute.orchestrator import AetherRouteOrchestrator

async def main():
    orchestrator = AetherRouteOrchestrator()
    await orchestrator.start()

    response = await orchestrator.query(
        messages=[{"role": "user", "content": "Summarise the history of the Roman Empire."}],
        query="Summarise the history of the Roman Empire.",
        session_id="my-session"
    )
    print(response["text"])
    await orchestrator.close()

asyncio.run(main())

Structured Output (with Pydantic)

from pydantic import BaseModel, Field

class Ticket(BaseModel):
    ticket_id: int
    sentiment: str = Field(description="positive, neutral, or negative")
    urgency: str   = Field(description="high, medium, or low")
    summary: str

ticket, raw = await orchestrator.query_structured(
    messages=[{"role": "user", "content": "Ticket #42: customer is very angry, system is down."}],
    query="Extract ticket info.",
    response_model=Ticket,
    session_id="my-session"
)
print(ticket.urgency)   # "high"

---

🎬 Running the Demo

The included demo.py showcases all 6 scenarios using mock providers (runs fully offline):

python demo.py

Demonstrates:

• Routing decisions based on query classification
• Provider outage hot-failover
• Output validator repair loops
• Security injection blocking
• Session cost governor enforcement
• Fuzzy semantic cache hits

---

📊 CLI Observability Report

View routing analytics, cost breakdowns, and request traces directly in your terminal:

python -m aetherroute.observability.report

# Or if installed via pip:
aetherroute-report

# Options:
#   --db PATH     SQLite database path (default: aetherroute.db)
#   --traces N    Number of recent traces to display (default: 15)
aetherroute-report --db aetherroute.db --traces 25

The report shows:

• Summary panel — total requests, cost, average latency, success rate
• Provider routing table — request count, total cost, avg latency per provider
• Task category breakdown — coding, summarization, reasoning, creative, extraction
• Session cost totals — cost aggregated per session ID
• Request trace table — last N requests with colour-coded success/fail indicators

---

🧪 Running Tests

pip install aetherroute[dev]
pytest tests/ -v

---

📁 Project Structure

aetherroute/
├── adapters/          # Prompt normalization for each provider
├── cache/             # Semantic + exact caching (Redis/in-memory)
├── config.py          # YAML config loader (AetherRouteConfig)
├── context/           # Context curation (TF-IDF, summarization)
├── cost/              # Cost governor & SQLite transaction logging
├── observability/
│   ├── logger.py      # SQLite request logger
│   └── report.py      # Rich CLI observability report
├── orchestrator.py    # Main entry-point: AetherRouteOrchestrator
├── providers/         # OpenAI, Anthropic, Mistral, Ollama wrappers
├── router/            # Task classifier & routing engine
├── security/          # Input sanitizer & RBAC permission guard
└── validation/        # Pydantic validator & self-consistency checker

---

🤝 Contributing

Contributions, issues, and feature requests are welcome!

1. Fork the repository
2. Create a feature branch: git checkout -b feature/my-feature
3. Commit your changes: git commit -m "Add my feature"
4. Push to the branch: git push origin feature/my-feature
5. Open a Pull Request

Please ensure all tests pass (pytest tests/) and update the CHANGELOG.md.

---

👤 Author

Mithun Barath M R

• 📧 barathmithun1548@gmail.com
• 💼 LinkedIn
• 🐙 GitHub
• 🏥 Co-founder, MedClara — Healthcare AI Automation

---

📄 License

This project is licensed under the MIT License — see LICENSE for details.