Memperbaiki Kesalahan Pengambilan RAG dengan CRAG, LangGraph, dan Milvus

Ketika aplikasi LLM mulai diproduksi, tim semakin membutuhkan model mereka untuk menjawab pertanyaan yang didasarkan pada data pribadi atau informasi real-time. Retrieval-augmented generation (RAG) - di mana model diambil dari basis pengetahuan eksternal pada waktu kueri - adalah pendekatan standar. Pendekatan ini mengurangi halusinasi dan menjaga jawaban tetap terkini.

Namun, inilah masalah yang muncul dengan cepat dalam praktiknya: sebuah dokumen dapat memiliki nilai kesamaan yang tinggi dan tetap saja salah dalam menjawab pertanyaan. Pipeline RAG tradisional menyamakan kemiripan dengan relevansi. Dalam produksi, asumsi itu rusak. Hasil yang berada di peringkat teratas mungkin saja sudah usang, hanya berhubungan secara garis besar, atau tidak memiliki detail yang dibutuhkan pengguna.

CRAG (Corrective Retrieval-Augmented Generation) mengatasi hal ini dengan menambahkan evaluasi dan koreksi antara pencarian dan pembuatan. Alih-alih mempercayai skor kemiripan secara membabi buta, sistem akan memeriksa apakah konten yang diambil benar-benar menjawab pertanyaan-dan memperbaiki situasi jika tidak.

Artikel ini akan menjelaskan cara membangun sistem CRAG yang siap produksi dengan menggunakan LangChain, LangGraph, dan Milvus.

Tiga Masalah Pengambilan yang Tidak Dapat Dipecahkan oleh RAG Tradisional

Sebagian besar kegagalan RAG dalam produksi berakar pada salah satu dari tiga masalah:

Ketidakcocokan pengambilan. Dokumen tersebut secara topik serupa tetapi tidak benar-benar menjawab pertanyaan. Tanyakan cara mengonfigurasi sertifikat HTTPS di Nginx, dan sistem mungkin akan memberikan panduan penyiapan Apache, panduan 2019, atau penjelasan umum tentang cara kerja TLS. Secara semantik hampir sama, secara praktis tidak berguna.

Konten basi. Pencarian vektor tidak memiliki konsep kemutakhiran. Kueri "praktik terbaik asinkronisasi Python" dan Anda akan mendapatkan campuran pola tahun 2018 dan pola tahun 2024, yang diberi peringkat murni berdasarkan jarak penyematan. Sistem tidak dapat membedakan mana yang sebenarnya dibutuhkan pengguna.

Kontaminasi memori. Masalah yang satu ini bertambah seiring berjalannya waktu dan sering kali paling sulit untuk diperbaiki. Katakanlah sistem mengambil referensi API yang sudah ketinggalan zaman dan menghasilkan kode yang salah. Hasil yang buruk tersebut akan disimpan kembali ke dalam memori. Pada kueri serupa berikutnya, sistem akan mengambilnya lagi-memperkuat kesalahan tersebut. Informasi yang sudah basi dan yang baru secara bertahap bercampur, dan keandalan sistem terkikis di setiap siklus.

Ini bukan kasus yang jarang terjadi. Kasus-kasus ini muncul secara teratur setelah sistem RAG menangani lalu lintas yang sebenarnya. Itulah yang membuat pemeriksaan kualitas pengambilan menjadi sebuah kebutuhan, bukan keinginan.

Apa itu CRAG? Evaluasi Dulu, Lalu Hasilkan

Corrective Retrieval-Augmented Generation (CRAG ) adalah metode yang menambahkan langkah evaluasi dan koreksi antara pengambilan dan pembangkitan dalam pipa RAG. Metode ini diperkenalkan dalam makalah Corrective Retrieval Augmented Generation (Yan et al., 2024). Tidak seperti RAG tradisional, yang membuat keputusan biner - menggunakan dokumen atau membuangnya - RAG menilai setiap hasil yang diambil untuk relevansi dan merutekannya melalui salah satu dari tiga jalur koreksi sebelum mencapai model bahasa.

RAG tradisional mengalami kesulitan ketika hasil pencarian berada di zona abu-abu: sebagian relevan, agak ketinggalan zaman, atau kehilangan bagian penting. Gerbang ya/tidak yang sederhana akan membuang sebagian informasi yang berguna atau membiarkan konten yang berisik masuk. CRAG membingkai ulang pipeline dari ambil → hasilkan menjadi ambil → evaluasi → perbaiki → hasilkan, memberikan sistem kesempatan untuk memperbaiki kualitas hasil sebelum pembuatan dimulai.

Alur kerja empat langkah CRAG: Pengambilan → Evaluasi → Koreksi → Pembangkitan, yang menunjukkan bagaimana dokumen dinilai dan dirutekan

Hasil pencarian diklasifikasikan ke dalam salah satu dari tiga kategori:

• Benar: langsung menjawab pertanyaan; dapat digunakan setelah perbaikan ringan
• Ambigu: sebagian relevan; membutuhkan informasi tambahan
• Salah: tidak relevan; buang dan kembalikan ke sumber alternatif

Keputusan	Keyakinan	Tindakan
Benar	> 0.9	Memperbaiki konten dokumen
Ambigu	0.5-0.9	Sempurnakan dokumen + lengkapi dengan pencarian web
Salah	< 0.5	Buang hasil pencarian; kembali sepenuhnya ke pencarian web

Penyempurnaan Konten

CRAG juga menangani masalah yang lebih halus dengan RAG standar: sebagian besar sistem memberikan seluruh dokumen yang diambil ke model. Hal ini membuang token dan melemahkan sinyal - model harus mengarungi paragraf yang tidak relevan untuk menemukan satu kalimat yang benar-benar penting. CRAG menyaring konten yang diambil terlebih dahulu, mengekstrak bagian yang relevan dan membuang sisanya.

Makalah asli menggunakan strip pengetahuan dan aturan heuristik untuk ini. Dalam praktiknya, pencocokan kata kunci dapat digunakan untuk banyak kasus penggunaan, dan sistem produksi dapat menambahkan ringkasan berbasis LLM atau ekstraksi terstruktur untuk kualitas yang lebih tinggi.

Proses penyempurnaan memiliki tiga bagian:

• Penguraiandokumen: mengekstrak bagian-bagian penting dari dokumen yang lebih panjang
• Penulisanulang kueri: mengubah kueri yang tidak jelas atau ambigu menjadi kueri yang lebih tepat sasaran
• Seleksi pengetahuan: menggandakan, memberi peringkat, dan hanya mempertahankan konten yang paling berguna

Proses penyempurnaan dokumen dalam tiga langkah: Dekomposisi Dokumen (2000 → 500 token), Penulisan Ulang Kueri (meningkatkan ketepatan pencarian), dan Seleksi Pengetahuan (menyaring, memberi peringkat, dan memangkas)

Evaluator

Evaluator adalah inti dari CRAG. Evaluator tidak dimaksudkan untuk penalaran yang mendalam - ini adalah gerbang triase yang cepat. Diberikan sebuah kueri dan sekumpulan dokumen yang diambil, evaluator akan memutuskan apakah konten tersebut cukup baik untuk digunakan.

Makalah asli memilih model T5-Large yang telah disetel dengan baik daripada LLM tujuan umum. Alasannya: kecepatan dan ketepatan lebih penting daripada fleksibilitas untuk tugas khusus ini.

Atribut	T5-Besar yang disetel dengan baik	GPT-4
Latensi	10-20 ms	200 ms+
Akurasi	92% (percobaan di atas kertas)	TBD
Kesesuaian Tugas	Tinggi - tugas tunggal yang disetel dengan baik, presisi yang lebih tinggi	Sedang - tujuan umum, lebih fleksibel tetapi kurang terspesialisasi

Pengembalian Penelusuran Web

Ketika pengambilan internal ditandai sebagai salah atau ambigu, CRAG dapat memicu pencarian web untuk menarik informasi yang lebih segar atau tambahan. Ini bertindak sebagai jaring pengaman untuk kueri yang sensitif terhadap waktu dan topik-topik di mana basis pengetahuan internal memiliki kesenjangan.

Mengapa Milvus Sangat Cocok untuk CRAG dalam Produksi

Efektivitas CRAG bergantung pada apa yang ada di bawahnya. Basis data vektor perlu melakukan lebih dari sekadar pencarian kemiripan dasar - ia perlu mendukung isolasi multi-penyewa, pengambilan hibrida, dan fleksibilitas skema yang dituntut oleh sistem CRAG produksi.

Setelah mengevaluasi beberapa opsi, kami memilih Milvus karena tiga alasan.

Isolasi Multi-Penyewa

Dalam sistem berbasis agen, setiap pengguna atau sesi membutuhkan ruang memorinya sendiri. Pendekatan naif-satu koleksi per penyewa-menjadi sangat memusingkan secara operasional, terutama dalam skala besar.

Milvus menangani hal ini dengan Kunci Partisi. Tetapkan is_partition_key=True pada bidang agent_id, dan Milvus merutekan kueri ke partisi yang tepat secara otomatis. Tidak ada koleksi yang melebar, tidak ada kode perutean manual.

Dalam tolok ukur kami dengan 10 juta vektor di 100 penyewa, Milvus dengan Pemadatan Clustering menghasilkan QPS 3-5x lebih tinggi dibandingkan dengan baseline yang tidak dioptimalkan.

Pencarian Hibrida

Pencarian vektor murni tidak cukup untuk SKU produk-konten yang sama persis seperti SKU-2024-X5, string versi, atau terminologi tertentu.

Milvus 2.5 mendukung pencarian h ibrida secara native: vektor padat untuk kemiripan semantik, vektor jarang untuk pencocokan kata kunci gaya BM25, dan pemfilteran metadata skalar-semuanya dalam satu kueri. Hasilnya digabungkan menggunakan Reciprocal Rank Fusion (RRF), sehingga Anda tidak perlu membangun dan menggabungkan pipeline pencarian yang terpisah.

Pada set data 1 juta vektor, latensi pengambilan Milvus Sparse-BM25 mencapai 6 ms, dengan dampak yang dapat diabaikan pada kinerja CRAG end-to-end.

Skema Fleksibel untuk Memori yang Terus Berkembang

Seiring dengan berkembangnya pipeline CRAG, model data pun ikut berkembang. Kami perlu menambahkan bidang seperti confidence, verified, dan source saat melakukan iterasi pada logika evaluasi. Pada sebagian besar database, hal ini berarti skrip migrasi dan waktu henti.

Milvus mendukung bidang JSON dinamis, sehingga metadata dapat diperluas dengan cepat tanpa gangguan layanan.

Berikut ini adalah sebuah skema umum:

fields = [
    FieldSchema(name="agent_id", dtype=DataType.VARCHAR, is_partition_key=True),  # multi-tenancy
    FieldSchema(name="dense_embedding", dtype=DataType.FLOAT_VECTOR, dim=1536),   # semantic retrieval
    FieldSchema(name="sparse_embedding", dtype=DataType.SPARSE_FLOAT_VECTOR),# BM25
    FieldSchema(name="metadata", dtype=DataType.JSON),# dynamic schema
]
# hybrid retrieval + metadata filtering
results = collection.hybrid_search(
reqs=[
AnnSearchRequest(data=[dense_vec], anns_field=“dense_embedding”, limit=20),
AnnSearchRequest(data=[sparse_vec], anns_field=“sparse_embedding”, limit=20)
],
rerank=RRFRanker(),
output_fields=[“metadata”],
expr=‘metadata[“confidence”] > 0.9’,# CRAG confidence filtering
limit=5
)

Milvus juga menyederhanakan penskalaan penerapan. Menawarkan mode Lite, Standalone, dan Distributed yang kompatibel dengan kode - peralihan dari pengembangan lokal ke klaster produksi hanya membutuhkan perubahan string koneksi.

Praktik Langsung: Membangun Sistem CRAG dengan Middleware LangGraph dan Milvus

Mengapa Pendekatan Middleware?

Cara umum untuk membangun CRAG dengan LangGraph adalah dengan menyambungkan state graph dengan node dan edge yang mengendalikan setiap langkah. Cara ini berhasil, tetapi graf menjadi kusut seiring dengan bertambahnya kompleksitas, dan debugging menjadi sangat memusingkan.

Kami memilih pola Middleware di LangGraph 1.0. Pola ini mencegat permintaan sebelum pemanggilan model, sehingga pengambilan, evaluasi, dan koreksi ditangani di satu tempat yang kohesif. Dibandingkan dengan pendekatan state-graph:

• Lebih sedikit kode: logika terpusat, tidak tersebar di seluruh node grafik
• Lebih mudah diikuti: aliran kontrol terbaca secara linier
• Lebih mudah di-debug: kegagalan mengarah ke satu lokasi, bukan ke penjelajahan grafik

Alur Kerja Inti

Pipeline berjalan dalam empat langkah:

1. Pengambilan: mengambil 3 dokumen teratas yang relevan dari Milvus, dengan cakupan ke penyewa saat ini
2. Evaluasi: menilai kualitas dokumen dengan model yang ringan
3. Koreksi: menyempurnakan, melengkapi dengan pencarian web, atau kembali sepenuhnya - berdasarkan keputusan
4. Injeksi: memberikan konteks yang telah diselesaikan ke model melalui perintah sistem yang dinamis

Pengaturan Lingkungan dan Persiapan Data

Variabel lingkungan

export OPENAI_API_KEY="your-api-key"
export TAVILY_API_KEY="your-tavily-key"

Membuat koleksi Milvus

Sebelum menjalankan kode, buatlah koleksi di Milvus dengan skema yang sesuai dengan logika pengambilan.

# filename: crag_agent.py
# ============ Import dependencies ============
from typing import Literal, List
from langchain.agents import create_agent
from langchain.agents.middleware import AgentMiddleware, before_model, dynamic_prompt
from langchain.chat_models import init_chat_model
from langchain_milvus import Milvus
from langchain_openai import OpenAIEmbeddings
from langchain_core.documents import Document
from langchain_core.messages import SystemMessage, HumanMessage
from langchain_community.tools.tavily_search import TavilySearchResults
# ============ CRAG Middleware (minimal-change version) ============
class CRAGMiddleware(AgentMiddleware):
"""CRAG evaluation and correction middleware (uses official decorator-based hooks to avoid permanently polluting the message stack)“"”
<span class="hljs-keyword">def</span> <span class="hljs-title function_">__init__</span>(<span class="hljs-params">self, vector_store: Milvus, agent_id: <span class="hljs-built_in">str</span></span>):
    <span class="hljs-built_in">super</span>().__init__()
    <span class="hljs-variable language_">self</span>.vector_store = vector_store
    <span class="hljs-variable language_">self</span>.agent_id = agent_id  <span class="hljs-comment"># multi-tenant isolation</span>
    <span class="hljs-comment"># Lightweight evaluator: used for relevance judgment (can be replaced with the structured version introduced later)</span>
    <span class="hljs-variable language_">self</span>.evaluator = init_chat_model(<span class="hljs-string">&quot;openai:gpt-4o-mini&quot;</span>, temperature=<span class="hljs-number">0</span>)
    <span class="hljs-comment"># Web search fallback</span>
    <span class="hljs-variable language_">self</span>.web_search = TavilySearchResults(max_results=<span class="hljs-number">3</span>)

    @before_model
def run_crag(self, state):
“""Run retrieval -> evaluation -> correction before model invocation and prepare the final context""”
# Get the last user message
last_msg = state[“messages”][-1]
query = getattr(last_msg, “content”, “”) if hasattr(last_msg, “content”) else last_msg.get(“content”, “”)
    <span class="hljs-comment"># 1. Retrieval: get documents from Milvus (PartitionKey + confidence filtering)</span>
    docs = <span class="hljs-variable language_">self</span>._retrieve_from_milvus(query)

    <span class="hljs-comment"># 2. Evaluation: three-way decision</span>
    verdict = <span class="hljs-variable language_">self</span>._evaluate_relevance(query, docs)

    <span class="hljs-comment"># 3. Correction: choose the handling strategy based on the verdict</span>
    <span class="hljs-keyword">if</span> verdict == <span class="hljs-string">&quot;incorrect&quot;</span>:
        <span class="hljs-comment"># Retrieval failed, rely entirely on Web search</span>
        web_results = <span class="hljs-variable language_">self</span>._web_search_fallback(query)
        final_context = <span class="hljs-variable language_">self</span>._format_web_results(web_results)
    <span class="hljs-keyword">elif</span> verdict == <span class="hljs-string">&quot;ambiguous&quot;</span>:
        <span class="hljs-comment"># Retrieval is ambiguous, refine documents + supplement with Web search</span>
        refined_docs = <span class="hljs-variable language_">self</span>._refine_documents(docs, query)
        web_results = <span class="hljs-variable language_">self</span>._web_search_fallback(query)
        final_context = <span class="hljs-variable language_">self</span>._merge_context(refined_docs, web_results)
    <span class="hljs-keyword">else</span>:
        <span class="hljs-comment"># Retrieval quality is good, only refine the documents</span>
        refined_docs = <span class="hljs-variable language_">self</span>._refine_documents(docs, query)
        final_context = <span class="hljs-variable language_">self</span>._format_internal_docs(refined_docs)

    <span class="hljs-comment"># 4. Put the context into a temporary key, used only for dynamic prompt assembly in the current model call</span>
    state[<span class="hljs-string">&quot;_crag_context&quot;</span>] = final_context
    <span class="hljs-keyword">return</span> state

    @dynamic_prompt
def attach_context(self, state, prompt_messages: List):
“""Inject the CRAG-generated context as a SystemMessage before the prompt for the current model call""”
final_context = state.get(“_crag_context”)
if final_context:
sys_msg = SystemMessage(
content=f"Here is some relevant background information. Please answer the user’s question based on this information:\n\n{final_context}"
)
# Applies only to the current call and is not permanently written into state[“messages”]
prompt_messages = [sys_msg] + prompt_messages
return prompt_messages
<span class="hljs-comment"># ======== Internal methods: retrieval / evaluation / refinement / formatting ========</span>

<span class="hljs-keyword">def</span> <span class="hljs-title function_">_retrieve_from_milvus</span>(<span class="hljs-params">self, query: <span class="hljs-built_in">str</span></span>) -&gt; <span class="hljs-built_in">list</span>:
    <span class="hljs-string">&quot;&quot;&quot;Retrieve documents from Milvus (Partition Key + confidence filtering)&quot;&quot;&quot;</span>
    <span class="hljs-keyword">try</span>:
        <span class="hljs-comment"># Note: different adapter versions may place filter parameters differently; here expr is passed through search_kwargs</span>
        docs = <span class="hljs-variable language_">self</span>.vector_store.similarity_search(
            query,
            k=<span class="hljs-number">3</span>,
            search_kwargs={<span class="hljs-string">&quot;expr&quot;</span>: <span class="hljs-string">f&#x27;agent_id == &quot;<span class="hljs-subst">{self.agent_id}</span>&quot;&#x27;</span>}
        )
        <span class="hljs-comment"># Confidence filtering (to avoid low-quality memory contamination)</span>
        filtered_docs = [
            doc <span class="hljs-keyword">for</span> doc <span class="hljs-keyword">in</span> docs
            <span class="hljs-keyword">if</span> (doc.metadata <span class="hljs-keyword">or</span> {}).get(<span class="hljs-string">&quot;confidence&quot;</span>, <span class="hljs-number">0.0</span>) &gt; <span class="hljs-number">0.7</span>
        ]
        <span class="hljs-keyword">return</span> filtered_docs <span class="hljs-keyword">or</span> docs  <span class="hljs-comment"># If there are no high-confidence results, fall back to the original results for evaluator judgment</span>
    <span class="hljs-keyword">except</span> Exception <span class="hljs-keyword">as</span> e:
        <span class="hljs-built_in">print</span>(<span class="hljs-string">f&quot;[CRAG] Retrieval failed: <span class="hljs-subst">{e}</span>&quot;</span>)
        <span class="hljs-keyword">return</span> []

<span class="hljs-keyword">def</span> <span class="hljs-title function_">_evaluate_relevance</span>(<span class="hljs-params">self, query: <span class="hljs-built_in">str</span>, docs: <span class="hljs-built_in">list</span></span>) -&gt; <span class="hljs-type">Literal</span>[<span class="hljs-string">&quot;relevant&quot;</span>, <span class="hljs-string">&quot;ambiguous&quot;</span>, <span class="hljs-string">&quot;incorrect&quot;</span>]:
    <span class="hljs-string">&quot;&quot;&quot;Evaluate document relevance (three-way decision), simplified version: the LLM returns the verdict directly&quot;&quot;&quot;</span>
    <span class="hljs-keyword">if</span> <span class="hljs-keyword">not</span> docs:
        <span class="hljs-keyword">return</span> <span class="hljs-string">&quot;incorrect&quot;</span>

    <span class="hljs-comment"># Evaluate only the Top-3 documents, taking the first 500 characters of each</span>
    doc_content = <span class="hljs-string">&quot;\n\n&quot;</span>.join([
        <span class="hljs-string">f&quot;[Document <span class="hljs-subst">{i+<span class="hljs-number">1</span>}</span>] <span class="hljs-subst">{(doc.page_content <span class="hljs-keyword">or</span> <span class="hljs-string">&#x27;&#x27;</span>)[:<span class="hljs-number">500</span>]}</span>...&quot;</span>
        <span class="hljs-keyword">for</span> i, doc <span class="hljs-keyword">in</span> <span class="hljs-built_in">enumerate</span>(docs[:<span class="hljs-number">3</span>])
    ])

    prompt = <span class="hljs-string">f&quot;&quot;&quot;You are an expert in document relevance evaluation. Assess whether the following documents can answer the query.

Query: {query}
Document content:
{doc_content}
Evaluation criteria:

relevant: the document directly contains the answer and is highly relevant
ambiguous: the document is partially relevant and needs external knowledge
incorrect: the document is irrelevant and cannot answer the query

Return only one word: relevant or ambiguous or incorrect
“"”
try:
result = self.evaluator.invoke(prompt)
verdict = (getattr(result, “content”, “”) or “”).strip().lower()
if verdict not in {“relevant”, “ambiguous”, “incorrect”}:
verdict = “ambiguous”
return verdict
except Exception as e:
print(f"[CRAG] Evaluation failed: {e}")
return “ambiguous”
<span class="hljs-keyword">def</span> <span class="hljs-title function_">_refine_documents</span>(<span class="hljs-params">self, docs: <span class="hljs-built_in">list</span>, query: <span class="hljs-built_in">str</span></span>) -&gt; <span class="hljs-built_in">list</span>:
    <span class="hljs-string">&quot;&quot;&quot;Refine documents (simplified strips: sentence filtering based on keywords)&quot;&quot;&quot;</span>
    refined = []
    <span class="hljs-comment"># Simple Chinese-period replacement + rough English sentence splitting</span>
    keywords = [kw.strip() <span class="hljs-keyword">for</span> kw <span class="hljs-keyword">in</span> query.split() <span class="hljs-keyword">if</span> kw.strip()]

    <span class="hljs-keyword">for</span> doc <span class="hljs-keyword">in</span> docs:
        text = doc.page_content <span class="hljs-keyword">or</span> <span class="hljs-string">&quot;&quot;</span>
        sentences = (
            text.replace(<span class="hljs-string">&quot;。&quot;</span>, <span class="hljs-string">&quot;。\n&quot;</span>)
                .replace(<span class="hljs-string">&quot;. &quot;</span>, <span class="hljs-string">&quot;.\n&quot;</span>)
                .replace(<span class="hljs-string">&quot;! &quot;</span>, <span class="hljs-string">&quot;!\n&quot;</span>)
                .replace(<span class="hljs-string">&quot;? &quot;</span>, <span class="hljs-string">&quot;?\n&quot;</span>)
                .split(<span class="hljs-string">&quot;\n&quot;</span>)
        )
        sentences = [s.strip() <span class="hljs-keyword">for</span> s <span class="hljs-keyword">in</span> sentences <span class="hljs-keyword">if</span> s.strip()]

        <span class="hljs-comment"># Match any keyword</span>
        relevant_sentences = [
            s <span class="hljs-keyword">for</span> s <span class="hljs-keyword">in</span> sentences
            <span class="hljs-keyword">if</span> <span class="hljs-built_in">any</span>(keyword <span class="hljs-keyword">in</span> s <span class="hljs-keyword">for</span> keyword <span class="hljs-keyword">in</span> keywords)
        ]

        <span class="hljs-keyword">if</span> relevant_sentences:
            refined_text = <span class="hljs-string">&quot;。&quot;</span>.join(relevant_sentences[:<span class="hljs-number">3</span>])
            refined.append(Document(page_content=refined_text, metadata=doc.metadata <span class="hljs-keyword">or</span> {}))

    <span class="hljs-keyword">return</span> refined <span class="hljs-keyword">if</span> refined <span class="hljs-keyword">else</span> docs  <span class="hljs-comment"># If nothing is extracted, fall back to the original documents</span>

<span class="hljs-keyword">def</span> <span class="hljs-title function_">_web_search_fallback</span>(<span class="hljs-params">self, query: <span class="hljs-built_in">str</span></span>) -&gt; <span class="hljs-built_in">list</span>:
    <span class="hljs-string">&quot;&quot;&quot;Web search fallback&quot;&quot;&quot;</span>
    <span class="hljs-keyword">try</span>:
        <span class="hljs-keyword">return</span> <span class="hljs-variable language_">self</span>.web_search.invoke(query) <span class="hljs-keyword">or</span> []
    <span class="hljs-keyword">except</span> Exception <span class="hljs-keyword">as</span> e:
        <span class="hljs-built_in">print</span>(<span class="hljs-string">f&quot;[CRAG] Web search failed: <span class="hljs-subst">{e}</span>&quot;</span>)
        <span class="hljs-keyword">return</span> []

<span class="hljs-keyword">def</span> <span class="hljs-title function_">_merge_context</span>(<span class="hljs-params">self, internal_docs: <span class="hljs-built_in">list</span>, web_results: <span class="hljs-built_in">list</span></span>) -&gt; <span class="hljs-built_in">str</span>:
    <span class="hljs-string">&quot;&quot;&quot;Merge internal memory and external knowledge into the final context&quot;&quot;&quot;</span>
    parts = []
    <span class="hljs-keyword">if</span> internal_docs:
        parts.append(<span class="hljs-string">&quot;[Internal Memory]&quot;</span>)
        <span class="hljs-keyword">for</span> i, doc <span class="hljs-keyword">in</span> <span class="hljs-built_in">enumerate</span>(internal_docs, <span class="hljs-number">1</span>):
            parts.append(<span class="hljs-string">f&quot;<span class="hljs-subst">{i}</span>. <span class="hljs-subst">{doc.page_content}</span>&quot;</span>)
    <span class="hljs-keyword">if</span> web_results:
        parts.append(<span class="hljs-string">&quot;[External Knowledge]&quot;</span>)
        <span class="hljs-keyword">for</span> i, result <span class="hljs-keyword">in</span> <span class="hljs-built_in">enumerate</span>(web_results, <span class="hljs-number">1</span>):
            content = (result <span class="hljs-keyword">or</span> {}).get(<span class="hljs-string">&quot;content&quot;</span>, <span class="hljs-string">&quot;&quot;</span>)
            url = (result <span class="hljs-keyword">or</span> {}).get(<span class="hljs-string">&quot;url&quot;</span>, <span class="hljs-string">&quot;&quot;</span>)
            parts.append(<span class="hljs-string">f&quot;<span class="hljs-subst">{i}</span>. <span class="hljs-subst">{content}</span>\n   Source: <span class="hljs-subst">{url}</span>&quot;</span>)
    <span class="hljs-keyword">return</span> <span class="hljs-string">&quot;\n\n&quot;</span>.join(parts) <span class="hljs-keyword">if</span> parts <span class="hljs-keyword">else</span> <span class="hljs-string">&quot;No relevant information found&quot;</span>

<span class="hljs-keyword">def</span> <span class="hljs-title function_">_format_internal_docs</span>(<span class="hljs-params">self, docs: <span class="hljs-built_in">list</span></span>) -&gt; <span class="hljs-built_in">str</span>:
    <span class="hljs-string">&quot;&quot;&quot;Format internal documents&quot;&quot;&quot;</span>
    <span class="hljs-keyword">if</span> <span class="hljs-keyword">not</span> docs:
        <span class="hljs-keyword">return</span> <span class="hljs-string">&quot;No relevant information found&quot;</span>
    parts = [<span class="hljs-string">&quot;[Internal Memory]&quot;</span>]
    <span class="hljs-keyword">for</span> i, doc <span class="hljs-keyword">in</span> <span class="hljs-built_in">enumerate</span>(docs, <span class="hljs-number">1</span>):
        parts.append(<span class="hljs-string">f&quot;<span class="hljs-subst">{i}</span>. <span class="hljs-subst">{doc.page_content}</span>&quot;</span>)
    <span class="hljs-keyword">return</span> <span class="hljs-string">&quot;\n\n&quot;</span>.join(parts)

<span class="hljs-keyword">def</span> <span class="hljs-title function_">_format_web_results</span>(<span class="hljs-params">self, results: <span class="hljs-built_in">list</span></span>) -&gt; <span class="hljs-built_in">str</span>:
    <span class="hljs-string">&quot;&quot;&quot;Format Web search results&quot;&quot;&quot;</span>
    <span class="hljs-keyword">if</span> <span class="hljs-keyword">not</span> results:
        <span class="hljs-keyword">return</span> <span class="hljs-string">&quot;No relevant information found&quot;</span>
    parts = [<span class="hljs-string">&quot;[External Knowledge]&quot;</span>]
    <span class="hljs-keyword">for</span> i, result <span class="hljs-keyword">in</span> <span class="hljs-built_in">enumerate</span>(results, <span class="hljs-number">1</span>):
        content = (result <span class="hljs-keyword">or</span> {}).get(<span class="hljs-string">&quot;content&quot;</span>, <span class="hljs-string">&quot;&quot;</span>)
        url = (result <span class="hljs-keyword">or</span> {}).get(<span class="hljs-string">&quot;url&quot;</span>, <span class="hljs-string">&quot;&quot;</span>)
        parts.append(<span class="hljs-string">f&quot;<span class="hljs-subst">{i}</span>. <span class="hljs-subst">{content}</span>\n   Source: <span class="hljs-subst">{url}</span>&quot;</span>)
    <span class="hljs-keyword">return</span> <span class="hljs-string">&quot;\n\n&quot;</span>.join(parts)

# ============ Initialize the Milvus vector database ============
vector_store = Milvus(
embedding_function=OpenAIEmbeddings(),
connection_args={“host”: “localhost”, “port”: “19530”},
collection_name=“agent_memory”
)
# ============ Create Agent ============
agent = create_agent(
model=“openai:gpt-4o”,
tools=[TavilySearchResults(max_results=3)],  # Web search tool
middleware=[
CRAGMiddleware(
vector_store=vector_store,
agent_id=“user_123_session_456”  # multi-tenant isolation: each Agent instance uses its own ID
)
]
)
# ============ Example run ============
if name == "main":
# Example query: use HumanMessage to ensure compatibility
response = agent.invoke({
“messages”: [
HumanMessage(content=“What were the operating expenses in Nike’s latest quarterly earnings report?”)
]
})
print(response[“messages”][-1].content)

Catatan Versi: Kode ini menggunakan fitur-fitur Middleware terbaru di LangGraph dan LangChain. API ini dapat berubah seiring dengan perkembangan kerangka kerja-periksa dokumentasi LangGraph untuk mengetahui penggunaan terbaru.

Modul Utama

1. Desain evaluator tingkat produksi

Metode _evaluate_relevance() pada kode di atas sengaja disederhanakan untuk pengujian cepat. Untuk produksi, Anda akan menginginkan output yang terstruktur dengan penilaian kepercayaan dan penjelasan:

from pydantic import BaseModel
from langchain.prompts import PromptTemplate
class RelevanceVerdict(BaseModel):
“""Structured output for the evaluation result""”
verdict: Literal[“relevant”, “ambiguous”, “incorrect”]
confidence: float  # confidence score (used for memory quality monitoring)
reasoning: str     # reason for the judgment (used for debugging and review)
# Note: the CRAG paper uses a fine-tuned T5-Large evaluator (10-20 ms latency)
# Here, gpt-4o-mini is used as the engineering implementation option (easier to deploy, but with slightly higher latency)
grader_llm = ChatOpenAI(model=“gpt-4o-mini”, temperature=0)
grader_prompt = PromptTemplate(
template="""You are an expert in document relevance evaluation. Assess whether the following documents can answer the query.
Query: {query}
Document content:
{document}
Evaluation criteria:

relevant: the document directly contains the answer, confidence > 0.9
ambiguous: the document is partially relevant, confidence 0.5-0.9
incorrect: the document is irrelevant, confidence < 0.5

Return in JSON format: {{"verdict": "…", "confidence": 0.xx, "reasoning": "…"}}
“"”,
input_variables=[“query”, “document”]
)
grader_chain = grader_prompt | grader_llm.with_structured_output(RelevanceVerdict)
# Replace the _evaluate_relevance() method in CRAGMiddleware
def _evaluate_relevance(self, query: str, docs: list) -> Literal[“relevant”, “ambiguous”, “incorrect”]:
"""Evaluate document relevance (returns structured result)“"”
if not docs:
return “incorrect”
<span class="hljs-comment"># Evaluate only the Top-3 documents, taking the first 500 characters of each</span>
doc_content = <span class="hljs-string">&quot;\n\n&quot;</span>.join([
    <span class="hljs-string">f&quot;[Document <span class="hljs-subst">{i+<span class="hljs-number">1</span>}</span>] <span class="hljs-subst">{doc.page_content[:<span class="hljs-number">500</span>]}</span>...&quot;</span>
    <span class="hljs-keyword">for</span> i, doc <span class="hljs-keyword">in</span> <span class="hljs-built_in">enumerate</span>(docs[:<span class="hljs-number">3</span>])
])

result = grader_chain.invoke({
    <span class="hljs-string">&quot;query&quot;</span>: query,
    <span class="hljs-string">&quot;document&quot;</span>: doc_content
})

<span class="hljs-comment"># Store the confidence score in logs or a monitoring system</span>
<span class="hljs-built_in">print</span>(<span class="hljs-string">f&quot;[CRAG Evaluation] verdict=<span class="hljs-subst">{result.verdict}</span>, confidence=<span class="hljs-subst">{result.confidence:<span class="hljs-number">.2</span>f}</span>&quot;</span>)
<span class="hljs-built_in">print</span>(<span class="hljs-string">f&quot;[CRAG Reasoning] <span class="hljs-subst">{result.reasoning}</span>&quot;</span>)

<span class="hljs-comment"># Optional: store the evaluation result in Milvus for memory quality analysis</span>
<span class="hljs-variable language_">self</span>._store_evaluation_metrics(query, result)

<span class="hljs-keyword">return</span> result.verdict

def _store_evaluation_metrics(self, query: str, verdict_result: RelevanceVerdict):
"""Store evaluation metrics in Milvus (for memory quality monitoring)“"”
# Example: store the evaluation result in a separate Collection for analysis
# In actual use, you need to create the evaluation_metrics Collection
pass

2. Penyempurnaan dan pengembalian pengetahuan

Tiga mekanisme bekerja sama untuk menjaga konteks model tetap berkualitas tinggi:

• Penyempurnaan pengetahuan mengekstrak kalimat yang paling relevan dengan kueri dan menghilangkan noise.
• Pencarian fallback dipicu ketika pencarian lokal tidak mencukupi, menarik pengetahuan eksternal melalui Tavily.
• Penggabungan konteks menggabungkan memori internal dengan hasil eksternal ke dalam satu blok konteks yang diduplikasi sebelum mencapai model.

Kiat untuk Menjalankan CRAG dalam Produksi

Ada tiga area yang paling penting ketika Anda bergerak di luar pembuatan prototipe.

1. Biaya: Pilih Evaluator yang Tepat

Evaluator berjalan pada setiap kueri, menjadikannya pengungkit terbesar untuk latensi dan biaya.

• Beban kerja dengan konkurensi tinggi: Model ringan yang disetel dengan baik seperti T5-Large menjaga latensi pada 10-20 ms dan biaya yang dapat diprediksi.
• Lalu lintas rendah atau pembuatan prototipe: Model yang di-host seperti gpt-4o-mini lebih cepat disiapkan dan membutuhkan lebih sedikit pekerjaan operasional, tetapi latensi dan biaya per panggilan menjadi lebih tinggi.

2. Dapat diamati: Instrumen dari Hari Pertama

Masalah produksi yang paling sulit adalah masalah yang tidak dapat Anda lihat hingga kualitas jawaban sudah menurun.

• Pemantauan infrastruktur: Milvus terintegrasi dengan Prometheus. Mulailah dengan tiga metrik: milvus_query_latency_seconds, milvus_search_qps, dan milvus_insert_throughput.
• Pemantauan aplikasi: Lacak distribusi putusan CRAG, tingkat pemicu penelusuran web, dan distribusi skor kepercayaan. Tanpa sinyal-sinyal ini, Anda tidak dapat mengetahui apakah penurunan kualitas disebabkan oleh pengambilan yang buruk atau kesalahan penilaian evaluator.

3. Pemeliharaan Jangka Panjang: Mencegah Kontaminasi Memori

Semakin lama sebuah agen berjalan, semakin banyak data yang basi dan berkualitas rendah terakumulasi dalam memori. Siapkan pagar pembatas lebih awal:

• Pra-penyaringan: Hanya memori permukaan dengan confidence > 0.7 sehingga konten berkualitas rendah diblokir sebelum mencapai evaluator.
• Peluruhan waktu: Secara bertahap mengurangi bobot memori yang lebih lama. Tiga puluh hari adalah default awal yang wajar, dapat disesuaikan per kasus penggunaan.
• Pembersihan terjadwal: Jalankan pekerjaan mingguan untuk membersihkan memori lama yang tidak terverifikasi. Hal ini mencegah lingkaran umpan balik di mana data yang sudah basi diambil, digunakan, dan disimpan kembali.

Penutup - dan Beberapa Pertanyaan Umum

CRAG mengatasi salah satu masalah yang paling sering muncul dalam RAG produksi: hasil pencarian yang terlihat relevan namun sebenarnya tidak. Dengan menyisipkan langkah evaluasi dan koreksi antara pengambilan dan pembuatan, CRAG menyaring hasil yang buruk, mengisi kesenjangan dengan pencarian eksternal, dan memberikan konteks yang lebih bersih kepada model untuk digunakan.

Membuat CRAG bekerja dengan andal dalam produksi membutuhkan lebih dari sekadar logika pengambilan yang baik. Dibutuhkan basis data vektor yang menangani isolasi multi-penyewa, pencarian hibrida, dan skema yang terus berkembang-di sinilah Milvus cocok. Di sisi aplikasi, memilih evaluator yang tepat, menginstruksikan pengamatan lebih awal, dan secara aktif mengelola kualitas memori adalah hal yang membedakan demo dengan sistem yang dapat Anda percayai.

Jika Anda sedang membangun RAG atau sistem agen dan mengalami masalah kualitas pengambilan, kami ingin membantu:

• Bergabunglah dengan komunitas Milvus Slack untuk mengajukan pertanyaan, berbagi arsitektur Anda, dan belajar dari pengembang lain yang menangani masalah serupa.
• Pesan sesi Milvus Office Hours selama 20 menit untuk membahas kasus penggunaan Anda bersama tim-apakah itu desain CRAG, pengambilan hibrida, atau penskalaan multi-penyewa.
• Jika Anda lebih suka melewatkan penyiapan infrastruktur dan langsung membangun, Zilliz Cloud (dikelola Milvus) menawarkan tingkat gratis untuk memulai.

---

Beberapa pertanyaan yang sering muncul ketika tim mulai menerapkan CRAG:

Apa perbedaan CRAG dengan hanya menambahkan reranker ke RAG?

Reranker menyusun ulang hasil berdasarkan relevansi namun tetap mengasumsikan dokumen yang diambil dapat digunakan. CRAG melangkah lebih jauh lagi - CRAG mengevaluasi apakah konten yang diambil benar-benar menjawab kueri, dan mengambil tindakan korektif jika tidak: menyempurnakan kecocokan parsial, melengkapi dengan penelusuran web, atau membuang hasil sepenuhnya. Ini adalah lingkaran kontrol kualitas, bukan hanya jenis yang lebih baik.

Mengapa skor kemiripan yang tinggi terkadang menghasilkan dokumen yang salah?

Menyematkan kemiripan mengukur kedekatan semantik dalam ruang vektor, tetapi itu tidak sama dengan menjawab pertanyaan. Sebuah dokumen tentang mengkonfigurasi HTTPS di Apache secara semantik dekat dengan pertanyaan tentang HTTPS di Nginx-tetapi tidak akan membantu. CRAG menangkap hal ini dengan mengevaluasi relevansi dengan kueri yang sebenarnya, bukan hanya jarak vektor.

Apa yang harus saya cari dalam basis data vektor untuk CRAG?

Tiga hal yang paling penting: pengambilan hibrida (sehingga Anda dapat menggabungkan pencarian semantik dengan pencocokan kata kunci untuk istilah yang tepat), isolasi multi-penyewa (sehingga setiap sesi pengguna atau agen memiliki ruang memorinya sendiri), dan skema yang fleksibel (sehingga Anda dapat menambahkan bidang seperti confidence atau verified tanpa waktu henti saat pipeline Anda berkembang).

Apa yang terjadi jika tidak ada dokumen yang diambil yang relevan?

CRAG tidak menyerah begitu saja. Ketika keyakinan turun di bawah 0,5, ia akan kembali ke pencarian web. Ketika hasilnya ambigu (0,5-0,9), CRAG akan menggabungkan dokumen internal yang telah disempurnakan dengan hasil pencarian eksternal. Model ini selalu mendapatkan beberapa konteks untuk digunakan, bahkan ketika basis pengetahuan memiliki kesenjangan.