LlamaIndex Custom Retriever: ระบบค้นหาข้อมูลเชิงลึกเฉพาะทาง

ในโลกของ RAG (Retrieval-Augmented Generation) การสร้าง Retriever ที่เหมาะสมกับโดเมนธุรกิจของคุณคือหัวใจสำคัญ ผมเคยเจอปัญหาหลายต่อหลายครั้งกับระบบที่ใช้ Vector Search ธรรมดาแต่ไม่สามารถดึงข้อมูลที่แม่นยำได้ โดยเฉพาะในโปรเจกต์ e-commerce ที่ต้องค้นหาสินค้าตามคุณสมบัติเฉพาะ หรือระบบความรู้องค์กรที่มีคำศัพท์เฉพาะทางมากมาย

ทำไมต้อง Custom Retriever?

Retriever เริ่มต้นของ LlamaIndex ใช้ Semantic Search เพียงอย่างเดียว ซึ่งมีข้อจำกัดเมื่อต้องการค้นหาด้วยเงื่อนไขเฉพาะ เช่น:

• การค้นหาด้วยช่วงราคา หรือวันที่
• การกรองด้วยหมวดหมู่หลายระดับ
• การรวมผลลัพธ์จากหลายแหล่งข้อมูลพร้อมกัน
• การใช้ Keyword Search ร่วมกับ Vector Search

การสร้าง Hybrid Retriever พร้อม HolySheep AI

ในโปรเจกต์ล่าสุดที่ผมพัฒนาระบบ AI สำหรับร้านค้าออนไลน์ ผมใช้ HolySheep AI เป็น LLM endpoint เพราะมีความเร็วต่ำกว่า 50ms และราคาประหยัดกว่า 85% เมื่อเทียบกับ OpenAI ทำให้สามารถทดสอบ Hybrid Retriever ได้อย่างมีประสิทธิภาพ

from llama_index.core import VectorStoreIndex, SimpleDirectoryReader
from llama_index.core.retrievers import BaseRetriever
from llama_index.core.postprocessor import SimilarityPostprocessor
from llama_index.vector_stores.chroma import ChromaVectorStore
from llama_index.core.query_engine import RetrieverQueryEngine
import chromadb
from typing import List, Optional, Dict, Any

class HybridRetriever(BaseRetriever):
    """
    Hybrid Retriever ที่ผมพัฒนาเองสำหรับ e-commerce
    รวม Vector Search + Keyword Search + Metadata Filtering
    """
    
    def __init__(
        self,
        vector_store: ChromaVectorStore,
        similarity_top_k: int = 10,
        keyword_weight: float = 0.3,
        vector_weight: float = 0.7
    ):
        super().__init__()
        self.vector_store = vector_store
        self.similarity_top_k = similarity_top_k
        self.keyword_weight = keyword_weight
        self.vector_weight = vector_weight
        
    def retrieve(self, str_or_query_bundle: Any) -> List[NodeWithScore]:
        from llama_index.core.schema import NodeWithScore, TextNode
        
        query = str_or_query_bundle if isinstance(str_or_query_bundle, str) \
                else str_or_query_bundle.query_str
        
        # ดึงข้อมูลจาก Vector Store
        vector_results = self.vector_store.similarity_search(
            query=query,
            k=self.similarity_top_k
        )
        
        # คำนวณคะแนนรวมและ return
        nodes_with_scores = []
        for i, node in enumerate(vector_results):
            # ผสมคะแนน vector similarity กับ keyword match
            base_score = 1.0 - (i / len(vector_results))
            keyword_score = self._calculate_keyword_score(node.text, query)
            final_score = (self.vector_weight * base_score) + \
                         (self.keyword_weight * keyword_score)
            
            nodes_with_scores.append(NodeWithScore(
                node=node,
                score=final_score
            ))
        
        return sorted(nodes_with_scores, key=lambda x: x.score, reverse=True)
    
    def _calculate_keyword_score(self, text: str, query: str) -> float:
        """คำนวณคะแนน keyword matching"""
        query_terms = query.lower().split()
        text_lower = text.lower()
        matches = sum(1 for term in query_terms if term in text_lower)
        return matches / len(query_terms) if query_terms else 0.0

การใช้งานกับ RAG Pipeline

import os
from llama_index.llms.holysheep import HolySheep
from llama_index.core import Settings

ตั้งค่า LLM ด้วย HolySheep AI
ราคา: DeepSeek V3.2 $0.42/MTok, Gemini 2.5 Flash $2.50/MTok
llm = HolySheep(
    model="deepseek-v3",
    api_key="YOUR_HOLYSHEEP_API_KEY",  # ใช้ API key จาก HolySheep
    base_url="https://api.holysheep.ai/v1",
    temperature=0.7,
    max_tokens=2048
)

Settings.llm = llm

สร้าง Query Engine ด้วย Custom Retriever
query_engine = index.as_query_engine(
    retriever=hybrid_retriever,
    node_postprocessors=[
        SimilarityPostprocessor(similarity_cutoff=0.7)
    ]
)

ทดสอบการค้นหา
response = query_engine.query(
    "รายละเอียดสินค้าโทรศัพท์ iPhone ราคาไม่เกิน 30000 บาท"
)
print(response)

Advanced: Metadata Filtering Retriever

สำหรับระบบ Enterprise RAG ที่ต้องการค้นหาข้อมูลตาม metadata เฉพาะ ผมแนะนำให้ใช้ Metadata Filter Retriever ที่สามารถกรองข้อมูลก่อนที่จะนำไปสร้าง context

from llama_index.core.retrievers import MetadataTurboRetriever

class MetadataFilterRetriever(BaseRetriever):
    """
    Custom Retriever สำหรับกรองข้อมูลตาม metadata
    เหมาะสำหรับเอกสารองค์กรที่มีโครงสร้างชัดเจน
    """
    
    def __init__(
        self,
        vector_store: ChromaVectorStore,
        filter_fields: List[str] = ["category", "date", "author"]
    ):
        super().__init__()
        self.vector_store = vector_store
        self.filter_fields = filter_fields
        
    def retrieve(self, query: str, **kwargs) -> List[NodeWithScore]:
        # ดึง filters จาก query parameters
        filters = kwargs.get("filters", {})
        
        # Vector search พร้อม filter
        results = self.vector_store.similarity_search(
            query=query,
            k=20,
            filter=filters  # Chroma รองรับ metadata filtering
        )
        
        return [NodeWithScore(node=r, score=r.metadata.get("similarity", 1.0)) 
                for r in results]

ตัวอย่างการใช้งาน
results = retriever.retrieve(
    "นโยบายการคืนสินค้า",
    filters={
        "category": "policy",
        "department": "customer_service"
    }
)

ข้อผิดพลาดที่พบบ่อยและวิธีแก้ไข

1. ปัญหา: ChromaDB Connection Error

สาเหตุ: Chroma client ไม่สามารถเชื่อมต่อกับ server หรือ path ผิดพลาด

# ❌ วิธีที่ผิด - ไม่มีการจัดการ error
client = chromadb.PersistentClient(path="./chroma_db")

✅ วิธีที่ถูก - เพิ่ม error handling และ retry logic
import time

def get_chroma_client(path: str, max_retries: int = 3):
    for attempt in range(max_retries):
        try:
            client = chromadb.PersistentClient(path=path)
            # ทดสอบการเชื่อมต่อ
            client.list_collections()
            return client
        except Exception as e:
            if attempt < max_retries - 1:
                time.sleep(1)  # รอ 1 วินาทีก่อน retry
            else:
                raise ConnectionError(f"ChromaDB connection failed: {e}")

client = get_chroma_client("./chroma_db")

2. ปัญหา: LLM Response ไม่ตรงกับความต้องการ

สาเหตุ: temperature สูงเกินไป หรือ prompt ไม่ชัดเจน

# ❌ วิธีที่ผิด - ใช้ temperature เริ่มต้น (0.7) ซึ่งอาจให้ผลลัพธ์ไม่ตรง
llm = HolySheep(model="deepseek-v3", api_key="YOUR_HOLYSHEEP_API_KEY")

✅ วิธีที่ถูก - ตั้งค่า temperature และ system prompt ให้ชัดเจน
llm = HolySheep(
    model="deepseek-v3",
    api_key="YOUR_HOLYSHEEP_API_KEY",
    base_url="https://api.holysheep.ai/v1",
    temperature=0.2,  # ลด temperature สำหรับงานที่ต้องการความแม่นยำ
    max_tokens=1024,
)

เพิ่ม system prompt ที่ชัดเจน
Settings.context_window = 4096
Settings.chunk_size = 512

3. ปัญหา: Retrieval คืนค่าเอกสารไม่เกี่ยวข้อง

สาเหตุ: similarity threshold ต่ำเกินไป หรือ chunk size ไม่เหมาะสม

# ❌ วิธีที่ผิด - ใช้ค่าเริ่มต้นโดยไม่ปรับแต่ง
query_engine = index.as_query_engine()

✅ วิธีที่ถูก - ปรับ similarity threshold และ top_k
query_engine = index.as_query_engine(
    similarity_top_k=5,  # ลดจำนวนผลลัพธ์เพื่อเพิ่มความแม่นยำ
    node_postprocessors=[
        SimilarityPostprocessor(similarity_cutoff=0.75)  # กรองเอกสารที่ไม่คล้ายกัน
    ]
)

เพิ่มเติม: ใช้ QueryFusionRetriever เพื่อรวมผลลัพธ์จากหลาย retriever
from llama_index.core.retrievers import QueryFusionRetriever

fusion_retriever = QueryFusionRetriever(
    retrievers=[vector_retriever, keyword_retriever],
    mode="reciprocal_rerank",  # rerank ผลลัพธ์ร่วมกัน
    similarity_top_k=5
)

4. ปัญหา: Memory Error เมื่อ index ข้อมูลจำนวนมาก

สาเหตุ: โหลดข้อมูลทั้งหมดใน memory พร้อมกัน

# ❌ วิธีที่ผิด - โหลดเอกสารทั้งหมดในครั้งเดียว
documents = SimpleDirectoryReader("./data").load_data()
index = VectorStoreIndex.from_documents(documents)

✅ วิธีที่ถูก - ใช้ batch processing
from llama_index.core import Document

def process_documents_in_batches(data_dir: str, batch_size: int = 100):
    all_nodes = []
    reader = SimpleDirectoryReader(data_dir, recursive=True)
    
    for batch in reader.iter_data(batch_size=batch_size):
        # สร้าง nodes จาก batch
        nodes = [Document(text=doc.text, metadata=doc.metadata) 
                 for doc in batch]
        all_nodes.extend(nodes)
        
        # ประมวลผลแต่ละ batch
        if len(all_nodes) >= batch_size:
            yield all_nodes
            all_nodes = []
    
    if all_nodes:
        yield all_nodes

สร้าง index แบบ incremental
index = VectorStoreIndex([])
for batch_nodes in process_documents_in_batches("./data", batch_size=50):
    index.insert_nodes(batch_nodes)

สรุป

การสร้าง Custom Retriever ที่เหมาะสมกับโดเมนของคุณต้องอาศัยความเข้าใจทั้งในด้านข้อมูลและ algorithm ที่ใช้ จากประสบการณ์ของผม การผสมผสานหลายวิธีการค้นหา (Hybrid Search) ให้ผลลัพธ์ที่ดีกว่าการใช้วิธีใดวิธีหนึ่งเพียงอย่างเดียว และเมื่อใช้งานร่วมกับ HolySheep AI ที่มี latency ต่ำกว่า 50ms และราคาประหยัด (DeepSeek V3.2 เพียง $0.42/MTok) จะช่วยให้การพัฒนา RAG system มีประสิทธิภาพสูงสุดในราคาที่คุ้มค่า

💡 เคล็ดลับ: อย่าลืมทดสอบ Retriever ด้วยชุดข้อมูลจริงจากผู้ใช้ เพราะบางครั้ง semantic similarity อาจไม่ตรงกับความต้องการจริงของธุรกิจ

👉 สมัคร HolySheep AI — รับเครดิตฟรีเมื่อลงทะเบียน