Async Multi-LLM Gateway

A FastAPI backend that fans a single prompt out to multiple LLM providers concurrently, applies a shared timeout so a slow model can't hold up the whole response, retries transient failures per-provider, and logs every prompt/response pair to a database in the background — without adding that logging time to the API's response latency.

It's built around Groq, Grok (xAI), and Cohere as the target providers. To make the concurrency, timeout, and fault-tolerance behavior easy to test without real API keys, the repo ships with three mock LLM services that simulate fast, medium, and slow responses.

How it works

1. A client sends a single prompt to POST /query.
2. The gateway calls every configured LLM backend in parallel with asyncio + httpx, each call wrapped in a tenacity retry decorator (3 attempts, exponential backoff) to absorb transient failures.
3. A shared 10-second timeout applies to the whole parallel batch — a slow provider is marked as timed out instead of blocking the others.
4. The response (results from every model, each tagged success / timeout / error) is returned to the client immediately.
5. Only after the response is sent, a FastAPI BackgroundTask logs the prompt and every result to the database — so persistence never adds latency to the request.

Client
  │  POST /query {"prompt": "..."}
  ▼
FastAPI Gateway (main.py)
  │
  ├─▶ call_llm1 ─┐
  ├─▶ call_llm2 ─┼── asyncio.gather, 10s shared timeout, tenacity retries per call
  └─▶ call_llm3 ─┘
  │
  ├─▶ Return results to client (success / timeout / error per model)
  └─▶ BackgroundTask: log prompt + results to DB (non-blocking)

Mock LLM services

Since hitting real provider APIs isn't required to demonstrate the gateway's behavior, three standalone FastAPI services simulate LLM backends with deliberately different latencies:

Service	Port	Simulated latency	Purpose
mock_llm1.py	8001	2s	Represents a fast provider
mock_llm2.py	8002	5s	Represents a medium-latency provider
mock_llm3.py	8003	15s	Deliberately exceeds the 10s gateway timeout, to exercise timeout handling

Tech stack

FastAPI, httpx (async HTTP), asyncio, SQLAlchemy (SQLite/PostgreSQL), tenacity (retries), Pydantic, python-dotenv

Project structure

Aync-LLM-Gateway/
├── main.py                  # FastAPI app: POST /query, GET /health, GET /
├── services.py               # Per-provider LLM call functions
├── mock_llm1.py               # Mock fast LLM backend (port 8001)
├── mock_llm2.py               # Mock medium-latency LLM backend (port 8002)
├── mock_llm3.py               # Mock slow LLM backend (port 8003, exceeds timeout)
├── test_client.py             # Manual test script for the gateway's endpoints
├── models/
│   └── models.py              # SQLAlchemy chatlogs table
├── schemas/
│   └── schemas.py             # Pydantic request/response models
├── database/
│   ├── database.py            # Engine, session factory, declarative Base
│   └── dependencies.py        # get_db() FastAPI dependency
├── utils/
│   └── retries.py             # Shared tenacity retry decorator
├── requirements.txt
└── .env.example

Setup

1. Clone and install

git clone https://github.com/koushiikk/Aync-LLM-Gateway.git
cd Aync-LLM-Gateway
pip install -r requirements.txt

2. Configure environment

cp .env.example .env

Fill in:

• LLM1_key, LLM2_key, LLM3_key — any placeholder string works against the mock services
• DATABASE_URL — e.g. sqlite:///./chat.db for local dev, or a PostgreSQL URL for production

3. Start the mock LLM backends (three separate terminals)

python mock_llm1.py   # port 8001
python mock_llm2.py   # port 8002
python mock_llm3.py   # port 8003

4. Start the gateway

python main.py
# or: uvicorn main:app --reload

5. Test it

python test_client.py

or directly:

curl -X POST http://localhost:8000/query \
  -H "Content-Type: application/json" \
  -d '{"prompt": "What are the main differences between supervised and unsupervised learning?"}'

API

Endpoint	Method	Description
/query	POST	Sends {"prompt": str} to all configured LLMs in parallel; returns per-model results with status: success | timeout | error
/health	GET	Health check
/	GET	Service info and endpoint list

Status

This project is under active development — the design above is the intended end-to-end behavior, but a few pieces aren't wired up yet:

• services.py currently implements only call_llm1_app. call_llm2_app, call_llm3_app, and the call_all_llms_parallel orchestrator that main.py imports still need to be written.
• A few import paths need to match the actual package layout: services.py imports from retries import retry_decorator (should be from utils.retries import retry_decorator), and models/models.py / database/dependencies.py import from database import ... (should be from database.database import ...).
• schemas.py doesn't yet define LLMResult, which main.py imports, and chatResponse's response: list[str] field doesn't match how main.py builds it (results=...).

Fixing these is on the list before the gateway runs end-to-end.