Belajar FastAPI - AI/ML Integration & Streaming
Membangun endpoint AI dengan FastAPI: streaming response token-by-token, tool-calling LLM dengan schema dari Pydantic, mengintegrasikan klien OpenAI-compatible secara async, serta pola serving model yang efisien.
#Pendahuluan
Di episode 1 kita menyebut FastAPI sebagai framework de facto untuk AI backends di 2026. Sekarang kita buktikan. Episode ini membangun endpoint yang berinteraksi dengan LLM: streaming token yang menjawab langsung ke klien, tool-calling dengan schema yang lahir dari Pydantic, dan pola serving model yang skala-nya benar.
Mengapa episode ini penting? Karena arsitektur yang kita bangun selama 22 episode — async, streaming, Pydantic, dependency — adalah persis perangkat yang dibutuhkan AI backends. StreamingResponse (episode 7) yang dulu terasa abstrak sekarang menjadi jalur hidup interaksi LLM.
#Streaming Response: Fondasi LLM
LLM modern tidak menunggu jawaban selesai — mereka mengeluarkan token satu per satu, dan klien ingin melihatnya begitu muncul. Kita sudah punya alatnya:
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import httpx
app = FastAPI()
LLM_ENDPOINT = "http://localhost:11434/api/chat" # Ollama
@app.post("/chat/stream")
async def chat_stream(prompt: str) -> StreamingResponse:
async def token_stream():
async with httpx.AsyncClient() as client:
async with client.stream(
"POST",
LLM_ENDPOINT,
json={"model": "qwen3", "messages": [{"role": "user", "content": prompt}], "stream": True},
) as response:
async for line in response.aiter_lines():
if line.strip():
yield f"data: {line}\n\n"
return StreamingResponse(
token_stream(),
media_type="text/event-stream",
headers={"Cache-Control": "no-cache"},
)Alurnya: klien POST /chat/stream → FastAPI membuka stream ke LLM → tiap baris JSON dari LLM diteruskan sebagai event SSE. Klien EventSource (atau fetch streaming) menampilkan token begitu tiba.
Important
Streaming end-to-end adalah satu jalur async penuh: async def handler → async with client.stream(...) → async for iterasi → yield. Satu time.sleep atau HTTP call sinkron di tengah akan memutus aliran dan membuat klien menunggu seluruh jawaban.
#Format SSE untuk LLM
Agar klien bisa memproses, bungkus token dalam format SSE (dari episode 13) — banyak backend memakai format OpenAI-compatible:
import json
def sse_chunk(delta: str, finish: bool = False) -> str:
payload = {
"choices": [{"delta": {"content": delta}, "finish_reason": "stop" if finish else None}]
}
return f"data: {json.dumps(payload, ensure_ascii=False)}\n\n"Klien JavaScript standar (misal OpenAI SDK) bisa langsung mengkonsumsi format ini tanpa adaptor.
#Tool-Calling: Schema dari Pydantic
Tool-calling (function calling) membuat LLM memutuskan memanggil fungsi dengan argumen terstruktur. Dan karena argumen harus valid JSON dengan skema ketat, Pydantic adalah pilihan alami — schema yang kita definisikan langsung dikirim ke LLM:
from pydantic import BaseModel, Field
from typing import Literal
class SearchProduct(BaseModel):
"""Cari produk di katalog berdasarkan kata kunci dan harga."""
query: str = Field(description="kata kunci pencarian")
max_price: float | None = Field(default=None, description="harga maksimal")
class GetStock(BaseModel):
"""Cek stok produk."""
product_id: int = Field(description="ID produk")
ToolName = Literal["search_product", "get_stock"]
TOOLS = {
"search_product": SearchProduct,
"get_stock": GetStock,
}
def tool_schemas() -> list[dict]:
return [
{
"type": "function",
"function": {
"name": name,
"description": model.__doc__,
"parameters": model.model_json_schema(),
},
}
for name, model in TOOLS.items()
]model.model_json_schema() mengubah Pydantic model menjadi JSON Schema — persis yang dipahami LLM. Deskripsi yang kalian tulis di Field dan docstring menjadi instruksi untuk model.
#Menjalankan Tool-Calling
Endpoint yang menggabungkan semuanya:
import json
import httpx
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
class ChatRequest(BaseModel):
message: str
class ChatResponse(BaseModel):
reply: str
@app.post("/chat/tools", response_model=ChatResponse)
async def chat_with_tools(req: ChatRequest) -> ChatResponse:
messages = [{"role": "user", "content": req.message}]
async with httpx.AsyncClient(timeout=30) as client:
resp = await client.post(
"http://localhost:11434/api/chat",
json={
"model": "qwen3",
"messages": messages,
"tools": tool_schemas(),
},
)
data = resp.json()
tool_calls = data.get("message", {}).get("tool_calls", [])
if tool_calls:
return await execute_tool_calls(tool_calls, messages, client)
return ChatResponse(reply=data["message"]["content"])Pola dua langkah: minta LLM menganalisis → jika ada tool_calls, eksekusi tool sungguhan (query database!) → kembalikan hasil ke LLM untuk jawaban final. Inilah cara membangun agent dengan FastAPI.
Tip
Tool-calling adalah alasan utama arsitektur FastAPI (typing-first + Pydantic) menang untuk AI backends: schema tool yang akurat langsung turun dari model yang sama dengan validasi request API kalian. Tidak ada konversi ganda yang bisa melenceng.
#Serving Model Sendiri
Tidak semua model harus lewat API eksternal. Untuk model kecil/terbuka, serve sendiri — Ollama, vLLM, atau Triton — lalu FastAPI jadi gerbangnya:
ollama pull qwen3:8b
ollama serveimport httpx
from typing import Annotated
from fastapi import Depends
LLM_URL = "http://localhost:11434"
async def get_llm_client() -> httpx.AsyncClient:
client = httpx.AsyncClient(base_url=LLM_URL, timeout=60)
try:
yield client
finally:
await client.aclose()
LLMClient = Annotated[httpx.AsyncClient, Depends(get_llm_client)]
@app.get("/models")
async def list_models(llm: LLMClient) -> list[str]:
resp = await llm.get("/api/tags")
resp.raise_for_status()
return [m["name"] for m in resp.json()["models"]]Pola dependency yield (episode 8) menjaga lifecycle klien HTTP — dibuat per request, ditutup setelah selesai. base_url memusatkan konfigurasi (yang bisa dipindah ke Settings di episode 16).
#Memisahkan Model dari Endpoint
Arsitektur yang bisa diskalakan memisahkan dua lapisan:
Gateway (FastAPI) menangani auth, rate limit, validasi, dan routing; model service menghabiskan GPU; business API dieksekusi saat tool-calling. Pemisahan ini memungkinkan skala independen — lebih banyak request ke gateway, lebih banyak GPU ke model service.
#Common Pitfalls
| Pitfall | Solusi |
|---|---|
Memakai requests sync di streaming | httpx.AsyncClient + client.stream |
| Timeout pendek untuk LLM | timeout=60+ di klien |
| Menunggu jawaban penuh sebelum kirim | Stream token satu per satu |
| Schema tool tidak akurat | Lahirkan dari Pydantic model_json_schema |
| Endpoint AI tanpa rate limit | Terapkan episode 19 — LLM mahal |
#Penutup
Inti yang harus dibawa pulang:
- Streaming end-to-end async →
StreamingResponse+httpxstream → token mengalir realtime. - Format SSE / OpenAI-compatible membuat klien standar langsung bisa pakai.
- Tool-calling: schema JSON dari
model_json_schema()Pydantic, eksekusi dua langkah. - Pisahkan gateway, model service, dan business API agar skala independen.
- LLM itu mahal → rate limit ketat di endpoint AI.
Di episode 24 selanjutnya kita akan membahas deployment — uvicorn/gunicorn dengan worker yang benar, multi-stage Dockerfile yang ramping, hingga deploy ke Fly, Render, Railway, dan Kubernetes. Saatnya API kalian tampil di dunia nyata!