Belajar AI Engineer - Backend & API untuk AI Apps

#Pendahuluan

Selama lima belas episode, kalian membangun logika AI yang hebat — di dalam script yang berjalan di terminal kalian. Episode ini mengubahnya menjadi produk: API yang bisa dipanggil aplikasi web, mobile, atau integrasi lain, dengan keamanan dan batasan yang layak produksi.

Mengapa penting? Karena prototipe dan produk dipisahkan oleh satu hal: batasan (boundaries). Produk AI harus membatasi siapa yang boleh memanggil, berapa sering, berapa biaya, dan berapa lama waktu respons. Tanpa batasan ini, satu bug di frontend bisa menghabiskan ratusan dolar token dalam hitungan menit. Episode ini memasang batasan-batasan itu.

#Struktur API AI Modern

Backend AI yang baik memisahkan tiga lapisan:

1. API layer (FastAPI/Express): routing, auth, rate limit, validasi input.
2. Service layer: logika AI — prompt building, RAG, memory, tool calls.
3. Data layer: database pengguna, vector DB, cache.

Kunci utamanya: API layer tidak boleh tahu detail LLM, dan service layer tidak boleh menyentuh HTTP. Pemisahan ini membuat sistem bisa diuji dan di-scale.

#Membangun API dengan FastAPI

Contoh API chat dengan streaming dan autentikasi sederhana:

app.py - FastAPI untuk chat AI

import time
from fastapi import FastAPI, Depends, Header, HTTPException
from fastapi.responses import StreamingResponse
from openai import OpenAI
 
app = FastAPI()
client = OpenAI()
 
VALID_KEYS = {"dev_key_123"}  # demo; pakai DB/secret manager di produksi
 
def verify_key(authorization: str | None = Header(None)) -> str:
    if not authorization or not authorization.startswith("Bearer "):
        raise HTTPException(401, "missing token")
    token = authorization.removeprefix("Bearer ")
    if token not in VALID_KEYS:
        raise HTTPException(403, "invalid token")
    return token
 
RATE: dict[str, list[float]] = {}
def rate_limit(api_key: str, max_req: int = 10, window: int = 60) -> None:
    now = time.time()
    RATE.setdefault(api_key, []).append(now)
    RATE[api_key] = [t for t in RATE[api_key] if t > now - window]
    if len(RATE[api_key]) > max_req:
        raise HTTPException(429, "too many requests")
 
@app.post("/chat")
async def chat(prompt: str, api_key: str = Depends(verify_key)):
    rate_limit(api_key)
    def stream():
        r = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": prompt}],
            stream=True,
        )
        for chunk in r:
            d = chunk.choices[0].delta.content
            if d:
                yield d
    return StreamingResponse(stream(), media_type="text/plain")

Empat komponen yang patut dicermati:

• verify_key: setiap request wajib token; tanpa ini API terbuka untuk disalahgunakan.
• rate_limit: membatasi request per pengguna — ini juga guardrail biaya pertama.
• StreamingResponse: jawaban mengalir ke klien token demi token (episode 3).
• Pemisahan: logic AI dipanggil langsung di sini demi contoh, tapi di produksi dipindah ke service layer.

#Streaming yang Benar

Streaming dari backend ke frontend butuh dua lapisan: (1) backend meng-stream dari LLM (di atas), dan (2) frontend membaca stream. Untuk aplikasi web, pola umum adalah Server-Sent Events (SSE) atau WebSocket:

SSE untuk chat

import json
 
def stream_events():
    r = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": "Jelaskan RAG."}],
        stream=True,
    )
    for chunk in r:
        d = chunk.choices[0].delta.content
        if d:
            yield f"data: {json.dumps({'delta': d})}\n\n"
    yield "data: [DONE]\n\n"
 
@app.post("/chat/stream")
async def chat_stream(prompt: str, api_key: str = Depends(verify_key)):
    return StreamingResponse(stream_events(), media_type="text/event-stream")

SSE punya keunggulan dibanding polling: klien menerima delta secara real-time, dan koneksi bisa ditutup (cancel) kapan saja — yang juga menghemat token.

#Auth dan Keamanan API

Untuk produksi, ganti VALID_KEYS di atas dengan sistem yang benar:

• JWT / session: setiap pengguna punya identitas; token dikeluarkan oleh auth service.
• Scope per key: key untuk read-only, key untuk admin — jangan satu key untuk semua.
• Secret manager: simpan LLM API key di secret manager (env, Vault), bukan di kode.
• Rate limit per user + per key: dua lapis — mencegah abuse individual dan kebocoran key.
• Audit log: catat siapa memanggil apa, kapan, berapa token — kebutuhan observability (episode 17) dan compliance (episode 20).

#Guardrail Biaya di Backend

Selain rate limit, pasang batas biaya eksplisit:

Budget guard per pengguna

def check_budget(user_id: str, cost: float) -> None:
    total = get_daily_spend(user_id)  # dari DB / observability
    if total + cost > BUDGET_HARIAN:
        raise HTTPException(429, "budget harian tercapai, coba besok")

Kombinasi rate limit + budget guard + auth adalah minimum untuk API AI produksi. Tanpa ketiganya, produk kalian satu request malang dari tagihan mengejutkan.

#Common Pitfalls

• API tanpa rate limit: satu loop bug di klien = biaya meledak.
• Memanggil LLM dari frontend: kebocoran key dan tidak terkontrol.
• Streaming tanpa cancel: klien yang pergi tidak membatalkan request = token terbuang.
• API layer penuh logika AI: sulit diuji, sulit diganti model, sulit di-scale.

#Penutup

Inti yang harus dibawa pulang:

• Backend AI = API layer + service layer + data layer, dipisahkan dengan jelas.
• FastAPI + streaming (SSE) + rate limit + auth = minimum produksi.
• Jangan pernah memanggil LLM dari frontend; lewatkan semua lewat backend.
• Budget guard per pengguna melindungi dari pembengkakan biaya.

Di episode 17 selanjutnya kita memasang mata pada sistem: observability & monitoring AI — tracing langkah agent, tracking biaya per request, dan monitoring kualitas jawaban di produksi. Sampai jumpa di episode 17!