Panduan Praktis Setup FastAPI untuk Server LLM Lokal

Verdict Cepat: Menjalankan inference LLM lokal lewat script terminal biasa sering kali kurang fleksibel saat aplikasi frontend, agen AI, atau script otomatisasi lain membutuhkan akses bersamaan. Membangun API wrapper mandiri dengan FastAPI dan library Uvicorn memberikan kontrol penuh atas streaming token SSE (Server-Sent Events), pembatasan throughput (rate limiting), serta format response yang kompatibel dengan standar OpenAI API tanpa ketergantungan pada SaaS pihak ketiga.

Kebutuhan integrasi model bahasa besar pada infrastruktur internal terus meningkat. Banyak pengembang menjalankan model open-source seperti Llama, Mistral, atau Qwen menggunakan runtime lokal seperti Ollama atau vLLM. Kendati runtime tersebut menyediakan antarmuka bawaan, lapisan proxy atau custom backend sering diperlukan untuk menyisipkan otentikasi API key kustom, logging payload ke database, validasi schema request via Pydantic, serta routing dinamis ke berbagai engine backend.

FastAPI menjadi pilihan standar industri karena performa asynchronous berbasis ASGI yang sangat cepat, dokumentasi Swagger UI otomatis, dan kemudahan mengelola background tasks saat menangani antrean prompt yang padat. Artikel ini menyajikan panduan langkah demi langkah membangun server inferensi API lokal yang siap pakai, hemat sumber daya, dan terstruktur rapi.

Arsitektur Server LLM Berbasis FastAPI

Sebelum menulis baris kode pertama, pemahaman mengenai alur data antara client, web framework, dan inference engine sangat penting. Server yang dibangun bertindak sebagai middleware cerdas yang menerima request HTTP, memvalidasi parameter generasi teks (seperti temperature, top_p, max_tokens), lalu meneruskannya ke engine model lokal.

Struktur arsitektur ini memisahkan layer presentasi dan layer komputasi berat. Ketika request streaming masuk, FastAPI memanfaatkan generator Python asynchronous untuk menghasilkan data chunk per token secara realtime menggunakan protokol HTTP Server-Sent Events. Hal ini mencegah lonjakan memori di sisi server karena response tidak ditahan seluruhnya di RAM sebelum dikirimkan ke pemanggil.

Persiapan Lingkungan dan Instalasi Dependensi

Langkah awal dimulai dengan menyiapkan virtual environment Python terisolasi pada server Linux atau mesin lokal. Penggunaan isolated environment mencegah konflik paket dengan sistem operasi utama.

# Buat direktori proyek dan masuk ke dalamnya
mkdir -p ~/fastapi-llm-server && cd ~/fastapi-llm-server

# Buat virtual environment dengan Python 3.10+
python3 -m venv venv
source venv/bin/activate

# Install dependensi utama
pip install fastapi uvicorn pydantic httpx sse-starlette python-dotenv

Komponen dependensi yang digunakan memiliki fungsi spesifik:

• FastAPI: Framework web asynchronous performa tinggi untuk membangun REST API modern.
• Uvicorn: Web server ASGI berbasis uvloop untuk menjalankan aplikasi Python secara konkuren.
• Pydantic: Validasi data tipe ketat untuk payload request dan response JSON.
• HTTPX: Client HTTP asynchronous untuk berkomunikasi dengan backend model internal secara non-blocking.
• sse-starlette: Modul handler untuk streaming Server-Sent Events yang stabil dan efisien.

Menentukan Data Schema dengan Pydantic

Standarisasi request format sangat penting agar klien AI yang biasa menggunakan OpenAI SDK dapat langsung terhubung tanpa perubahan kode masif. Buat file baru bernama schemas.py untuk mendefinisikan struktur data pesan dan parameter inferensi.

from pydantic import BaseModel, Field
from typing import List, Optional, Literal, Dict, Any

class ChatMessage(BaseModel):
    role: Literal["system", "user", "assistant"] = Field(..., description="Role pengirim pesan")
    content: str = Field(..., min_length=1, description="Isi teks pesan")

class ChatCompletionRequest(BaseModel):
    model: str = Field("default-model", description="Identifier model target")
    messages: List[ChatMessage] = Field(..., min_items=1, description="Daftar riwayat percakapan")
    temperature: Optional[float] = Field(0.7, ge=0.0, le=2.0)
    top_p: Optional[float] = Field(0.9, ge=0.0, le=1.0)
    max_tokens: Optional[int] = Field(1024, ge=1, le=8192)
    stream: Optional[bool] = Field(False, description="Aktifkan Server-Sent Events streaming")

class ChatChoice(BaseModel):
    index: int
    message: ChatMessage
    finish_reason: Optional[str] = "stop"

class ChatCompletionResponse(BaseModel):
    id: str
    object: str = "chat.completion"
    created: int
    model: str
    choices: List[ChatChoice]
    usage: Dict[str, int]

Skema di atas menerapkan validasi ketat. Jika client mengirim nilai temperature di atas 2.0 atau pesan kosong, FastAPI otomatis mengembalikan response HTTP 422 Unprocessable Entity lengkap dengan detail field yang bermasalah tanpa membebani inference engine.

Membangun Engine Core dan Endpoint FastAPI

Selanjutnya, buat file aplikasi utama bernama main.py. Pada implementasi ini, server akan memproses permintaan non-streaming dan streaming token secara bersamaan dengan memanggil backend engine lokal (misalnya Ollama yang berjalan pada port 11434 atau vLLM pada port 8000).

import time
import uuid
import json
import httpx
from fastapi import FastAPI, HTTPException, Security, status
from fastapi.security.api_key import APIKeyHeader
from sse_starlette.sse import EventSourceResponse
from schemas import ChatCompletionRequest, ChatCompletionResponse, ChatChoice, ChatMessage

app = FastAPI(
    title="Custom Local LLM Gateway",
    version="1.0.0",
    description="FastAPI Wrapper Mandiri untuk Model LLM Lokal"
)

API_KEY_NAME = "Authorization"
api_key_header = APIKeyHeader(name=API_KEY_NAME, auto_error=False)
VALID_API_KEY = "Bearer rahasia-kustom-lokal-123"
BACKEND_LLM_URL = "http://127.0.0.1:11434/api/chat"

async def verify_token(api_key: str = Security(api_key_header)):
    if api_key != VALID_API_KEY:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="API Key tidak valid atau tidak disertakan."
        )
    return api_key

@app.get("/health", tags=["Monitoring"])
async def health_check():
    return {"status": "healthy", "timestamp": int(time.time())}

async def generate_stream_tokens(request_data: ChatCompletionRequest):
    req_id = f"chatcmpl-{uuid.uuid4().hex[:12]}"
    created_ts = int(time.time())
    
    payload = {
        "model": request_data.model,
        "messages": [m.model_dump() for m in request_data.messages],
        "stream": True,
        "options": {
            "temperature": request_data.temperature,
            "top_p": request_data.top_p,
            "num_predict": request_data.max_tokens
        }
    }
    
    async with httpx.AsyncClient(timeout=120.0) as client:
        async with client.stream("POST", BACKEND_LLM_URL, json=payload) as response:
            if response.status_code != 200:
                yield {
                    "event": "error",
                    "data": json.dumps({"error": "Backend model tidak merespons dengan benar."})
                }
                return

            async for chunk in response.aiter_lines():
                if not chunk.strip():
                    continue
                try:
                    data = json.loads(chunk)
                    delta_text = data.get("message", {}).get("content", "")
                    done = data.get("done", False)
                    
                    chunk_payload = {
                        "id": req_id,
                        "object": "chat.completion.chunk",
                        "created": created_ts,
                        "model": request_data.model,
                        "choices": [{
                            "index": 0,
                            "delta": {"content": delta_text} if not done else {},
                            "finish_reason": "stop" if done else None
                        }]
                    }
                    yield {"data": json.dumps(chunk_payload)}
                    
                    if done:
                        yield {"data": "[DONE]"}
                except Exception:
                    continue

@app.post("/v1/chat/completions", tags=["LLM Inference"])
async def create_chat_completion(
    request: ChatCompletionRequest,
    auth: str = Security(verify_token)
):
    if request.stream:
        return EventSourceResponse(generate_stream_tokens(request))

    req_id = f"chatcmpl-{uuid.uuid4().hex[:12]}"
    created_ts = int(time.time())
    
    payload = {
        "model": request.model,
        "messages": [m.model_dump() for m in request.messages],
        "stream": False,
        "options": {
            "temperature": request.temperature,
            "top_p": request.top_p,
            "num_predict": request.max_tokens
        }
    }
    
    async with httpx.AsyncClient(timeout=120.0) as client:
        try:
            resp = await client.post(BACKEND_LLM_URL, json=payload)
            if resp.status_code != 200:
                raise HTTPException(
                    status_code=resp.status_code,
                    detail="Inference backend error"
                )
            result = resp.json()
        except httpx.RequestError as exc:
            raise HTTPException(
                status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
                detail=f"Gagal menghubungi backend engine: {str(exc)}"
            )

    assistant_content = result.get("message", {}).get("content", "")
    prompt_eval_count = result.get("prompt_eval_count", 0)
    eval_count = result.get("eval_count", 0)

    return ChatCompletionResponse(
        id=req_id,
        created=created_ts,
        model=request.model,
        choices=[
            ChatChoice(
                index=0,
                message=ChatMessage(role="assistant", content=assistant_content),
                finish_reason="stop"
            )
        ],
        usage={
            "prompt_tokens": prompt_eval_count,
            "completion_tokens": eval_count,
            "total_tokens": prompt_eval_count + eval_count
        }
    )

Tabel Komparasi Metode Deployment Server LLM

Dalam menentukan pendekatan deployment untuk kebutuhan produksi maupun riset tim internal, pertimbangan performa dan fleksibilitas menjadi faktor utama. Berikut tabel perbandingannya:

Konfigurasi Uvicorn untuk Produksi

Pada lingkungan pengujian, perintah pengembangan biasa cukup memadai. Namun untuk beban kerja produksi, jalankan server dengan konfigurasi worker proses ganda dan bind ke host lokal yang aman.

# Jalankan dengan 4 worker pada port 8080
uvicorn main:app --host 0.0.0.0 --port 8080 --workers 4 --log-level info

Jika server ditempatkan di VPS publik, pasang reverse proxy seperti Nginx di depan Uvicorn. Reverse proxy bertugas menangani terminasi SSL/TLS HTTPS, rate limiting per IP, serta proteksi terhadap request berukuran abnormal sebelum diteruskan ke aplikasi FastAPI.

Integrasi Pengujian dengan Python Client

Setelah server aktif, verifikasi fungsionalitas streaming dan non-streaming dapat dilakukan menggunakan script client Python standar.

import httpx
import json

URL = "http://127.0.0.1:8080/v1/chat/completions"
HEADERS = {
    "Authorization": "Bearer rahasia-kustom-lokal-123",
    "Content-Type": "application/json"
}

payload = {
    "model": "qwen2.5-coder:7b",
    "messages": [
        {"role": "system", "content": "Jawab secara ringkas dan lugas."},
        {"role": "user", "content": "Jelaskan fungsi asyncio di Python dalam 2 kalimat."}
    ],
    "temperature": 0.3,
    "stream": False
}

with httpx.Client(timeout=30.0) as client:
    response = client.post(URL, headers=HEADERS, json=payload)
    print("Response JSON:")
    print(json.dumps(response.json(), indent=2))

Koneksi ini juga kompatibel dengan library ekosistem seperti LangChain, LlamaIndex, atau software automasi coding cukup dengan mengarahkan parameter base_url ke endpoint lokal tersebut.

Kesalahan Umum yang Sering Terjadi

Dalam mengelola server inferensi mandiri, beberapa kendala teknis sering menurunkan stabilitas layanan:

• Timeout HTTP yang Terlalu Pendek: Pemrosesan prompt panjang pada GPU kelas menengah membutuhkan waktu beberapa detik sebelum token pertama keluar. Setting timeout default 5 detik pada client HTTP internal akan memicu error 504. Pastikan parameter timeout diatur minimal 60 hingga 120 detik.
• Blokir Event Loop Asyncio: Menjalankan fungsi blocking synchronous (seperti komputasi berat murni CPU atau pemanggilan sleep biasa) di dalam route asynchronous akan menghentikan proses event loop worker, sehingga request client lain ikut tertahan. Gunakan modul asyncio.to_thread jika harus mengeksekusi library synchronous.
• Kebocoran Memori SSE: Mengabaikan penutupan koneksi saat client memutuskan koneksi di tengah streaming dapat meninggalkan koneksi menggantung di server. Pustaka sse-starlette menangani pembersihan event stream secara otomatis selama generator function keluar dengan benar.

Rekomendasi Terkait di Grafisify

Bagi yang mendalami arsitektur infrastruktur kecerdasan buatan mandiri, pelajari juga topik relevan berikut:

• Panduan Praktis Setup Qdrant Vector DB Lokal untuk RAG
• Panduan Praktis Setup ChromaDB Lokal untuk Proyek AI
• Panduan Praktis Setup LiteLLM Proxy untuk Multi-Model
• Panduan Praktis Setup Open-WebUI untuk LLM Lokal
• Panduan Praktis Browser Use untuk Otomasi Web Berbasis AI

Dokumentasi resmi FastAPI dan standar OpenAI API dapat dipelajari secara mendalam melalui dokumentasi FastAPI dan portal dokumentasi OpenAI API Reference.

Membangun antarmuka API sendiri dengan FastAPI membuka peluang integrasi tanpa batas, fleksibilitas kontrol data, dan keamanan jaringan internal yang terjamin.