Upload files to "/"
This commit is contained in:
@@ -0,0 +1,128 @@
|
||||
import random
|
||||
from pathlib import Path
|
||||
|
||||
# FUNGSI MEMBACA PDF
|
||||
def baca_pdf(path):
|
||||
reader = PdfReader(path)
|
||||
return "\n".join(
|
||||
(p.extract_text() or "")
|
||||
for p in reader.pages
|
||||
)
|
||||
|
||||
# FUNGSI MEMOTONG TEKS (CHUNKING)
|
||||
def potong(teks, ukuran=800, tumpang=100):
|
||||
hasil = []
|
||||
mulai = 0
|
||||
while mulai < len(teks):
|
||||
hasil.append(teks[mulai:mulai + ukuran])
|
||||
mulai += ukuran - tumpang
|
||||
return [h for h in hasil if h.strip()]
|
||||
|
||||
# MENCARI PDF DI FOLDER DOKUMEN
|
||||
daftar_pdf = list(Path("dokumen").glob("*.pdf"))
|
||||
|
||||
if not daftar_pdf:
|
||||
print("❌ Tidak ada file PDF di folder 'dokumen'! Pastikan Anda membuat folder bernama 'dokumen' dan memasukkan file PDF terkait KAI di dalamnya.")
|
||||
else:
|
||||
# Mengambil PDF pertama
|
||||
pdf = daftar_pdf[0]
|
||||
|
||||
# Membaca isi PDF
|
||||
teks = baca_pdf(pdf)
|
||||
|
||||
print(f"\n📄 Nama PDF: {pdf.name}")
|
||||
print(f"📊 Total karakter: {len(teks)}")
|
||||
|
||||
# Menampilkan 500 karakter pertama
|
||||
print("\n===== 500 KARAKTER PERTAMA =====")
|
||||
print(teks[:500])
|
||||
|
||||
# Eksperimen tiga ukuran chunk
|
||||
for ukuran in (500, 1000, 1500):
|
||||
overlap = int(ukuran * 0.1)
|
||||
chunks = potong(
|
||||
teks,
|
||||
ukuran=ukuran,
|
||||
tumpang=overlap
|
||||
)
|
||||
|
||||
print("\n" + "=" * 50)
|
||||
print(f"UKURAN CHUNK: {ukuran}")
|
||||
print(f"OVERLAP: {overlap}")
|
||||
print(f"JUMLAH CHUNK: {len(chunks)}")
|
||||
|
||||
# Mengambil maksimal 3 chunk secara acak
|
||||
sampel = random.sample(
|
||||
range(len(chunks)),
|
||||
min(3, len(chunks))
|
||||
)
|
||||
|
||||
for nomor in sampel:
|
||||
print(f"\n--- CHUNK {nomor} ---")
|
||||
print(chunks[nomor])
|
||||
|
||||
|
||||
# LAB 2: EMBEDDING DAN CHROMADB
|
||||
if daftar_pdf:
|
||||
# Model embedding multilingual
|
||||
emb = embedding_functions.SentenceTransformerEmbeddingFunction(
|
||||
model_name="paraphrase-multilingual-MiniLM-L12-v2"
|
||||
)
|
||||
|
||||
# Membuat database lokal
|
||||
client = chromadb.PersistentClient(path="./chroma_db")
|
||||
|
||||
def bangun_indeks(nama_koleksi, ukuran):
|
||||
# Menghapus koleksi lama agar tidak tercampur
|
||||
try:
|
||||
client.delete_collection(name=nama_koleksi)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
# Membuat koleksi baru
|
||||
koleksi = client.create_collection(
|
||||
name=nama_koleksi,
|
||||
embedding_function=emb,
|
||||
metadata={"hnsw:space": "cosine"}
|
||||
)
|
||||
|
||||
for pdf in daftar_pdf:
|
||||
teks = baca_pdf(pdf)
|
||||
overlap = int(ukuran * 0.1)
|
||||
chunks = potong(
|
||||
teks,
|
||||
ukuran=ukuran,
|
||||
tumpang=overlap
|
||||
)
|
||||
|
||||
if not chunks:
|
||||
print(pdf.name, "tidak memiliki teks.")
|
||||
continue
|
||||
|
||||
koleksi.upsert(
|
||||
ids=[
|
||||
f"{pdf.stem}-{ukuran}-{i}"
|
||||
for i in range(len(chunks))
|
||||
],
|
||||
documents=chunks,
|
||||
metadatas=[
|
||||
{
|
||||
"sumber": pdf.name,
|
||||
"chunk": i,
|
||||
"ukuran": ukuran
|
||||
}
|
||||
for i in range(len(chunks))
|
||||
]
|
||||
)
|
||||
|
||||
print(pdf.name, "->", len(chunks), "chunk")
|
||||
|
||||
print(f"\n✅ Koleksi '{nama_koleksi}' berhasil dibuat.")
|
||||
print("Total data dalam koleksi:", koleksi.count())
|
||||
|
||||
# Membuat dua koleksi untuk eksperimen
|
||||
bangun_indeks("pedoman_500", 500)
|
||||
bangun_indeks("pedoman_1000", 1000)
|
||||
|
||||
else:
|
||||
print("⚠️ Indeks gagal dibuat karena tidak ada file PDF.")
|
||||
Binary file not shown.
+146
@@ -0,0 +1,146 @@
|
||||
import math
|
||||
import re
|
||||
from collections import Counter
|
||||
|
||||
|
||||
class KAIGapekaRAG:
|
||||
"""Sistem RAG Sederhana berbasis Pure Python untuk topik GAPEKA 2026 PT KAI."""
|
||||
|
||||
def __init__(self, knowledge_base):
|
||||
print("[INFO] Menginisialisasi Basis Pengetahuan KAI Gapeka 2026...")
|
||||
self.documents = knowledge_base
|
||||
self.vocab = self._build_vocab(self.documents)
|
||||
# Pra-penghitungan vektor TF-IDF untuk dokumen lokal
|
||||
self.doc_vectors = [
|
||||
self._compute_tfidf(doc, self.vocab, self.documents)
|
||||
for doc in self.documents
|
||||
]
|
||||
|
||||
def _tokenize(self, text):
|
||||
"""Membersihkan teks dan memecahnya menjadi kata-kata (token)."""
|
||||
return re.findall(r"\b\w+\b", text.lower())
|
||||
|
||||
def _build_vocab(self, docs):
|
||||
"""Membangun kamus kata unik dari seluruh dokumen."""
|
||||
vocab = set()
|
||||
for doc in docs:
|
||||
vocab.update(self._tokenize(doc))
|
||||
return sorted(list(vocab))
|
||||
|
||||
def _compute_tfidf(self, text, vocab, all_docs):
|
||||
"""Menghitung bobot TF-IDF secara manual tanpa library eksternal."""
|
||||
tokens = self._tokenize(text)
|
||||
tf = Counter(tokens)
|
||||
total_tokens = len(tokens) if tokens else 1
|
||||
|
||||
vector = {}
|
||||
for word in vocab:
|
||||
# Term Frequency (TF)
|
||||
term_freq = tf[word] / total_tokens
|
||||
|
||||
# Inverse Document Frequency (IDF)
|
||||
doc_count = sum(1 for d in all_docs if word in self._tokenize(d))
|
||||
idf = (
|
||||
math.log(len(all_docs) / (doc_count + 1)) + 1 if doc_count > 0 else 0
|
||||
)
|
||||
|
||||
vector[word] = term_freq * idf
|
||||
return vector
|
||||
|
||||
def _cosine_similarity(self, vec1, vec2):
|
||||
"""Menghitung kemiripan kosinus antara dua vektor TF-IDF."""
|
||||
intersection = set(vec1.keys()).intersection(set(vec2.keys()))
|
||||
numerator = sum(vec1[x] * vec2[x] for x in intersection)
|
||||
|
||||
sum1 = sum(val ** 2 for val in vec1.values())
|
||||
sum2 = sum(val ** 2 for val in vec2.values())
|
||||
denominator = math.sqrt(sum1) * math.sqrt(sum2)
|
||||
|
||||
if denominator == 0:
|
||||
return 0.0
|
||||
return numerator / denominator
|
||||
|
||||
def retrieve(self, query, top_k=1):
|
||||
"""Mencari dokumen paling relevan menggunakan Cosine Similarity."""
|
||||
query_vector = self._compute_tfidf(query, self.vocab, self.documents)
|
||||
scores = []
|
||||
|
||||
for i, doc_vec in enumerate(self.doc_vectors):
|
||||
score = self._cosine_similarity(query_vector, doc_vec)
|
||||
scores.append((score, self.documents[i]))
|
||||
|
||||
# Urutkan berdasarkan skor tertinggi
|
||||
scores.sort(key=lambda x: x[0], reverse=True)
|
||||
return [doc for score, doc in scores[:top_k]]
|
||||
|
||||
def generate_answer(self, query):
|
||||
"""Tahap Generation: Menggabungkan hasil retrieval dengan format jawaban RAG."""
|
||||
retrieved_docs = self.retrieve(query, top_k=1)
|
||||
context = (
|
||||
retrieved_docs[0]
|
||||
if retrieved_docs
|
||||
else "Maaf, informasi tidak ditemukan."
|
||||
)
|
||||
|
||||
# Template Generator Extractive RAG
|
||||
answer = (
|
||||
f"==================================================\n"
|
||||
f"HASIL GENERASI RAG - PT KERETA API INDONESIA (GAPEKA 2026)\n"
|
||||
f"==================================================\n"
|
||||
f"Pertanyaan : {query}\n\n"
|
||||
f"Konteks Relevan (Retrieved):\n-> {context}\n\n"
|
||||
f"Jawaban Sistem:\n"
|
||||
f"Berdasarkan informasi resmi kebijakan GAPEKA 2026 PT KAI, "
|
||||
f"berikut adalah detail informasinya:\n"
|
||||
f'"{context}"\n'
|
||||
f"=================================================="
|
||||
)
|
||||
return answer, retrieved_docs
|
||||
|
||||
|
||||
# ==========================================
|
||||
# EKSEKUSI PROGRAM UTAMA
|
||||
# ==========================================
|
||||
if __name__ == "__main__":
|
||||
# 1. Dataset / Basis Pengetahuan Lokal KAI GAPEKA 2026
|
||||
kai_kb = [
|
||||
(
|
||||
"PT Kereta Api Indonesia (KAI) resmi memberlakukan Grafik Perjalanan "
|
||||
"Kereta Api (GAPEKA) 2026 mulai tanggal 1 Desember 2026."
|
||||
),
|
||||
(
|
||||
"Tiket perjalanan kereta api untuk jadwal GAPEKA 2026 sudah dapat "
|
||||
"dipesan secara online melalui aplikasi Access by KAI."
|
||||
),
|
||||
(
|
||||
"Terdapat penyesuaian serta penyederhanaan nama pada beberapa "
|
||||
"perjalanan kereta api agar lebih simpel dan mudah diingat oleh "
|
||||
"penumpang."
|
||||
),
|
||||
(
|
||||
"KAI menambah titik pemberhentian di sejumlah stasiun untuk "
|
||||
"mendukung mobilitas antarkota dan kawasan aglomerasi."
|
||||
),
|
||||
(
|
||||
"KAI menghadirkan layanan rute baru seperti KA Borobudur "
|
||||
"(Lempuyangan–Gambir PP, fakultatif) dan KA Pandawa "
|
||||
"(Yogyakarta–Semarang Tawang PP) yang mulai beroperasi pada 1 "
|
||||
"Desember 2026."
|
||||
),
|
||||
]
|
||||
|
||||
# 2. Inisialisasi Sistem RAG
|
||||
rag = KAIGapekaRAG(kai_kb)
|
||||
|
||||
# 3. Contoh Daftar Pengujian Pertanyaan Tugas RAG
|
||||
daftar_pertanyaan = [
|
||||
"Kapan GAPEKA 2026 mulai diberlakukan dan di mana bisa pesan tiketnya?",
|
||||
"Apa saja rute kereta api baru yang dihadirkan KAI dalam GAPEKA 2026?",
|
||||
"Bagaimana kebijakan terkait nama kereta api dalam GAPEKA 2026?"
|
||||
]
|
||||
|
||||
# 4. Menjalankan Pipeline Pengujian
|
||||
for idx, sample_query in enumerate(daftar_pertanyaan, 1):
|
||||
print(f"\n[PENGUJIAN {idx}]")
|
||||
final_answer, sources = rag.generate_answer(sample_query)
|
||||
print(final_answer)
|
||||
@@ -0,0 +1,206 @@
|
||||
import math
|
||||
import re
|
||||
from collections import Counter
|
||||
|
||||
|
||||
class KAIGapekaRAG:
|
||||
"""Sistem RAG Sederhana berbasis Pure Python untuk topik GAPEKA 2026 PT KAI."""
|
||||
|
||||
def __init__(self, knowledge_base):
|
||||
print("[INFO] Menginisialisasi Basis Pengetahuan KAI Gapeka 2026...")
|
||||
self.documents = knowledge_base
|
||||
self.vocab = self._build_vocab(self.documents)
|
||||
# Pra-penghitungan vektor TF-IDF untuk dokumen lokal
|
||||
self.doc_vectors = [
|
||||
self._compute_tfidf(doc, self.vocab, self.documents)
|
||||
for doc in self.documents
|
||||
]
|
||||
|
||||
def _tokenize(self, text):
|
||||
"""Membersihkan teks dan memecahnya menjadi kata-kata (token)."""
|
||||
return re.findall(r"\b\w+\b", text.lower())
|
||||
|
||||
def _build_vocab(self, docs):
|
||||
"""Membangun kamus kata unik dari seluruh dokumen."""
|
||||
vocab = set()
|
||||
for doc in docs:
|
||||
vocab.update(self._tokenize(doc))
|
||||
return sorted(list(vocab))
|
||||
|
||||
def _compute_tfidf(self, text, vocab, all_docs):
|
||||
"""Menghitung bobot TF-IDF secara manual tanpa library eksternal."""
|
||||
tokens = self._tokenize(text)
|
||||
tf = Counter(tokens)
|
||||
total_tokens = len(tokens) if tokens else 1
|
||||
|
||||
vector = {}
|
||||
for word in vocab:
|
||||
# Term Frequency (TF)
|
||||
term_freq = tf[word] / total_tokens
|
||||
|
||||
# Inverse Document Frequency (IDF)
|
||||
doc_count = sum(1 for d in all_docs if word in self._tokenize(d))
|
||||
idf = (
|
||||
math.log(len(all_docs) / (doc_count + 1)) + 1 if doc_count > 0 else 0
|
||||
)
|
||||
|
||||
vector[word] = term_freq * idf
|
||||
return vector
|
||||
|
||||
def _cosine_similarity(self, vec1, vec2):
|
||||
"""Menghitung kemiripan kosinus antara dua vektor TF-IDF."""
|
||||
intersection = set(vec1.keys()).intersection(set(vec2.keys()))
|
||||
numerator = sum(vec1[x] * vec2[x] for x in intersection)
|
||||
|
||||
sum1 = sum(val ** 2 for val in vec1.values())
|
||||
sum2 = sum(val ** 2 for val in vec2.values())
|
||||
denominator = math.sqrt(sum1) * math.sqrt(sum2)
|
||||
|
||||
if denominator == 0:
|
||||
return 0.0
|
||||
return numerator / denominator
|
||||
|
||||
def retrieve_with_scores(self, query, top_k=5):
|
||||
"""Mengembalikan daftar dokumen dan skornya berdasarkan Cosine Similarity."""
|
||||
query_vector = self._compute_tfidf(query, self.vocab, self.documents)
|
||||
scores = []
|
||||
|
||||
for i, doc_vec in enumerate(self.doc_vectors):
|
||||
score = self._cosine_similarity(query_vector, doc_vec)
|
||||
scores.append((score, self.documents[i]))
|
||||
|
||||
# Urutkan berdasarkan skor tertinggi
|
||||
scores.sort(key=lambda x: x[0], reverse=True)
|
||||
return scores[:top_k]
|
||||
|
||||
def generate_answer(self, query):
|
||||
"""Tahap Generation: Menggabungkan hasil retrieval dengan format jawaban RAG."""
|
||||
retrieved_scores = self.retrieve_with_scores(query, top_k=1)
|
||||
context = (
|
||||
retrieved_scores[0][1]
|
||||
if retrieved_scores and retrieved_scores[0][0] > 0
|
||||
else "Maaf, informasi tidak ditemukan."
|
||||
)
|
||||
|
||||
answer = (
|
||||
f"==================================================\n"
|
||||
f"HASIL GENERASI RAG - PT KERETA API INDONESIA (GAPEKA 2026)\n"
|
||||
f"==================================================\n"
|
||||
f"Pertanyaan : {query}\n\n"
|
||||
f"Konteks Relevan (Retrieved):\n-> {context}\n\n"
|
||||
f"Jawaban Sistem:\n"
|
||||
f"Berdasarkan informasi resmi kebijakan GAPEKA 2026 PT KAI, "
|
||||
f"berikut adalah detail informasinya:\n"
|
||||
f'"{context}"\n'
|
||||
f"=================================================="
|
||||
)
|
||||
return answer, context
|
||||
|
||||
|
||||
# ==========================================
|
||||
# MODUL EVALUASI RETRIEVAL (Hit@K & MRR)
|
||||
# ==========================================
|
||||
|
||||
# Daftar Uji (Pertanyaan dan kata kunci yang harus ada di dokumen yang diretrubusi)
|
||||
UJI = [
|
||||
("Kapan GAPEKA 2026 mulai diberlakukan?", "1 desember 2026"),
|
||||
("Di mana bisa pesan tiket kereta GAPEKA 2026?", "access by kai"),
|
||||
("Apa ada perubahan nama kereta api?", "penyederhanaan nama"),
|
||||
("Apakah ada penambahan titik pemberhentian stasiun?", "titik pemberhentian"),
|
||||
("Sebutkan rute kereta api baru yang beroperasi!", "ka borobudur"),
|
||||
]
|
||||
|
||||
|
||||
def cari_peringkat(rag_system, pertanyaan, kunci, k=5):
|
||||
"""Mencari peringkat dokumen pertama yang mengandung kata kunci."""
|
||||
hasil = rag_system.retrieve_with_scores(pertanyaan, top_k=k)
|
||||
kunci_bersih = " ".join(kunci.lower().split())
|
||||
|
||||
for peringkat, (score, doc) in enumerate(hasil, start=1):
|
||||
doc_bersih = " ".join(doc.lower().split())
|
||||
if kunci_bersih in doc_bersih and score > 0:
|
||||
return peringkat
|
||||
return None
|
||||
|
||||
|
||||
def hit_at_k(rag_system, uji_data, k=3):
|
||||
"""Menghitung metrik Hit@K."""
|
||||
benar = 0
|
||||
for pertanyaan, kunci in uji_data:
|
||||
peringkat = cari_peringkat(rag_system, pertanyaan, kunci, k)
|
||||
if peringkat is not None:
|
||||
benar += 1
|
||||
return benar / len(uji_data)
|
||||
|
||||
|
||||
def hitung_mrr(rag_system, uji_data):
|
||||
"""Menghitung metrik Mean Reciprocal Rank (MRR@5)."""
|
||||
total = 0
|
||||
for pertanyaan, kunci in uji_data:
|
||||
peringkat = cari_peringkat(rag_system, pertanyaan, kunci, 5)
|
||||
if peringkat is not None:
|
||||
total += 1 / peringkat
|
||||
return total / len(uji_data)
|
||||
|
||||
|
||||
# ==========================================
|
||||
# EKSEKUSI PROGRAM UTAMA
|
||||
# ==========================================
|
||||
if __name__ == "__main__":
|
||||
# 1. Dataset / Basis Pengetahuan Lokal KAI GAPEKA 2026
|
||||
kai_kb = [
|
||||
(
|
||||
"PT Kereta Api Indonesia (KAI) resmi memberlakukan Grafik Perjalanan "
|
||||
"Kereta Api (GAPEKA) 2026 mulai tanggal 1 Desember 2026."
|
||||
),
|
||||
(
|
||||
"Tiket perjalanan kereta api untuk jadwal GAPEKA 2026 sudah dapat "
|
||||
"dipesan secara online melalui aplikasi Access by KAI."
|
||||
),
|
||||
(
|
||||
"Terdapat penyesuaian serta penyederhanaan nama pada beberapa "
|
||||
"perjalanan kereta api agar lebih simpel dan mudah diingat oleh penumpang."
|
||||
),
|
||||
(
|
||||
"KAI menambah titik pemberhentian di sejumlah stasiun untuk "
|
||||
"mendukung mobilitas antarkota dan kawasan aglomerasi."
|
||||
),
|
||||
(
|
||||
"KAI menghadirkan layanan rute baru seperti KA Borobudur "
|
||||
"(Lempuyangan–Gambir PP, fakultatif) dan KA Pandawa "
|
||||
"(Yogyakarta–Semarang Tawang PP) yang mulai beroperasi pada 1 Desember 2026."
|
||||
),
|
||||
]
|
||||
|
||||
# 2. Inisialisasi Sistem RAG
|
||||
rag = KAIGapekaRAG(kai_kb)
|
||||
|
||||
# 3. Contoh Uji Jalankan Pipeline RAG Tunggal
|
||||
sample_query = "Kapan GAPEKA 2026 mulai diberlakukan dan di mana bisa pesan tiketnya?"
|
||||
final_answer, sources = rag.generate_answer(sample_query)
|
||||
print("\n" + final_answer)
|
||||
|
||||
# 4. Evaluasi Performa Retrieval (Hit@K & MRR)
|
||||
print("\n" + "=" * 50)
|
||||
print("EVALUASI PERFORMA RETRIEVAL KAI GAPEKA 2026")
|
||||
print("=" * 50)
|
||||
print(f"Jumlah Pertanyaan Uji : {len(UJI)}")
|
||||
|
||||
for k in (1, 3, 5):
|
||||
nilai_hit = hit_at_k(rag, UJI, k)
|
||||
print(f"Hit@{k} = {nilai_hit:.2f}")
|
||||
|
||||
mrr_score = hitung_mrr(rag, UJI)
|
||||
print(f"MRR@5 = {mrr_score:.2f}")
|
||||
|
||||
print("\nPERTANYAAN YANG GAGAL DITEMUKAN (Jika ada):")
|
||||
gagal = False
|
||||
for pertanyaan, kunci in UJI:
|
||||
peringkat = cari_peringkat(rag, pertanyaan, kunci, 5)
|
||||
if peringkat is None:
|
||||
gagal = True
|
||||
print(f"- {pertanyaan}")
|
||||
|
||||
if not gagal:
|
||||
print("Semua pertanyaan berhasil ditemukan oleh sistem retrieval!")
|
||||
print("=" * 50)
|
||||
Reference in New Issue
Block a user