Bagian III - Bab 13

Retrieval-Augmented Generation (RAG)

Mengapa RAG diperlukan, strategi chunking, membangun pipeline RAG end-to-end, evaluasi kualitas retrieval dan generation, serta pola halusinasi spesifik di RAG.

13.1 Mengapa RAG Diperlukan

LLM memiliki tiga keterbatasan mendasar yang membuat mereka tidak bisa langsung digunakan sebagai sistem jawab-pertanyaan untuk domain spesifik.

• Pengetahuan terbatas pada data training. LLM tidak tahu tentang dokumen internal perusahaanmu, kebijakan terbaru, atau data yang dibuat setelah tanggal cutoff training-nya.
• Context window terbatas. Kamu tidak bisa memasukkan seluruh basis pengetahuan ke dalam satu prompt, bahkan model dengan context window 1 juta token pun memiliki batas, dan biayanya akan sangat mahal.
• Halusinasi. Ketika model tidak tahu jawabannya, ia cenderung mengarang jawaban yang terdengar meyakinkan daripada mengaku tidak tahu.

RAG menyelesaikan ketiga masalah ini sekaligus: ambil hanya bagian dokumen yang relevan, sisipkan ke prompt, dan instruksikan model untuk menjawab hanya berdasarkan konteks yang diberikan.

13.2 Strategi Chunking

Chunking adalah langkah yang paling berdampak pada kualitas retrieval dan paling sering diremehkan. Ukuran dan cara memotong dokumen menentukan seberapa relevan chunk yang berhasil diambil untuk setiap query.

Panduan praktis menentukan nilai chunk_size dan chunk_overlap:

Jenis Dokumen	chunk_size	chunk_overlap	Alasan
FAQ, kebijakan pendek	300-500	50-100	Setiap FAQ biasanya sudah mandiri, chunk kecil meningkatkan presisi
Artikel, laporan paragraf panjang	700-1000	150-200	Paragraf panjang butuh chunk lebih besar agar konteks tidak terputus
Buku, dokumentasi teknis panjang	1000-1500	200-300	Struktur hierarki dalam perlu chunk lebih besar untuk konteks bab/subbab
Dokumen berisi tabel	Khusus	0	Jangan gunakan text splitter biasa, gunakan Unstructured atau camelot untuk ekstrak tabel utuh

13.3 Membangun Pipeline RAG Minimal

Berikut implementasi pipeline RAG lengkap yang bisa langsung dijalankan, menggabungkan semua komponen dari bab-bab sebelumnya menjadi satu sistem yang berfungsi.

Indexing Pipeline

Pipeline indexing: dari file PDF hingga tersimpan di vector database. Jalankan ini sekali saat setup, dan ulangi hanya ketika dokumen berubah.

import os
from pathlib import Path
from dotenv import load_dotenv
import pdfplumber
from langchain.text_splitter import RecursiveCharacterTextSplitter
from sentence_transformers import SentenceTransformer
from pinecone import Pinecone, ServerlessSpec

load_dotenv()

# Konfigurasi
EMBEDDING_MODEL = "paraphrase-multilingual-MiniLM-L12-v2"
INDEX_NAME = "rag-dokumen"
DIMENSION = 384
CHUNK_SIZE = 700
CHUNK_OVERLAP = 150

# Inisialisasi
model = SentenceTransformer(EMBEDDING_MODEL)
pc = Pinecone(api_key=os.getenv("PINECONE_API_KEY"))

if INDEX_NAME not in pc.list_indexes().names():
    pc.create_index(
        name=INDEX_NAME, dimension=DIMENSION, metric="cosine",
        spec=ServerlessSpec(cloud="aws", region="us-east-1")
    )
index = pc.Index(INDEX_NAME)

splitter = RecursiveCharacterTextSplitter(
    chunk_size=CHUNK_SIZE,
    chunk_overlap=CHUNK_OVERLAP,
    separators=["\n\n", "\n", ". ", " "]
)

def index_pdf(path_file: str) -> int:
    \"\"\"Ekstrak, chunk, embed, dan index satu file PDF.\"\"\"
    nama_file = Path(path_file).name
    halaman_teks = []

    with pdfplumber.open(path_file) as pdf:
        for i, hal in enumerate(pdf.pages, 1):
            teks = hal.extract_text()
            if teks and teks.strip():
                halaman_teks.append((i, teks.strip()))

    # Chunking per halaman agar metadata halaman akurat
    semua_chunks = []
    for nomor_hal, teks in halaman_teks:
        chunks = splitter.split_text(teks)
        for j, chunk in enumerate(chunks):
            semua_chunks.append({
                "id": f"{nama_file}-hal{nomor_hal}-chunk{j}",
                "teks": chunk,
                "sumber": nama_file,
                "halaman": nomor_hal
            })

    # Embedding dan upsert ke Pinecone (batch 100)
    for i in range(0, len(semua_chunks), 100):
        batch = semua_chunks[i:i+100]
        embeddings = model.encode([c["teks"] for c in batch]).tolist()
        vectors = [
            (c["id"], emb, {"teks": c["teks"], "sumber": c["sumber"], "halaman": c["halaman"]})
            for c, emb in zip(batch, embeddings)
        ]
        index.upsert(vectors=vectors)

    print(f"{nama_file}: {len(semua_chunks)} chunk diindeks")
    return len(semua_chunks)

Querying Pipeline

Fungsi retrieval: ubah query pengguna menjadi embedding dan ambil chunk paling relevan dari vector database.

def retrieve(query: str, top_k: int = 4, min_skor: float = 0.45) -> list[dict]:
    \"\"\"
    Ambil chunk paling relevan untuk query.

    Args:
        query: pertanyaan pengguna
        top_k: jumlah chunk yang diambil
        min_skor: threshold minimum cosine similarity (0-1)

    Returns:
        list of {"teks", "skor", "sumber", "halaman"}
    \"\"\"
    # Embed query dengan model yang SAMA seperti indexing
    vec_query = model.encode(query).tolist()

    hasil = index.query(
        vector=vec_query,
        top_k=top_k,
        include_metadata=True
    )

    chunk_relevan = []
    for match in hasil["matches"]:
        if match["score"] >= min_skor:
            chunk_relevan.append({
                "teks": match["metadata"]["teks"],
                "skor": round(match["score"], 3),
                "sumber": match["metadata"].get("sumber", "-"),
                "halaman": match["metadata"].get("halaman", "-")
            })

    print(f"Retrieval '{query[:40]}...': {len(chunk_relevan)} chunk relevan")
    return chunk_relevan

Inject Konteks ke LLM

Fungsi augmentasi dan generasi: susun prompt dengan konteks yang sudah diambil, lalu kirim ke LLM untuk menghasilkan jawaban akhir.

import google.generativeai as genai

genai.configure(api_key=os.getenv("GEMINI_API_KEY"))

SYSTEM_PROMPT = \"\"\"
Kamu adalah asisten yang menjawab pertanyaan berdasarkan dokumen yang diberikan.

Aturan:
1. Jawab HANYA berdasarkan isi [KONTEKS] di bawah.
2. Jika jawaban tidak ada di [KONTEKS], katakan:
   "Maaf, saya tidak menemukan informasi tersebut dalam dokumen."
3. Jangan mengarang atau menambahkan informasi di luar [KONTEKS].
4. Sebutkan sumber dokumen jika relevan.
5. Gunakan Bahasa Indonesia yang jelas dan ringkas.
\"\"\"

def generate(query: str, chunks: list[dict]) -> str:
    \"\"\"Augment prompt dengan konteks lalu generate jawaban.\"\"\"
    if not chunks:
        return "Maaf, saya tidak menemukan informasi yang relevan dalam dokumen."

    # Susun konteks dari semua chunk yang diambil
    bagian_konteks = []
    for i, chunk in enumerate(chunks, 1):
        bagian_konteks.append(
            f"[Sumber: {chunk['sumber']}, Halaman {chunk['halaman']}]\n"
            f"{chunk['teks']}"
        )
    konteks = "\n\n---\n\n".join(bagian_konteks)

    # Susun prompt final dengan konteks + pertanyaan
    prompt_user = f\"\"\"
[KONTEKS]
{konteks}

[PERTANYAAN]
{query}
\"\"\"

    model_llm = genai.GenerativeModel(
        model_name="gemini-1.5-flash",
        system_instruction=SYSTEM_PROMPT
    )

    response = model_llm.generate_content(
        prompt_user,
        generation_config={"temperature": 0.1}  # rendah untuk RAG
    )
    return response.text

Jalankan Semua

Fungsi utama yang menggabungkan semua langkah menjadi satu pipeline, plus contoh penggunaan lengkap dari indexing hingga mengajukan pertanyaan.

def tanya(query: str) -> dict:
    \"\"\"
    Pipeline RAG lengkap: retrieve -> augment -> generate.
    Kembalikan jawaban beserta sumber yang digunakan.
    \"\"\"
    # Langkah 1: Retrieve chunk relevan
    chunks = retrieve(query, top_k=4, min_skor=0.45)

    # Langkah 2: Generate jawaban dengan konteks
    jawaban = generate(query, chunks)

    # Kembalikan jawaban + metadata untuk transparansi sumber
    return {
        "pertanyaan": query,
        "jawaban": jawaban,
        "sumber": [
            f"{c['sumber']} (hal. {c['halaman']}, skor: {c['skor']})"
            for c in chunks
        ]
    }

# Contoh penggunaan lengkap
# 1. Index dokumen (jalankan sekali)
index_pdf("dokumen/kebijakan_toko.pdf")
index_pdf("dokumen/panduan_produk.pdf")

# 2. Ajukan pertanyaan
hasil = tanya("Bagaimana kebijakan pengembalian barang?")

print("Pertanyaan:", hasil["pertanyaan"])
print("Jawaban:", hasil["jawaban"])
print("Sumber:")
for s in hasil["sumber"]:
    print(f" - {s}")

13.4 Mengevaluasi Kualitas Pipeline RAG

Sistem RAG yang sudah berjalan perlu dievaluasi secara sistematis sebelum digunakan di produksi. Ada dua lapisan evaluasi yang perlu dilakukan secara terpisah karena kegagalan bisa terjadi di salah satu atau keduanya.

Untuk memastikan sistem RAG bekerja dengan baik, kita perlu melakukan evaluasi. Evaluasi dilakukan dengan membuat test set yang berisi pertanyaan dan jawaban yang diharapkan. Test set ini dibuat secara manual oleh tim yang memahami domain.

Berikut contoh kode untuk melakukan evaluasi retrieval dan generation:

# Test set: daftar pertanyaan + jawaban yang diharapkan
# Ini dibuat secara manual oleh tim yang memahami domain
TEST_SET = [
    {
        "pertanyaan": "Berapa lama kebijakan pengembalian barang?",
        "jawaban_ekspektasi": "30 hari",
        "chunk_ekspektasi": "kebijakan.pdf"
    },
    {
        "pertanyaan": "Berapa minimal pembelian untuk gratis ongkir?",
        "jawaban_ekspektasi": "Rp 200.000",
        "chunk_ekspektasi": "kebijakan.pdf"
    },
    {
        "pertanyaan": "Siapa presiden Indonesia saat ini?",
        "jawaban_ekspektasi": "tidak ada informasi",
        "chunk_ekspektasi": None  # harus fallback
    },
]

def evaluasi_retrieval(test_item: dict, chunks: list[dict]) -> bool:
    \"\"\"
    Periksa apakah chunk yang diharapkan berhasil diambil.
    Retrieval dianggap berhasil jika sumber yang diharapkan
    ada di antara chunk yang diambil.
    \"\"\"
    if test_item["chunk_ekspektasi"] is None:
        return len(chunks) == 0  # ekspektasi: tidak ada chunk relevan

    sumber_diambil = {c["sumber"] for c in chunks}
    return test_item["chunk_ekspektasi"] in sumber_diambil

def evaluasi_generation(jawaban: str, ekspektasi: str) -> bool:
    \"\"\"
    Periksa apakah jawaban mengandung informasi yang diharapkan.
    Evaluasi sederhana berbasis substring, untuk produksi
    gunakan LLM-as-a-Judge (dibahas di Bab 15).
    \"\"\"
    return ekspektasi.lower() in jawaban.lower()

def jalankan_evaluasi():
    \"\"\"Jalankan evaluasi pada seluruh test set.\"\"\"
    hasil_retrieval = []
    hasil_generation = []

    for item in TEST_SET:
        chunks = retrieve(item["pertanyaan"])
        jawaban = generate(item["pertanyaan"], chunks)

        ret_ok = evaluasi_retrieval(item, chunks)
        gen_ok = evaluasi_generation(jawaban, item["jawaban_ekspektasi"])

        hasil_retrieval.append(ret_ok)
        hasil_generation.append(gen_ok)

        status = "OK" if (ret_ok and gen_ok) else "GAGAL"
        print(f"{status} [{'R-ok' if ret_ok else 'R-gagal'}|{'G-ok' if gen_ok else 'G-gagal'}] {item['pertanyaan'][:50]}")

    ret_score = sum(hasil_retrieval) / len(hasil_retrieval) * 100
    gen_score = sum(hasil_generation) / len(hasil_generation) * 100

    print(f"Skor Retrieval: {ret_score:.0f}%")
    print(f"Skor Generation: {gen_score:.0f}%")
    print("Siap produksi" if ret_score >= 80 and gen_score >= 80 else "Perlu perbaikan sebelum produksi")

jalankan_evaluasi()

13.5 Pola Halusinasi Spesifik di Sistem RAG

Sistem RAG mengurangi halusinasi tapi tidak menghilangkannya sepenuhnya. Ada tiga pola halusinasi yang spesifik muncul di sistem RAG dan perlu diketahui cara mengatasinya.

1. Chunk Relevan Tidak Diambil, Retrieval Miss

Gejala: model menjawab "tidak ada informasi" padahal jawabannya ada di dokumen, atau memberikan jawaban yang tidak grounded sama sekali.

Penyebab umum: chunk terlalu besar sehingga embedding-nya "terlalu general". Model embedding tidak cocok dengan bahasa dokumen. Threshold min_skor terlalu tinggi sehingga chunk relevan tersaring.

Solusi: kurangi chunk_size, ganti model embedding yang lebih sesuai dengan bahasa dokumen, turunkan threshold sementara untuk melihat chunk apa yang sebenarnya diambil dengan retrieve(query, min_skor=0.3) # debug mode.

2. Model Mengabaikan Konteks, Generation Miss

Gejala: chunk yang relevan berhasil diambil (terlihat di log retrieval), tapi jawaban model tidak mencerminkan isi chunk tersebut. Model menjawab dari "pengetahuan sendiri".

Penyebab umum: System Prompt tidak cukup tegas melarang model menjawab di luar konteks. Temperature terlalu tinggi. Terlalu banyak chunk yang disertakan sehingga model "kebingungan".

Solusi: perkuat instruksi di System Prompt dengan larangan eksplisit, turunkan temperature ke 0.0-0.1, kurangi jumlah chunk yang disertakan (coba top_k=2 bukan 4).

3. Chunk Noise, Chunk Tidak Relevan Ikut Masuk

Gejala: jawaban model mencampur informasi dari dua topik berbeda, atau memberikan jawaban yang terlihat relevan tapi sebenarnya dari dokumen yang tidak tepat.

Penyebab umum: threshold min_skor terlalu rendah sehingga chunk yang tidak relevan ikut masuk ke konteks. Dokumen yang diindeks mencakup topik yang terlalu luas tanpa filtering.

Solusi: naikkan threshold min_skor, gunakan metadata filtering untuk membatasi pencarian pada dokumen yang relevan, log dan periksa chunk apa yang sebenarnya dikirim ke LLM dengan print([c["teks"][:80] for c in chunks]).

Kesalahan Umum

Ringkasan Bab 13