Kenapa RAG Pipeline Masih Jadi Tulang Punggung Aplikasi AI di 2026?
Kalau kamu pernah bertanya-tanya kenapa chatbot AI perusahaanmu masih sering "berhalusinasi" — memberikan jawaban yang terdengar meyakinkan tapi ternyata salah total — jawabannya hampir selalu soal bagaimana sistem itu mengakses dan memproses informasi. Nah, di sinilah Retrieval-Augmented Generation (RAG) masuk sebagai penyelamat.
RAG sendiri bukan konsep baru. Pertama kali diperkenalkan oleh Lewis et al. dari Meta AI di tahun 2020, konsep ini menggabungkan kemampuan generatif LLM dengan mekanisme retrieval dari basis data eksternal. Tapi jujur, di tahun 2026, RAG sudah berevolusi jauh banget melampaui arsitektur aslinya. Kita sekarang bicara tentang Agentic RAG, GraphRAG, hybrid search dengan reranking, dan adaptive chunking — teknik-teknik yang bisa meningkatkan akurasi retrieval hingga 40% dibandingkan pendekatan naif.
Kalau mau disederhanakan: 2024 adalah tahun RAG, 2025 adalah tahun Agents, dan 2026? Ini adalah tahun Agentic RAG — di mana kedua paradigma ini akhirnya menyatu jadi sistem yang benar-benar cerdas.
Artikel ini akan memandu kamu membangun pipeline RAG tingkat produksi dari nol. Lengkap dengan kode praktis dan strategi optimasi terkini. So, mari kita mulai dari fondasi.
Arsitektur Pipeline RAG Modern: 7 Lapisan yang Wajib Kamu Pahami
Pipeline RAG produksi di 2026 bukan lagi sekadar "embed dokumen, simpan di vektor database, lalu retrieve." Arsitektur modern terdiri dari tujuh lapisan yang masing-masing punya peran krusial:
- Klasifikasi Intent — Memahami apa yang sebenarnya ditanyakan pengguna
- Transformasi Query — Mengubah query mentah jadi query yang optimal untuk retrieval
- Deteksi Bahasa — Menangani input multibahasa dengan tepat
- Input Guardrails — Memfilter query yang berbahaya atau di luar cakupan
- Hybrid Search — Menggabungkan pencarian leksikal dan semantik
- Reranking — Menyusun ulang hasil berdasarkan relevansi sebenarnya
- Generasi dengan Metadata Filtering — Menghasilkan jawaban yang akurat dan terkontrol
Setiap lapisan ini saling terhubung dan memengaruhi kualitas output akhir. Kegagalan di satu lapisan bakal berdampak cascading ke lapisan berikutnya — percayalah, saya sudah mengalaminya sendiri saat debugging pipeline yang tiba-tiba menghasilkan jawaban ngawur karena satu masalah kecil di layer transformasi query.
Jadi, mari kita bahas satu per satu.
Langkah 1: Pemrosesan Dokumen dan Strategi Chunking
Sebelum dokumen bisa di-retrieve, dokumen harus dipecah menjadi potongan-potongan (chunks) yang bisa diproses. Dan honestly, di sinilah banyak pipeline RAG gagal — karena menggunakan strategi chunking yang terlalu sederhana.
Fixed-Size Chunking vs Semantic Chunking
Pendekatan paling dasar adalah fixed-size chunking: memotong dokumen setiap 512 atau 1024 token dengan overlap tertentu. Simpel, cepat, tapi sering memutus konteks di tengah paragraf atau bahkan di tengah kalimat. Bukan ideal.
Benchmark terbaru menunjukkan bahwa pendekatan fixed-size berkinerja 15-20% lebih buruk dibandingkan metode semantik dalam task question-answering yang kompleks. Ini angka yang nggak bisa diabaikan.
Semantic chunking jauh lebih cerdas. Alih-alih memotong berdasarkan jumlah token, pendekatan ini menganalisis makna dan struktur teks untuk menentukan batas potongan yang natural. Hasilnya? Konteks terjaga, informasi tidak terpecah, dan retrieval jadi lebih akurat.
from langchain_experimental.text_splitter import SemanticChunker
from langchain_openai import OpenAIEmbeddings
from langchain.text_splitter import RecursiveCharacterTextSplitter
# Pendekatan 1: Fixed-size chunking (baseline)
fixed_splitter = RecursiveCharacterTextSplitter(
chunk_size=512,
chunk_overlap=50,
separators=["\n\n", "\n", ". ", " ", ""]
)
# Pendekatan 2: Semantic chunking (direkomendasikan)
embeddings = OpenAIEmbeddings(model="text-embedding-3-large")
semantic_splitter = SemanticChunker(
embeddings=embeddings,
breakpoint_threshold_type="percentile",
breakpoint_threshold_amount=85 # Threshold untuk menentukan batas chunk
)
# Contoh penggunaan
dokumen = """
Retrieval-Augmented Generation (RAG) adalah teknik yang menggabungkan
kemampuan LLM dengan retrieval dari basis data eksternal. Teknik ini
pertama kali diperkenalkan oleh Lewis et al. pada tahun 2020.
Dalam implementasi modern, RAG menggunakan vector database untuk
menyimpan embedding dokumen. Proses pencarian dilakukan dengan
menghitung kesamaan kosinus antara embedding query dan embedding
dokumen yang tersimpan.
"""
# Fixed-size chunks
chunks_fixed = fixed_splitter.split_text(dokumen)
print(f"Fixed chunks: {len(chunks_fixed)} potongan")
# Semantic chunks
chunks_semantic = semantic_splitter.split_text(dokumen)
print(f"Semantic chunks: {len(chunks_semantic)} potongan")
# Semantic chunking akan memisahkan berdasarkan topik:
# Chunk 1: Definisi dan sejarah RAG
# Chunk 2: Implementasi modern dengan vector database
for i, chunk in enumerate(chunks_semantic):
print(f"\n--- Chunk {i+1} ---")
print(chunk)Adaptive Chunking: Pendekatan Terdepan di 2026
Nah, ini yang menarik. Riset terbaru menunjukkan bahwa adaptive chunking menghasilkan akurasi tertinggi — 87% dibandingkan baseline 50% — dengan relevansi mencapai 93%. Angka yang cukup fantastis kalau kamu tanya saya.
Adaptive chunking bekerja dengan menggabungkan threshold kesamaan dan ukuran jendela variabel untuk menjaga kontinuitas topik sambil menghindari pengenceran informasi. Intinya, chunk-nya menyesuaikan diri dengan konten — bukan sebaliknya.
import numpy as np
from sentence_transformers import SentenceTransformer
class AdaptiveChunker:
"""Chunker adaptif yang menyesuaikan ukuran chunk berdasarkan konten."""
def __init__(
self,
model_name: str = "all-MiniLM-L6-v2",
similarity_threshold: float = 0.75,
min_chunk_size: int = 100,
max_chunk_size: int = 1000
):
self.model = SentenceTransformer(model_name)
self.similarity_threshold = similarity_threshold
self.min_chunk_size = min_chunk_size
self.max_chunk_size = max_chunk_size
def _hitung_kesamaan(self, emb1: np.ndarray, emb2: np.ndarray) -> float:
"""Hitung cosine similarity antara dua embedding."""
return np.dot(emb1, emb2) / (np.linalg.norm(emb1) * np.linalg.norm(emb2))
def chunk(self, teks: str) -> list[str]:
"""Pecah teks menjadi chunks adaptif berdasarkan kesamaan semantik."""
# Pisahkan menjadi kalimat
kalimat_list = [s.strip() for s in teks.split('.') if s.strip()]
if not kalimat_list:
return [teks]
# Hitung embedding untuk setiap kalimat
embeddings = self.model.encode(kalimat_list)
chunks = []
chunk_saat_ini = [kalimat_list[0]]
emb_chunk = embeddings[0]
for i in range(1, len(kalimat_list)):
kesamaan = self._hitung_kesamaan(emb_chunk, embeddings[i])
panjang_chunk = len('. '.join(chunk_saat_ini))
# Tentukan apakah harus mulai chunk baru
mulai_baru = (
kesamaan < self.similarity_threshold
and panjang_chunk >= self.min_chunk_size
) or panjang_chunk >= self.max_chunk_size
if mulai_baru:
chunks.append('. '.join(chunk_saat_ini) + '.')
chunk_saat_ini = [kalimat_list[i]]
emb_chunk = embeddings[i]
else:
chunk_saat_ini.append(kalimat_list[i])
# Update embedding chunk dengan rata-rata
emb_chunk = np.mean(
[emb_chunk, embeddings[i]], axis=0
)
# Tambahkan chunk terakhir
if chunk_saat_ini:
chunks.append('. '.join(chunk_saat_ini) + '.')
return chunks
# Penggunaan
chunker = AdaptiveChunker(
similarity_threshold=0.75,
min_chunk_size=100,
max_chunk_size=800
)
chunks = chunker.chunk(dokumen_panjang)
print(f"Dihasilkan {len(chunks)} chunks adaptif")Contextual Headers: Jangan Lupakan Metadata
Satu hal yang sering di-skip oleh developer (padahal dampaknya besar): hierarchical chunking dengan header kontekstual. Artinya, setiap potongan teks nggak cuma berisi konten, tapi juga informasi tentang di mana posisinya dalam dokumen asli — judul bab, sub-bab, dan hierarki struktural.
Hasilnya? 30-40% lebih sedikit retrieval call yang diperlukan untuk menjawab pertanyaan yang sama. Lumayan banget kan?
def tambahkan_header_kontekstual(chunks: list[str], metadata: dict) -> list[dict]:
"""Tambahkan header kontekstual ke setiap chunk untuk meningkatkan retrieval."""
enriched_chunks = []
for i, chunk in enumerate(chunks):
enriched = {
"content": chunk,
"metadata": {
"source": metadata.get("filename", "unknown"),
"chapter": metadata.get("current_chapter", ""),
"section": metadata.get("current_section", ""),
"chunk_index": i,
"total_chunks": len(chunks),
"contextual_header": f"{metadata.get('title', '')} > "
f"{metadata.get('current_chapter', '')} > "
f"{metadata.get('current_section', '')}"
}
}
# Prepend contextual header ke konten untuk embedding
enriched["content_for_embedding"] = (
f"[{enriched['metadata']['contextual_header']}]\n\n"
f"{chunk}"
)
enriched_chunks.append(enriched)
return enriched_chunksLangkah 2: Memilih dan Mengonfigurasi Vector Database
Setelah dokumen di-chunk, langkah selanjutnya adalah menyimpan embedding-nya di vector database. Di 2026, ada tren besar yang perlu kamu perhatikan: vektor bukan lagi tipe database tersendiri, melainkan tipe data yang bisa diintegrasikan ke database yang sudah ada.
Berikut perbandingan vector database populer di 2026:
- Pinecone — Fully managed, mudah digunakan, cocok untuk startup dan prototyping cepat. Mendukung metadata filtering dan hybrid search bawaan
- Milvus / Zilliz — Open-source, performa tinggi, cocok untuk skala enterprise. Mendukung GPU-accelerated search
- Qdrant — Open-source dengan performa sangat baik, filtering canggih, dan arsitektur yang efisien secara memori
- Weaviate — Open-source dengan dukungan bawaan untuk knowledge graph dan hybrid search
- pgvector (PostgreSQL) — Ekstensi PostgreSQL, ideal kalau kamu sudah pakai Postgres dan nggak mau tambah infrastruktur baru
- ChromaDB — Ringan, cocok untuk pengembangan lokal dan prototyping
Pilihan mana yang terbaik? Tergantung kebutuhan. Tapi secara pribadi, untuk proyek baru saya biasanya mulai dengan Qdrant atau pgvector (kalau sudah ada Postgres), lalu migrasi ke solusi managed kalau sudah butuh skalabilitas lebih.
Setup Pinecone dengan LangChain
from pinecone import Pinecone, ServerlessSpec
from langchain_pinecone import PineconeVectorStore
from langchain_openai import OpenAIEmbeddings
# Inisialisasi Pinecone
pc = Pinecone(api_key="your-api-key")
# Buat index jika belum ada
index_name = "rag-produksi"
if index_name not in pc.list_indexes().names():
pc.create_index(
name=index_name,
dimension=3072, # Dimensi text-embedding-3-large
metric="cosine",
spec=ServerlessSpec(
cloud="aws",
region="us-east-1"
)
)
# Inisialisasi embedding model
embeddings = OpenAIEmbeddings(
model="text-embedding-3-large",
dimensions=3072
)
# Buat vector store
vector_store = PineconeVectorStore(
index=pc.Index(index_name),
embedding=embeddings,
text_key="content"
)
# Indexing dokumen
from langchain_core.documents import Document
documents = [
Document(
page_content=chunk["content_for_embedding"],
metadata=chunk["metadata"]
)
for chunk in enriched_chunks
]
# Batch upsert untuk performa optimal
vector_store.add_documents(documents, batch_size=100)
print(f"Berhasil mengindeks {len(documents)} dokumen")Setup Qdrant untuk Penggunaan Open-Source
from qdrant_client import QdrantClient
from qdrant_client.models import (
VectorParams, Distance, PointStruct,
Filter, FieldCondition, MatchValue
)
from langchain_community.vectorstores import Qdrant
# Koneksi ke Qdrant (lokal atau cloud)
client = QdrantClient(
url="http://localhost:6333",
# Atau untuk Qdrant Cloud:
# url="https://your-cluster.qdrant.io",
# api_key="your-api-key"
)
# Buat collection
client.create_collection(
collection_name="dokumen_produksi",
vectors_config=VectorParams(
size=3072,
distance=Distance.COSINE
)
)
# Buat vector store dengan LangChain
qdrant_store = Qdrant(
client=client,
collection_name="dokumen_produksi",
embeddings=embeddings,
)
# Indexing dengan metadata filtering
qdrant_store.add_documents(documents)
# Pencarian dengan filter metadata
hasil = qdrant_store.similarity_search(
query="Bagaimana cara mengoptimasi RAG pipeline?",
k=5,
filter=Filter(
must=[
FieldCondition(
key="metadata.chapter",
match=MatchValue(value="Optimasi")
)
]
)
)
for doc in hasil:
print(f"Skor: {doc.metadata.get('score', 'N/A')}")
print(f"Konten: {doc.page_content[:200]}...")
print("---")Langkah 3: Hybrid Search — Menggabungkan BM25 dan Pencarian Semantik
Ini salah satu pelajaran terpenting yang saya dapat dari deploy RAG di produksi: jangan hanya mengandalkan pencarian semantik.
Pencarian vektor memang bagus untuk menangkap makna, tapi sering melewatkan kecocokan kata kunci yang tepat. Sebaliknya, BM25 (pencarian leksikal) sangat jago untuk kecocokan eksak tapi nggak ngerti sinonim atau parafrase. Dua-duanya punya kelemahan masing-masing.
Solusinya? Hybrid search yang menggabungkan keduanya. Dan datanya cukup meyakinkan — hybrid search dengan reranking mencapai MRR 66,43% dibandingkan 56,72% untuk pencarian semantik saja. Peningkatan 9,3 poin persentase yang sangat berarti di dunia nyata.
from rank_bm25 import BM25Okapi
from sentence_transformers import SentenceTransformer
import numpy as np
class HybridRetriever:
"""Retriever hybrid yang menggabungkan BM25 dan pencarian semantik."""
def __init__(
self,
documents: list[str],
model_name: str = "all-MiniLM-L6-v2",
alpha: float = 0.5 # Bobot: 0=hanya BM25, 1=hanya semantik
):
self.documents = documents
self.alpha = alpha
# Setup BM25
tokenized_docs = [doc.lower().split() for doc in documents]
self.bm25 = BM25Okapi(tokenized_docs)
# Setup model semantik
self.model = SentenceTransformer(model_name)
self.doc_embeddings = self.model.encode(documents)
def _normalize_scores(self, scores: np.ndarray) -> np.ndarray:
"""Normalisasi skor ke rentang [0, 1] menggunakan min-max scaling."""
min_s, max_s = scores.min(), scores.max()
if max_s - min_s == 0:
return np.zeros_like(scores)
return (scores - min_s) / (max_s - min_s)
def retrieve(self, query: str, top_k: int = 10) -> list[dict]:
"""Lakukan hybrid retrieval dengan penggabungan skor."""
# Skor BM25
tokenized_query = query.lower().split()
bm25_scores = self.bm25.get_scores(tokenized_query)
bm25_normalized = self._normalize_scores(bm25_scores)
# Skor semantik (cosine similarity)
query_embedding = self.model.encode([query])[0]
semantic_scores = np.dot(self.doc_embeddings, query_embedding) / (
np.linalg.norm(self.doc_embeddings, axis=1)
* np.linalg.norm(query_embedding)
)
semantic_normalized = self._normalize_scores(semantic_scores)
# Gabungkan skor dengan weighted combination
hybrid_scores = (
(1 - self.alpha) * bm25_normalized
+ self.alpha * semantic_normalized
)
# Urutkan dan ambil top-k
top_indices = np.argsort(hybrid_scores)[::-1][:top_k]
results = []
for idx in top_indices:
results.append({
"document": self.documents[idx],
"hybrid_score": float(hybrid_scores[idx]),
"bm25_score": float(bm25_normalized[idx]),
"semantic_score": float(semantic_normalized[idx]),
"rank": len(results) + 1
})
return results
# Penggunaan
retriever = HybridRetriever(
documents=semua_chunks,
alpha=0.6 # Sedikit lebih berat ke semantik
)
hasil = retriever.retrieve(
query="Bagaimana cara mengoptimasi performa RAG?",
top_k=10
)
for r in hasil[:5]:
print(f"Rank {r['rank']}: BM25={r['bm25_score']:.3f}, "
f"Semantik={r['semantic_score']:.3f}, "
f"Hybrid={r['hybrid_score']:.3f}")
print(f" {r['document'][:100]}...")Langkah 4: Reranking dengan Cross-Encoder — Presisi Tingkat Tinggi
Setelah hybrid search menghasilkan kandidat dokumen, langkah selanjutnya adalah reranking. Dan ini sering jadi game-changer yang banyak orang lewatkan.
Cross-encoder adalah model neural yang dilatih pada jutaan penilaian relevansi manusia, membuatnya jauh lebih akurat dalam menilai relevansi pasangan query-dokumen dibandingkan pencarian bi-encoder biasa.
Kenapa butuh reranking? Karena pencarian awal (baik BM25 maupun semantik) menggunakan representasi independen untuk query dan dokumen. Cross-encoder, sebaliknya, memproses pasangan query-dokumen secara bersamaan, sehingga bisa menangkap hubungan yang lebih halus — nuansa yang nggak tertangkap oleh embedding terpisah.
from sentence_transformers import CrossEncoder
class RerankerPipeline:
"""Pipeline reranking menggunakan cross-encoder."""
def __init__(
self,
model_name: str = "cross-encoder/ms-marco-MiniLM-L-12-v2"
):
self.cross_encoder = CrossEncoder(model_name)
def rerank(
self,
query: str,
candidates: list[dict],
top_k: int = 5
) -> list[dict]:
"""Rerank kandidat dokumen menggunakan cross-encoder."""
# Siapkan pasangan query-dokumen
pairs = [
[query, candidate["document"]]
for candidate in candidates
]
# Hitung skor cross-encoder
ce_scores = self.cross_encoder.predict(pairs)
# Tambahkan skor ke kandidat
for i, candidate in enumerate(candidates):
candidate["rerank_score"] = float(ce_scores[i])
# Urutkan berdasarkan skor reranking
reranked = sorted(
candidates,
key=lambda x: x["rerank_score"],
reverse=True
)
return reranked[:top_k]
# Pipeline lengkap: Hybrid Search → Reranking
reranker = RerankerPipeline()
# Step 1: Ambil 20 kandidat dari hybrid search
kandidat = retriever.retrieve(query="optimasi RAG pipeline", top_k=20)
# Step 2: Rerank menjadi 5 dokumen paling relevan
hasil_final = reranker.rerank(
query="optimasi RAG pipeline",
candidates=kandidat,
top_k=5
)
print("Hasil setelah reranking:")
for r in hasil_final:
print(f" Rerank Score: {r['rerank_score']:.4f}")
print(f" Hybrid Score: {r['hybrid_score']:.4f}")
print(f" Dokumen: {r['document'][:150]}...")
print()Satu catatan penting: cross-encoder memang lebih mahal secara komputasi dibandingkan bi-encoder. Strategi yang umum (dan yang saya pakai juga) adalah mengambil 20-50 kandidat dari pencarian awal yang cepat, lalu mererank menjadi 3-5 dokumen final. Pendekatan dua tahap ini memberikan keseimbangan optimal antara akurasi dan latensi.
Langkah 5: Query Transformation — Menjembatani Gap antara Query dan Dokumen
Ada masalah tersembunyi dalam RAG yang jarang dibahas di tutorial pemula: vocabulary mismatch. Pengguna bertanya dengan cara yang berbeda dari bagaimana informasi disimpan dalam dokumen. Transformasi query mengatasi masalah ini.
Hypothetical Document Embedding (HyDE)
Teknik HyDE ini cara kerjanya unik banget. Alih-alih langsung mencari berdasarkan query pengguna, LLM diminta untuk menghasilkan jawaban hipotetis terlebih dahulu. Jawaban hipotetis ini kemudian di-embed dan digunakan untuk retrieval.
Logikanya? Jawaban hipotetis kemungkinan mengandung terminologi yang mirip dengan dokumen asli, jadi hasil retrieval-nya seringkali lebih relevan. Clever, kan?
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_core.prompts import ChatPromptTemplate
class HyDERetriever:
"""Retriever dengan Hypothetical Document Embedding."""
def __init__(self, vector_store, llm=None):
self.vector_store = vector_store
self.llm = llm or ChatOpenAI(model="gpt-4o-mini", temperature=0.7)
self.embeddings = OpenAIEmbeddings(model="text-embedding-3-large")
self.prompt = ChatPromptTemplate.from_template("""
Kamu adalah ahli di bidang yang relevan. Berdasarkan pertanyaan
berikut, tulis paragraf informatif yang menjawab pertanyaan
tersebut seolah-olah kamu sedang menulis dokumentasi teknis.
Pertanyaan: {query}
Jawaban hipotetis:
""")
def retrieve(self, query: str, k: int = 5) -> list:
"""Retrieve dokumen menggunakan HyDE."""
# Step 1: Generate jawaban hipotetis
chain = self.prompt | self.llm
jawaban_hipotetis = chain.invoke({"query": query}).content
# Step 2: Gunakan jawaban hipotetis untuk retrieval
hasil = self.vector_store.similarity_search(
jawaban_hipotetis,
k=k
)
return hasil
# Penggunaan
hyde_retriever = HyDERetriever(vector_store=vector_store)
hasil = hyde_retriever.retrieve(
"Bagaimana cara mengurangi halusinasi pada sistem RAG?"
)Multi-Query Retrieval
Teknik lain yang nggak kalah efektif: multi-query retrieval. Idenya sederhana — hasilkan beberapa variasi query dari pertanyaan asli, jalankan retrieval untuk masing-masing, lalu gabungkan hasilnya. Ini meningkatkan recall secara signifikan karena setiap variasi query bisa menangkap dokumen yang berbeda.
from langchain.retrievers import MultiQueryRetriever
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.7)
multi_retriever = MultiQueryRetriever.from_llm(
retriever=vector_store.as_retriever(search_kwargs={"k": 5}),
llm=llm
)
# Secara otomatis menghasilkan variasi query dan menggabungkan hasil
hasil = multi_retriever.invoke(
"Apa saja best practice untuk deployment RAG di produksi?"
)
# LLM akan menghasilkan variasi seperti:
# 1. "Strategi deployment RAG pipeline di lingkungan production"
# 2. "Panduan implementasi RAG system untuk enterprise"
# 3. "Checklist produksi untuk sistem retrieval-augmented generation"Langkah 6: GraphRAG — Ketika Vektor Saja Tidak Cukup
Oke, ini bagian yang menurut saya paling exciting. Di tahun 2026, GraphRAG muncul sebagai evolusi signifikan dari RAG tradisional. Alih-alih hanya menyimpan dokumen sebagai embedding vektor, GraphRAG merepresentasikan dokumen sebagai knowledge graph — mengekstrak entitas dan hubungan antar entitas, lalu mengorganisasinya jadi struktur graf yang saling terhubung.
Kenapa ini penting? Karena banyak pertanyaan yang butuh pemahaman relasional.
"Siapa yang bekerja dengan siapa?", "Proyek apa yang terkait dengan teknologi ini?", "Bagaimana konsep A memengaruhi konsep B?" — pertanyaan-pertanyaan kayak gini sulit dijawab hanya dengan pencarian kesamaan vektor. Kamu butuh pemahaman tentang hubungan antar entitas.
Arsitektur Hybrid: Vector + Graph
Pendekatan terbaik di 2026 bukan memilih antara vektor dan graf, melainkan menggabungkan keduanya:
- Vector search untuk mendapatkan kandidat awal berdasarkan kesamaan semantik
- Graph traversal untuk memperluas konteks dengan entitas dan hubungan terkait
- Reranking untuk menyusun ulang hasil gabungan
from neo4j import GraphDatabase
from langchain_community.graphs import Neo4jGraph
from langchain_openai import ChatOpenAI
class HybridGraphVectorRAG:
"""RAG hybrid yang menggabungkan vector search dan graph traversal."""
def __init__(self, vector_store, neo4j_uri, neo4j_auth):
self.vector_store = vector_store
self.graph = Neo4jGraph(
url=neo4j_uri,
username=neo4j_auth[0],
password=neo4j_auth[1]
)
self.llm = ChatOpenAI(model="gpt-4o", temperature=0)
def _extract_entities(self, query: str) -> list[str]:
"""Ekstrak entitas dari query menggunakan LLM."""
response = self.llm.invoke(
f"Ekstrak entitas kunci dari query berikut sebagai "
f"daftar dipisahkan koma: {query}"
)
return [e.strip() for e in response.content.split(",")]
def _graph_search(self, entities: list[str], depth: int = 2) -> str:
"""Cari informasi relasional dari knowledge graph."""
context_parts = []
for entity in entities:
# Query graph untuk entitas dan relasinya
result = self.graph.query(f"""
MATCH (n)-[r]-(m)
WHERE toLower(n.name) CONTAINS toLower('{entity}')
RETURN n.name as source, type(r) as relation,
m.name as target, m.description as desc
LIMIT 20
""")
for record in result:
context_parts.append(
f"{record['source']} --[{record['relation']}]--> "
f"{record['target']}: {record.get('desc', '')}"
)
return "\n".join(context_parts)
def retrieve(self, query: str, k: int = 5) -> dict:
"""Lakukan retrieval hybrid vector + graph."""
# Step 1: Vector search
vector_results = self.vector_store.similarity_search(query, k=k)
vector_context = "\n\n".join(
[doc.page_content for doc in vector_results]
)
# Step 2: Graph search
entities = self._extract_entities(query)
graph_context = self._graph_search(entities)
# Step 3: Gabungkan konteks
combined_context = (
f"=== Konteks dari Pencarian Dokumen ===\n"
f"{vector_context}\n\n"
f"=== Konteks dari Knowledge Graph ===\n"
f"{graph_context}"
)
return {
"context": combined_context,
"vector_results": vector_results,
"graph_entities": entities,
"graph_context": graph_context
}Langkah 7: Evaluasi Pipeline RAG — Metrik yang Harus Kamu Ukur
Membangun pipeline RAG tanpa sistem evaluasi yang baik itu seperti nyetir malam-malam tanpa lampu — mungkin bisa sampai tujuan, tapi kemungkinan besar bakal nabrak sesuatu. Di 2026, evaluasi RAG sudah jadi disiplin tersendiri dengan metrik-metrik standar yang wajib dipantau.
Empat Metrik Utama RAG
- Answer Relevancy — Seberapa relevan jawaban yang dihasilkan terhadap pertanyaan yang diajukan?
- Faithfulness — Apakah jawaban benar-benar didasarkan pada konteks yang di-retrieve, atau ada "halusinasi"?
- Context Precision — Seberapa banyak dari konteks yang di-retrieve yang benar-benar relevan?
- Context Recall — Apakah semua informasi yang diperlukan berhasil di-retrieve?
Keempat metrik ini saling melengkapi. Faithfulness tinggi tapi recall rendah? Berarti jawaban kamu akurat tapi nggak lengkap. Relevancy tinggi tapi precision rendah? Berarti kamu retrieve terlalu banyak dokumen yang nggak relevan (dan itu buang-buang token).
from ragas import evaluate
from ragas.metrics import (
faithfulness,
answer_relevancy,
context_precision,
context_recall
)
from datasets import Dataset
# Siapkan dataset evaluasi
eval_data = {
"question": [
"Apa itu RAG?",
"Bagaimana cara kerja hybrid search?",
"Apa keuntungan GraphRAG?"
],
"answer": [
jawaban_dari_pipeline_1,
jawaban_dari_pipeline_2,
jawaban_dari_pipeline_3
],
"contexts": [
[konteks_retrieve_1],
[konteks_retrieve_2],
[konteks_retrieve_3]
],
"ground_truth": [
"RAG adalah teknik yang menggabungkan retrieval dengan generation...",
"Hybrid search menggabungkan BM25 dan pencarian semantik...",
"GraphRAG menggunakan knowledge graph untuk retrieval relasional..."
]
}
dataset = Dataset.from_dict(eval_data)
# Jalankan evaluasi
hasil_evaluasi = evaluate(
dataset=dataset,
metrics=[
faithfulness,
answer_relevancy,
context_precision,
context_recall
]
)
print("Hasil Evaluasi RAG Pipeline:")
print(f" Faithfulness: {hasil_evaluasi['faithfulness']:.4f}")
print(f" Answer Relevancy: {hasil_evaluasi['answer_relevancy']:.4f}")
print(f" Context Precision: {hasil_evaluasi['context_precision']:.4f}")
print(f" Context Recall: {hasil_evaluasi['context_recall']:.4f}")Monitoring di Produksi
Evaluasi satu kali nggak cukup — itu cuma snapshot. Di lingkungan produksi, kamu perlu monitoring terus-menerus. Platform seperti Maxim, Evidently AI, dan Braintrust menyediakan dashboard untuk melacak metrik RAG secara real-time, mendeteksi degradasi kualitas, dan mengidentifikasi pola kegagalan sebelum pengguna mulai komplain.
import logging
from datetime import datetime
class RAGMonitor:
"""Monitor sederhana untuk melacak kualitas pipeline RAG di produksi."""
def __init__(self):
self.logger = logging.getLogger("rag_monitor")
self.metrics_history = []
def log_query(
self,
query: str,
retrieved_docs: list,
answer: str,
latency_ms: float,
user_feedback: float = None
):
"""Catat metrik untuk setiap query yang diproses."""
metric = {
"timestamp": datetime.utcnow().isoformat(),
"query": query,
"num_docs_retrieved": len(retrieved_docs),
"answer_length": len(answer),
"latency_ms": latency_ms,
"user_feedback": user_feedback,
"has_empty_context": len(retrieved_docs) == 0,
}
self.metrics_history.append(metric)
# Alert jika latensi terlalu tinggi
if latency_ms > 5000:
self.logger.warning(
f"Latensi tinggi: {latency_ms}ms untuk query: {query[:50]}"
)
# Alert jika tidak ada dokumen yang ditemukan
if len(retrieved_docs) == 0:
self.logger.error(
f"Tidak ada konteks ditemukan untuk: {query[:50]}"
)
return metric
def get_summary(self) -> dict:
"""Dapatkan ringkasan metrik."""
if not self.metrics_history:
return {}
latencies = [m["latency_ms"] for m in self.metrics_history]
feedbacks = [
m["user_feedback"] for m in self.metrics_history
if m["user_feedback"] is not None
]
return {
"total_queries": len(self.metrics_history),
"avg_latency_ms": sum(latencies) / len(latencies),
"p95_latency_ms": sorted(latencies)[int(len(latencies) * 0.95)],
"empty_context_rate": sum(
1 for m in self.metrics_history if m["has_empty_context"]
) / len(self.metrics_history),
"avg_user_satisfaction": (
sum(feedbacks) / len(feedbacks) if feedbacks else None
),
}Langkah 8: Agentic RAG — Evolusi Menuju Sistem yang Benar-Benar Cerdas
Nah, sekarang kita masuk ke bagian yang paling seru. Di tahun 2026, Agentic RAG jadi paradigma dominan. Berbeda dengan RAG tradisional yang mengikuti pipeline linear (query, retrieve, generate — selesai), Agentic RAG menambahkan pola penalaran seperti planning, reflection, tool use, dan multi-agent collaboration.
Dan angkanya berbicara: riset menunjukkan bahwa sistem multi-agent mengungguli pendekatan single-agent hingga 90,2%. Ini bukan peningkatan kecil — ini perubahan paradigma yang fundamental.
Ciri Khas Agentic RAG
- Iterative Retrieval — Agen nggak cuma retrieve sekali, tapi bisa melakukan beberapa putaran retrieval berdasarkan informasi yang sudah didapat
- Self-Verification — Agen memverifikasi klaim sebelum merespons, mengurangi halusinasi secara drastis
- Tool Use — Agen bisa menggunakan alat selain retriever, seperti kalkulator, code executor, atau API eksternal
- Task Decomposition — Pertanyaan kompleks dipecah jadi sub-tugas yang masing-masing ditangani secara optimal
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from typing import TypedDict, Annotated
import operator
class AgenticRAGState(TypedDict):
"""State untuk Agentic RAG pipeline."""
query: str
sub_queries: list[str]
retrieved_contexts: Annotated[list[str], operator.add]
verification_results: list[dict]
final_answer: str
iteration: int
def decompose_query(state: AgenticRAGState) -> AgenticRAGState:
"""Pecah query kompleks menjadi sub-query."""
llm = ChatOpenAI(model="gpt-4o", temperature=0)
response = llm.invoke(
f"Pecah pertanyaan berikut menjadi 2-4 sub-pertanyaan "
f"yang lebih spesifik:\n\n{state['query']}"
)
sub_queries = [
q.strip() for q in response.content.split("\n")
if q.strip()
]
return {"sub_queries": sub_queries, "iteration": 0}
def iterative_retrieve(state: AgenticRAGState) -> AgenticRAGState:
"""Retrieve konteks untuk setiap sub-query."""
contexts = []
for sub_q in state["sub_queries"]:
# Gunakan hybrid retriever + reranker
kandidat = retriever.retrieve(sub_q, top_k=10)
reranked = reranker.rerank(sub_q, kandidat, top_k=3)
for r in reranked:
contexts.append(r["document"])
return {
"retrieved_contexts": contexts,
"iteration": state["iteration"] + 1
}
def verify_and_generate(state: AgenticRAGState) -> AgenticRAGState:
"""Verifikasi konteks dan hasilkan jawaban."""
llm = ChatOpenAI(model="gpt-4o", temperature=0)
context_str = "\n\n---\n\n".join(state["retrieved_contexts"])
response = llm.invoke(
f"Berdasarkan konteks berikut, jawab pertanyaan ini: "
f"{state['query']}\n\n"
f"Konteks:\n{context_str}\n\n"
f"PENTING: Hanya jawab berdasarkan konteks yang diberikan. "
f"Jika informasi tidak tersedia, katakan dengan jelas."
)
return {"final_answer": response.content}
def should_continue(state: AgenticRAGState) -> str:
"""Tentukan apakah perlu iterasi lagi atau selesai."""
if state["iteration"] >= 3:
return "generate"
if len(state["retrieved_contexts"]) >= 10:
return "generate"
return "retrieve"
# Bangun graf Agentic RAG
workflow = StateGraph(AgenticRAGState)
workflow.add_node("decompose", decompose_query)
workflow.add_node("retrieve", iterative_retrieve)
workflow.add_node("generate", verify_and_generate)
workflow.set_entry_point("decompose")
workflow.add_edge("decompose", "retrieve")
workflow.add_conditional_edges(
"retrieve",
should_continue,
{"retrieve": "retrieve", "generate": "generate"}
)
workflow.add_edge("generate", END)
# Kompilasi dan jalankan
app = workflow.compile()
result = app.invoke({
"query": "Bandingkan performa GraphRAG vs vector RAG tradisional "
"dan rekomendasikan pendekatan terbaik untuk dokumen teknis",
"sub_queries": [],
"retrieved_contexts": [],
"verification_results": [],
"final_answer": "",
"iteration": 0
})
print(result["final_answer"])Guardrails dan Keamanan: Aspek yang Sering Diabaikan
Di lingkungan enterprise, RAG menangani data sensitif dan menghasilkan konten yang dilihat oleh pengguna dan pelanggan. Keamanan dan guardrails bukan pilihan — melainkan keharusan mutlak. Ini bukan area yang bisa kamu "nanti dulu".
Aspek Keamanan Kritis
- Access Control Sinkronisasi — Kontrol akses RAG harus disinkronkan dengan sistem IAM enterprise secara real-time. Ketika seorang karyawan keluar atau berganti peran, izin akses harus langsung diperbarui di seluruh sistem RAG. Satu celah di sini bisa jadi masalah besar
- Input Sanitization — Cegah prompt injection melalui query pengguna yang bisa memanipulasi perilaku LLM
- Output Filtering — Pastikan respons tidak mengandung informasi sensitif yang bocor dari konteks
- Audit Trail — Catat setiap query, konteks yang digunakan, dan jawaban yang dihasilkan untuk keperluan audit dan compliance
import re
class RAGGuardrails:
"""Guardrails untuk pipeline RAG produksi."""
PATTERNS_BERBAHAYA = [
r"ignore previous instructions",
r"forget your system prompt",
r"you are now",
r"act as if",
r"pretend you",
]
DATA_SENSITIF = [
r"\b\d{16}\b", # Nomor kartu kredit
r"\b\d{3}-\d{2}-\d{4}\b", # SSN format
r"password\s*[:=]\s*\S+", # Password dalam teks
]
def cek_input(self, query: str) -> tuple[bool, str]:
"""Periksa apakah input pengguna aman."""
query_lower = query.lower()
for pattern in self.PATTERNS_BERBAHAYA:
if re.search(pattern, query_lower):
return False, f"Query terdeteksi mengandung pola berbahaya"
if len(query) > 10000:
return False, "Query terlalu panjang"
return True, "OK"
def filter_output(self, response: str) -> str:
"""Filter informasi sensitif dari output."""
filtered = response
for pattern in self.DATA_SENSITIF:
filtered = re.sub(
pattern, "[DATA DISAMARKAN]", filtered
)
return filtered
def validasi_konteks(
self,
konteks: list[str],
user_role: str,
allowed_sources: list[str]
) -> list[str]:
"""Filter konteks berdasarkan hak akses pengguna."""
return [
ctx for ctx in konteks
if any(src in ctx for src in allowed_sources)
]Checklist Deployment: Dari Development ke Produksi
Sebelum men-deploy pipeline RAG ke produksi, pastikan kamu sudah melewati checklist berikut. Saya sudah beberapa kali menyesal karena skip satu atau dua poin di sini — jadi, serius, jangan di-skip:
- Chunking Strategy — Sudah menguji setidaknya 3 strategi chunking dan memilih yang terbaik berdasarkan metrik evaluasi?
- Embedding Model — Model embedding yang dipilih sesuai dengan domain dan bahasa data kamu?
- Hybrid Search — Sudah mengimplementasikan kombinasi BM25 + semantik search?
- Reranking — Cross-encoder reranking sudah diintegrasikan?
- Evaluasi — Punya pipeline evaluasi otomatis dengan metrik faithfulness, relevancy, precision, dan recall?
- Monitoring — Ada dashboard untuk melacak latensi, tingkat kegagalan, dan kepuasan pengguna?
- Guardrails — Input sanitization dan output filtering sudah aktif?
- Scalability — Pipeline sudah diuji dengan beban produksi yang realistis?
- Fallback — Apa yang terjadi ketika retriever nggak menemukan konteks yang relevan?
- Versioning — Ada mekanisme untuk melacak versi indeks, model, dan konfigurasi?
Penutup: RAG Bukan Tujuan Akhir, Melainkan Fondasi
Pipeline RAG di 2026 sudah jauh lebih canggih dibandingkan satu tahun lalu. Dari adaptive chunking yang meningkatkan akurasi hingga 87%, hybrid search yang mengungguli pencarian semantik saja sebesar 9,3 poin persentase, hingga Agentic RAG yang memberikan peningkatan performa 90,2% — ini semua bukan sekadar teori akademis. Teknik-teknik ini sudah diimplementasikan di produksi oleh perusahaan-perusahaan terdepan di seluruh dunia.
Tapi yang penting untuk diingat: RAG hanyalah fondasi. Ia jadi benar-benar powerful ketika digabungkan dengan pola-pola lain — seperti workflow agentik yang sudah kita bahas. Bayangkan sebuah agen ReAct yang menggunakan pipeline RAG hybrid sebagai salah satu tool-nya, atau sistem multi-agent di mana setiap agen punya akses ke knowledge graph yang berbeda.
Kunci sukses implementasi RAG di produksi bukan soal mengejar teknologi terbaru atau framework paling populer. Kuncinya adalah memahami setiap komponen secara mendalam, mengukur dampaknya secara kuantitatif, dan mengiterasi berdasarkan data nyata dari pengguna.
Mulailah dari yang sederhana, ukur, lalu tingkatkan. Itulah jalan menuju pipeline RAG yang benar-benar andal di produksi. Semoga panduan ini membantu kamu memulai — atau memperbaiki — pipeline RAG-mu.