Upload files to "/"

tugas week 10..py

@@ -0,0 +1,278 @@
+import hashlib
+import random
+import re
+from pathlib import Path
+
+import chromadb
+from chromadb.utils import embedding_functions
+from langchain_text_splitters import RecursiveCharacterTextSplitter
+from pypdf import PdfReader
+
+
+BASE_DIR = Path(__file__).resolve().parent
+DOCUMENT_DIR = BASE_DIR / "dokumen"
+DB_DIR = BASE_DIR / "chroma_db_data"
+CHUNK_SIZES = (500, 1000, 1500)
+OVERLAP_RATIO = 0.1
+EMBEDDING_MODEL = "paraphrase-multilingual-MiniLM-L12-v2"
+
+# Ubah pertanyaan dan kata kunci agar sesuai dengan dokumen yang diuji.
+EVALUATION_QUESTIONS = [
+    ("Berapa SKS maksimal jika IP semester sebelumnya di atas 3?", "24 SKS"),
+    ("Berapa tingkat kehadiran minimum untuk mengikuti UAS?", "75%"),
+    ("Di mana mahasiswa merencanakan studi setiap awal semester?", "SIAKAD"),
+    ("Apa bukti yang harus diserahkan untuk ketidakhadiran karena sakit?", "surat keterangan resmi"),
+    ("Kapan surat keterangan sakit atau izin harus diserahkan?", "tiga hari"),
+    ("Berapa semester masa studi aktif yang harus ditempuh sebelum mengajukan cuti?", "minimal 2"),
+    ("Apa yang diperlukan jika mahasiswa penerima beasiswa ingin mengajukan cuti?", "surat izin tertulis"),
+    ("Melalui apa formulir cuti akademik diajukan?", "portal SIAKAD"),
+    ("Seberapa awal cuti harus diajukan sebelum registrasi semester baru?", "2 minggu"),
+    ("Kewajiban apa saja yang harus dilunasi sebelum cuti?", "uang kuliah"),
+    ("Siapa yang harus menyetujui permohonan cuti akademik?", "Dosen Pembimbing Akademik"),
+    ("Berapa lama cuti akademik boleh diambil selama masa studi?", "maksimal 2"),
+    ("Apakah masa cuti dihitung dalam batas waktu evaluasi kelulusan?", "tidak akan diperhitungkan"),
+    ("Berapa minimal SKS yang umumnya harus diselesaikan untuk lulus sarjana?", "144 SKS"),
+    ("Berapa IPK minimum untuk dinyatakan lulus?", "2.00"),
+]
+
+# Lima pertanyaan langsung dan dua parafrasa untuk Lab 2.
+RETRIEVAL_QUERIES = [
+    "Apa syarat untuk mengajukan cuti akademik?",
+    "Berapa batas kehadiran untuk bisa mengikuti ujian akhir?",
+    "Berapa jumlah SKS yang dapat diambil mahasiswa dengan IP tinggi?",
+    "Apa persyaratan untuk dinyatakan lulus sarjana?",
+    "Apa yang harus dilakukan mahasiswa pada awal semester?",
+    "Bagaimana cara berhenti kuliah sementara secara resmi?",  # Parafrasa cuti akademik
+    "Berapa banyak kredit kuliah yang dapat diambil jika nilai semester bagus?",  # Parafrasa SKS
+]
+
+
+def find_pdf_files():
+    files = []
+    handbook = BASE_DIR / "Buku Pedoman Akademik.pdf"
+    if handbook.is_file():
+        files.append(handbook)
+    if DOCUMENT_DIR.is_dir():
+        files.extend(DOCUMENT_DIR.glob("*.pdf"))
+
+    unique_files = sorted({path.resolve() for path in files})
+    if not unique_files:
+        raise FileNotFoundError(
+            "PDF tidak ditemukan. Letakkan 2-3 PDF berbasis teks di folder "
+            f"'{DOCUMENT_DIR.name}' atau sediakan 'Buku Pedoman Akademik.pdf'."
+        )
+    if len(unique_files) > 3:
+        raise ValueError(
+            f"Ditemukan {len(unique_files)} PDF. Lab menggunakan 2-3 PDF; "
+            f"pilih maksimal 3 PDF di '{DOCUMENT_DIR.name}'."
+        )
+    if len(unique_files) < 2:
+        print(
+            "PERHATIAN: baru ditemukan 1 PDF. Untuk memenuhi Lab 2, tambahkan "
+            "1-2 PDF lain ke folder 'dokumen'."
+        )
+    return unique_files
+
+
+def read_pdf(path):
+    reader = PdfReader(str(path))
+    text = "\n".join((page.extract_text() or "") for page in reader.pages).strip()
+    if not text:
+        raise ValueError(
+            f"Tidak ada teks yang dapat diekstrak dari '{path.name}'. "
+            "Pastikan PDF bukan hasil scan, atau lakukan OCR terlebih dahulu."
+        )
+    return text
+
+
+def make_chunks(text, size):
+    splitter = RecursiveCharacterTextSplitter(
+        chunk_size=size,
+        chunk_overlap=int(size * OVERLAP_RATIO),
+        length_function=len,
+    )
+    return splitter.split_text(text)
+
+
+def show_chunking_experiment(documents):
+    print("\n" + "=" * 70)
+    print("LAB 1 - EKSPERIMEN CHUNKING")
+    print("=" * 70)
+
+    for path, text in documents.items():
+        print(f"\nPDF: {path.name}")
+        print("500 karakter pertama:")
+        print(text[:500] or "(tidak ada teks)")
+
+        for size in CHUNK_SIZES:
+            chunks = make_chunks(text, size)
+            print(f"\nUkuran {size}, overlap {int(size * OVERLAP_RATIO)}: "
+                  f"{len(chunks)} chunk")
+            sample = random.Random(size).sample(chunks, min(3, len(chunks)))
+            for index, chunk in enumerate(sample, start=1):
+                print(f"  Contoh {index} ({len(chunk)} karakter):")
+                print(f"  {chunk[:350]}{'...' if len(chunk) > 350 else ''}")
+
+    print(
+        "\nPeriksa contoh di atas: catat apakah kalimat, tabel, atau daftar "
+        "terpotong pada batas chunk."
+    )
+
+
+def create_indexes(client, embedding_function, documents):
+    collections = {}
+    chunks_by_size = {}
+
+    for size in CHUNK_SIZES:
+        collection_name = f"rag_koleksi_chunk_{size}"
+        existing_names = {collection.name for collection in client.list_collections()}
+        if collection_name in existing_names:
+            client.delete_collection(name=collection_name)
+
+        collection = client.create_collection(
+            name=collection_name,
+            embedding_function=embedding_function,
+            metadata={"hnsw:space": "cosine"},
+        )
+        all_chunks = []
+        all_ids = []
+        all_metadatas = []
+
+        for path, text in documents.items():
+            chunks = make_chunks(text, size)
+            source = path.relative_to(BASE_DIR).as_posix()
+            source_hash = hashlib.sha1(source.encode("utf-8")).hexdigest()[:10]
+            all_chunks.extend(chunks)
+            all_ids.extend(
+                f"{source_hash}-{index}" for index in range(len(chunks))
+            )
+            all_metadatas.extend(
+                {
+                    "source": source,
+                    "chunk_size": size,
+                    "chunk_index": index,
+                }
+                for index in range(len(chunks))
+            )
+            chunks_by_size.setdefault(size, {})[source] = chunks
+            print(f"  {source}: {len(chunks)} chunk (ukuran {size})")
+
+        if all_chunks:
+            collection.add(
+                documents=all_chunks,
+                metadatas=all_metadatas,
+                ids=all_ids,
+            )
+        collections[size] = collection
+
+    return collections, chunks_by_size
+
+
+def show_retrieval_queries(collection):
+    print("\n" + "=" * 70)
+    print("LAB 2 - BANGUN INDEKS & UJI PENCARIAN")
+    print("=" * 70)
+    print("Hasil berikut memakai koleksi chunk 1000 dan menampilkan top-3.")
+
+    for query in RETRIEVAL_QUERIES:
+        results = collection.query(query_texts=[query], n_results=3)
+        print(f"\nPertanyaan: {query}")
+        for rank, (document, metadata, distance) in enumerate(
+            zip(
+                results["documents"][0],
+                results["metadatas"][0],
+                results["distances"][0],
+            ),
+            start=1,
+        ):
+            print(
+                f"  Top {rank} | jarak={distance:.4f} | "
+                f"{metadata['source']} | chunk {metadata['chunk_index']}"
+            )
+            print(f"  {document[:350]}{'...' if len(document) > 350 else ''}")
+
+    print(
+        "\nPeriksa relevansi hasil top-1. Dua pertanyaan terakhir adalah "
+        "parafrasa; catat apakah tetap menemukan potongan yang tepat."
+    )
+
+
+def normalize(text):
+    return re.sub(r"\s+", " ", text.lower()).strip()
+
+
+def hit_at_k(collection, questions, k):
+    hits = 0
+    misses = []
+    for question, keyword in questions:
+        results = collection.query(query_texts=[question], n_results=k)
+        documents = results["documents"][0] or []
+        if any(normalize(keyword) in normalize(document) for document in documents):
+            hits += 1
+        else:
+            misses.append(question)
+    return hits / len(questions), misses
+
+
+def evaluate_indexes(collections):
+    print("\n" + "=" * 70)
+    print("LAB 3 - EVALUASI HIT@K")
+    print("=" * 70)
+    print(f"Jumlah pertanyaan uji: {len(EVALUATION_QUESTIONS)}")
+
+    scores = {}
+    misses_at_three = {}
+    for size, collection in collections.items():
+        scores[size] = {}
+        for k in (1, 3, 5):
+            score, misses = hit_at_k(collection, EVALUATION_QUESTIONS, k)
+            scores[size][k] = score
+            if k == 3:
+                misses_at_three[size] = misses
+            print(f"Chunk {size:>4} | Hit@{k} = {score:.2f}")
+
+    best_size = max(scores, key=lambda size: scores[size][3])
+    failed_questions = misses_at_three[best_size]
+    print("\nBahan kesimpulan:")
+    print(
+        f"- Hit@3 tertinggi diperoleh ukuran chunk {best_size} "
+        f"({scores[best_size][3]:.2f})."
+    )
+    if failed_questions:
+        print(
+            "- Pertanyaan yang belum menemukan kata kunci pada top-3: "
+            + "; ".join(failed_questions)
+        )
+    else:
+        print("- Semua kata kunci jawaban ditemukan pada top-3 ukuran terbaik.")
+    print(
+        "- Hit@k berbasis kata kunci adalah evaluasi praktis; periksa hasil "
+        "secara manual karena kata kunci yang muncul belum tentu menjawab "
+        "pertanyaan dengan benar."
+    )
+
+
+def main():
+    pdf_files = find_pdf_files()
+    documents = {path: read_pdf(path) for path in pdf_files}
+
+    print("PDF yang akan diproses:")
+    for path in pdf_files:
+        print(f"- {path.relative_to(BASE_DIR)}")
+
+    show_chunking_experiment(documents)
+
+    print("\nMenyiapkan model embedding multilingual...")
+    embedding_function = embedding_functions.SentenceTransformerEmbeddingFunction(
+        model_name=EMBEDDING_MODEL
+    )
+    client = chromadb.PersistentClient(path=str(DB_DIR))
+
+    print("\nMembangun indeks ChromaDB dengan metrik cosine...")
+    collections, _ = create_indexes(client, embedding_function, documents)
+    show_retrieval_queries(collections[1000])
+    evaluate_indexes(collections)
+
+
+if __name__ == "__main__":
+    main()