Seseorang buka fitur chat kamu, tekan kirim, dan dapat 500. Nggak ada yang berubah di repo. Satu provider balas 429 selama empat puluh detik, dan aplikasi kamu cuma punya satu tempat buat beli token.
Tambalan yang biasa dipakai: try/except di setiap service yang memanggil API. Lama-lama logikanya beda sendiri. Satu service retry tiga kali, satu nggak pernah retry, satu lagi retry di error 400 dan bikin outage-nya lebih panjang. Urusan billing juga kacau: delapan service, delapan key, dan nggak ada satu tempat yang bisa jawab fitur mana yang bakar empat juta token kemarin.
Gateway menyelesaikan tiga hal itu sekaligus dalam satu deploy. Ada provider kedua buat fallback, cuma satu key yang perlu dirotasi, dan satu baris log per request berisi model, status, latency, dan jumlah token. Nggak perlu framework. Sekitar 150 baris FastAPI dan httpx, dan semua client OpenAI SDK bisa diarahkan ke situ cuma dengan ganti base URL.
Yang perlu disiapkan
- Python 3.11+
pip install fastapi uvicorn httpx- Key untuk dua provider yang kompatibel OpenAI (OpenAI, Groq, Together, Fireworks, atau vLLM punyamu sendiri)
- curl
Pakai httpx, bukan requests, karena gateway-nya async. Client yang blocking bikin semua request lain ikut nunggu selama satu generasi masih jalan.
Langkah 1: provider yang gagal sesuai permintaan
Provider asli nggak bisa diminta balas 503 di jam yang pas. Jadi jalur fallback diuji dulu pakai upstream palsu: dua instance dari aplikasi 30 baris yang sama, satu selalu balas 503, satu balas normal.
# mock_upstream.py
import json
import os
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse, StreamingResponse
mode = os.environ.get("MOCK_MODE", "ok")
app = FastAPI()
@app.post("/chat/completions")
async def chat(request: Request):
if mode == "down":
return JSONResponse(
{"error": {"message": "upstream overloaded", "type": "rate_limit"}},
status_code=503,
headers={"retry-after": "0"},
)
body = await request.json()
if body.get("stream"):
async def gen():
for token in ["Hello", " from", " the", " fallback", " provider"]:
frame = {"choices": [{"index": 0, "delta": {"content": token}}]}
yield f"data: {json.dumps(frame)}\n\n".encode()
yield b"data: [DONE]\n\n"
return StreamingResponse(gen(), media_type="text/event-stream")
return {
"id": "chatcmpl-mock",
"object": "chat.completion",
"model": body.get("model", "mock-1"),
"choices": [{
"index": 0,
"message": {"role": "assistant", "content": "Hello from the fallback provider"},
"finish_reason": "stop",
}],
"usage": {"prompt_tokens": 11, "completion_tokens": 7, "total_tokens": 18},
}
Dua instance:
MOCK_MODE=down python3 -m uvicorn mock_upstream:app --port 9101 &
MOCK_MODE=ok python3 -m uvicorn mock_upstream:app --port 9102 &
Langkah 2: gateway-nya
# gateway.py
import asyncio
import json
import logging
import os
import time
from contextlib import asynccontextmanager
import httpx
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse, Response, StreamingResponse
logging.basicConfig(level=logging.INFO, format="%(message)s")
log = logging.getLogger("gateway")
# Urutan penting: entri pertama adalah primary, sisanya fallback.
PROVIDERS = json.loads(
os.environ.get(
"GATEWAY_PROVIDERS",
json.dumps([
{
"name": "openai",
"base_url": "https://api.openai.com/v1",
"key_env": "OPENAI_API_KEY",
},
{
"name": "groq",
"base_url": "https://api.groq.com/openai/v1",
"key_env": "GROQ_API_KEY",
},
]),
)
)
RETRY_STATUS = {408, 409, 425, 429, 500, 502, 503, 504}
ATTEMPTS_PER_PROVIDER = 2
@asynccontextmanager
async def lifespan(app: FastAPI):
app.state.client = httpx.AsyncClient(
timeout=httpx.Timeout(connect=5.0, read=120.0, write=10.0, pool=10.0),
limits=httpx.Limits(max_connections=200, max_keepalive_connections=50),
)
yield
await app.state.client.aclose()
app = FastAPI(lifespan=lifespan)
def headers_for(provider: dict) -> dict:
headers = {"Content-Type": "application/json"}
key = os.environ.get(provider["key_env"], "")
if key:
headers["Authorization"] = f"Bearer {key}"
return headers
def backoff_seconds(resp: httpx.Response, attempt: int) -> float:
raw = resp.headers.get("retry-after")
if raw:
try:
return min(float(raw), 5.0)
except ValueError:
pass
return 0.25 * (2 ** attempt)
def log_call(provider: str, status: int, started: float, usage: dict | None) -> None:
log.info(json.dumps({
"provider": provider,
"status": status,
"latency_ms": round((time.perf_counter() - started) * 1000),
"prompt_tokens": (usage or {}).get("prompt_tokens"),
"completion_tokens": (usage or {}).get("completion_tokens"),
}))
async def call_with_fallback(client: httpx.AsyncClient, body: bytes):
"""Balikin (nama_provider, response) dari provider pertama yang menjawab."""
last = None
for provider in PROVIDERS:
for attempt in range(ATTEMPTS_PER_PROVIDER):
started = time.perf_counter()
try:
resp = await client.post(
f"{provider['base_url']}/chat/completions",
content=body,
headers=headers_for(provider),
)
except httpx.HTTPError as exc:
log_call(provider["name"], 0, started, None)
log.info(json.dumps({"error": f"{type(exc).__name__}: {exc}"}))
continue
usage = None
try:
usage = resp.json().get("usage")
except ValueError:
pass
log_call(provider["name"], resp.status_code, started, usage)
if resp.status_code not in RETRY_STATUS:
return provider["name"], resp
last = (provider["name"], resp)
if attempt + 1 < ATTEMPTS_PER_PROVIDER:
await asyncio.sleep(backoff_seconds(resp, attempt))
return last
async def open_stream_with_fallback(client: httpx.AsyncClient, body: bytes):
"""Buka stream, tapi cuma boleh fallback sebelum ada byte yang dikirim."""
last = None
for provider in PROVIDERS:
started = time.perf_counter()
request = client.build_request(
"POST",
f"{provider['base_url']}/chat/completions",
content=body,
headers=headers_for(provider),
)
resp = await client.send(request, stream=True)
log_call(provider["name"], resp.status_code, started, None)
if resp.status_code not in RETRY_STATUS:
return provider["name"], resp
last = resp
await resp.aclose()
raise httpx.HTTPStatusError(
"all providers failed",
request=request,
response=last or httpx.Response(502),
)
@app.post("/v1/chat/completions")
async def chat_completions(request: Request):
raw = await request.body()
try:
payload = json.loads(raw)
except ValueError:
return JSONResponse({"error": {"message": "invalid JSON body"}}, status_code=400)
client: httpx.AsyncClient = request.app.state.client
if not payload.get("stream"):
result = await call_with_fallback(client, raw)
if result is None:
return JSONResponse({"error": {"message": "all providers failed"}}, status_code=502)
name, resp = result
return Response(
content=resp.content,
status_code=resp.status_code,
media_type="application/json",
headers={"x-gateway-provider": name},
)
try:
name, upstream = await open_stream_with_fallback(client, raw)
except httpx.HTTPStatusError:
return JSONResponse({"error": {"message": "all providers failed"}}, status_code=502)
async def relay():
try:
async for chunk in upstream.aiter_raw():
if await request.is_disconnected():
break
yield chunk
finally:
await upstream.aclose()
return StreamingResponse(
relay(),
media_type="text/event-stream",
headers={
"x-gateway-provider": name,
"Cache-Control": "no-cache",
"X-Accel-Buffering": "no",
},
)
@app.get("/healthz")
async def healthz():
return {
"providers": [
{"name": p["name"], "configured": bool(os.environ.get(p["key_env"]))}
for p in PROVIDERS
]
}
Jalankan dengan dua upstream palsu tadi:
export GATEWAY_PROVIDERS='[{"name":"broken","base_url":"http://127.0.0.1:9101","key_env":"BROKEN_KEY"},{"name":"good","base_url":"http://127.0.0.1:9102","key_env":"GOOD_KEY"}]'
python3 -m uvicorn gateway:app --port 9200
Ada empat keputusan di file itu yang bobotnya paling besar.
Timeout dipisah dengan sengaja. httpx menganggap timeout sebagai lima detik tanpa aktivitas. Read timeout lima detik bakal motong generasi panjang di tengah kalimat, jadi connect dibiarkan 5 dan read dinaikkan ke 120. Kalau diganti satu angka global yang besar, socket mati dan DNS lambat juga ikut tersembunyi di balik 120 detik.
Cuma retry yang masuk akal untuk diulang. 429 atau 502 artinya "coba lagi nanti". 400 artinya payload kamu salah, dan retry dua kali cuma menambah latency ke sebuah bug. Jadi set retry-nya 408, 409, 425, 429 dan keluarga 5xx, dua percobaan per provider, dan Retry-After dihormati maksimal lima detik. Provider yang minta jeda 30 detik itu alasan buat pindah ke provider berikutnya, bukan alasan buat tidur.
Streaming cuma fallback sebelum byte pertama. Begitu gateway sudah mengirim satu chunk, client sudah pegang separuh jawaban dan failover nggak bisa menyelamatkan apa pun. Jadi jalur stream memeriksa status code sebelum mengirim apa-apa, baru setelah itu menyalin byte. Lewat dari titik itu, upstream gagal berarti stream selesai.
Body diteruskan apa adanya. Nggak ada yang mem-parse messages, jadi tools, response_format, temperature, dan field khusus vendor sampai ke provider tanpa diubah. Gateway cuma menambah auth dan header x-gateway-provider di response.
Langkah 3: buktikan jalan
Cek health, tanpa key sama sekali di sandbox ini (dua-duanya configured: false):
$ curl -s http://127.0.0.1:9200/healthz
{"providers":[{"name":"broken","configured":false},{"name":"good","configured":false}]}
Request non-streaming saat primary sedang mati:
$ curl -s -D- -o /dev/null -X POST http://127.0.0.1:9200/v1/chat/completions \
-H 'content-type: application/json' \
-d '{"model":"mock-1","messages":[{"role":"user","content":"hi"}]}'
HTTP/1.1 200 OK
x-gateway-provider: good
content-type: application/json
Request yang sama dengan "stream": true:
HTTP/1.1 200 OK
x-gateway-provider: good
data: {"choices": [{"index": 0, "delta": {"content": "Hello"}}]}
data: {"choices": [{"index": 0, "delta": {"content": " from"}}]}
data: {"choices": [{"index": 0, "delta": {"content": " the"}}]}
data: {"choices": [{"index": 0, "delta": {"content": " fallback"}}]}
Log gateway untuk request itu, satu baris JSON per percobaan:
{"provider": "broken", "status": 503, "latency_ms": 3, "prompt_tokens": null, "completion_tokens": null}
{"provider": "broken", "status": 503, "latency_ms": 2, "prompt_tokens": null, "completion_tokens": null}
{"provider": "good", "status": 200, "latency_ms": 2, "prompt_tokens": null, "completion_tokens": null}
Dua entri untuk broken (itu retry-nya), lalu fallback. Request non-streaming mencatat prompt_tokens: 11 dan completion_tokens: 7 dari blok usage milik mock, angka yang nanti dikalikan tabel harga kamu.
Yang paling penting: OpenAI SDK asli jalan tanpa diubah.
from openai import OpenAI
client = OpenAI(api_key="anything", base_url="http://127.0.0.1:9200/v1")
resp = client.chat.completions.create(
model="mock-1",
messages=[{"role": "user", "content": "say hi"}],
)
print(resp.choices[0].message.content)
# Hello from the fallback provider
stream = client.chat.completions.create(
model="mock-1",
messages=[{"role": "user", "content": "say hi"}],
stream=True,
)
print("".join(c.choices[0].delta.content or "" for c in stream if c.choices))
# Hello from the fallback provider
Kalau semua provider mati, request non-streaming meneruskan error upstream terakhir apa adanya (503 dengan body dari provider, dan itu yang kamu mau di log), sementara request streaming dapat 502 yang bersih.
Bug yang pasti kamu temui
Versi pertama gateway ini memasang header tanpa syarat:
headers = {"Authorization": f"Bearer {os.environ.get(key_env, '')}"}
httpx menolak mengirim header yang isinya cuma Bearer dan melempar:
LocalProtocolError: Illegal header value b'Bearer '
Karena handler menangkap httpx.HTTPError secara luas, kegagalan itu kelihatan seperti provider yang tumbang: status 599 di log, body kosong ke client, dan dua provider dianggap mati padahal dua-duanya sehat. Header sekarang dibangun hanya kalau key-nya ada. Key yang hilang jadi dilewati, bukan merusak request.
Langkah 4: jaga streaming tetap hidup di belakang nginx
nginx mem-buffer response upstream secara default. Itu mengubah aliran token jadi satu kiriman besar saat generasi selesai, dan time to first token kamu berubah jadi durasi generasi penuh. X-Accel-Buffering: no sudah dipasang di response, sisanya di location block:
location /v1/ {
proxy_pass http://127.0.0.1:9200;
proxy_buffering off;
proxy_read_timeout 300s;
proxy_http_version 1.1;
proxy_set_header Connection "";
}
Langkah 5: hitung token untuk request streaming
Response non-streaming membawa usage, jadi baris log-nya gratis. Response streaming nggak, kecuali request-nya meminta. Endpoint yang kompatibel OpenAI menerima stream_options:
{
"model": "gpt-4.1-mini",
"messages": [{"role": "user", "content": "hi"}],
"stream": true,
"stream_options": {"include_usage": true}
}
Dengan flag itu provider mengirim satu chunk tambahan tepat sebelum data: [DONE], dengan choices: [] dan total di usage. Semua chunk sebelumnya membawa usage: null. Kalau parser kamu menganggap choices[0] selalu ada, chunk ekstra itu adalah crash yang menunggu request produksi pertama. Gateway meneruskannya seperti frame lain, jadi perhitungan biaya dibaca di sisi client atau lewat parser kecil yang ditambahkan ke loop relay.
Itu juga alasan angka token di log streaming tadi null: mock-nya nggak mengirim chunk usage, jadi nggak ada yang bisa dicatat.
Kapan pakai ini, kapan pakai alternatif
LiteLLM proxy kalau kamu butuh terjemahan antar API provider, virtual key per tim, budget, atau strategi routing seperti latency-based routing. Cakupan providernya jauh lebih luas dari yang bakal kamu sambung manual, dan itu pilihan yang tepat begitu gateway-mu harus jadi produk, bukan 150 baris di repo.
Gateway hosted (Cloudflare AI Gateway, Portkey) kalau kamu nggak mau menjalankan apa pun. Kamu tukar kendali atas logika routing dengan nol maintenance.
Kode ini kalau semua provider-mu sudah bicara /v1/chat/completions dan kamu mau aturan failover-nya ada di repo sendiri, bisa di-review lewat pull request, dan di-deploy bareng aplikasi.
Satu peringatan soal mencampur provider. Anthropic punya lapisan kompatibilitas OpenAI SDK, dan dokumentasi mereka menyebutnya untuk pengujian dan perbandingan, bukan solusi produksi jangka panjang. Kalau Claude salah satu backend dan kamu butuh fitur penuhnya (citations, input PDF, prompt caching), tulis adapter native untuk /v1/messages atau biarkan LiteLLM yang menerjemahkan. Jangan asumsikan lapisan kompatibilitas itu mencakup semuanya.
Langkah berikutnya
Tiga tambahan yang layak dikerjakan, berurutan: rate limit per pemanggil di depan /v1/chat/completions (satu tim nggak boleh bisa menghabiskan fallback bersama), counter metrik per provider dan status code supaya Grafana bisa menunjukkan failover tanpa grep log, dan peta model per provider supaya request untuk model yang nggak ada di provider itu diarahkan ke padanannya, bukan 404.