Mengintegrasikan model bahasa besar (LLM) ke dalam arsitektur asynchronous worker queue membawa risiko integritas data yang signifikan. Masalah utama muncul ketika worker mengeksekusi inferensi AI dan langsung menulis payload ke shared cache (seperti Redis) dengan Time-To-Live (TTL) panjang tanpa validasi ketat. Ketika LLM menghasilkan halusinasi—seperti identifier fiktif, angka di luar batas kewajaran, atau struktur JSON yang rusak—cache produksi teracuni (cache poisoning). Layanan hilir (downstream services) yang mengonsumsi cache tersebut akan menyebarkan data korup ke seluruh ekosistem aplikasi.

Anatomi Masalah: Dari Halusinasi Menjadi Cache Poisoning

LLM bersifat probabilistik, sedangkan sistem downstream (database transaksional, sistem akuntansi, pipeline analitik) bersifat deterministik. Pola naif yang sering ditemukan di sistem produksi adalah sebagai berikut:

  1. Worker mengambil task dari queue (misal: ekstraksi laporan keuangan).
  2. Worker memanggil API model AI.
  3. Worker langsung mengeksekusi SET cache_key model_output EX 86400.

Pola ini membuka celah fatal. Kasus nyata seperti anomali data pada sistem audit korporat terjadi saat model menghasilkan entitas bisnis yang tidak pernah terdaftar di database primer. Jika payload halusinasi masuk ke cache dengan TTL 24 jam, sistem reporting akan membaca data palsu tersebut secara berulang tanpa menyentuh database verifikasi. Akibatnya, integritas cache rusak dan perbaikannya memerlukan cache invalidation massal yang memicu cache stampede.

Pola Arsitektur: Validation Gate & Two-Phase Write

Untuk mencegah data halusinasi masuk ke cache publik, pisahkan proses komputasi AI dari fase publikasi cache menggunakan pendekatan Two-Phase Write yang ditengahi oleh Validation Gate.

1. Phase 1: Staging Cache

Payload mentah dari inferensi AI tidak boleh langsung ditulis ke production key. Tuliskan output ke staging key sementara dengan TTL sangat singkat (misalnya 60 detik) atau tahan di memori lokal worker. Staging key ini diisolasi dari konsumsi service publik.

2. Validation Gate: Verifikasi Deterministik

Sebelum data dipromosikan, jalankan tiga lapis verifikasi deterministik:

  • Structural Validation (JSON Schema): Memastikan seluruh tipe data, struktur key, dan format string sesuai kontrak tanpa ada field tambahan yang dihalusinasikan.
  • Domain Range Check: Memastikan nilai numerik berada dalam batas logika bisnis (misalnya rasio pajak berada antara 0.0 hingga 1.0, nilai transaksi tidak bernilai negatif).
  • Referential Integrity Check: Melakukan lookup langsung ke database primer (PostgreSQL/MySQL/Read Replica) untuk memvalidasi keberadaan foreign key, ID entitas, atau referensi relasional yang diklaim oleh LLM.

3. Phase 2: Atomic Promotion

Jika validasi lolos, data dipindahkan secara atomik dari staging ke production key. Jika validasi gagal, fase promosi dibatalkan seketika.

Penanganan Anomali: Quarantine Queue (DLQ)

Payload yang gagal validasi tidak boleh sekadar di-drop atau memicu retry tak terbatas. Worker yang mencoba mengeksekusi ulang prompt yang sama ke LLM kemungkinan besar akan menghasilkan halusinasi serupa dan membuang kuota token komputasi.

Arahkan payload gagal ke Quarantine Queue (atau Dead Letter Queue khusus AI). Tujuannya adalah:

  • Menyediakan jejak audit (audit trail) untuk analisis deviasi output LLM.
  • Mengumpulkan sampel prompt untuk fine-tuning atau evaluasi few-shot prompt engineering.
  • Memberikan sinyal telemetri/alerting ke tim engineer jika rasio halusinasi melonjak drastis setelah pembaruan model.

Implementasi: Atomic Promotion via Redis

Untuk menghindari race condition saat validasi selesai, promosi cache harus atomik. Di Redis, operasi ini paling efisien dilakukan menggunakan perintah RENAME atau kombinasi transaksi MULTI/EXEC.

Perintah RENAME staging_key prod_key bersifat atomik di Redis: key produksi akan langsung digantikan oleh key staging tanpa jeda waktu di mana key bernilai kosong atau korup terbaca oleh client lain.

Contoh Kode Worker Python

Berikut implementasi minimalis worker queue yang mengamankan penulisan cache dari output AI menggunakan validasi Pydantic, referential check sederhana, dan promosi Redis atomik.

import json
import redis
from pydantic import BaseModel, Field, ValidationError

r = redis.Redis(host='localhost', port=6379, db=0, decode_responses=True)

# 1. Kontrak Skema Deterministik
class AIAnalysisPayload(BaseModel):
    account_id: int
    confidence_score: float = Field(ge=0.0, le=1.0)
    anomaly_detected: bool
    risk_level: str = Field(pattern="^(LOW|MEDIUM|HIGH)$")

def mock_database_account_exists(account_id: int) -> bool:
    # ponytail: stub DB check; ganti dengan query SQL SELECT 1 ke DB primer
    valid_accounts = {1001, 1002, 1003, 1004}
    return account_id in valid_accounts

def quarantine_payload(task_id: str, raw_data: str, reason: str):
    quarantine_entry = {
        "task_id": task_id,
        "raw_payload": raw_data,
        "failure_reason": reason
    }
    r.rpush("queue:ai:quarantine", json.dumps(quarantine_entry))

def process_ai_task(task_id: str, raw_ai_json: str):
    staging_key = f"staging:analysis:{task_id}"
    production_key = f"cache:analysis:{task_id}"

    # Phase 1: Tulis ke staging key dengan TTL pendek
    r.set(staging_key, raw_ai_json, ex=60)

    try:
        # Deterministic Check 1 & 2: JSON Schema & Range Checks
        parsed_data = json.loads(raw_ai_json)
        validated = AIAnalysisPayload(**parsed_data)

        # Deterministic Check 3: Referential Integrity
        if not mock_database_account_exists(validated.account_id):
            raise ValueError(f"Halusinasi terdeteksi: account_id {validated.account_id} tidak valid.")

    except (ValidationError, ValueError, json.JSONDecodeError) as err:
        # Validasi gagal: batalkan promosi, hapus staging, alihkan ke DLQ
        r.delete(staging_key)
        quarantine_payload(task_id, raw_ai_json, str(err))
        return False

    # Phase 2: Atomic Promotion via Redis Transaction
    pipe = r.pipeline()
    pipe.rename(staging_key, production_key)
    pipe.expire(production_key, 86400)  # TTL produksi 24 jam
    pipe.execute()

    return True

# Verifikasi Worker
if __name__ == "__main__":
    # Kasus 1: Output Halusinasi (account_id tidak terdaftar di DB)
    hallucinated_output = json.dumps({
        "account_id": 9999,
        "confidence_score": 0.95,
        "anomaly_detected": False,
        "risk_level": "LOW"
    })
    success = process_ai_task("task_01", hallucinated_output)
    assert success is False
    assert r.exists("cache:analysis:task_01") == 0
    assert r.llen("queue:ai:quarantine") == 1

    # Kasus 2: Output Valid
    valid_output = json.dumps({
        "account_id": 1001,
        "confidence_score": 0.88,
        "anomaly_detected": True,
        "risk_level": "HIGH"
    })
    success = process_ai_task("task_02", valid_output)
    assert success is True
    assert r.exists("cache:analysis:task_02") == 1

    print("Self-check passed: Cache terlindungi dari halusinasi.")

Trade-offs dan Pertimbangan Sistem

Menerapkan validation gate pada background worker menghasilkan beberapa konsekuensi teknis yang harus diperhitungkan:

  • Latensi Worker Tambahan: Melakukan referential check ke database menambah I/O load. Pastikan pengecekan ID memanfaatkan query index yang optimal atau dibaca dari Redis Bloom Filter / local cache entitas primer.
  • Staging Key Leakage: Jika worker mengalami crash tepat sebelum fase validasi, staging key akan tertinggal. Tetapkan TTL staging yang sangat pendek (30-60 detik) agar Redis mengosongkan memori secara otomatis.
  • Penanganan DLQ: Tim engineering harus memiliki mekanisme metrik (Prometheus/Grafana) untuk memantau laju pertumbuhan Quarantine Queue agar antrean tidak menyebabkan memory exhaustion pada instans broker.