Bagian VI - Bab 19

Optimasi Performa dan Biaya

Peta sumber pemborosan di sistem AI, caching berlapis (exact, semantic, Redis), dan model routing untuk efisiensi biaya.

19.1 Peta Sumber Pemborosan di Sistem AI

Sebelum mengoptimalkan, penting memahami di mana biaya sebenarnya berasal. Sebagian besar pemborosan di sistem AI produksi berasal dari empat sumber yang bisa diatasi secara teknis.

19.2 Caching: Optimasi Paling Cepat Diimplementasikan

Caching adalah teknik dengan rasio effort-to-impact terbaik. Untuk sistem RAG dengan pola pertanyaan yang berulang (FAQ, dokumentasi produk, customer service), caching bisa memotong biaya LLM hingga 60-80% dengan implementasi yang relatif sederhana.

Ada dua level caching yang relevan untuk sistem AI: exact cache (respons disimpan berdasarkan teks query yang identik persis) dan semantic cache (respons disimpan berdasarkan kemiripan semantik, query "cara retur barang" dan "bagaimana pengembalian produk" akan menghasilkan jawaban yang sama dari cache).

Exact Cache (In-Memory)

Menggunakan dictionary Python paling sederhana, hanya cocok untuk query yang identik persis. Berguna sebagai langkah pertama sebelum semantic cache.

import hashlib
import time
from functools import wraps

class ExactCache:
    """
    In-memory exact cache dengan TTL (Time To Live).
    Cache ini hilang saat server restart — gunakan Redis
    untuk persistensi antar restart.
    """
    def __init__(self, ttl_detik: int = 3600):
        self._cache: dict = {}
        self.ttl = ttl_detik

    def _key(self, teks: str) -> str:
        # Hash query agar key pendek dan konsisten
        return hashlib.md5(teks.lower().strip().encode()).hexdigest()

    def get(self, query: str):
        key = self._key(query)
        if key in self._cache:
            nilai, timestamp = self._cache[key]
            if time.time() - timestamp < self.ttl:
                return nilai  # cache hit
            del self._cache[key]  # expired
        return None  # cache miss

    def set(self, query: str, nilai) -> None:
        key = self._key(query)
        self._cache[key] = (nilai, time.time())

# Integrasikan ke pipeline RAG
cache = ExactCache(ttl_detik=3600)  # cache 1 jam

def tanya_dengan_cache(query: str) -> dict:
    # Cek cache dulu sebelum memanggil LLM
    cached = cache.get(query)
    if cached:
        print(f"Cache hit: {query[:40]}")
        return {**cached, "dari_cache": True}

    # Cache miss — panggil pipeline normal
    hasil = tanya(query)
    cache.set(query, hasil)
    return {**hasil, "dari_cache": False}

Semantic Cache

Menggunakan embedding dan cosine similarity untuk menemukan jawaban yang sudah tersimpan meski query berbeda kata. Menangkap jauh lebih banyak cache hit dari exact cache.

import numpy as np
from sentence_transformers import SentenceTransformer

class SemanticCache:
    """
    Cache berbasis kemiripan semantik.
    Query yang berbeda kata tapi bermakna sama
    akan mendapat jawaban dari cache yang sama.
    """
    def __init__(self, threshold: float = 0.92, ttl_detik: int = 3600):
        self.model     = SentenceTransformer("paraphrase-multilingual-MiniLM-L12-v2")
        self.threshold = threshold  # ketat: 0.92, longgar: 0.85
        self.ttl       = ttl_detik
        self._entries: list = []    # [(embedding, nilai, timestamp)]

    def get(self, query: str):
        if not self._entries:
            return None

        emb_query = self.model.encode(query, normalize_embeddings=True)
        sekarang  = time.time()

        skor_maks = -1
        kandidat  = None

        for emb, nilai, ts in self._entries:
            if sekarang - ts > self.ttl:
                continue
            skor = float(np.dot(emb_query, emb))
            if skor > skor_maks:
                skor_maks = skor
                kandidat  = nilai

        if skor_maks >= self.threshold:
            print(f"Semantic cache hit (skor: {skor_maks:.3f})")
            return kandidat
        return None

    def set(self, query: str, nilai) -> None:
        emb = self.model.encode(query, normalize_embeddings=True)
        self._entries.append((emb, nilai, time.time()))

# Gunakan persis seperti ExactCache
sem_cache = SemanticCache(threshold=0.92)

def tanya_semantic_cache(query: str) -> dict:
    cached = sem_cache.get(query)
    if cached:
        return {**cached, "dari_cache": True}
    hasil = tanya(query)
    sem_cache.set(query, hasil)
    return {**hasil, "dari_cache": False}

Cache dengan Redis

Cache berbasis Redis, persisten antar restart server, bisa dibagi antar instance, dan mendukung TTL secara native. Ini adalah pendekatan yang tepat untuk produksi dengan lebih dari satu container.

# pip install redis
import redis
import json
import hashlib

# Inisialisasi Redis client
# Di produksi, gunakan Redis Cloud atau Railway Redis
r = redis.Redis(
    host=os.getenv("REDIS_HOST", "localhost"),
    port=int(os.getenv("REDIS_PORT", 6379)),
    password=os.getenv("REDIS_PASSWORD"),
    decode_responses=True
)

def cache_key(query: str) -> str:
    return f"rag:cache:{hashlib.md5(query.lower().strip().encode()).hexdigest()}"

def tanya_redis_cache(query: str, ttl: int = 3600) -> dict:
    key = cache_key(query)

    # Cek Redis
    cached_str = r.get(key)
    if cached_str:
        print(f"Redis cache hit: {query[:40]}")
        return {**json.loads(cached_str), "dari_cache": True}

    # Cache miss
    hasil = tanya(query)

    # Simpan ke Redis dengan TTL
    r.setex(key, ttl, json.dumps(hasil, ensure_ascii=False))
    return {**hasil, "dari_cache": False}

# Pantau efektivitas cache
def cache_stats() -> dict:
    keys = r.keys("rag:cache:*")
    return {
        "total_cache_entries": len(keys),
        "memory_usage_mb": r.info("memory")["used_memory_human"]
    }

19.3 Model Routing: Model yang Tepat untuk Tugas yang Tepat

Tidak semua pertanyaan membutuhkan model yang sama. Menggunakan GPT-4o untuk semua request termasuk FAQ sederhana adalah pemborosan yang bisa dihindari dengan model routing: sebuah layer tipis yang menentukan model mana yang paling tepat untuk setiap jenis pertanyaan.

Implementasi model routing: router mengklasifikasikan kompleksitas query menggunakan model ringan, lalu mengarahkan ke model yang tepat. Biaya routing (Flash) jauh lebih kecil dari penghematan yang dihasilkan.

import google.generativeai as genai
from enum import Enum

class Kompleksitas(Enum):
    SEDERHANA = "sederhana"  # FAQ, salam, pertanyaan langsung
    SEDANG    = "sedang"     # analisis, multi-step ringan
    KOMPLEKS  = "kompleks"   # reasoning mendalam, dokumen panjang

# Mapping kompleksitas ke model
MODEL_MAP = {
    Kompleksitas.SEDERHANA: "gemini-1.5-flash",
    Kompleksitas.SEDANG:    "gpt-4o-mini",
    Kompleksitas.KOMPLEKS:  "gpt-4o",
}

def klasifikasi_kompleksitas(query: str) -> Kompleksitas:
    """
    Gunakan model ringan untuk mengklasifikasikan query.
    Biaya klasifikasi ini sangat kecil (~10 token output).
    """
    model_router = genai.GenerativeModel("gemini-1.5-flash")
    prompt = f"""
Klasifikasikan query berikut berdasarkan kompleksitas yang dibutuhkan:
- sederhana: FAQ, salam, pertanyaan dengan jawaban langsung
- sedang: analisis ringan, perbandingan, pertanyaan multi-bagian
- kompleks: reasoning mendalam, analisis dokumen panjang, keputusan kritis

Query: "{query}"

Balas HANYA dengan satu kata: sederhana / sedang / kompleks
"""
    response = model_router.generate_content(prompt)
    hasil = response.text.strip().lower()

    if "sederhana" in hasil:
        return Kompleksitas.SEDERHANA
    elif "kompleks" in hasil:
        return Kompleksitas.KOMPLEKS
    return Kompleksitas.SEDANG
def tanya_dengan_routing(query: str) -> dict:
    """Pipeline RAG dengan model routing."""
    # Langkah 1: Klasifikasikan kompleksitas
    level = klasifikasi_kompleksitas(query)
    model_dipilih = MODEL_MAP[level]

    print(f"Query: '{query[:40]}' → {level.value} → {model_dipilih}")

    # Langkah 2: Retrieve chunks (sama untuk semua model)
    chunks = retrieve(query)

    # Langkah 3: Generate dengan model yang sesuai
    jawaban = generate(query, chunks, model=model_dipilih)

    return {
        "jawaban":        jawaban,
        "model_digunakan": model_dipilih,
        "kompleksitas":    level.value
    }

Kesalahan Umum

Ringkasan Bab 19