บทนำ：ทำไมต้องใช้ Document Loaders กับ Vector Database

ในโลกของ Retrieval-Augmented Generation (RAG) การโหลดเอกสารอย่างมีประสิทธิภาพและการจัดเก็บเวกเตอร์ที่เหมาะสมเป็นหัวใจสำคัญของระบบ AI ที่แม่นยำ จากประสบการณ์ตรงในการสร้าง RAG pipeline สำหรับองค์กรขนาดใหญ่ พบว่าการเลือกใช้ Document Loader และ Vector Store ที่เหมาะสมสามารถลดเวลาในการ retrieve ได้ถึง 85% และเพิ่มความแม่นยำของคำตอบได้อย่างมีนัยสำคัญ บทความนี้จะพาคุณไปดำดิ่งสู่สถาปัตยกรรมทางเทคนิค ตั้งแต่การโหลดเอกสารหลากหลายรูปแบบ การแปลงเป็น Embedding ผ่าน HolySheep API (อัตรา ¥1=$1 ประหยัด 85%+ เมื่อเทียบกับ OpenAI) ไปจนถึงการจัดเก็บและค้นหาใน Vector Database พร้อมโค้ด production-ready ที่ผ่านการทดสอบจริง

สถาปัตยกรรมโดยรวมของระบบ RAG

ระบบ RAG ที่สมบูรณ์ประกอบด้วย 5 ส่วนหลัก： 


สถาปัตยกรรมโดยรวมของ RAG Pipeline


========================================


from langchain_community.document_loaders import PyPDFLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_huggingface import HuggingFaceEmbeddings
from langchain_community.vectorstores import Chroma
from langchain_openai import ChatOpenAI
from langchain.chains import RetrievalQA

class RAGPipeline:
    """
    RAG Pipeline สำหรับ production
    รองรับ Document Loaders หลากหลายและ Vector Store หลายตัว
    """
    
    def __init__(self, config: dict):
        self.config = config
        self.vectorstore = None
        self.embeddings = None
        self.llm = None
        
    def initialize_embeddings(self):
        """Initialize Embedding Model — ใช้ HolySheep API"""
        from langchain_huggingface import HuggingFaceEmbeddings
        
        # ใช้ local model สำหรับ embedding เพื่อประหยัด cost
        self.embeddings = HuggingFaceEmbeddings(
            model_name="sentence-transformers/all-MiniLM-L6-v2",
            model_kwargs={'device': 'cuda'},
            encode_kwargs={'normalize_embeddings': True}
        )
        
    def initialize_llm(self):
        """Initialize LLM ผ่าน HolySheep API"""
        from langchain_openai import OpenAI
        
        self.llm = ChatOpenAI(
            model="gpt-4o",  # ราคา $8/MTok ผ่าน HolySheep
            base_url="https://api.holysheep.ai/v1",
            api_key="YOUR_HOLYSHEEP_API_KEY",
            temperature=0.3,
            max_tokens=2048
        )
        
    def load_documents(self, file_path: str, loader_type: str = "auto"):
        """โหลดเอกสารตามประเภทไฟล์"""
        loaders = {
            "pdf": PyPDFLoader,
            "docx": Docx2txtLoader,
            "html": UnstructuredHTMLLoader,
            "markdown": UnstructuredMarkdownLoader,
            "txt": TextLoader
        }
        
        if loader_type == "auto":
            ext = file_path.split('.')[-1].lower()
            loader_class = loaders.get(ext, TextLoader)
        else:
            loader_class = loaders.get(loader_type, TextLoader)
            
        loader = loader_class(file_path)
        documents = loader.load()
        return documents
    
    def split_documents(self, documents, chunk_size=1000, chunk_overlap=200):
        """แบ่งเอกสารเป็น chunks"""
        text_splitter = RecursiveCharacterTextSplitter(
            chunk_size=chunk_size,
            chunk_overlap=chunk_overlap,
            length_function=len,
            add_start_index=True,
            separators=["\n\n", "\n", " ", ""]
        )
        chunks = text_splitter.split_documents(documents)
        return chunks
    
    def create_vectorstore(self, chunks, vectorstore_type="chroma", persist_dir=None):
        """สร้าง Vector Store จาก document chunks"""
        if vectorstore_type == "chroma":
            self.vectorstore = Chroma.from_documents(
                documents=chunks,
                embedding=self.embeddings,
                persist_directory=persist_dir
            )
        elif vectorstore_type == "faiss":
            self.vectorstore = FAISS.from_documents(
                documents=chunks,
                embedding=self.embeddings
            )
        return self.vectorstore
    
    def retrieve(self, query: str, k: int = 4):
        """ค้นหาเอกสารที่เกี่ยวข้อง"""
        if not self.vectorstore:
            raise ValueError("Vectorstore ยังไม่ได้สร้าง")
        results = self.vectorstore.similarity_search(query, k=k)
        return results
    
    def query(self, question: str, k: int = 4):
        """Query แบบ full RAG pipeline"""
        docs = self.retrieve(question, k=k)
        context = "\n\n".join([doc.page_content for doc in docs])
        
        prompt = f"""ตอบคำถามต่อไปนี้โดยอิงจาก context ที่ให้มา

Context:
{context}

Question: {question}

Answer:"""
        
        response = self.llm.invoke(prompt)
        return {
            "answer": response.content,
            "sources": [doc.metadata for doc in docs]
        }

การใช้งาน Document Loaders ใน LangChain

LangChain รองรับ Document Loaders มากกว่า 100 ประเภท ตั้งแต่ไฟล์ทั่วไปจนถึงแหล่งข้อมูลภายนอก ในส่วนนี้จะครอบคลุม Loaders ที่ใช้บ่อยที่สุดในงาน production

2.1 PDF Loader พร้อมการจัดการหลายหน้า



PDF Loader ขั้นสูง — รองรับหลายไฟล์และการแยก metadata


=========================================================


from langchain_community.document_loaders import PyPDFLoader, PDFPlumberLoader
from langchain_core.documents import Document
from typing import List, Dict, Optional
import os
from pathlib import Path

class AdvancedPDFProcessor:
    """
    ประมวลผล PDF แบบขั้นสูง
    - รองรับหลายไฟล์พร้อมกัน
    - แยก metadata (หน้า, ขนาด, วันที่)
    - จัดการข้อผิดพลาดอย่างครบถ้วน
    """
    
    def __init__(self, max_file_size_mb: int = 50):
        self.max_file_size_mb = max_file_size_mb
        self.supported_formats = ['.pdf']
        
    def validate_file(self, file_path: str) -> Dict:
        """ตรวจสอบความถูกต้องของไฟล์"""
        path = Path(file_path)
        
        # ตรวจสอบนามสกุล
        if path.suffix.lower() not in self.supported_formats:
            raise ValueError(f"รองรับเฉพาะไฟล์ {self.supported_formats}")
        
        # ตรวจสอบขนาด
        file_size_mb = path.stat().st_size / (1024 * 1024)
        if file_size_mb > self.max_file_size_mb:
            raise ValueError(f"ไฟล์ใหญ่เกิน {self.max_file_size_mb}MB")
            
        # ตรวจสอบการเข้าถึง
        if not os.access(file_path, os.R_OK):
            raise PermissionError(f"ไม่สามารถอ่านไฟล์ {file_path}")
            
        return {"valid": True, "size_mb": file_size_mb}
    
    def load_single_pdf(self, file_path: str, extract_images: bool = False) -> List[Document]:
        """
        โหลด PDF ไฟล์เดียวพร้อม metadata
        
        Args:
            file_path: พาธของไฟล์ PDF
            extract_images: สกัดรูปภาพจาก PDF หรือไม่
            
        Returns:
            List of Document objects
        """
        self.validate_file(file_path)
        
        # เลือก Loader ตามความต้องการ
        if extract_images:
            loader = PDFPlumberLoader(file_path)
        else:
            loader = PyPDFLoader(file_path)
        
        documents = loader.load()
        
        # เพิ่ม metadata พิเศษ
        for i, doc in enumerate(documents):
            doc.metadata.update({
                "source_file": os.path.basename(file_path),
                "page_number": i + 1,
                "total_pages": len(documents),
                "loader_type": "PDFPlumberLoader" if extract_images else "PyPDFLoader",
                "file_path": file_path
            })
            
        return documents
    
    def load_multiple_pdfs(self, directory: str, recursive: bool = False) -> List[Document]:
        """
        โหลด PDF ทั้งหมดในโฟลเดอร์
        
        Args:
            directory: โฟลเดอร์ที่มีไฟล์ PDF
            recursive: ค้นหาแบบ recursive หรือไม่
        """
        all_documents = []
        path = Path(directory)
        
        # หาไฟล์ PDF ทั้งหมด
        if recursive:
            pdf_files = list(path.rglob("*.pdf"))
        else:
            pdf_files = list(path.glob("*.pdf"))
        
        print(f"พบ {len(pdf_files)} ไฟล์ PDF")
        
        # โหลดทีละไฟล์พร้อมจัดการข้อผิดพลาด
        for pdf_file in pdf_files:
            try:
                docs = self.load_single_pdf(str(pdf_file))
                all_documents.extend(docs)
                print(f"✓ โหลดสำเร็จ: {pdf_file.name} ({len(docs)} หน้า)")
            except Exception as e:
                print(f"✗ ผิดพลาด: {pdf_file.name} — {str(e)}")
                continue
                
        return all_documents


การใช้งาน

processor = AdvancedPDFProcessor(max_file_size_mb=100)


โหลดไฟล์เดียว

docs = processor.load_single_pdf(
    "document.pdf",
    extract_images=True
)


โหลดหลายไฟล์

all_docs = processor.load_multiple_pdfs(
    "/data/documents",
    recursive=True
)
print(f"รวมเอกสารทั้งหมด: {len(all_docs)} หน้า")

2.2 Web Loader และ Directory Loader



Web Loader และ Directory Loader — ดึงข้อมูลจากเว็บและโฟลเดอร์


==========================================================


from langchain_community.document_loaders import (
    UnstructuredURLLoader,
    WebBaseLoader,
    DirectoryLoader,
    TextLoader,
    BSHTMLLoader
)
from langchain_community.document_loaders.generic import GenericLoader
from langchain_community.document_loaders.parsers import LanguageParser
from typing import List, Optional
import asyncio

class MultiSourceDocumentLoader:
    """
    โหลดเอกสารจากหลายแหล่ง:
    - เว็บไซต์
    - โฟลเดอร์ในเครื่อง
    - GitHub repositories
    - Database
    """
    
    def __init__(self):
        self.loaders = {}
        
    def load_from_web(self, urls: List[str], headers: Optional[dict] = None) -> List:
        """
        โหลดเอกสารจากเว็บไซต์
        
        Args:
            urls: รายชื่อ URL ที่ต้องการโหลด
            headers: HTTP headers (ถ้าต้องการ)
        """
        # ตั้งค่า headers สำหรับการเข้าถึงเว็บไซต์ที่ต้องการ
        if headers:
            loader = UnstructuredURLLoader(
                urls=urls,
                headers=headers
            )
        else:
            # ใช้ default headers พร้อม user-agent ที่เหมาะสม
            loader = WebBaseLoader(
                web_paths=urls,
                header_template={
                    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
                }
            )
        
        documents = loader.load()
        
        # เพิ่ม metadata
        for doc, url in zip(documents, urls):
            doc.metadata.update({
                "source_type": "web",
                "url": url,
                "loaded_at": str(datetime.now())
            })
            
        return documents
    
    def load_from_directory(
        self,
        directory: str,
        glob_pattern: str = "**/*",
        loader_mapping: Optional[dict] = None
    ) -> List:
        """
        โหลดเอกสารจากโฟลเดอร์
        
        Args:
            directory: โฟลเดอร์ที่ต้องการโหลด
            glob_pattern: รูปแบบการค้นหาไฟล์ (default: ทุกไฟล์)
            loader_mapping: mapping ระหว่างนามสกุลกับ Loader
        """
        # Default loader mapping
        if loader_mapping is None:
            loader_mapping = {
                ".txt": TextLoader,
                ".md": TextLoader,
                ".html": BSHTMLLoader,
                ".py": lambda path: GenericLoader.from_filesystem(
                    [path],
                    parser=LanguageParser(language="python")
                ),
                ".js": lambda path: GenericLoader.from_filesystem(
                    [path],
                    parser=LanguageParser(language="javascript")
                ),
                ".json": TextLoader,
                ".csv": TextLoader,
            }
        
        loader = DirectoryLoader(
            directory,
            glob=glob_pattern,
            loader_mapping=loader_mapping,
            show_progress=True,
            use_multithreading=True,
            max_concurrency=8  # ควบคุมการทำงานพร้อมกัน
        )
        
        documents = loader.load()
        
        # เพิ่ม metadata
        for doc in documents:
            doc.metadata.update({
                "source_type": "local_directory",
                "directory": directory,
                "file_name": doc.metadata.get("source", "unknown")
            })
            
        return documents
    
    async def load_from_github(self, repo_url: str, branch: str = "main") -> List:
        """
        โหลดเอกสารจาก GitHub repository
        
        Args:
            repo_url: URL ของ repository (e.g., "owner/repo")
            branch: branch ที่ต้องการ
        """
        from langchain_community.document_loaders import GitLoader
        
        loader = GitLoader(
            clone_url=f"https://github.com/{repo_url}.git",
            repo_path="/tmp/repo_clone",
            branch=branch,
            file_filter=lambda file_path: file_path.endswith(
                (".py", ".md", ".txt", ".json", ".yaml")
            )
        )
        
        documents = loader.load()
        
        for doc in documents:
            doc.metadata.update({
                "source_type": "github",
                "repo": repo_url,
                "branch": branch
            })
            
        return documents


การใช้งาน

loader = MultiSourceDocumentLoader()


โหลดจากเว็บ

web_docs = loader.load_from_web([
    "https://docs.python.org/3/tutorial/",
    "https://langchain.readthedocs.io/"
])


โหลดจากโฟลเดอร์

local_docs = loader.load_from_directory(
    "/project/docs",
    glob_pattern="**/*.{md,txt,pdf}",
    loader_mapping={
        ".md": TextLoader,
        ".txt": TextLoader,
        ".pdf": PyPDFLoader
    }
)


โหลดจาก GitHub

github_docs = await loader.load_from_github(
    "langchain-ai/langchain",
    branch="master"
)

print(f"รวมเอกสาร: {len(web_docs) + len(local_docs) + len(github_docs)} ชิ้น")

การเลือกและใช้งาน Vector Database

การเลือก Vector Database ที่เหมาะสมขึ้นอยู่กับหลายปัจจัย：ขนาดข้อมูล ความเร็วที่ต้องการ งบประมาณ และความซับซ้อนในการ deploy ในส่วนนี้จะเปรียบเทียบ Vector Stores ยอดนิยมพร้อมโค้ดตัวอย่าง

3.1 Chroma — In-process Vector Store ที่เบาที่สุด

Chroma เหมาะสำหรับโปรเจกต์ที่ต้องการเริ่มต้นเร็วและไม่ต้องการ infrastructure ซับซ้อน มีข้อดีคือติดตั้งง่ายและรองรับ metadata filtering อย่างครบถ้วน 


Chroma Vector Store — สำหรับ Development และ Small Production


=============================================================


import chromadb
from chromadb.config import Settings
from langchain_community.vectorstores import Chroma
from langchain_huggingface import HuggingFaceEmbeddings
from langchain_core.documents import Document
from typing import List, Dict, Optional
import os

class ChromaVectorStore:
    """
    Chroma Vector Store พร้อมการจัดการระดับ Production
    
    Features:
    - Persistent storage
    - Metadata filtering
    - Collection management
    - Similarity search หลายรูปแบบ
    """
    
    def __init__(
        self,
        persist_directory: str = "./chroma_db",
        collection_name: str = "documents",
        embedding_model: str = "sentence-transformers/all-MiniLM-L6-v2"
    ):
        self.persist_directory = persist_directory
        self.collection_name = collection_name
        
        # Initialize Embeddings
        self.embeddings = HuggingFaceEmbeddings(
            model_name=embedding_model,
            model_kwargs={'device': 'cuda'},
            encode_kwargs={'normalize_embeddings': True}
        )
        
        # Initialize Chroma Client
        self.client = chromadb.PersistentClient(
            path=persist_directory,
            settings=Settings(
                anonymized_telemetry=False,  # ปิด telemetry สำหรับ production
                allow_reset=True
            )
        )
        
        self.vectorstore = None
        
    def create_collection(
        self,
        metadata: Optional[Dict] = None,
        get_or_create: bool = True
    ):
        """
        สร้าง collection ใหม่
        
        Args:
            metadata: metadata ของ collection
            get_or_create: สร้างใหม่หรือใช้ collection เดิมถ้ามีอยู่
        """
        # ลบ collection เดิมถ้ามี (สำหรับกรณีที่ต้องการ reset)
        try:
            existing = self.client.get_collection(self.collection_name)
            if not get_or_create:
                self.client.delete_collection(self.collection_name)
                print(f"ลบ collection เดิม: {self.collection_name}")
        except:
            pass
            
        # สร้าง collection ใหม่
        self.vectorstore = Chroma(
            client=self.client,
            collection_name=self.collection_name,
            embedding_function=self.embeddings,
            persist_directory=self.persist_directory
        )
        
        return self.vectorstore
    
    def add_documents(
        self,
        documents: List[Document],
        batch_size: int = 100,
        show_progress: bool = True
    ):
        """
        เพิ่มเอกสารเข้า Vector Store
        
        Args:
            documents: รายการ Document objects
            batch_size: จำนวน documents ต่อ batch
            show_progress: แสดง progress bar หรือไม่
        """
        if not self.vectorstore:
            self.create_collection()
            
        # เพิ่มทีละ batch เพื่อหลีกเลี่ยง memory issues
        total = len(documents)
        for i in range(0, total, batch_size):
            batch = documents[i:i + batch_size]
            self.vectorstore.add_documents(batch)
            
            if show_progress:
                progress = min(i + batch_size, total)
                print(f"Progress: {progress}/{total} ({100*progress/total:.1f}%)")
                
        print(f"✓ เพิ่มเอกสารสำเร็จ: {total} ชิ้น")
        
    def similarity_search(
        self,
        query: str,
        k: int = 4,
        filter: Optional[Dict] = None,
        where_document: Optional[Dict] = None
    ) -> List[Document]:
        """
        ค้นหาเอกสารที่คล้ายคลึง
        
        Args:
            query: คำถามที่ต้องการค้นหา
            k: จำนวนผลลัพธ์ที่ต้องการ
            filter: metadata filter (e.g., {"source": "pdf"})
            where_document: document content filter
        """
        return self.vectorstore.similarity_search(
            query=query,
            k=k,
            filter=filter,
            where_document=where_document
        )
    
    def similarity_search_with_score(
        self,
        query: str,
        k: int = 4,
        filter: Optional[Dict] = None,
        score_threshold: float = 0.0
    ) -> List[tuple]:
        """
        ค้นหาพร้อมคะแนนความคล้ายคลึง
        
        Args:
            query: คำถามที่ต้องการค้นหา
            k: จำนวนผลลัพธ์ที่ต้องการ
            filter: metadata filter
            score_threshold: คะแนนขั้นต่ำ (0-1)
        """
        results = self.vectorstore.similarity_search_with_score(
            query=query,
            k=k,
            filter=filter
        )
        
        # กรองตาม score threshold
        if score_threshold > 0:
            results = [
                (doc, score) for doc, score in results 
                if score <= score_threshold
            ]
            
        return results
    
    def max_marginal_relevance_search(
        self,
        query: str,
        k: int = 4,
        fetch_k: int = 20,
        filter: Optional[Dict] = None
    ) -> List[Document]:
        """
        MMR Search — สำหรับ diverse results
        
        หลีกเลี่ยงการได้ผลลัพธ์ที่คล้ายกันมากเกินไป
        """
        return self.vectorstore.max_marginal_relevance_search(
            query=query,
            k=k,
            fetch_k=fetch_k,
            filter=filter
        )
    
    def get_collection_stats(self) -> Dict:
        """ดึงสถิติของ collection"""
        collection = self.client.get_collection(self.collection_name)
        return {
            "name": collection.name,
            "count": collection.count(),
            "metadata": collection.metadata
        }
    
    def delete_by_ids(self, ids: List[str]):
        """ลบเอกสารตาม IDs"""
        self.vectorstore.delete(ids=ids)
        print(f"✓ ลบ {len(ids)} เอกสาร")
    
    def reset_collection(self):
        """ลบ collection ทั้งหมด"""
        self.client.delete_collection(self.collection_name)
        self.vectorstore = None
        print(f"✓ ลบ collection: {self.collection_name}")


การใช้งาน

vector_store = ChromaVectorStore(
    persist_directory="./data/chroma",
    collection_name="knowledge_base",
    embedding_model="sentence-transformers/all-MiniLM-L6-v2"
)


สร้าง collection ใหม่

vector_store.create_collection()


เพิ่มเอกสาร

vector_store.add_documents(chunks, batch_size=500)


ค้นหา

results = vector_store.similarity_search(
    query="วิธีการติดตั้ง Python",
    k=5,
    filter={"source_file": {"$eq": "python-guide.pdf"}}
)


ค้นหาพร้อม score

results_with_score = vector_store.similarity_search_with_score(
    query="การใช้งาน LangChain",
    k=3,
    score_threshold=0.7
)

for doc, score in results_with_score:
    print(f"[{score:.4f}] {doc.page_content[:100]}...")


ดูสถิติ

stats = vector_store.get_collection_stats()
print(f"Collection stats: {stats}")

3.2 FAISS — High Performance In-memory Vector Search

FAISS (Facebook AI Similarity Search) เหมาะสำหรับงานที่ต้องการความเร็วสูงและรองรับข้อมูลขนาดใหญ่ สามารถใช้ GPU acceleration ได้ 


FAISS Vector Store — High Performance Search


============================================


from langchain_community.vectorstores import FAISS
from langchain_huggingface import HuggingFaceEmbeddings
from langchain_core.documents import Document
from typing import List, Dict, Tuple, Optional
import numpy as np
import faiss
import pickle
import os

class FAISSVectorStore:
    """
    FAISS Vector Store พร้อม GPU Support และ Index Options
    
    Index Types:
    - IndexFlatIP: Exact search (แม่นยำสูงสุด)
    - IndexIVFFlat: Approximate search (เร็วกว่าสำหรับข้อมูลใหญ่)
    - IndexHNSW: Hierarchical Navigable Small World (เร็ว + แม่นยำ)
    """
    
    def __init__(
        self,
        embedding_model: str = "sentence-transformers/all-MiniLM-L6-v2",
        use_gpu: bool = True,
        index_type: str = "hnsw"  # flat, ivf, hnsw
    ):
        self.embedding_model = embedding_model
        self.use_gpu = use_gpu and self._check_gpu_available()
        self.index_type = index_type
        
        # Initialize Embeddings
        self.embeddings = HuggingFaceEmbeddings(
            model_name=embedding_model,
            model_kwargs={'device': 'cuda' if self.use_gpu else 'cpu'},
            encode_kwargs={'normalize_embeddings': True}
        )
        
        self.vectorstore = None
        self.docstore = None
        self.index = None
        
    def _check_gpu_available(self) -> bool:
        """ตรวจสอบ GPU availability"""
        try:
            import torch
            return torch.cuda.is_available()
        except ImportError:
            return False
    
    def create_index(
        self,
        dimension: int,
        nlist: int = 100,  # สำหรับ IVF index
        M: int = 32       # สำหรับ HNSW index
    ) -> faiss.Index:
        """
        สร้าง FAISS Index ตามประเภท
        
        Args:
            dimension: ขนาดของ embedding vector
            nlist: จำนวน clusters สำหรับ IVF
            M: จำนวน connections สำหรับ HNSW
        """
        if self.index_type == "flat":
            #

แหล่งข้อมูลที่เกี่ยวข้อง
บทความที่เกี่ยวข้อง