RAG Tingkat Lanjut - Membangun Retrieval-Augmented Generation Kelas Produksi
Daftar Isi
Pendahuluan
Retrieval-Augmented Generation (RAG) telah menjadi pendekatan standar untuk mendasarkan Large Language Models (LLM) pada pengetahuan domain-spesifik. Meskipun sistem RAG dasar mudah dibangun, RAG kelas produksi membutuhkan teknik canggih untuk menangani nuansa pencarian informasi dunia nyata.
Keterbatasan RAG dasar yang ditangani oleh teknik lanjutan:
- Kesenjangan semantik: Embedding dense melewatkan pencocokan leksikal; retrieval sparse melewatkan kesamaan semantik
- Noise retrieval: Hasil top-k sering menyertakan dokumen yang tidak relevan
- Ambiguitas query: Query pengguna mungkin samar, multi-aspek, atau terbentuk buruk
- Fragmentasi konteks: Chunk berukuran tetap kehilangan struktur dan konteks dokumen
- Pengukuran kualitas: Tidak ada cara sistematis untuk mengevaluasi performa pipeline RAG
Dalam tutorial ini, Anda akan mempelajari teknik RAG lanjutan yang secara signifikan meningkatkan kualitas retrieval, akurasi jawaban, dan keandalan sistem secara keseluruhan. Setiap teknik telah teruji di produksi dan dapat dikombinasikan untuk efektivitas maksimal.
Prasyarat
- Python 3.10 atau lebih tinggi
- Pemahaman dasar arsitektur RAG (embedding, vector store, LLM)
- Keakraban dengan LangChain atau framework serupa
- Kunci API OpenAI (atau penyedia LLM lain)
- RAM 8GB+ direkomendasikan untuk model embedding lokal
Instalasi dan Pengaturan
# Dependensi inti
pip install langchain langchain-openai langchain-community
pip install chromadb faiss-cpu
pip install sentence-transformers
pip install rank-bm25
Untuk evaluasi
pip install ragas
Untuk reranking cross-encoder
pip install transformers torch
Utilitas tambahan
pip install tiktoken numpy pandas
Pengaturan lingkungan:
import os
os.environ["OPENAIAPIKEY"] = "kunci-api-anda"
from langchainopenai import ChatOpenAI, OpenAIEmbeddings
from langchain.textsplitter import RecursiveCharacterTextSplitter
from langchaincommunity.vectorstores import Chroma, FAISS
Verifikasi pengaturan
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
print("Pengaturan selesai.")
Pencarian Hybrid: Retrieval Dense + Sparse
Pencarian hybrid menggabungkan kekuatan retrieval vektor dense (pemahaman semantik) dengan retrieval sparse (pencocokan kata kunci eksak) untuk retrieval dokumen yang lebih kokoh.
import numpy as np
from rankbm25 import BM25Okapi
from langchainopenai import OpenAIEmbeddings
from langchaincommunity.vectorstores import FAISS
from langchain.schema import Document
from typing import List, Tuple
class RetrieverPencarianHybrid:
"""
Menggabungkan retrieval dense (embedding) dan sparse (BM25)
dengan bobot fusi yang dapat dikonfigurasi.
"""
def init(
self,
dokumen: List[Document],
embeddings,
bobotdense: float = 0.6,
bobotsparse: float = 0.4,
):
self.dokumen = dokumen
self.bobotdense = bobotdense
self.bobotsparse = bobotsparse
# Bangun indeks dense
self.vectorstore = FAISS.fromdocuments(dokumen, embeddings)
# Bangun indeks sparse (BM25)
dokumentokenisasi = [doc.pagecontent.lower().split() for doc in dokumen]
self.bm25 = BM25Okapi(dokumentokenisasi)
def normalisasiskor(self, skor: List[float]) -> List[float]:
"""Normalisasi min-max skor ke [0, 1]."""
if not skor:
return skor
mins, maxs = min(skor), max(skor)
if maxs == mins:
return [0.5] len(skor)
return [(s - mins) / (maxs - mins) for s in skor]
def retrieve(self, query: str, topk: int = 10) -> List[Tuple[Document, float]]:
"""
Mengambil dokumen menggunakan reciprocal rank fusion
dari hasil dense dan sparse.
"""
# Retrieval dense
hasildense = self.vectorstore.similaritysearchwithscore(
query, k=topk 2
)
skordokdense = {
doc.pagecontent: 1.0 / (rank + 1)
for rank, (doc, skor) in enumerate(hasildense)
}
# Retrieval sparse (BM25)
querytokenisasi = query.lower().split()
skorbm25 = self.bm25.getscores(querytokenisasi)
sparseterurut = sorted(
enumerate(skorbm25), key=lambda x: -x[1]
)[:topk 2]
skordoksparse = {
self.dokumen[idx].pagecontent: 1.0 / (rank + 1)
for rank, (idx, skor) in enumerate(sparseterurut)
}
# Reciprocal Rank Fusion
semuakonten = set(skordokdense.keys()) | set(skordoksparse.keys())
skorfusi = {}
for konten in semuakonten:
skord = skordokdense.get(konten, 0.0)
skors = skordoksparse.get(konten, 0.0)
skorfusi[konten] = (
self.bobotdense skord +
self.bobotsparse skors
)
# Urutkan berdasarkan skor fusi dan kembalikan topk
hasilterurut = sorted(skorfusi.items(), key=lambda x: -x[1])[:topk]
hasil = []
kontenkedok = {doc.pagecontent: doc for doc in self.dokumen}
for konten, skor in hasilterurut:
if konten in kontenkedok:
hasil.append((kontenkedok[konten], skor))
return hasil
Penggunaan
dokumen = [
Document(pagecontent="Machine learning menggunakan metode statistik untuk belajar dari data."),
Document(pagecontent="Python adalah bahasa pemrograman paling populer untuk ML."),
Document(pagecontent="Neural network terinspirasi dari struktur otak biologis."),
Document(pagecontent="Algoritma BM25 banyak digunakan dalam pencarian informasi."),
Document(pagecontent="Transformer merevolusi pemrosesan bahasa alami pada 2017."),
]
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
retrieverhybrid = RetrieverPencarianHybrid(
dokumen=dokumen,
embeddings=embeddings,
bobotdense=0.6,
bobotsparse=0.4,
)
hasil = retrieverhybrid.retrieve("algoritma BM25 pencarian informasi", topk=3)
for doc, skor in hasil:
print(f" [{skor:.3f}] {doc.pagecontent[:80]}...")
Reranking dengan Cross-Encoder
Reranking cross-encoder adalah salah satu peningkatan paling berdampak yang dapat Anda lakukan pada sistem RAG. Berbeda dengan bi-encoder (digunakan untuk retrieval awal), cross-encoder mengkodekan query dan dokumen secara bersamaan, menghasilkan skor relevansi yang jauh lebih akurat.
from transformers import AutoModelForSequenceClassification, AutoTokenizer
import torch
from langchain.schema import Document
from typing import List, Tuple
class RerankerCrossEncoder:
"""
Melakukan reranking dokumen yang diambil menggunakan model cross-encoder.
Ini memberikan skor relevansi yang jauh lebih akurat dibandingkan
kesamaan embedding saja.
"""
def init(
self,
namamodel: str = "cross-encoder/ms-marco-MiniLM-L-12-v2",
device: str = None,
ukuranbatch: int = 32,
):
self.device = device or ("cuda" if torch.cuda.isavailable() else "cpu")
self.ukuranbatch = ukuranbatch
self.tokenizer = AutoTokenizer.frompretrained(namamodel)
self.model = AutoModelForSequenceClassification.frompretrained(namamodel)
self.model.to(self.device)
self.model.eval()
@torch.nograd()
def skorpasangan(self, query: str, dokumen: List[str]) -> List[float]:
"""Skor pasangan query-dokumen menggunakan cross-encoder."""
skorlist = []
for i in range(0, len(dokumen), self.ukuranbatch):
batchdocs = dokumen[i:i + self.ukuranbatch]
pasangan = [[query, doc] for doc in batchdocs]
inputs = self.tokenizer(
pasangan,
padding=True,
truncation=True,
maxlength=512,
returntensors="pt",
).to(self.device)
outputs = self.model(inputs)
skorbatch = outputs.logits.squeeze(-1).cpu().tolist()
if isinstance(skorbatch, float):
skorbatch = [skorbatch]
skorlist.extend(skorbatch)
return skorlist
def rerank(
self,
query: str,
dokumen: List[Document],
topk: int = 5,
thresholdskor: float = None,
) -> List[Tuple[Document, float]]:
"""Rerank dokumen dan kembalikan topk paling relevan."""
teksdok = [doc.pagecontent for doc in dokumen]
skorlist = self.skorpasangan(query, teksdok)
# Pasangkan dokumen dengan skor dan urutkan
dokberskor = list(zip(dokumen, skorlist))
dokberskor.sort(key=lambda x: -x[1])
# Terapkan filter threshold jika ditentukan
if thresholdskor is not None:
dokberskor = [
(doc, skor) for doc, skor in dokberskor
if skor >= thresholdskor
]
return dokberskor[:topk]
Penggunaan: Retrieval dua tahap + reranking
reranker = RerankerCrossEncoder()
Tahap 1: Ambil kandidat (ambil lebih banyak dari kebutuhan)
hasilawal = retrieverhybrid.retrieve(
"Bagaimana transformer bekerja di NLP?",
topk=20 # Ambil lebih dari yang dibutuhkan
)
dokkandidat = [doc for doc, in hasilawal]
Tahap 2: Rerank dengan cross-encoder
hasilrerank = reranker.rerank(
query="Bagaimana transformer bekerja di NLP?",
dokumen=dokkandidat,
topk=5,
thresholdskor=0.0,
)
print("Hasil setelah reranking:")
for doc, skor in hasilrerank:
print(f" [{skor:.4f}] {doc.pagecontent[:80]}...")
Teknik Transformasi Query
Query mentah dari pengguna seringkali tidak optimal untuk retrieval. Teknik transformasi query memformulasikan ulang query untuk meningkatkan kualitas retrieval.
from langchainopenai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from typing import List
class TransformerQuery:
"""
Mentransformasi query pengguna untuk meningkatkan kualitas
retrieval menggunakan beberapa strategi.
"""
def init(self, llm=None):
self.llm = llm or ChatOpenAI(model="gpt-4o-mini", temperature=0.3)
def multiquery(self, query: str, jumlahquery: int = 3) -> List[str]:
"""
Hasilkan beberapa reformulasi query untuk menangkap
aspek dan frasa yang berbeda.
"""
prompt = ChatPromptTemplate.fromtemplate(
"Anda ahli dalam memformulasikan ulang query pencarian. "
"Hasilkan {jumlahquery} versi berbeda dari pertanyaan berikut "
"yang akan membantu mengambil dokumen relevan. "
"Setiap versi harus mendekati pertanyaan dari sudut yang berbeda.\n\n"
"Pertanyaan asli: {query}\n\n"
"Berikan setiap query pada baris baru, dinomori 1-{jumlahquery}."
)
respons = self.llm.invoke(
prompt.formatmessages(query=query, jumlahquery=jumlahquery)
)
# Parse query bernomor
daftarquery = []
for baris in respons.content.strip().split("\n"):
baris = baris.strip()
if baris and baris[0].isdigit():
bersih = baris.split(".", 1)[-1].strip() if "." in baris else baris
daftarquery.append(bersih)
return daftarquery[:jumlahquery]
def stepback(self, query: str) -> str:
"""
Hasilkan query 'step-back' yang lebih abstrak yang menangkap
konteks lebih luas yang diperlukan untuk menjawab pertanyaan spesifik.
"""
prompt = ChatPromptTemplate.fromtemplate(
"Anda ahli memahami konteks mendalam di balik pertanyaan. "
"Diberikan pertanyaan spesifik berikut, hasilkan satu pertanyaan "
"yang lebih luas dan abstrak yang akan membantu mengumpulkan "
"pengetahuan latar belakang yang diperlukan untuk menjawab "
"pertanyaan asli secara lengkap.\n\n"
"Pertanyaan asli: {query}\n\n"
"Pertanyaan step-back:"
)
respons = self.llm.invoke(prompt.formatmessages(query=query))
return respons.content.strip()
def hyde(self, query: str) -> str:
"""
Hypothetical Document Embeddings (HyDE): Hasilkan jawaban hipotetis
yang dapat digunakan sebagai query untuk pencocokan semantik yang lebih baik.
"""
prompt = ChatPromptTemplate.fromtemplate(
"Tulis paragraf pendek dan faktual (100-150 kata) yang akan "
"menjadi jawaban sempurna untuk pertanyaan berikut. Tulis seolah-olah "
"berasal dari dokumen referensi, bukan sebagai respons langsung.\n\n"
"Pertanyaan: {query}\n\n"
"Paragraf:"
)
respons = self.llm.invoke(prompt.formatmessages(query=query))
return respons.content.strip()
def dekomposisi(self, query: str) -> List[str]:
"""
Dekomposisi query kompleks menjadi sub-query yang lebih sederhana
yang dapat dijawab secara independen.
"""
prompt = ChatPromptTemplate.fromtemplate(
"Pecah pertanyaan kompleks berikut menjadi 2-4 sub-pertanyaan "
"yang lebih sederhana yang bersama-sama akan memberikan jawaban "
"lengkap untuk pertanyaan asli.\n\n"
"Pertanyaan kompleks: {query}\n\n"
"Sub-pertanyaan (satu per baris, dinomori):"
)
respons = self.llm.invoke(prompt.formatmessages(query=query))
subquery = []
for baris in respons.content.strip().split("\n"):
baris = baris.strip()
if baris and baris[0].isdigit():
bersih = baris.split(".", 1)[-1].strip() if "." in baris else baris
subquery.append(bersih)
return subquery
Contoh penggunaan
transformer = TransformerQuery()
queryasli = "Bagaimana RAG menangani informasi yang bertentangan dari berbagai sumber?"
Multi-query: ambil dengan beberapa reformulasi
multiqueries = transformer.multiquery(queryasli)
print("Reformulasi multi-query:")
for i, q in enumerate(multiqueries, 1):
print(f" {i}. {q}")
Step-back: dapatkan konteks lebih luas
querystepback = transformer.stepback(queryasli)
print(f"\nQuery step-back: {querystepback}")
HyDE: hasilkan dokumen hipotetis
dokhyde = transformer.hyde(queryasli)
print(f"\nDokumen HyDE: {dokhyde[:200]}...")
Dekomposisi: pecah menjadi sub-pertanyaan
subqueries = transformer.dekomposisi(queryasli)
print("\nSub-pertanyaan terdekomposisi:")
for i, q in enumerate(subqueries, 1):
print(f" {i}. {q}")
Retrieval dengan Fusi Multi-Query
def retrievedenganmultiquery(
retriever: RetrieverPencarianHybrid,
reranker: RerankerCrossEncoder,
transformer: TransformerQuery,
query: str,
topk: int = 5,
) -> List[Tuple[Document, float]]:
"""Pipeline retrieval lengkap dengan fusi multi-query dan reranking."""
# Hasilkan varian query
daftarquery = [query] + transformer.multiquery(query, jumlahquery=3)
# Ambil kandidat dari semua query
semuakandidat = {}
for q in daftarquery:
hasil = retriever.retrieve(q, topk=10)
for doc, skor in hasil:
konten = doc.pagecontent
if konten not in semuakandidat or skor > semuakandidat[konten][1]:
semuakandidat[konten] = (doc, skor)
dokkandidat = [doc for doc, in semuakandidat.values()]
print(f"Kandidat unik dari {len(daftarquery)} query: {len(dokkandidat)}")
# Rerank semua kandidat
hasilrerank = reranker.rerank(query, dokkandidat, topk=topk)
return hasilrerank
Strategi Chunking Parent-Child
Chunking parent-child menyelesaikan masalah fragmentasi konteks dengan mengindeks chunk kecil untuk retrieval yang presisi sambil mengembalikan chunk parent yang lebih besar untuk konteks yang komprehensif.
from langchain.schema import Document
from langchain.textsplitter import RecursiveCharacterTextSplitter
from typing import Dict, List, Optional
import hashlib
class ChunkerParentChild:
"""
Mengimplementasikan chunking parent-child: chunk kecil untuk retrieval,
chunk besar untuk konteks dalam prompt LLM.
"""
def init(
self,
ukuranchunkparent: int = 2000,
overlapparent: int = 200,
ukuranchunkchild: int = 400,
overlapchild: int = 50,
):
self.pemisahparent = RecursiveCharacterTextSplitter(
chunksize=ukuranchunkparent,
chunkoverlap=overlapparent,
separators=["\n\n", "\n", ". ", " ", ""],
)
self.pemisahchild = RecursiveCharacterTextSplitter(
chunksize=ukuranchunkchild,
chunkoverlap=overlapchild,
separators=["\n\n", "\n", ". ", " ", ""],
)
self.penyimpananparent: Dict[str, Document] = {}
self.childkeparent: Dict[str, str] = {}
def buatid(self, teks: str) -> str:
"""Buat ID deterministik dari konten."""
return hashlib.md5(teks.encode()).hexdigest()[:12]
def prosesdokumen(self, dokumen: List[Document]) -> List[Document]:
"""
Pecah dokumen menjadi chunk parent dan child.
Mengembalikan chunk child (untuk pengindeksan) dengan referensi parent.
"""
dokumenchild = []
for doc in dokumen:
# Buat chunk parent
chunkparent = self.pemisahparent.splittext(doc.pagecontent)
for teksparent in chunkparent:
idparent = self.buatid(teksparent)
dokparent = Document(
pagecontent=teksparent,
metadata={
doc.metadata,
"tipechunk": "parent",
"idparent": idparent,
}
)
self.penyimpananparent[idparent] = dokparent
# Buat chunk child dari parent ini
chunkchild = self.pemisahchild.splittext(teksparent)
for tekschild in chunkchild:
idchild = self.buatid(tekschild)
dokchild = Document(
pagecontent=tekschild,
metadata={
doc.metadata,
"tipechunk": "child",
"idchild": idchild,
"idparent": idparent,
}
)
dokumenchild.append(dokchild)
self.childkeparent[idchild] = idparent
print(f"Dibuat {len(self.penyimpananparent)} chunk parent")
print(f"Dibuat {len(dokumenchild)} chunk child")
return dokumenchild
def ambilparent(self, dokchild: Document) -> Optional[Document]:
"""Ambil chunk parent untuk chunk child yang diberikan."""
idparent = dokchild.metadata.get("idparent")
if idparent and idparent in self.penyimpananparent:
return self.penyimpananparent[idparent]
return None
def retrievedenganparent(
self,
hasilchild: List[Document],
deduplikasi: bool = True,
) -> List[Document]:
"""
Diberikan chunk child yang diambil, kembalikan chunk parent
mereka untuk konteks yang lebih kaya.
"""
idparentterlihat = set()
hasilparent = []
for dokchild in hasilchild:
dokparent = self.ambilparent(dokchild)
if dokparent:
idparent = dokparent.metadata["idparent"]
if not deduplikasi or idparent not in idparentterlihat:
idparentterlihat.add(idparent)
hasilparent.append(dokparent)
return hasilparent
Penggunaan
chunker = ChunkerParentChild(
ukuranchunkparent=2000,
ukuranchunkchild=400,
)
Proses dokumen
dokumenmentah = [
Document(
pagecontent="Teks dokumen panjang di sini..." 100,
metadata={"sumber": "manualteknis.pdf", "halaman": 1}
),
]
dokchild = chunker.prosesdokumen(dokumenmentah)
Indeks chunk child di vector store
vectorstore = FAISS.fromdocuments(dokchild, embeddings)
Saat retrieval: cari children, kembalikan parent
query = "Bagaimana mengonfigurasi sistem autentikasi?"
hasilchild = vectorstore.similaritysearch(query, k=5)
hasilparent = chunker.retrievedenganparent(hasilchild)
print(f"Diambil {len(hasilchild)} chunk child")
print(f"Dipetakan ke {len(hasilparent)} chunk parent unik")
Recursive Retrieval
Recursive retrieval secara iteratif memperbaiki hasil dengan mengambil, menganalisis, dan melakukan query ulang berdasarkan celah informasi yang ditemukan.
from langchainopenai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langchain.schema import Document
from typing import List, Tuple, Dict
import json
class RetrieverRekursif:
"""
Mengimplementasikan retrieval rekursif yang secara iteratif
memperbaiki pencarian berdasarkan analisis dokumen yang diambil.
"""
def init(self, retriever, reranker, llm=None, maksiterasi: int = 3):
self.retriever = retriever
self.reranker = reranker
self.llm = llm or ChatOpenAI(model="gpt-4o-mini", temperature=0)
self.maksiterasi = maksiterasi
def analisiscakupan(
self,
query: str,
dokdiambil: List[Document],
) -> Dict:
"""Analisis apakah dokumen yang diambil cukup menjawab query."""
konteks = "\n\n---\n\n".join([doc.pagecontent for doc in dokdiambil])
prompt = ChatPromptTemplate.fromtemplate(
"Analisis apakah dokumen yang diambil berikut cukup "
"menjawab query. Identifikasi celah informasi.\n\n"
"Query: {query}\n\n"
"Dokumen yang Diambil:\n{konteks}\n\n"
"Jawab dalam format JSON:\n"
'{{\n'
' "skorcakupan": <0.0-1.0>,\n'
' "sudahcukup": ,\n'
' "celah": ["celah1", "celah2"],\n'
' "querylanjutan": ["query1", "query2"]\n'
'}}'
)
respons = self.llm.invoke(
prompt.formatmessages(query=query, konteks=konteks[:4000])
)
try:
return json.loads(respons.content)
except json.JSONDecodeError:
return {
"skorcakupan": 0.5,
"sudahcukup": True,
"celah": [],
"querylanjutan": [],
}
def retrieve(
self,
query: str,
topk: int = 5,
) -> Tuple[List[Document], Dict]:
"""
Retrieval rekursif sampai cakupan mencukupi
atau iterasi maksimum tercapai.
"""
semuadokumen = []
kontenterlihat = set()
logretrieval = {"iterasi": []}
querysaatini = [query]
for iterasi in range(self.maksiterasi):
print(f"\nIterasi {iterasi + 1}/{self.maksiterasi}")
infoiterasi = {
"iterasi": iterasi + 1,
"query": querysaatini,
"dokbaru": 0,
}
# Ambil untuk semua query saat ini
for q in querysaatini:
hasil = self.retriever.retrieve(q, topk=topk)
for doc, skor in hasil:
if doc.pagecontent not in kontenterlihat:
kontenterlihat.add(doc.pagecontent)
semuadokumen.append(doc)
infoiterasi["dokbaru"] += 1
print(f" Query: {len(querysaatini)}, "
f"Dok baru: {infoiterasi['dokbaru']}, "
f"Total dok: {len(semuadokumen)}")
# Rerank semua dokumen yang terkumpul
hasilrerank = self.reranker.rerank(
query=query,
dokumen=semuadokumen,
topk=min(topk 2, len(semuadokumen)),
)
dokteratas = [doc for doc, in hasilrerank]
# Analisis cakupan
analisis = self.analisiscakupan(query, dokteratas[:topk])
infoiterasi["skorcakupan"] = analisis["skorcakupan"]
infoiterasi["sudahcukup"] = analisis["sudahcukup"]
logretrieval["iterasi"].append(infoiterasi)
print(f" Cakupan: {analisis['skorcakupan']:.2f}, "
f"Cukup: {analisis['sudahcukup']}")
if analisis["sudahcukup"]:
break
# Hasilkan query lanjutan untuk iterasi berikutnya
querysaatini = analisis.get("querylanjutan", [])
if not querysaatini:
break
# Reranking akhir
hasilakhir = self.reranker.rerank(
query=query,
dokumen=semuadokumen,
topk=topk,
)
dokakhir = [doc for doc, in hasilakhir]
logretrieval["totaldokumendiambil"] = len(semuadokumen)
logretrieval["dokumenakhir"] = len(dokakhir)
return dokakhir, logretrieval
Evaluasi dengan RAGAS
RAGAS (Retrieval-Augmented Generation Assessment) menyediakan metrik evaluasi otomatis untuk sistem RAG.
from ragas import evaluate
from ragas.metrics import (
faithfulness,
answerrelevancy,
contextprecision,
contextrecall,
)
from datasets import Dataset
from typing import List, Dict
def buatdatasetevaluasi(
pertanyaan: List[str],
jawabanbenar: List[str],
pipelinerag,
) -> Dataset:
"""
Buat dataset evaluasi dengan menjalankan pipeline RAG
pada sekumpulan pertanyaan dengan jawaban yang diketahui.
"""
dataeval = {
"question": [],
"answer": [],
"contexts": [],
"groundtruth": [],
}
for pertanyaanitem, kebenaran in zip(pertanyaan, jawabanbenar):
# Jalankan pipeline RAG
hasil = pipelinerag.query(pertanyaanitem)
dataeval["question"].append(pertanyaanitem)
dataeval["answer"].append(hasil["answer"])
dataeval["contexts"].append(
[doc.pagecontent for doc in hasil["sourcedocuments"]]
)
dataeval["groundtruth"].append(kebenaran)
return Dataset.fromdict(dataeval)
def jalankanevaluasiragas(dataseteval: Dataset) -> Dict:
"""
Jalankan evaluasi RAGAS komprehensif.
Metrik:
- faithfulness: Apakah jawaban berdasarkan konteks yang diambil?
- answerrelevancy: Apakah jawaban relevan dengan pertanyaan?
- contextprecision: Apakah konteks berperingkat teratas relevan?
- contextrecall: Apakah konteks mengandung info jawaban benar?
"""
hasil = evaluate(
dataseteval,
metrics=[
faithfulness,
answerrelevancy,
contextprecision,
contextrecall,
],
)
return hasil
Definisikan pertanyaan dan jawaban benar untuk evaluasi
pertanyaaneval = [
"Apa itu pencarian hybrid dalam RAG?",
"Bagaimana reranking cross-encoder meningkatkan retrieval?",
"Apa itu strategi chunking parent-child?",
"Bagaimana cara mengevaluasi pipeline RAG?",
]
jawabanbenareval = [
"Pencarian hybrid menggabungkan retrieval vektor dense dengan retrieval "
"sparse BM25 untuk memanfaatkan pemahaman semantik dan pencocokan kata kunci eksak.",
"Reranking cross-encoder mengkodekan query dan dokumen secara bersamaan untuk "
"menghasilkan skor relevansi yang lebih akurat dibandingkan kesamaan bi-encoder.",
"Chunking parent-child mengindeks chunk kecil untuk retrieval presisi sambil "
"mengembalikan chunk parent yang lebih besar untuk konteks komprehensif.",
"Pipeline RAG dapat dievaluasi menggunakan metrik seperti faithfulness, answer "
"relevancy, context precision, dan context recall dari RAGAS.",
]
Jalankan evaluasi
dataseteval = buatdatasetevaluasi(
pertanyaaneval, jawabanbenareval, pipelinerag
)
hasil = jalankanevaluasiragas(dataseteval)
print("\nHasil Evaluasi RAGAS:")
print(f" Faithfulness: {hasil['faithfulness']:.4f}")
print(f" Answer Relevancy: {hasil['answerrelevancy']:.4f}")
print(f" Context Precision: {hasil['contextprecision']:.4f}")
print(f" Context Recall: {hasil['contextrecall']:.4f}")
Fungsi evaluasi kustom untuk monitoring berkelanjutan
def evaluasiquerytunggal(
pertanyaan: str,
jawaban: str,
kontekslist: List[str],
jawabanbenar: str,
) -> Dict:
"""Evaluasi satu query RAG untuk monitoring."""
dataset = Dataset.fromdict({
"question": [pertanyaan],
"answer": [jawaban],
"contexts": [kontekslist],
"groundtruth": [jawabanbenar],
})
hasil = evaluate(
dataset,
metrics=[faithfulness, answerrelevancy],
)
return {
"faithfulness": hasil["faithfulness"],
"answerrelevancy": hasil["answerrelevancy"],
}
Pipeline RAG Produksi
Menyatukan semua teknik menjadi pipeline siap produksi.
from langchainopenai import ChatOpenAI, OpenAIEmbeddings
from langchain.prompts import ChatPromptTemplate
from langchain.schema import Document
from typing import List, Dict, Optional
import time
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(name)
class PipelineRAGProduksi:
"""
Pipeline RAG kelas produksi yang menggabungkan:
- Pencarian hybrid (dense + sparse)
- Reranking cross-encoder
- Transformasi query
- Chunking parent-child
- Logging dan metrik komprehensif
"""
def init(
self,
dokumen: List[Document],
model
embedding: str = "text-embedding-3-small",
modelllm: str = "gpt-4o",
modelreranker: str = "cross-encoder/ms-marco-MiniLM-L-12-v2",
ukuranchunkparent: int = 2000,
ukuranchunkchild: int = 400,
bobotdense: float = 0.6,
bobotsparse: float = 0.4,
):
self.llm = ChatOpenAI(model=modelllm, temperature=0)
self.embeddings = OpenAIEmbeddings(model=modelembedding)
# Inisialisasi chunker
self.chunker = ChunkerParentChild(
ukuranchunkparent=ukuranchunkparent,
ukuranchunkchild=ukuranchunkchild,
)
dokchild = self.chunker.prosesdokumen(dokumen)
# Inisialisasi retriever hybrid dengan dok child
self.retriever = RetrieverPencarianHybrid(
dokumen=dokchild,
embeddings=self.embeddings,
bobotdense=bobotdense,
bobotsparse=bobotsparse,
)
# Inisialisasi reranker
self.reranker = RerankerCrossEncoder(namamodel=modelreranker)
# Inisialisasi transformer query
self.transformerquery = TransformerQuery(self.llm)
# Pelacakan metrik
self.metrik = {
"totalquery": 0,
"rataratalatensims": 0,
"rataratapanjangkonteks": 0,
}
logger.info("Pipeline RAG produksi diinisialisasi")