Biar RAG nggak kena 504: pisahin jalur query dan ingestion

Bikin RAG di masa hackathon itu terasa instan. Satu container FastAPI, terima file PDF, ekstrak teks, bikin embedding, simpan ke vector database, semua dalam satu request HTTP yang sama. Pamer ke teman, keliatan keren.

Begitu aplikasinya ketemu pengguna beneran, masalahnya datang dari arah yang di demo belum kelihatan. Ada yang upload laporan keuangan 200 halaman penuh tabel. Parsing, chunking, dan panggilan API embedding-nya makan puluhan detik. Load balancer menyerah duluan dengan 504 Gateway Timeout. Yang lebih nyebelin lagi: proses parsing yang rakus CPU menyandera thread server. Kueri chat pengguna lain ikut macet.

Terus gimana cara bikin RAG yang tahan banting, tanpa harus nyewa klaster Kubernetes yang ribet dan mahal? Jawabannya bukan mesin yang lebih besar, tapi memisahkan dua beban yang sifatnya memang beda.

Dua jalur yang sifatnya kebalikan

Di dalam sistem RAG ada dua beban kerja yang berlawanan karakternya:

• Jalur kueri: pengguna bertanya, kueri di-embed (di bawah 100 ms), vector search (di bawah 50 ms), LLM men-streaming jawaban. Sifatnya I/O-bound dan minta latensi serendah mungkin.
• Jalur ingestion: pengguna upload dokumen, parsing layout, chunking, ratusan embedding dibikin, masuk index. Sifatnya CPU-bound, makan detik sampai menit.

Kalau dua jalur ini berbagi CPU di service yang sama, uptime-mu diserahkan ke pengguna yang paling hobi upload file gede. Aturan emasnya satu: jalur ingestion wajib dipisah jadi worker asinkron berbasis queue.

Kenapa Cloud Run dan Cloud Tasks

Banyak tim langsung pasang Kafka atau RabbitMQ di VM sendiri buat ngatur background worker. Sah-sah aja sih, tapi buat tim kecil itu beban operasionalnya kebablasan. Duet serverless di GCP menyelesaikan masalah yang sama tanpa server yang harus dirawat:

• Cloud Run: jalanin container apa pun, scale-to-zero saat nggak ada trafik (saat sepi, biayanya nol 💸), dan bisa scale-out ke ratusan instance otomatis waktu queue membludak 🚀
• Cloud Tasks: queue yang dikelola penuh, dengan rate limiting bawaan (misalnya maksimal 10 task paralel biar kuota TPM/RPM provider embedding nggak jebol), retry dengan exponential backoff (bawaannya max_attempts 100, backoff sampai 3600 detik), dan deduplikasi task. Yang terakhir ini ada syaratnya, nanti kita bahas.

Perjalanan satu dokumen, dari upload sampai siap

Begini siklus hidup sebuah dokumen kalau kita telusuri di RAG yang sudah matang:

Pertama, upload langsung ke storage. File 50 MB nggak lewat backend. Backend cuma bikin GCS signed URL, lalu browser upload file binernya langsung ke bucket. Bandwidth dan memori server API aman.

Kedua, enqueue lalu jawab cepat. Setelah file mendarat di GCS, backend membuat task ke Cloud Tasks dengan payload metadata (document_id, gcs_uri). Lalu dia langsung membalas HTTP 202 Accepted ke pengguna. Kueri chat pengguna lain nggak kena imbas apa pun.

Ketiga, worker jalan di container terpisah. Cloud Tasks mengirim HTTP POST ke endpoint worker di service Cloud Run yang lain. Langkahnya satu chain: unduh file, parsing, chunking, batch embedding. Lalu tulis ke vector store (Qdrant, Vertex AI Vector Search, atau pgvector), update status dokumen jadi READY, dan kabari pengguna lewat webhook atau websocket.

Kodenya, bagian yang bisa kupastikan jalan

Sekarang kita lihat kodenya. Ini bagian enqueue-nya, pakai google-cloud-tasks 2.26.0:

from google.cloud import tasks_v2
import json
import time
from fastapi import FastAPI, status

app = FastAPI()
tasks_client = tasks_v2.CloudTasksClient()

PROJECT_ID = "my-gcp-ai-project"
LOCATION_ID = "asia-southeast2"  # region Jakarta
QUEUE_ID = "document-ingestion-queue"

queue_path = tasks_client.queue_path(PROJECT_ID, LOCATION_ID, QUEUE_ID)

@app.post("/api/v1/documents/process", status_code=status.HTTP_202_ACCEPTED)
async def enqueue_document_processing(doc_id: str, gcs_uri: str):
    # 1. Siapkan payload task
    payload = {
        "document_id": doc_id,
        "gcs_uri": gcs_uri
    }

    # 2. Buat task HTTP target ke worker Cloud Run
    task = {
        # Nama eksplisit = deduplikasi: nama sama ditolak ALREADY_EXISTS
        "name": f"{queue_path}/tasks/{doc_id}-{int(time.time() // 60)}",
        "http_request": {
            "http_method": tasks_v2.HttpMethod.POST,
            "url": "https://rag-worker-xyz-as.a.run.app/internal/tasks/ingest",
            "headers": {"Content-Type": "application/json"},
            "body": json.dumps(payload).encode(),
            # OIDC token untuk otentikasi aman antar-layanan di GCP!
            "oidc_token": {
                "service_account_email": "tasks-invoker@my-gcp-ai-project.iam.gserviceaccount.com"
            }
        }
    }

    # 3. Masukkan task ke queue Cloud Tasks
    response = tasks_client.create_task(request={"parent": queue_path, "task": task})

    return {
        "status": "queued",
        "message": "Dokumen berhasil dimasukkan ke antrean pemrosesan asinkron.",
        "task_id": response.name
    }

Yang bisa kupastikan tanpa kredensial GCP: bentuk request-nya valid. queue_path kebentuk projects/my-gcp-ai-project/locations/asia-southeast2/queues/document-ingestion-queue, dan objek task-nya terserialisasi benar jadi httpMethod: POST, oidcToken dengan service account yang benar, dan body JSON yang utuh. create_task yang beneran mengirim ke queue butuh kredensial, jadi bagian itu sengaja nggak kujalankan di sini.

Dua catatan tambahan buat kode ini. Pertama, oidc_token di task itu tuh yang bikin worker Cloud Run boleh diproteksi, selama service account-nya pegang peran run.invoker. Tapi ingress internal ada syaratnya: Cloud Tasks baru dihitung internal kalau project-nya sama (atau satu perimeter VPC-SC) dan kamu pakai URL bawaan run.app. Kalau URL bawaannya kamu matikan atau kamu pakai custom domain, pemanggilannya berhenti jalan. Jadi buat worker privat, yang jadi tumpuan itu wajib autentikasinya, bukan cuma ingress internal-nya. Kedua, Cloud Tasks menganggap task berhasil kalau worker membalas 2xx sebelum timeout bawaan 10 menit (maksimal bisa dinaikkan ke 30). Kalau parsing PDF 200 halaman-mu lebih lama dari itu, task-nya dianggap gagal. Dia di-retry, padahal pekerjaannya belum tentu salah. Angka timeout ini wajib kamu sesuaikan dengan beban parsing-mu.

Dua detail yang nggak kelihatan di diagram

Diagram arsitektur selalu kelihatan rapi, tapi dua hal ini nih yang biasanya baru ketahuan setelah tumbang:

Cloud Tasks itu at-least-once, bukan exactly-once. Dokumentasi resminya bilang terus terang: dalam keadaan langka, satu task bisa dieksekusi lebih dari sekali, dan kamu sendiri yang harus memastikan pengulangan itu aman. Artinya kamu wajib bikin worker-nya idempoten: task yang sama masuk dua kali, hasilnya harus tetap sama. Ini prinsip yang sama dengan checkpoint idempoten di sistem orkestrasi agent, cuma beda konteks.

Deduplikasi cuma berlaku untuk task bernama eksplisit. Tanpa nama, task yang sama bisa kekirim dua kali kalau sisi pemanggil me-retry create_task. Dengan nama, Cloud Tasks menolak duplikatnya dengan error ALREADY_EXISTS. Tapi ada jebakannya. Nama task diingat sampai 24 jam setelah task-nya selesai. Bahkan 9 hari kalau queue-nya dibikin dari queue.yaml. Jadi kalau namanya cuma document_id, pengguna yang upload ulang dokumen yang sama dalam 24 jam bakal ditolak, bukan di-dedup. Dari luar kelihatannya seperti queue-nya rusak. Karena itu di kode di atas namanya {document_id}-{epoch_menit}: upload yang sama tetap ke-dedup dalam menit yang sama, tapi upload ulang di hari berikutnya tetap jalan.

Cold start: satu angka per service

Tantangan khas serverless itu cold start, apalagi kalau container-nya memuat pustaka parsing yang berat. Obatnya beda-beda per service:

• Layanan kueri: pasang min-instances: 1. Selalu ada satu instance hangat, jadi pengguna pertama nggak pernah kena jeda beberapa detik. Konsekuensinya instance idle itu tetap dibayar dengan tarif idle yang lebih murah (di mode request-based billing, yang jadi bawaan; di mode instance-based kamu bayar tarif penuh selama instance-nya hidup).
• Worker ingestion: biarkan di min-instances: 0. Nggak ada pengguna yang nunggu di layar, jadi aman scale-to-zero, dan biayanya nol waktu sepi.
• Image-nya dirampingkan: multi-stage build, cache pip nggak ikut, base image slim (misalnya python:3.13-slim), supaya waktu narik image-nya tetap sebentar.

Yang bikin RAG kuat bukan modelnya

Sistem RAG yang bertahan di produksi bukan yang LLM-nya paling gede, tapi yang disiplin soal siapa boleh pakai CPU kapan. Pisah jalur cepat dan jalur berat, percayakan queue-nya ke layanan yang emang dibuat buat itu, dan biaya yang kamu bayar cuma komputasi pas dokumen beneran diproses.

Kalau kueri chat-mu mulai lemot tiap ada yang upload dokumen, yang disalahkan bukan model embedding-nya, tapi arsitekturnya. Pisah jalurnya dulu, sisanya tinggal urusan angka. Semoga 504-nya pamit 😄