Tutorial ini membangun layanan chatbot yang kecil namun siap produksi dengan FastAPI, SSE streaming, memori percakapan, dan pergantian model. Tujuannya bukan untuk membuat demo mainan. Tujuannya adalah untuk mendapatkan backend yang benar-benar bisa Anda tempatkan di balik antarmuka produk dan diiterasi dengan aman.
Jika Anda sudah mengarahkan satu OpenAI-compatible SDK ke LemonData, artikel ini melanjutkan dari sana. Jika Anda belum melakukan penggantian base URL, baca panduan migrasi terlebih dahulu. Jika kekhawatiran utama Anda adalah request shaping dan backoff saat beban tinggi, pasangkan panduan ini dengan panduan rate limiting AI API.
Apa yang Kita Bangun
Layanan yang sudah jadi memiliki enam bagian yang bergerak:
- Sebuah endpoint
/chatsinkron untuk smoke test. - Sebuah endpoint
/chat/streamstreaming untuk UI yang sebenarnya. - Status percakapan yang dikunci oleh
conversation_id. - Sebuah model allowlist agar frontend tidak dapat meminta ID sembarangan.
- Penanganan error yang tidak runtuh pada 429 pertama.
- Jalur yang jelas dari prototipe in-memory ke Redis atau PostgreSQL.
Itu sudah cukup untuk menjalankan bot dukungan, asisten internal, atau versi pertama dari widget chat tersemat.
Instal Stack Minimum
pip install fastapi uvicorn openai pydantic redis
Anda dapat mengabaikan redis untuk tahap awal, tetapi berguna untuk menghubungkan import-nya sekarang agar jalur upgrade terlihat jelas.
Langkah 1: Mulai Dengan Endpoint Chat yang Kecil dan Sederhana
Cara tercepat untuk tersesat dalam pengerjaan chatbot adalah mulai dengan websockets, penggunaan tool, dan orkestrasi agent sebelum jalur request dasar stabil. Mulailah dengan satu endpoint kecil yang membuktikan bahwa key, base URL, dan perutean model Anda sudah benar.
from fastapi import FastAPI
from openai import OpenAI
from pydantic import BaseModel
app = FastAPI()
client = OpenAI(
api_key="sk-lemon-xxx",
base_url="https://api.lemondata.cc/v1"
)
class ChatRequest(BaseModel):
message: str
model: str = "gpt-4.1-mini"
conversation_id: str | None = None
@app.post("/chat")
async def chat(req: ChatRequest):
response = client.chat.completions.create(
model=req.model,
messages=[{"role": "user", "content": req.message}]
)
return {"reply": response.choices[0].message.content}
Jalankan satu smoke test. Jika ini gagal, jangan terus menumpuk fitur di atasnya.
Langkah 2: Tambahkan Streaming Karena Pengguna Merasakan Latency Sebelum Mereka Mengukurnya
Sebagian besar produk chatbot terasa lambat bukan karena modelnya lambat, tetapi karena UI tetap kosong sampai respons penuh tiba. SSE sudah cukup untuk banyak produk chat dan memiliki beban operasional yang lebih rendah daripada websockets.
from fastapi.responses import StreamingResponse
@app.post("/chat/stream")
async def chat_stream(req: ChatRequest):
def generate():
stream = client.chat.completions.create(
model=req.model,
messages=[{"role": "user", "content": req.message}],
stream=True
)
for chunk in stream:
delta = chunk.choices[0].delta
if delta.content:
yield f"data: {delta.content}\n\n"
yield "data: [DONE]\n\n"
return StreamingResponse(generate(), media_type="text/event-stream")
Di sisi frontend, client sisi browser yang paling sederhana pun masih cukup baik:
async function sendMessage(payload) {
const response = await fetch('/chat/stream', {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify(payload),
});
const reader = response.body.getReader();
const decoder = new TextDecoder();
while (true) {
const { done, value } = await reader.read();
if (done) break;
const chunk = decoder.decode(value, { stream: true });
console.log(chunk);
}
}
Jika produk Anda sudah menggunakan client browser dan HTTP standar, SSE menjaga arsitektur tetap lebih sederhana.
Langkah 3: Pindahkan Status Percakapan Keluar dari Request Body
Demo chatbot pertama biasanya menyimpan transkrip lengkap di browser dan mengirimkannya pada setiap giliran. Itu berhasil untuk prototipe. Ini menjadi berantakan saat Anda membutuhkan retry, sesi yang dapat dilanjutkan, atau tooling sisi server.
Penyimpanan in-memory cukup untuk memulai:
from collections import defaultdict
import uuid
conversations: dict[str, list] = defaultdict(list)
SYSTEM_PROMPT = "You are a helpful assistant. Be concise and direct."
def build_messages(conv_id: str, user_msg: str) -> list:
messages = [{"role": "system", "content": SYSTEM_PROMPT}]
history = conversations[conv_id][-20:]
messages.extend(history)
messages.append({"role": "user", "content": user_msg})
conversations[conv_id].append({"role": "user", "content": user_msg})
return messages
Jalur upgrade ke Redis sebagian besar adalah urusan penyimpanan:
import json
import redis
redis_client = redis.Redis(host="127.0.0.1", port=6379, decode_responses=True)
def load_history(conv_id: str) -> list:
raw = redis_client.get(f"chat:{conv_id}")
return json.loads(raw) if raw else []
def save_history(conv_id: str, history: list) -> None:
redis_client.setex(f"chat:{conv_id}", 60 * 60 * 24, json.dumps(history))
Gunakan Redis jika percakapan membutuhkan TTL, kemampuan untuk dilanjutkan, atau deployment multi-instance. Gunakan PostgreSQL jika transkrip itu sendiri adalah data produk.
Langkah 4: Perlakukan Error sebagai Perilaku Produk, Bukan Sekadar Exception
Jika chatbot Anda menghadap pelanggan, jalur kegagalan sama pentingnya dengan jalur keberhasilan. Seorang pengguna tidak peduli apakah kegagalan berasal dari rate limiting, saldo, atau pemadaman model. Mereka peduli apakah UI membeku.
from openai import APIConnectionError, APIError, RateLimitError
@app.post("/chat/stream")
async def chat_stream(req: ChatRequest):
conv_id = req.conversation_id or str(uuid.uuid4())
messages = build_messages(conv_id, req.message)
def generate():
full_response = []
try:
stream = client.chat.completions.create(
model=req.model,
messages=messages,
stream=True
)
for chunk in stream:
delta = chunk.choices[0].delta
if delta.content:
full_response.append(delta.content)
yield f"data: {delta.content}\n\n"
except RateLimitError:
yield "data: [ERROR] The model is busy. Please retry in a few seconds.\n\n"
except APIConnectionError:
yield "data: [ERROR] Temporary network issue. Please retry.\n\n"
except APIError as error:
yield f"data: [ERROR] {error.message}\n\n"
else:
conversations[conv_id].append(
{"role": "assistant", "content": "".join(full_response)}
)
finally:
yield "data: [DONE]\n\n"
return StreamingResponse(
generate(),
media_type="text/event-stream",
headers={"X-Conversation-ID": conv_id}
)
Jika Anda melayani beban yang berarti, Anda juga harus membentuk request sebelum mencapai upstream. Pola detailnya ada di panduan rate limiting, tetapi versi singkatnya adalah: gunakan bounded retries, gunakan jitter, dan jangan pernah except Exception: time.sleep(1).
Langkah 5: Pergantian Model Membutuhkan Allowlist, Bukan Kotak Teks Bebas
Satu API key dapat menjangkau ratusan model. Itu tidak berarti UI Anda harus mengekspos ratusan model. Backend harus mempublikasikan allowlist kecil yang sesuai dengan use case Anda.
AVAILABLE_MODELS = {
"fast": "gpt-4.1-mini",
"balanced": "claude-sonnet-4-6",
"reasoning": "o3",
"budget": "deepseek-chat",
}
@app.get("/models")
async def list_models():
return {"models": AVAILABLE_MODELS}
Ini melakukan tiga hal yang berguna:
- ini menghentikan frontend dari meminta ID model yang tidak valid atau usang
- ini memungkinkan Anda memetakan ulang tier nanti tanpa men-deploy ulang setiap client
- ini memberi Anda satu tempat untuk menegakkan kontrol biaya
Jika tim Anda masih memutuskan provider mana yang akan distandarisasi, perbandingan harga dan perbandingan OpenRouter vs LemonData adalah dua halaman yang layak dibaca sebelum Anda mengunci allowlist.
Langkah 6: Tambahkan Sisi Produksi Sebelum Traffic Tiba
Backend chatbot menjadi kelas produksi ketika sisi-sisi di sekitarnya ditangani, bukan ketika panggilan chat intinya cerdas.
Daftar periksanya singkat:
- tambahkan request ID agar Anda dapat menghubungkan kegagalan frontend ke log backend
- batasi konkurensi per pengguna dan ukuran request
- pangkas riwayat yang panjang sebelum mereka meledakkan anggaran token Anda
- log model, latency, ukuran input, dan finish reason
- pisahkan pesan error yang terlihat pengguna dari detail error internal
- uji satu model alternatif sehingga Anda tahu fallback berfungsi sebelum pemadaman pertama terjadi
Pemangkasan riwayat bisa tetap sederhana:
def trim_history(messages: list, max_tokens: int = 8000) -> list:
system = messages[0]
history = messages[1:]
total_chars = len(system["content"])
trimmed = []
for msg in reversed(history):
msg_chars = len(msg["content"])
if total_chars + msg_chars > max_tokens * 4:
break
trimmed.insert(0, msg)
total_chars += msg_chars
return [system] + trimmed
Intinya bukan penghitungan token yang sempurna. Intinya adalah menghentikan ledakan konteks yang jelas.
Dari Demo ke Produk
Setelah backend ini stabil, upgrade berikutnya jarang berupa “lebih banyak AI.” Biasanya berupa infrastruktur yang membosankan:
- auth agar satu pengguna tidak dapat membaca percakapan pengguna lain
- persistensi agar sesi bertahan saat deploy
- rate limiting agar satu pengguna yang berisik tidak menghabiskan kuota Anda
- billing atau atribusi penggunaan jika chatbot menghadap pelanggan
- ringkasan latar belakang jika percakapan membutuhkan memori jangka panjang
Itulah mengapa gateway terpadu membantu. Setelah Anda melakukan migrasi base URL, perubahan model tidak lagi menjadi penulisan ulang platform dan menjadi sekadar konfigurasi.
Smoke Test
uvicorn main:app --reload --port 8000
curl -N -X POST http://localhost:8000/chat/stream \
-H "Content-Type: application/json" \
-d '{"message": "Hello!", "model": "gpt-4.1-mini"}'
Jika Anda dapat melakukan streaming satu giliran, mempertahankan satu percakapan, dan mengembalikan error yang bersih pada kegagalan yang dipaksakan, Anda memiliki fondasi yang tepat.
Estimasi Biaya
Buat API key di LemonData, arahkan OpenAI SDK Anda ke https://api.lemondata.cc/v1, dan Anda dapat merilis versi produksi pertama chatbot Anda tanpa mengelola akun provider yang terpisah.
| Model | Biaya Harian | Biaya Bulanan |
|---|---|---|
| GPT-4.1-mini | ~$2.40 | ~$72 |
| GPT-4.1 | ~$12.00 | ~$360 |
| Claude Sonnet 4.6 | ~$18.00 | ~$540 |
| DeepSeek V3 | ~$1.68 | ~$50 |
Menggunakan GPT-4.1-mini untuk sebagian besar percakapan dan upgrade ke Claude Sonnet 4.6 hanya saat pengguna memintanya menjaga biaya di bawah $100/bulan untuk sebagian besar aplikasi.
Dapatkan API key Anda: lemondata.cc menyediakan 300+ model melalui satu endpoint. Kredit gratis $1 untuk mulai membangun.