Edge Case Integrasi AI: Pola Fallback Rule-Based dan Idempotency

Mengandalkan output Large Language Model (LLM) atau sistem agentic pada jalur kritis transaksi backend memperkenalkan risiko nondeterministik: latensi tinggi, parsing error, halusinasi skema, dan timeout. Masalah ini serupa dengan kasus otomasi Ford, di mana sistem berbasis model probabilistik gagal menangani kasus batas kritis sehingga perusahaan harus kembali mengintegrasikan heuristik deterministik dari para insinyur veteran. Untuk menjaga keandalan sistem produksi, integrasi AI memerlukan arsitektur isolasi kegagalan berbasis pola fallback rule-based dan jaminan idempotency.

Anatomi Kegagalan Integrasi AI pada API Gateway

Layanan inferensi AI memiliki profil kegagalan yang berbeda dari dependensi microservices standar:

• Latensi Berekor Panjang (Long-tail Latency): Inferensi token dapat mengalami lonjakan waktu respons mendadak di bawah beban antrean GPU.
• Validasi Skema Gagal: Mode structured output dapat menghasilkan JSON parsial atau field penting bernilai null saat model terpotong batas token (max tokens).
• Degradasi Semantik: Respons valid secara sintaksis tetapi melanggar batas aturan bisnis (misal: memberikan diskon melebihi pagu margin).

Pola Arsitektur: Circuit Breaker dan Dynamic Timeout

Mengintegrasikan model AI tanpa isolasi dapat menghabiskan connection pool backend. Implementasikan dynamic timeout yang disesuaikan dengan volume payload (estimasi token input) dan pasang Circuit Breaker untuk memutus rute ke AI service saat tingkat kegagalan melampaui batas ambang.

import time
from typing import Dict, Any, Optional

class AIServiceException(Exception):
    pass

class CircuitBreakerOpenException(Exception):
    pass

class SimpleCircuitBreaker:
    def __init__(self, failure_threshold: int = 3, recovery_time: float = 30.0):
        self.failure_threshold = failure_threshold
        self.recovery_time = recovery_time
        self.failure_count = 0
        self.last_failure_time = 0.0
        self.state = "CLOSED"

    def can_execute(self) -> bool:
        if self.state == "OPEN":
            if time.time() - self.last_failure_time > self.recovery_time:
                self.state = "HALF-OPEN"
                return True
            return False
        return True

    def record_success(self):
        self.failure_count = 0
        self.state = "CLOSED"

    def record_failure(self):
        self.failure_count += 1
        self.last_failure_time = time.time()
        if self.failure_count >= self.failure_threshold:
            self.state = "OPEN"

Menjamin Idempotency Saat Transisi Rute Fallback

Saat request AI mengalami timeout pada sisi client, ada kemungkinan model sebenarnya berhasil menyelesaikan komputasi di remote provider. Mengalihkan request langsung ke rule engine tanpa kontrol konkurensi berisiko menimbulkan mutasi ganda pada resource yang sama (double write).

Solusinya adalah memetakan transaksi menggunakan Idempotency-Key berbasis state machine di penyimpanan data terdistribusi (seperti Redis) dengan atomisitas CAS (Compare-And-Swap) atau distributed lock.

Status State Machine Idempotency

1. PENDING: Kunci dikunci oleh prosesor AI. Operasi sedang berlangsung.
2. FALLBACK_ROUTED: Panggilan AI timeout/gagal; rute dialihkan ke rule engine veteran. Status diperbarui dengan aman.
3. COMPLETED: Mutasi selesai dieksekusi, payload tersimpan bersama cache respons.

Implementasi: Fallback Rule-Based dengan Idempotency Control

Contoh berikut mendemonstrasikan orkestrasi panggilan inferensi AI dengan penanganan timeout, pemutusan sirkuit, dan fallback deterministik rule-based menggunakan locking berbasis idempotency key.

import json
import uuid

# Simulasi Redis cache/lock store
STATE_STORE: Dict[str, Dict[str, Any]] = {}

def acquire_idempotency_lock(key: str, owner_id: str) -> bool:
    entry = STATE_STORE.get(key)
    if entry and entry.get("status") in ["COMPLETED", "PROCESSING"]:
        return False
    STATE_STORE[key] = {"status": "PROCESSING", "owner": owner_id}
    return True

def release_or_update(key: str, status: str, result: Dict[str, Any]):
    STATE_STORE[key] = {"status": status, "data": result}

def legacy_rule_engine(payload: Dict[str, Any]) -> Dict[str, Any]:
    """Logika deterministik pengganti model jika AI timeout/invalid."""
    score = 0
    if payload.get("credit_score", 0) > 700:
        score += 50
    if payload.get("debt_to_income", 1.0) < 0.4:
        score += 50
    
    decision = "APPROVED" if score >= 70 else "MANUAL_REVIEW"
    return {"decision": decision, "risk_score": score, "engine": "deterministic_v1"}

def call_ai_service(payload: Dict[str, Any], timeout_seconds: float) -> Dict[str, Any]:
    # Simulasi interaksi eksternal LLM yang berpotensi error/timeout
    import random
    if payload.get("simulate_timeout"):
        raise TimeoutError("AI inference timeout limit reached")
    if random.choice([True, False]):
        raise ValueError("Schema validation error: missing JSON fields")
    return {"decision": "APPROVED", "risk_score": 85, "engine": "llm_agent_v2"}

def process_transaction(idempotency_key: str, payload: Dict[str, Any], cb: SimpleCircuitBreaker) -> Dict[str, Any]:
    existing = STATE_STORE.get(idempotency_key)
    if existing and existing.get("status") == "COMPLETED":
        return existing["data"]

    worker_id = str(uuid.uuid4())
    if not acquire_idempotency_lock(idempotency_key, worker_id):
        raise RuntimeError("Request concurrently processed or duplicate key.")

    # Dynamic timeout berdasarkan panjang input text (konsep token scale)
    base_timeout = 2.0
    text_length = len(str(payload))
    dynamic_timeout = base_timeout + (text_length / 1000.0)

    result = None
    use_fallback = False

    if cb.can_execute():
        try:
            result = call_ai_service(payload, timeout_seconds=dynamic_timeout)
            cb.record_success()
        except (TimeoutError, ValueError, AIServiceException) as err:
            cb.record_failure()
            use_fallback = True
    else:
        use_fallback = True

    if use_fallback:
        # Fallback ke rule-based deterministik (kasus batas Ford)
        result = legacy_rule_engine(payload)

    release_or_update(idempotency_key, "COMPLETED", result)
    return result

Strategi Retry dengan Exponential Backoff

Saat fallback terjadi karena timeout, do not retry AI calls indefinitely. Gunakan strategi retry terbatas hanya untuk respon HTTP transport tertentu (seperti rate limit 429 atau server reboot 503). Jika error berupa 400 Bad Request atau parsing exception, bypass retry langsung menuju fallback rule engine.

Peringatan Produksi: Selalu terapkan jitter acak pada interval backoff untuk menghindari thundering herd problem ke gateway LLM saat endpoint pulih dari kegagalan.

Trade-off dan Evaluasi

Pola fallback rule-based deterministik bukan tanda kemunduran teknologi, melainkan fondasi pertahanan sistem terdistribusi. Menjaga aturan deterministik tetap hidup berdampingan dengan skema idempotency yang ketat memastikan sistem backend tetap resilien meskipun AI service eksternal mengalami degradasi total.