Bagian IV - Bab 15

Membangun API Backend dengan FastAPI

Mengapa sistem AI butuh backend terpisah, validasi input dengan Pydantic, membangun endpoint RAG dan Agent, async/non-blocking, dan error handling yang baik.

15.1 Mengapa Sistem AI Membutuhkan Backend Terpisah

Ada prinsip desain software yang disebut Separation of Concerns, setiap bagian sistem sebaiknya memiliki satu tanggung jawab yang jelas dan terpisah dari bagian lain. Dalam konteks sistem AI, ini berarti memisahkan logika AI (RAG, Agent) dari cara sistem tersebut diakses.

Manfaat konkret dari pemisahan ini: frontend tim bisa membangun UI tanpa perlu memahami cara kerja RAG. Kamu bisa mengganti implementasi RAG (misalnya dari Pinecone ke FAISS) tanpa mengubah satu baris pun kode frontend. Dan yang paling penting, sistem AI bisa diakses oleh banyak jenis klien sekaligus: web, mobile, dan integrasi pihak ketiga, semua melalui satu API yang sama.

15.2 Anatomi API Endpoint

Sebelum menulis kode, penting memahami komponen dasar dari sebuah HTTP request dan response.

15.3 Validasi Input dengan Pydantic

Pydantic adalah library yang membuat FastAPI begitu kuat. Ia mendefinisikan kontrak, struktur data yang harus dipenuhi setiap request sebelum diproses lebih lanjut. Tanpa ini, input yang tidak valid akan menyebabkan error di lapisan AI yang jauh lebih sulit didiagnosis.

from pydantic import BaseModel, Field, field_validator
from typing import Optional

# Model untuk request masuk
class ChatRequest(BaseModel):
    pertanyaan: str = Field(
        ...,  # ... berarti wajib diisi
        min_length=1,
        max_length=1000,
        description="Pertanyaan pengguna dalam Bahasa Indonesia"
    )
    session_id: Optional[str] = Field(
        None,
        description="ID sesi untuk melacak riwayat percakapan"
    )

    # Validasi kustom, dijalankan otomatis oleh Pydantic
    @field_validator("pertanyaan")
    @classmethod
    def pertanyaan_tidak_kosong(cls, v: str) -> str:
        if not v.strip():
            raise ValueError("Pertanyaan tidak boleh hanya berisi spasi")
        return v.strip()

# Model untuk response keluar
class ChatResponse(BaseModel):
    jawaban: str
    sumber: list[str] = Field(default_factory=list)
    session_id: str

# Contoh: apa yang terjadi jika input tidak valid
# Request: {"pertanyaan": ""}
# FastAPI otomatis mengembalikan:
#   Status: 422 Unprocessable Entity
#   Body: {"detail": [{"loc": ["body", "pertanyaan"],
#          "msg": "Pertanyaan tidak boleh hanya berisi spasi"}]}
# Tanpa kode validasi manual apapun di endpoint!

15.4 Membangun Endpoint dengan FastAPI

Berikut implementasi lengkap yang menghubungkan FastAPI dengan sistem RAG dan Agent yang sudah dibangun di bab-bab sebelumnya.

Hello World

Aplikasi FastAPI paling minimal, cara terbaik memverifikasi bahwa server berjalan sebelum menambahkan logika yang lebih kompleks.

# pip install fastapi uvicorn
from fastapi import FastAPI

app = FastAPI(
    title="ReWork AI API",
    description="Backend untuk sistem RAG dan Agent ReWork Store",
    version="1.0.0"
)

@app.get("/")
def root():
    return {"status": "ok", "message": "ReWork AI API berjalan"}

@app.get("/health")
def health_check():
    """Endpoint untuk memverifikasi server hidup, penting untuk monitoring."""
    return {"status": "healthy"}

# Jalankan dengan: uvicorn main:app --reload
# Buka http://localhost:8000/docs untuk dokumentasi interaktif
# (Swagger UI dibuat otomatis oleh FastAPI)

Endpoint RAG dan Agent

Endpoint yang menghubungkan FastAPI dengan fungsi RAG dari Bab 13 dan Agent dari Bab 14. Perhatikan bagaimana endpoint hanya memanggil fungsi yang sudah ada, tidak menduplikasi logika AI.

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
import uuid

# Import fungsi yang sudah dibangun di bab sebelumnya
from rag_pipeline import tanya  # dari Bab 13
from agent_pipeline import tanya_agent  # dari Bab 14

app = FastAPI(title="ReWork AI API")

class ChatRequest(BaseModel):
    pertanyaan: str = Field(..., min_length=1, max_length=1000)

class ChatResponse(BaseModel):
    jawaban: str
    sumber: list[str] = []
    request_id: str

# Endpoint RAG: jawab dari dokumen
@app.post("/chat/rag", response_model=ChatResponse)
def chat_rag(request: ChatRequest):
    """
    Endpoint untuk pertanyaan yang dijawab berdasarkan dokumen
    menggunakan pipeline RAG (Bab 13).
    """
    hasil = tanya(request.pertanyaan)
    return ChatResponse(
        jawaban=hasil["jawaban"],
        sumber=hasil["sumber"],
        request_id=str(uuid.uuid4())
    )

# Endpoint Agent: butuh tool calling
@app.post("/chat/agent", response_model=ChatResponse)
def chat_agent(request: ChatRequest):
    """
    Endpoint untuk pertanyaan yang membutuhkan tool calling
    menggunakan LangGraph agent (Bab 14).
    """
    jawaban = tanya_agent(request.pertanyaan)
    return ChatResponse(
        jawaban=jawaban,
        request_id=str(uuid.uuid4())
    )

Async dan Non-blocking

Pemanggilan AI secara asinkron. Penting untuk responsivitas server saat menangani banyak request bersamaan, tanpa async, satu request yang lambat akan memblokir semua request lain.

import asyncio
from fastapi import FastAPI
import google.generativeai as genai

app = FastAPI()

# Tanpa async (blocking)
# Request kedua harus MENUNGGU request pertama selesai
@app.post("/chat-blocking")
def chat_blocking(request: ChatRequest):
    response = model.generate_content(request.pertanyaan)
    return {"jawaban": response.text}

# Dengan async (non-blocking)
# Request lain bisa diproses SAAT menunggu respons LLM
@app.post("/chat-async")
async def chat_async(request: ChatRequest):
    response = await model.generate_content_async(request.pertanyaan)
    return {"jawaban": response.text}

# Async dengan RAG: retrieve dan generate paralel
async def retrieve_async(query: str) -> list[dict]:
    # Wrap fungsi sync dalam thread pool jika library
    # tidak punya versi async native (umum untuk Pinecone)
    return await asyncio.to_thread(retrieve, query)

@app.post("/chat/rag-async")
async def chat_rag_async(request: ChatRequest):
    chunks = await retrieve_async(request.pertanyaan)
    jawaban = await model.generate_content_async(
        susun_prompt(request.pertanyaan, chunks)
    )
    return {"jawaban": jawaban.text}

Error Handling

Penanganan error yang baik mencegah server crash dan memberikan pesan yang jelas ke klien ketika sesuatu gagal, termasuk kegagalan dari API eksternal seperti Pinecone atau Gemini.

from fastapi import FastAPI, HTTPException
from fastapi.responses import JSONResponse
import logging

logger = logging.getLogger("rework-api")
app = FastAPI()

@app.post("/chat/rag", response_model=ChatResponse)
def chat_rag(request: ChatRequest):
    try:
        hasil = tanya(request.pertanyaan)
        return ChatResponse(jawaban=hasil["jawaban"], sumber=hasil["sumber"])

    except ConnectionError as e:
        # Vector database tidak bisa diakses
        logger.error(f"Pinecone connection error: {e}")
        raise HTTPException(
            status_code=503,
            detail="Layanan sedang tidak tersedia. Coba lagi beberapa saat."
        )

    except Exception as e:
        # Tangkap error tak terduga lainnya, JANGAN expose
        # detail teknis ke klien, tapi log untuk debugging
        logger.error(f"Unexpected error: {e}", exc_info=True)
        raise HTTPException(
            status_code=500,
            detail="Terjadi kesalahan internal. Tim kami sudah diberi tahu."
        )

# Global exception handler
# Menangkap error yang tidak tertangani di endpoint manapun
@app.exception_handler(Exception)
async def global_exception_handler(request, exc):
    logger.error(f"Unhandled exception: {exc}", exc_info=True)
    return JSONResponse(
        status_code=500,
        content={"detail": "Terjadi kesalahan tak terduga."}
    )

Kesalahan Umum

Ringkasan Bab 15