Mengelola sistem Retrieval-Augmented Generation (RAG) untuk korpus teks statis berukuran masif—seperti naskah literatur kuno, kanon hukum, atau teks teologi historis (seperti pada crosscanon.com)—memiliki karakteristik unik: dokumen referensi jarang atau tidak pernah berubah, namun beban query similarity search dan biaya komputasi embedding sangat tinggi. Mengabaikan optimalisasi pada arsitektur ini memicu bottleneck I/O pada database vektor dan lonjakan biaya inference model embedding.

Dua komponen utama untuk menstabilkan throughput sistem ini adalah worker queue untuk menangani chunking dan indexing batch secara decoupled, serta semantic caching layer berbasis Redis untuk mengeliminasi komputasi berulang dari query inference dan vector search.

Arsitektur Pemrosesan Asynchronous Korpus Statis

Memproses jutaan baris teks statis membutuhkan pipeline batching yang terisolasi dari proses web-serving utama. Memuat teks besar secara sinkron akan memicu timeout dan memory leak.

Pipeline pemrosesan dibagi menjadi tiga tahap via distributed message queue (misal: Celery atau BullMQ berbasis Redis):

  1. Ingestion & Chunking: Memecah kanon statis berdasarkan batas struktural semantik (pasal, ayat, paragraf naskah kuno), bukan sekadar token split kaku.
  2. Queue Dispatch: Menjadwalkan chunk ke dalam queue dengan payload minimal (hanya referensi chunk ID dan canonical hash).
  3. Vector Worker Execution: Worker mengambil batch, memanggil embedding model (secara lokal via TensorRT/ONNX atau remote endpoint), lalu menulis ke vector store (Milvus/Qdrant/pgvector).

Contoh Implementasi Enqueueing Batch Task (Python)

import hashlib
from celery import Celery

app = Celery('rag_tasks', broker='redis://localhost:6379/0')

def compute_chunk_hash(text: str, metadata: dict) -> str:
    payload = f"{text}:{sorted(metadata.items())}"
    return hashlib.sha256(payload.encode('utf-8')).hexdigest()

@app.task(bind=True, max_retries=3, default_retry_delay=10)
def process_corpus_chunk(self, chunk_id: str, text: str, metadata: dict):
    try:
        # ponytail: direct batch call without pooling; upgrade to local model worker pool if throughput degrades
        chunk_hash = compute_chunk_hash(text, metadata)
        
        # Eksekusi embedding model
        embedding = generate_embedding(text)
        
        # Simpan ke vector database bersama metadata kanonikal
        upsert_to_vector_db(
            vector_id=chunk_id,
            vector=embedding,
            metadata={**metadata, "sha256": chunk_hash}
        )
    except Exception as exc:
        raise self.retry(exc=exc)

Semantic Cache Menggunakan Redis Vector Search

Pada korpus teks referensi statis, pola query pengguna sering kali serupa atau berulang secara konseptual. Penggunaan semantic caching memungkinkan aplikasi mengembalikan context retrieval tanpa perlu mengeksekusi cosine similarity search ke database vektor primer.

Redis Stack mendukung indexing vektor (HNSW / Flat) langsung di RAM. Skema cache menyimpan embedding dari query yang pernah dieksekusi bersama pointer ke konteks chunk yang relevan.

Implementasi Semantic Query Caching

import redis
import numpy as np
from redis.commands.search.query import Query

r = redis.Redis(host='localhost', port=6379, db=1)
SIMILARITY_THRESHOLD = 0.95

def get_semantic_cached_context(query_vector: list[float]) -> str | None:
    query_bytes = np.array(query_vector, dtype=np.float32).tobytes()
    
    # Query HNSW index pada Redis
    q = (
        Query("*=>[KNN 1 @vector $vec AS score]")
        .sort_by("score")
        .return_fields("context", "score")
        .dialect(2)
    )
    results = r.ft("idx:query_cache").search(q, query_params={"vec": query_bytes})
    
    if results.docs:
        top_match = results.docs[0]
        # Redis return cosine distance: score = 1 - cosine_similarity
        similarity = 1.0 - float(top_match.score)
        if similarity >= SIMILARITY_THRESHOLD:
            return top_match.context
            
    return None

def set_semantic_cache(query_vector: list[float], context: str, ttl_seconds: int = 86400):
    query_bytes = np.array(query_vector, dtype=np.float32).tobytes()
    cache_id = f"cache:query:{hash(query_bytes)}"
    
    pipeline = r.pipeline()
    pipeline.hset(cache_id, mapping={
        "vector": query_bytes,
        "context": context
    })
    pipeline.expire(cache_id, ttl_seconds)
    pipeline.execute()

Distributed Locking pada Re-indexing Batch

Meskipun korpus kanonikal bersifat statis, operasi re-indexing tetap diperlukan ketika mengubah dimensi embedding model, chunking window size, atau metadata schema. Jika worker queue dijalankan pada banyak instans, risiko tumpang tindih komputasi (race condition) pada segmen kanonikal yang sama meningkat drastis.

Gunakan Redlock (distributed lock via Redis) untuk mengunci batch partisi naskah yang sedang diproses oleh worker tertentu:

import time
import uuid

def acquire_partition_lock(client: redis.Redis, partition_id: str, token: str, ttl_ms: int = 30000) -> bool:
    lock_key = f"lock:partition:{partition_id}"
    return bool(client.set(lock_key, token, nx=True, px=ttl_ms))

def release_partition_lock(client: redis.Redis, partition_id: str, token: str):
    lock_key = f"lock:partition:{partition_id}"
    # Lua script menjamin lock hanya dilepas oleh pemegang token sah
    lua_release = """
        if redis.call('get', KEYS[1]) == ARGV[1] then
            return redis.call('del', KEYS[1])
        else
            return 0
        end
    """
    client.eval(lua_release, 1, lock_key, token)

Mitigasi Cache Stampede

Cache stampede (thundering herd problem) terjadi saat TTL cache pada query populer habis secara simultan, menyebabkan ratusan request query paralel langsung membebani vector store dan embedding worker secara tiba-tiba.

Untuk korpus statis, dua strategi terbaik adalah:

  • Probabilistic Early Expiration (Algoritma XFetch): Menghitung probabilitas re-komputasi cache sebelum data kedaluwarsa secara aktual berdasarkan waktu eksekusi komputasi (compute delta) dan parameter varians ($eta$).
  • Mutex Lock Fetching: Jika cache miss terdeteksi, hanya satu thread/request yang diizinkan mengambil data baru ke primary store, sementara request lain menunggu atau membaca stale data sementara.

Implementasi Mutex pada Cache Miss

import time

def get_context_with_mutex(client: redis.Redis, query_key: str, fetch_fn, ttl: int = 3600) -> str:
    cached = client.get(query_key)
    if cached:
        return cached.decode('utf-8')

    lock_key = f"mutex:{query_key}"
    acquired = client.set(lock_key, "1", nx=True, ex=10)

    if acquired:
        try:
            # Re-fetch dan komputasi embedding
            fresh_data = fetch_fn()
            client.set(query_key, fresh_data, ex=ttl)
            return fresh_data
        finally:
            client.del(lock_key)
    else:
        # Tunggu fallback 100ms dan baca ulang dari cache yang sudah diisi request pertama
        time.sleep(0.1)
        fallback = client.get(query_key)
        return fallback.decode('utf-8') if fallback else fetch_fn()

Trade-off dan Limitasi

Pendekatan di atas memprioritaskan efisiensi resource dan determinisme data, tetapi memiliki sejumlah konsekuensi teknis:

  • Memory Footprint: Redis Vector Search memuat indeks ke dalam RAM. Menyimpan vektor embedding query dalam jumlah besar memerlukan perhitungan maxmemory dan alokasi swap yang ketat.
  • False Positive Semantic Match: Threshold cosine similarity yang terlalu rendah (< 0.90) pada semantic cache berisiko mengembalikan konteks naskah yang salah untuk query dengan nuansa bahasa yang halus. Lakukan tuning threshold secara empiris.
  • Network Latency Celery: Enqueueing chunk teks berukuran terlalu kecil (< 100 token) satu per satu menimbulkan overhead network I/O yang lebih tinggi dibanding proses embedding itu sendiri. Kelompokkan chunk dalam micro-batches (misal: 32–64 chunk per payload message) sebelum masuk ke broker.