Commit

main.py

@@ -0,0 +1,364 @@
+# -*- coding: utf-8 -*-
+"""
+PRAKTIKUM RAG  |  Lab 1 (Chunking)  ->  Lab 2 (Indeks & Cari)  ->  Lab 3 (Hit@k)
+
+Cara pakai:
+  1. python -m pip install pdfplumber PyPDF2 sentence-transformers numpy pandas
+  2. Buat folder `dokumen/` di samping file ini, isi 3 file PDF buku cerita.
+  3. Jalankan per-cell di VS Code (tanda `# %%`) atau jalankan langsung `python main.py`.
+
+Catatan desain:
+  - Embedding : sentence-transformers (model multilingual, cocok untuk Bahasa Indonesia).
+  - Vector DB : numpy + cosine similarity.
+"""
+
+# %% ============================================================
+# 0. IMPORT & KONFIGURASI
+# ===============================================================
+import os
+import re
+import random
+from glob import glob
+
+import numpy as np
+
+FOLDER_DOKUMEN = "dokumen"          # folder berisi PDF
+OVERLAP_PERSEN = 0.15               # overlap 15%
+MODEL_EMBEDDING = "paraphrase-multilingual-MiniLM-L12-v2"  # mendukung Bahasa Indonesia
+random.seed(42)                     # supaya chunk "acak" bisa direproduksi
+
+
+# %% ============================================================
+# 1. FUNGSI INTI (dipakai oleh Lab 1, 2, dan 3)
+# ===============================================================
+def baca_pdf(filepath):
+    """
+    Membaca 1 file PDF dan mengembalikan seluruh teksnya (string).
+
+    - Utama : pdfplumber (lebih baik untuk layout/tabel).
+    - Cadangan: PyPDF2, dipakai jika pdfplumber tidak terpasang / gagal.
+    - Teks dinormalisasi: spasi berlebih dirapikan, halaman digabung dengan newline.
+    """
+    halaman = []
+    try:
+        import pdfplumber
+        with pdfplumber.open(filepath) as pdf:
+            for p in pdf.pages:
+                halaman.append(p.extract_text() or "")
+    except Exception as e:  # fallback
+        print(f"[peringatan] pdfplumber gagal ({e}); beralih ke PyPDF2")
+        from PyPDF2 import PdfReader
+        reader = PdfReader(filepath)
+        halaman = [(pg.extract_text() or "") for pg in reader.pages]
+
+    teks = "\n".join(halaman)
+    teks = re.sub(r"[ \t]+", " ", teks)        # rapikan spasi/tab berulang
+    teks = re.sub(r"\n{3,}", "\n\n", teks)     # maksimal 1 baris kosong
+    return teks.strip()
+
+
+def potong(teks, chunk_size, overlap):
+    """
+    Memotong `teks` menjadi potongan (chunk) berukuran `chunk_size` karakter.
+
+    Parameter:
+      chunk_size : panjang tiap chunk (karakter).
+      overlap    : tumpang-tindih antar chunk.
+                   Jika 0 < overlap < 1 -> dianggap persentase (mis. 0.15 = 15%).
+                   Jika >= 1            -> dianggap jumlah karakter.
+    Return: list[str]
+    """
+    if 0 < overlap < 1:
+        overlap = int(chunk_size * overlap)
+    overlap = int(overlap)
+    if overlap >= chunk_size:
+        raise ValueError("overlap harus lebih kecil dari chunk_size")
+
+    langkah = chunk_size - overlap
+    hasil = []
+    for mulai in range(0, len(teks), langkah):
+        potongan = teks[mulai:mulai + chunk_size].strip()
+        if potongan:
+            hasil.append(potongan)
+        if mulai + chunk_size >= len(teks):   # sudah mencapai akhir teks
+            break
+    return hasil
+
+
+class IndeksVektor:
+    """Indeks vektor sederhana: simpan chunk + embedding, cari dengan cosine similarity."""
+
+    _model = None  # model dimuat sekali saja (cache di level kelas)
+
+    def __init__(self, nama="indeks"):
+        self.nama = nama
+        self.chunks = []        # list[dict]: {"teks", "sumber", "id"}
+        self.vektor = None      # np.ndarray (n_chunk, dim), sudah dinormalisasi
+
+    @classmethod
+    def model(cls):
+        if cls._model is None:
+            from sentence_transformers import SentenceTransformer
+            cls._model = SentenceTransformer(MODEL_EMBEDDING)
+        return cls._model
+
+    def tambah(self, daftar_teks, sumber):
+        """Menambahkan chunk dari satu dokumen ke indeks lalu meng-embed-nya."""
+        awal = len(self.chunks)
+        for i, t in enumerate(daftar_teks):
+            self.chunks.append({"teks": t, "sumber": sumber, "id": awal + i})
+        emb = self.model().encode(daftar_teks, normalize_embeddings=True,
+                                  show_progress_bar=False)
+        self.vektor = emb if self.vektor is None else np.vstack([self.vektor, emb])
+
+    def cari(self, pertanyaan, k=3):
+        """Mengembalikan top-k chunk paling mirip (list[dict] dengan kunci 'skor')."""
+        q = self.model().encode([pertanyaan], normalize_embeddings=True)[0]
+        skor = self.vektor @ q                       # cosine (vektor sudah ternormalisasi)
+        idx = np.argsort(-skor)[:k]
+        return [{**self.chunks[i], "skor": float(skor[i])} for i in idx]
+
+
+def bangun_indeks(folder, chunk_size, overlap=OVERLAP_PERSEN, nama="indeks"):
+    """
+    Membaca semua PDF di `folder`, memotong, meng-embed, lalu mengembalikan
+    (indeks, laporan) dimana laporan = {nama_file: jumlah_chunk}.
+    """
+    indeks = IndeksVektor(nama)
+    laporan = {}
+    file_pdf = sorted(glob(os.path.join(folder, "*.pdf")))
+    if not file_pdf:
+        raise FileNotFoundError(f"Tidak ada PDF di folder '{folder}/'")
+    for path in file_pdf:
+        teks = baca_pdf(path)
+        chunks = potong(teks, chunk_size, overlap)
+        indeks.tambah(chunks, sumber=os.path.basename(path))
+        laporan[os.path.basename(path)] = len(chunks)
+    return indeks, laporan
+
+
+INDEKS_AKTIF = None  # diisi di Lab 2; dipakai default oleh query()
+
+
+def query(pertanyaan, k=3, index=None, tampil=True):
+    """
+    Mencari chunk paling relevan untuk `pertanyaan`.
+
+    - index : objek IndeksVektor; jika None memakai INDEKS_AKTIF (hasil Lab 2).
+    - k     : jumlah hasil teratas.
+    - tampil: True -> cetak hasil dengan rapi.
+    Return  : list[dict] berisi teks, sumber, id, skor.
+    """
+    index = index or INDEKS_AKTIF
+    hasil = index.cari(pertanyaan, k)
+    if tampil:
+        print(f"\n❓ {pertanyaan}")
+        for r, h in enumerate(hasil, 1):
+            cuplikan = h["teks"][:300].replace("\n", " ")
+            print(f"  [{r}] skor={h['skor']:.3f} | {h['sumber']} #chunk{h['id']}\n      {cuplikan}...")
+    return hasil
+
+
+# %% ============================================================
+# LAB 1 - PREPROCESSING & CHUNKING
+# ===============================================================
+# --- 1.1 Baca PDF + cetak 500 karakter pertama ---
+file_pdf = sorted(glob(os.path.join(FOLDER_DOKUMEN, "*.pdf")))
+assert file_pdf, f"Letakkan PDF di folder '{FOLDER_DOKUMEN}/' dulu."
+FILE_LAB1 = file_pdf[0]
+
+teks_lab1 = baca_pdf(FILE_LAB1)
+print(f"File          : {FILE_LAB1}")
+print(f"Total karakter: {len(teks_lab1):,}")
+print("=" * 60)
+print(teks_lab1[:500])
+print("=" * 60)
+
+
+# %%
+# --- 1.2 Eksperimen chunking: 500 / 1000 / 1500 dengan overlap 15% ---
+UKURAN_CHUNK = [500, 1000, 1500]
+hasil_chunk = {}
+for ukuran in UKURAN_CHUNK:
+    hasil_chunk[ukuran] = potong(teks_lab1, ukuran, OVERLAP_PERSEN)
+
+print(f"{'Chunk size':>12} | {'Overlap':>8} | {'Jumlah chunk':>12}")
+print("-" * 40)
+for ukuran, daftar in hasil_chunk.items():
+    print(f"{ukuran:>12} | {int(ukuran*OVERLAP_PERSEN):>8} | {len(daftar):>12}")
+
+
+# %%
+# --- 1.3 Evaluasi kualitas: 3 chunk acak per ukuran + deteksi otomatis ---
+def periksa_potongan(chunk):
+    """Heuristik sederhana untuk membantu analisis manual."""
+    c = chunk.strip()
+    catatan = []
+    if c and c[0].islower():
+        catatan.append("diawali huruf kecil -> kalimat terpotong di awal")
+    if c and c[-1] not in ".!?:;)\"'":
+        catatan.append("tidak diakhiri tanda baca -> kalimat terpotong di akhir")
+    baris = c.split("\n")
+    tabel = [b for b in baris if b.count("  ") >= 2 or b.count("|") >= 2
+             or len(re.findall(r"\d+[.,]?\d*", b)) >= 4]
+    if tabel:
+        catatan.append(f"terindikasi tabel/angka berderet ({len(tabel)} baris) -> cek apakah tabel terbelah")
+    return catatan or ["tidak ada indikasi pemotongan"]
+
+for ukuran, daftar in hasil_chunk.items():
+    print("\n" + "#" * 70)
+    print(f"CHUNK SIZE {ukuran}  (total {len(daftar)} chunk)")
+    print("#" * 70)
+    for no in random.sample(range(len(daftar)), k=min(3, len(daftar))):
+        print(f"\n--- chunk #{no} ({len(daftar[no])} karakter) ---")
+        print(daftar[no])
+        print("  >> Deteksi otomatis:", "; ".join(periksa_potongan(daftar[no])))
+
+
+# %% ============================================================
+# LAB 2 - BANGUN INDEKS & CARI
+# ===============================================================
+# --- 2.1 Indexing multi-dokumen (2-3 PDF dalam folder dokumen/) ---
+INDEKS_AKTIF, laporan = bangun_indeks(FOLDER_DOKUMEN, chunk_size=1000,
+                                      overlap=OVERLAP_PERSEN, nama="Indeks-1000")
+print("LAPORAN INDEXING")
+print("-" * 50)
+for nama_file, jml in laporan.items():
+    print(f"{nama_file:<35} : {jml:>4} chunk")
+print("-" * 50)
+print(f"{'TOTAL':<35} : {sum(laporan.values()):>4} chunk")
+
+
+# %%
+# --- 2.2 Direct query: 5 pertanyaan spesifik (Cerita Rakyat) ---
+PERTANYAAN_LANGSUNG = [
+    "Siapa nama gadis baik hati dalam cerita Bawang Merah Bawang Putih?",
+    "Benda apa yang diberikan nenek di dalam gua kepada Bawang Putih?",
+    "Apa janji yang dilanggar oleh Toba kepada istrinya?",
+    "Siapa nama anak laki-laki Toba yang sangat lapar dan serakah?",
+    "Dikena hukuman atau dikutuk menjadi apa Malin Kundang oleh ibunya?"
+]
+for p in PERTANYAAN_LANGSUNG:
+    query(p, k=3)
+
+
+# %%
+# --- 2.3 Uji semantik / parafrase: 2 pertanyaan (Cerita Rakyat) ---
+PERTANYAAN_PARAFRASE = [
+    "Buah apa yang dibawa pulang oleh Bawang Merah yang berisi perhiasan atau hewan buas?",
+    "Bencana alam apa yang tercipta akibat luapan air mata dan amarah istri Toba?"
+]
+for p in PERTANYAAN_PARAFRASE:
+    query(p, k=3)
+
+
+# %% ============================================================
+# LAB 3 - UKUR HIT@K
+# ===============================================================
+# --- 3.1 Ground truth: 15 pertanyaan + kata kunci jawaban cerita rakyat ---
+DATASET_UJI = [
+    ("Siapa nama gadis manis yang baik hati dan rajin?", ["Bawang Putih"]),
+    ("Apa yang diberikan nenek tua di dalam gua kepada Bawang Putih?", ["labu"]),
+    ("Benda apa yang hilang terbawa arus sungai saat Bawang Putih mencuci?", ["selendang", "baju"]),
+    ("Siapa nama saudara tiri Bawang Putih yang malas dan serakah?", ["Bawang Merah"]),
+    ("Di mana Bawang Putih bertemu dengan nenek tua?", ["gua"]),
+    ("Siapa nama pemuda nelayan yang menangkap ikan mas ajaib?", ["Toba"]),
+    ("Ikan mas ajaib berubah menjadi sosok apa?", ["wanita", "perempuan"]),
+    ("Apa syarat yang diajukan jelita ikan mas sebelum menikah dengan Toba?", ["rahasia", "janji"]),
+    ("Siapa nama anak laki-laki Toba yang sangat pemakan dan lapar?", ["Samosir"]),
+    ("Danau apa yang terbentuk dari luapan air mata dan banjir besar?", ["Danau Toba"]),
+    ("Siapa nama pemuda miskin yang pergi merantau naik kapal pedagang?", ["Malin Kundang"]),
+    ("Di mana Malin Kundang tinggal bersama ibunya saat masih kecil?", ["desa", "pantai"]),
+    ("Bagaimana sikap Malin Kundang saat bertemu kembali dengan ibu kandungnya?", ["durhaka", "menolak"]),
+    ("Malin Kundang dikutuk oleh ibunya menjadi apa?", ["batu"]),
+    ("Apa yang hancur dihantam badai dahsyat di tengah laut?", ["kapal"])
+]
+
+pertanyaan_list = [p for p, _ in DATASET_UJI]
+ground_truth_list = [g for _, g in DATASET_UJI]
+assert len(DATASET_UJI) == 15, "Dataset uji harus 15 pertanyaan"
+
+
+# %%
+# --- 3.2 Fungsi hit_at_k ---
+def _cocok(teks_chunk, kata_kunci, min_cocok=1):
+    """True jika minimal `min_cocok` kata kunci ditemukan di chunk (case-insensitive)."""
+    t = teks_chunk.lower()
+    return sum(kk.lower() in t for kk in kata_kunci) >= min_cocok
+
+
+def hit_at_k(pertanyaan_list, ground_truth_list, index, k, min_cocok=1):
+    """
+    Menghitung Hit@k = (jumlah pertanyaan yang 'hit') / (total pertanyaan).
+
+    Sebuah pertanyaan dianggap HIT bila minimal satu dari top-k chunk hasil
+    pencarian memuat >= `min_cocok` kata kunci ground truth.
+
+    Return: (skor_hit, daftar_indeks_miss)
+    """
+    hit, miss = 0, []
+    for i, (p, gt) in enumerate(zip(pertanyaan_list, ground_truth_list)):
+        top_k = index.cari(p, k)
+        if any(_cocok(h["teks"], gt, min_cocok) for h in top_k):
+            hit += 1
+        else:
+            miss.append(i)
+    return hit / len(pertanyaan_list), miss
+
+
+# %%
+# --- 3.3 Jalankan pada 2 indeks: A (chunk 500) vs B (chunk 1000) ---
+indeks_A, _ = bangun_indeks(FOLDER_DOKUMEN, 500,  OVERLAP_PERSEN, nama="Indeks A (500)")
+indeks_B, _ = bangun_indeks(FOLDER_DOKUMEN, 1000, OVERLAP_PERSEN, nama="Indeks B (1000)")
+
+K_LIST = [1, 3, 5]
+tabel, miss_detail = {}, {}
+for idx in (indeks_A, indeks_B):
+    tabel[idx.nama] = {}
+    for k in K_LIST:
+        skor, miss = hit_at_k(pertanyaan_list, ground_truth_list, idx, k)
+        tabel[idx.nama][f"Hit@{k}"] = skor
+        miss_detail[(idx.nama, k)] = miss
+
+print("\nPERBANDINGAN HIT@K (15 pertanyaan uji)")
+print("-" * 50)
+print(f"{'Indeks':<20} | {'Hit@1':<8} | {'Hit@3':<8} | {'Hit@5':<8}")
+print("-" * 50)
+for nama_indeks, skor_dict in tabel.items():
+    h1 = f"{skor_dict.get('Hit@1', 0):.2%}"
+    h3 = f"{skor_dict.get('Hit@3', 0):.2%}"
+    h5 = f"{skor_dict.get('Hit@5', 0):.2%}"
+    print(f"{nama_indeks:<20} | {h1:<8} | {h3:<8} | {h5:<8}")
+print("-" * 50)
+
+
+# %%
+# --- 3.4 Analisis failure case (miss pada k=5) + draf kesimpulan ---
+for idx in (indeks_A, indeks_B):
+    miss = miss_detail[(idx.nama, 5)]
+    print(f"\n{'='*70}\n{idx.nama}: {len(miss)} pertanyaan MISS pada k=5\n{'='*70}")
+    for i in miss:
+        top1 = idx.cari(pertanyaan_list[i], 1)[0]
+        print(f"\n[Q{i+1}] {pertanyaan_list[i]}")
+        print(f"   Kata kunci   : {ground_truth_list[i]}")
+        print(f"   Top-1 diambil: ({top1['sumber']}, skor={top1['skor']:.3f}) "
+              f"{top1['teks'][:200].replace(chr(10), ' ')}...")
+
+
+def draf_kesimpulan(tabel_data, miss_detail, nama_a, nama_b):
+    """Membuat draf kesimpulan otomatis."""
+    rata_a = sum(tabel_data[nama_a].values()) / len(tabel_data[nama_a])
+    rata_b = sum(tabel_data[nama_b].values()) / len(tabel_data[nama_b])
+    terbaik = nama_a if rata_a >= rata_b else nama_b
+    miss_b = sorted({f"Q{i+1}" for (n, k), m in miss_detail.items() if k == 5 for i in m})
+    return (
+        f"Berdasarkan rata-rata Hit@1/3/5, {terbaik} memberikan performa terbaik "
+        f"({nama_a}: {rata_a:.1%} vs {nama_b}: {rata_b:.1%}). "
+        f"Kegagalan (miss) pada k=5 terjadi di pertanyaan: {', '.join(miss_b) or 'tidak ada'}. "
+        "Dugaan penyebab: jawaban tersebar/terpotong di batas chunk, istilah teknis atau kata kunci "
+        "terlalu spesifik/berbeda format, atau ukuran chunk yang kurang fleksibel menangkap konteks kalimat."
+    )
+
+print("\nDRAF KESIMPULAN:\n")
+print(draf_kesimpulan(tabel, miss_detail, indeks_A.nama, indeks_B.nama))
+# %%