Kalau Anda sudah membaca artikel FastAPI untuk aplikasi AI, bagian BackgroundTasks di sana cocok untuk tugas ringan seperti logging. Untuk tugas yang benar-benar lama — bisa puluhan detik sampai beberapa menit, seperti memproses dokumen besar lewat beberapa tahap RAG atau menjalankan alur agent multi-langkah — pola yang lebih tepat adalah job queue: endpoint langsung mengembalikan ID job begitu tugas diterima, worker terpisah yang mengerjakannya, dan endpoint lain untuk mengecek statusnya.

Kenapa Bukan Cukup Menunggu di Endpoint Saja

Pola Queue-Worker: Submit, Proses, Cek Status

import asyncio, enum, uuid, time

class StatusJob(str, enum.Enum):
    ANTRE = "antre"
    BERJALAN = "berjalan"
    SELESAI = "selesai"
    GAGAL = "gagal"

class Job:
    def __init__(self, tugas):
        self.id = uuid.uuid4().hex[:8]
        self.tugas = tugas
        self.status = StatusJob.ANTRE
        self.hasil = None
        self.error = None

class AntrianJob:
    def __init__(self, jumlah_worker=2):
        self.antrian = asyncio.Queue()
        self.job_map = {}
        self.jumlah_worker = jumlah_worker
        self._worker_tasks = []

    def submit(self, tugas) -> str:
        job = Job(tugas)
        self.job_map[job.id] = job
        self.antrian.put_nowait(job)
        return job.id

    def status_job(self, job_id):
        job = self.job_map.get(job_id)
        return None if job is None else {
            "id": job.id, "status": job.status.value,
            "hasil": job.hasil, "error": job.error,
        }

    async def _worker(self, nama_worker):
        while True:
            job = await self.antrian.get()
            job.status = StatusJob.BERJALAN
            try:
                job.hasil = await job.tugas()
                job.status = StatusJob.SELESAI
            except Exception as e:
                job.error = str(e)
                job.status = StatusJob.GAGAL
            finally:
                self.antrian.task_done()

    def mulai_worker(self):
        for i in range(self.jumlah_worker):
            self._worker_tasks.append(asyncio.create_task(self._worker(f"worker-{i}")))

Endpoint FastAPI di atasnya cukup sederhana: POST /proses-dokumen memanggil antrian.submit(tugas) dan langsung mengembalikan {"job_id": "..."}, sementara GET /status/{job_id} memanggil status_job() untuk polling dari sisi klien.

Empat Hal yang Diuji dan Terbukti Benar

Alur Kerja dari Sisi Klien

1. Klien mengirim POST /proses-dokumen, langsung menerima {"job_id": "a1b2c3d4"} dalam hitungan milidetik.
2. Klien melakukan polling ke GET /status/a1b2c3d4 setiap beberapa detik untuk mengecek status.
3. Selama status antre atau berjalan, UI bisa menampilkan progres tanpa menahan koneksi tunggal yang lama.
4. Begitu status jadi selesai atau gagal, klien mengambil hasil (atau pesan error) dan berhenti polling.

Batasan In-Memory Queue — dan Kapan Perlu yang Lebih Serius

Pertanyaan yang Sering Muncul

Kenapa tidak pakai BackgroundTasks FastAPI saja untuk semua kasus?

BackgroundTasks tidak memberi ID job atau cara mengecek status/progres dari luar — cocok untuk tugas “kirim lalu lupakan” seperti logging, tapi tidak untuk tugas yang hasilnya perlu diambil klien nanti. Job queue dengan status memberi visibilitas yang BackgroundTasks tidak punya secara bawaan.

Berapa jumlah worker yang ideal?

Tergantung sifat tugasnya. Untuk tugas yang didominasi menunggu jaringan (memanggil API model), jumlah worker bisa lebih banyak dari jumlah CPU core karena sebagian besar waktu dihabiskan menunggu, bukan menghitung. Untuk tugas yang berat secara komputasi lokal, jumlah worker yang terlalu banyak justru saling berebut CPU.

Bagaimana kalau pengguna perlu diberi tahu tanpa polling terus-menerus?

WebSocket atau server-sent events bisa menggantikan polling untuk push notifikasi saat status berubah, tapi menambah kompleksitas koneksi yang harus dikelola. Untuk kebanyakan kasus, polling setiap beberapa detik sudah cukup dan jauh lebih sederhana diimplementasikan dan di-debug.

Apakah job yang gagal bisa dicoba ulang otomatis?

Bisa ditambahkan sebagai lapisan tambahan — worker mengecek jumlah percobaan sebelum menandai job sebagai gagal permanen, dan mengembalikannya ke antrian kalau masih di bawah batas. Pola ini serupa dengan retry-backoff yang dibahas di artikel skill backend untuk AI engineering, hanya diterapkan di level job, bukan di level satu panggilan API.

Baca Juga:

Webhook untuk Hasil AI Async: Signature, Retry, Idempotensi

Kesimpulan

Job queue memisahkan “menerima tugas” dari “mengerjakan tugas” — pemisahan yang jadi penting begitu tugas AI Anda cukup lama untuk menabrak timeout HTTP atau butuh ditampilkan progresnya. Pengujian di atas menunjukkan bagian yang paling sering luput: tanpa penanganan error yang benar di dalam loop worker, satu job yang gagal bisa mematikan seluruh worker dan menghentikan semua job setelahnya tanpa peringatan. Pola in-memory di artikel ini cukup untuk memulai, tapi begitu aplikasi butuh job yang tetap ada meski proses restart atau berjalan di banyak instance, saatnya beralih ke queue yang persisten seperti Redis.

Artikel terkait: skill backend untuk AI engineering.