Mengelola fondasi data produk AI: data cleaning, strategi chunking yang benar, dan data quality untuk RAG. Kalian membangun pipeline data yang menjaga knowledge base tetap bersih, ter-versioned, dan siap di-retrieve dengan hasil maksimaltype: text

Setiap sistem yang kalian bangun sejak episode 6 — embeddings, RAG, memory — berdiri di atas satu fondasi: data. Data yang kotor menghasilkan retrieval yang kotor, dan retrieval yang kotor menghasilkan jawaban yang salah. Episode ini membahas lapisan yang paling sering diremehkan namun paling menentukan kualitas: preprocessing.
Mengapa penting? Karena di dunia AI, garbage in, garbage out bukan slogan — ini hukum. Chunking yang asal-asalan membuat RAG tidak pernah akurat meskipun prompt dan model sudah optimal. Menguasai data pipeline adalah pembeda antara demo yang "sepertinya jalan" dan produk yang "benar-benar akurat".
Sebelum dokumen masuk ke vector DB, bersihkan dulu. Sumber kotoran yang paling umum:
| Kotoran | Contoh | Dampak |
|---|---|---|
| Duplikat & near-duplicate | Dokumen di-copy 3 kali | Retrieval mengembalikan isi sama, biaya membengkak |
| Format tidak konsisten | Tanggal "12/08" vs "2026-08-12" | Model salah interpretasi |
| Noise & boilerplate | Header, footer, navigasi web | Menghabiskan konteks, menurunkan skor relevansi |
| HTML/markup mentah | Tag dan escaping tidak dibersihkan | Vektor tercemar teks bukan konten |
| Bahasa campuran | Indonesia dicampur Inggris acak | Embedding terpecah, retrieval bising |
import re
def clean_text(text: str) -> str:
text = re.sub(r"<[^>]+>", "", text) # hapus HTML tag
text = re.sub(r"\s+", " ", text).strip() # normalisasi spasi
text = re.sub(r"(copyright .*|^#{2,}\s*)", "", text,
flags=re.IGNORECASE) # boilerplate kasar
return text
def dedupe(docs: list[str], threshold: float = 0.95) -> list[str]:
hasil: list[str] = []
for d in docs:
bersih = clean_text(d)
if not any(similar(bersih, h) > threshold for h in hasil): # placeholder
hasil.append(bersih)
return hasilUntuk near-duplicate di skala besar, gunakan embedding similarity (episode 6) — dua dokumen dengan kesamaan di atas ambang adalah duplikat.
Chunking adalah seni memecah dokumen menjadi potongan yang di-retrieve. Tidak ada ukuran universal — semuanya trade-off:
| Pendekatan | Kelebihan | Kekurangan |
|---|---|---|
| Fixed-size (N token + overlap) | Sederhana, seragam | Memotong tengah konsep |
| Paragraf | Menjaga unit makna | Panjang tidak seragam |
| Berbasis struktur (heading/section) | Konteks jelas per bagian | Tergantung struktur dokumen |
| Berbasis semantik (LLM split) | Batas paling bermakna | Mahal dan lambat |
Aturan praktis yang umum:
import re
def chunk_markdown(text: str, max_len: int = 800) -> list[dict]:
"""Pecah markdown per heading, lalu lanjutkan ke chunk berikutnya."""
sections = re.split(r"(?=^## )", text, flags=re.MULTILINE)
chunks = []
for i, sec in enumerate(sections):
heading = sec.splitlines()[0][:60] if sec.strip() else "(awal)"
parts = [sec[j:j + max_len] for j in range(0, len(sec), max_len)]
for p in parts:
chunks.append({"text": p, "section": heading, "pos": i})
return chunksSelain kebersihan per dokumen, RAG butuh quality assurance pada level koleksi. Praktek yang wajib:
source dan updated_at — ini fondasi grounding dan compliance.col.add(
ids=[f"{doc_id}-{i}" for i in range(len(chunks))],
documents=[c["text"] for c in chunks],
metadatas=[{"source": doc_id, "section": c["section"],
"updated_at": "2026-08-16"} for c in chunks],
)Pipeline produksi yang sehat memiliki lima tahap berulang:
Tip
Otomatisasi pipeline ini (misal dengan cron atau CI schedule) menjadikannya data pipeline yang hidup — bukan skrip sekali jalan. Setiap perubahan sumber dokumen memicu ingest ulang. Kita bangun backend-nya di episode 16.
Inti yang harus dibawa pulang:
Di episode 16 selanjutnya kita mengubah sistem menjadi produk: backend & API untuk AI apps — FastAPI, streaming, rate limiting, dan auth untuk API yang dipakai aplikasi nyata. Sampai jumpa di episode 16!