Bài 21: Document Loaders — load PDF, web, Notion

Sau bài này bạn sẽ:

• Hiểu Document Loader là gì và tại sao cần nó trong pipeline RAG.
• Dùng được các loader phổ biến: TextLoader, PyPDFLoader, CSVLoader, JSONLoader, WebBaseLoader.
• Bulk load cả thư mục với DirectoryLoader.
• Biết sự khác biệt giữa .load() và .lazy_load() và khi nào dùng cái nào.
• Nhận ra các pitfall phổ biến: scanned PDF, HTML noise, CSV granularity, silent_errors.

Trong pipeline RAG (Retrieval-Augmented Generation), bước đầu tiên là đưa data vào hệ thống. Data có thể đến từ rất nhiều nguồn khác nhau: file PDF nội bộ, trang web, database Notion, Google Drive, code repo trên GitHub, Slack export... Mỗi nguồn có format riêng, API riêng, cách parse riêng.

Document Loader là lớp adapter trong LangChain — nó đọc data từ một nguồn cụ thể và chuyển về một định dạng thống nhất: object Document. Nhờ đó, phần còn lại của pipeline (text splitting, embedding, indexing) không cần biết data đến từ đâu.

LangChain Community 0.3.x có hơn 200 loader. Tất cả đều implement cùng interface:

class BaseLoader:
    def load(self) -> list[Document]: ...
    def lazy_load(self) -> Iterator[Document]: ...
    async def aload(self) -> list[Document]: ...

Bạn swap từ PDF sang web hay sang Notion mà không cần sửa code phía sau loader.

Document là dataclass từ langchain_core.documents, chỉ có hai field:

from langchain_core.documents import Document

doc = Document(
    page_content="Nội dung văn bản ở đây.",
    metadata={
        "source": "./report.pdf",
        "page": 3,
        "author": "Nguyen Van A",
    }
)

print(doc.page_content)   # "Nội dung văn bản ở đây."
print(doc.metadata)       # {"source": ..., "page": 3, ...}

• page_content: text thuần, là thứ sẽ được split và embed về sau.
• metadata: dict tùy ý, giữ thông tin nguồn gốc. Metadata này theo document xuyên suốt pipeline và có thể dùng cho metadata filtering khi search (đã đề cập ở bài 18).

Mỗi loader tự điền metadata phù hợp với nguồn của nó. PyPDFLoader điền source và page. WebBaseLoader điền source (URL) và title.

Core package chứa tất cả loader là langchain-community:

pip install langchain-community

Nhưng nhiều loader cần thêm dependency phụ tùy nguồn data. LangChain không bundle sẵn để tránh bloat. Cần cài theo loader bạn dùng:

Kiểm tra version sau khi cài:

python -c "import langchain_community; print(langchain_community.__version__)"
# 0.3.x

TextLoader

Loader đơn giản nhất — đọc file .txt thuần, trả về 1 Document duy nhất với toàn bộ nội dung file.

from langchain_community.document_loaders import TextLoader

loader = TextLoader("./notes.txt", encoding="utf-8")
docs = loader.load()
# docs là list[Document] với 1 phần tử
# docs[0].metadata = {"source": "./notes.txt"}

Lưu ý tham số encoding: mặc định là utf-8, nhưng file Windows cũ hay có cp1252 hoặc latin-1. Nếu gặp UnicodeDecodeError, thử autodetect_encoding=True.

PyPDFLoader

Load file PDF với pypdf. Mỗi page PDF trở thành 1 Document riêng biệt.

from langchain_community.document_loaders import PyPDFLoader

loader = PyPDFLoader("./report.pdf")
docs = loader.load()

print(len(docs))           # = số page trong PDF
print(docs[0].page_content[:200])  # nội dung page 1
print(docs[0].metadata)
# {"source": "./report.pdf", "page": 0}  ← page đánh từ 0

Nếu cần extract bảng chính xác hơn, dùng PDFPlumberLoader (chậm hơn nhưng parse bảng tốt hơn):

from langchain_community.document_loaders import PDFPlumberLoader

loader = PDFPlumberLoader("./report.pdf")
docs = loader.load()
# docs[0].metadata sẽ có thêm: "page_number", "file_path", "total_pages"

CSVLoader

Mỗi row trong CSV trở thành 1 Document. Nội dung document là tất cả column được format thành chuỗi column_name: value\n.

from langchain_community.document_loaders import CSVLoader

loader = CSVLoader(
    file_path="./products.csv",
    csv_args={"delimiter": ","},
    source_column="product_id",  # dùng column này làm source trong metadata
)
docs = loader.load()

# docs[0].page_content ≈
# "product_id: P001\nname: Widget A\nprice: 9.99\ndescription: ..."
# docs[0].metadata = {"source": "P001", "row": 0}

Tham số source_column hữu ích khi mỗi row có ID riêng — metadata source sẽ là giá trị của column đó thay vì đường dẫn file.

JSONLoader

Dùng cú pháp jq để extract field từ JSON hoặc JSONL. Cần cài pip install jq.

from langchain_community.document_loaders import JSONLoader

# JSON có dạng: [{"id": 1, "text": "...", "author": "..."}, ...]
loader = JSONLoader(
    file_path="./articles.json",
    jq_schema=".[].text",      # extract field "text" từ mỗi element
    text_content=True,         # kết quả là string thẳng
)
docs = loader.load()

# Nếu muốn giữ thêm metadata từ JSON:
loader = JSONLoader(
    file_path="./articles.json",
    jq_schema=".[].text",
    text_content=True,
    metadata_func=lambda record, meta: {**meta, "author": record.get("author")},
)

Với JSONL (mỗi dòng là 1 JSON object), thêm json_lines=True.

WebBaseLoader

Fetch URL qua HTTP, parse HTML bằng BeautifulSoup, extract text. Nhận một URL hoặc list URL.

from langchain_community.document_loaders import WebBaseLoader

# Load 1 URL
loader = WebBaseLoader("https://docs.python.org/3/tutorial/")
docs = loader.load()
# docs[0].metadata = {"source": "https://...", "title": "...", "language": "en", ...}

# Load nhiều URL cùng lúc
urls = [
    "https://docs.python.org/3/tutorial/introduction.html",
    "https://docs.python.org/3/tutorial/controlflow.html",
]
loader = WebBaseLoader(urls)
docs = loader.load()   # 2 Document

Mặc định WebBaseLoader lấy toàn bộ text từ HTML kể cả nav bar, footer, cookie banner. Để filter chỉ lấy phần nội dung chính, dùng bs_kwargs:

import bs4

loader = WebBaseLoader(
    web_paths=["https://example.com/blog/post-1"],
    bs_kwargs={
        "parse_only": bs4.SoupStrainer(
            class_=("post-content", "article-body")  # chỉ lấy các class này
        )
    },
)
docs = loader.load()

Cách này loại bỏ noise đáng kể so với parse toàn trang.

RecursiveUrlLoader

Crawl recursive từ một URL gốc, theo link depth-limited. Phù hợp khi cần index cả một docs site nhỏ.

from langchain_community.document_loaders.recursive_url_loader import RecursiveUrlLoader
from bs4 import BeautifulSoup

def bs4_extractor(html: str) -> str:
    soup = BeautifulSoup(html, "lxml")
    return soup.get_text(separator="\n", strip=True)

loader = RecursiveUrlLoader(
    url="https://docs.langchain.com/docs/",
    max_depth=2,           # chỉ crawl 2 cấp link
    extractor=bs4_extractor,
    prevent_outside=True,  # không ra ngoài domain gốc
)
docs = loader.load()
print(len(docs))  # số trang đã crawl

SitemapLoader

Load từ sitemap.xml — lấy danh sách URL rồi fetch từng URL. Hữu ích khi site có sitemap sẵn.

from langchain_community.document_loaders.sitemap import SitemapLoader

loader = SitemapLoader(
    web_path="https://example.com/sitemap.xml",
    filter_urls=["https://example.com/docs/"],  # chỉ load URL trong /docs/
)
docs = loader.load()

Các loader cloud cần credentials và có setup phức tạp hơn. Phần này mô tả interface và yêu cầu cơ bản.

NotionDBLoader

Load từ Notion database. Cần tạo Notion integration tại notion.so/my-integrations và lấy Integration Token, sau đó share database với integration đó.

from langchain_community.document_loaders import NotionDBLoader

loader = NotionDBLoader(
    integration_token="secret_xxxxxxxxxxxxxxxxxxxx",
    database_id="xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx",
    request_timeout_sec=30,
)
docs = loader.load()
# Mỗi page trong database = 1 Document
# metadata chứa: "id", "title", "created_time", "last_edited_time", properties Notion

Nếu cần load từ Notion page (không phải database), dùng NotionDirectoryLoader với Notion export ZIP.

GoogleDriveLoader

Cần service account credentials hoặc OAuth2. Load Google Docs, Google Sheets (convert sang text).

from langchain_community.document_loaders import GoogleDriveLoader

loader = GoogleDriveLoader(
    folder_id="1BdFxxxxxxxxxxxxxxxxxxxxxxxxx",
    credentials_path="./credentials.json",  # service account JSON
    token_path="./token.json",
    file_types=["document", "sheet"],
    recursive=False,
)
docs = loader.load()

SlackDirectoryLoader

Parse từ Slack export ZIP (export từ Slack admin). Không cần API token — chỉ cần file export.

from langchain_community.document_loaders import SlackDirectoryLoader

loader = SlackDirectoryLoader(
    zip_path="./slack-export-2024.zip",
    workspace_url="https://yourworkspace.slack.com",  # optional, dùng cho metadata
)
docs = loader.load()
# Mỗi message = 1 Document
# metadata: "channel", "timestamp", "source" (link đến message)

Các loader cloud khác

• ConfluenceLoader: load từ Confluence space, cần URL + username + API token.
• JiraLoader: load issue từ Jira project.
• AzureBlobStorageFileLoader: đọc file từ Azure Blob Storage.
• S3FileLoader / S3DirectoryLoader: đọc từ AWS S3.

Danh sách đầy đủ: python.langchain.com/docs/integrations/document_loaders

DirectoryLoader wrapper quanh một loader khác, tự scan folder theo glob pattern và gọi loader đó cho từng file match.

from langchain_community.document_loaders import DirectoryLoader, PyPDFLoader

loader = DirectoryLoader(
    path="./internal-docs",
    glob="**/*.pdf",          # recursive, tất cả .pdf trong subdirectory
    loader_cls=PyPDFLoader,
    show_progress=True,       # in progress bar khi load nhiều file
    use_multithreading=True,  # load song song (thread pool)
    max_concurrency=4,
)
docs = loader.load()
print(f"Loaded {len(docs)} documents")

Load nhiều loại file trong cùng thư mục — dùng nhiều DirectoryLoader rồi gộp:

from langchain_community.document_loaders import (
    DirectoryLoader, PyPDFLoader, TextLoader
)

pdf_loader = DirectoryLoader("./docs", glob="**/*.pdf", loader_cls=PyPDFLoader)
txt_loader = DirectoryLoader("./docs", glob="**/*.txt", loader_cls=TextLoader)

all_docs = pdf_loader.load() + txt_loader.load()

Tham số quan trọng:

• silent_errors=True: bỏ qua file bị lỗi thay vì dừng toàn bộ. Hữu ích khi ingest thư mục lớn có file bị corrupt.
• loader_kwargs: truyền thêm kwargs xuống loader con. Ví dụ loader_kwargs={"encoding": "utf-8"} cho TextLoader.
• use_multithreading=True với max_concurrency: tăng tốc khi số file nhiều. Lưu ý: không hữu ích với I/O bound loader network.

.load() đọc toàn bộ data vào memory ngay lập tức và trả về list[Document]. Với dataset nhỏ (vài chục file PDF, vài MB) thì không vấn đề gì.

Với dataset lớn (hàng nghìn document, tổng hàng GB text), .load() có thể gây OOM. Dùng .lazy_load() thay thế:

from langchain_community.document_loaders import DirectoryLoader, PyPDFLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter

loader = DirectoryLoader("./big-docs", glob="**/*.pdf", loader_cls=PyPDFLoader)
splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=50)

# Xử lý từng document mà không load toàn bộ vào RAM
for doc in loader.lazy_load():
    chunks = splitter.split_documents([doc])
    # gửi chunks vào vector DB ngay, không tích lũy
    # vector_store.add_documents(chunks)

.lazy_load() trả về Iterator[Document], yield từng Document một khi đọc xong. Memory tại mỗi thời điểm chỉ chứa 1 document + chunks của nó.

LangChain 0.1+ chuẩn hóa lazy_load trên tất cả loader. Nếu loader cũ không implement, nó fallback về gọi load() bên trong — vẫn safe để gọi, chỉ là không có lợi về memory.

Kiểm tra loader có native lazy_load không:

import inspect
loader = PyPDFLoader("./file.pdf")
# Xem source của lazy_load
print(inspect.getsource(loader.lazy_load))

GitLoader

Clone hoặc load từ local git repo, filter file theo pattern. Cần pip install gitpython.

from langchain_community.document_loaders import GitLoader

# Load từ repo local đã có
loader = GitLoader(
    repo_path="./my-repo",
    branch="main",
    file_filter=lambda file_path: file_path.endswith(".py"),  # chỉ Python file
)
docs = loader.load()
# docs[0].metadata = {"file_path": "src/utils.py", "file_name": "utils.py",
#                      "file_type": ".py", "source": "src/utils.py"}

# Clone từ remote URL
loader = GitLoader(
    clone_url="https://github.com/langchain-ai/langchain",
    repo_path="./langchain-clone",
    branch="master",
    file_filter=lambda p: p.endswith(".py") and "tests" not in p,
)

Use case phổ biến: xây code RAG cho team nội bộ — developer hỏi "function X làm gì", agent tìm trong codebase.

GitHubIssuesLoader

Load issue (kèm comment) từ GitHub repository qua API. Cần GitHub access token.

from langchain_community.document_loaders import GitHubIssuesLoader

loader = GitHubIssuesLoader(
    repo="langchain-ai/langchain",
    access_token="ghp_xxxxxxxxxxxx",  # hoặc đọc từ env
    include_prs=False,       # bỏ qua PR, chỉ lấy issue
    state="open",            # "open", "closed", hoặc "all"
)
docs = loader.load()
# docs[0].page_content = tiêu đề + body + comments của issue
# docs[0].metadata = {"url": "https://github.com/...", "title": "...", "labels": [...]}

Hữu ích cho support agent: index toàn bộ issue để trả lời câu hỏi "đã có ai báo bug này chưa".

WebBaseLoader hỗ trợ aload() — fetch nhiều URL song song bằng asyncio:

import asyncio
from langchain_community.document_loaders import WebBaseLoader

urls = [
    "https://docs.python.org/3/tutorial/introduction.html",
    "https://docs.python.org/3/tutorial/controlflow.html",
    "https://docs.python.org/3/tutorial/datastructures.html",
]

loader = WebBaseLoader(urls)
docs = asyncio.run(loader.aload())
# Fetch 3 URL song song thay vì tuần tự

Trong FastAPI endpoint (môi trường async sẵn), dùng await trực tiếp:

@app.post("/ingest-urls")
async def ingest_urls(urls: list[str]):
    loader = WebBaseLoader(urls)
    docs = await loader.aload()
    # xử lý tiếp...
    return {"count": len(docs)}

Với file loader (PDF, text), I/O là disk — parallel thread thường hiệu quả hơn asyncio. DirectoryLoader có use_multithreading=True cho mục đích này.

Để load nhiều PDF song song bằng asyncio thủ công:

import asyncio
from concurrent.futures import ThreadPoolExecutor
from langchain_community.document_loaders import PyPDFLoader

async def load_pdf(path: str) -> list:
    loop = asyncio.get_event_loop()
    with ThreadPoolExecutor() as pool:
        return await loop.run_in_executor(
            pool, lambda: PyPDFLoader(path).load()
        )

async def load_all(paths: list[str]):
    results = await asyncio.gather(*[load_pdf(p) for p in paths])
    return [doc for docs in results for doc in docs]  # flatten

Document Loader là bước 1 của pipeline ingest. Hai bước tiếp theo là Text Splitting (bài 22) và Embedding + Indexing (đã đề cập ở bài 14–18 cho vector DB).

from langchain_community.document_loaders import PyPDFLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma

# 1. Load
loader = PyPDFLoader("./technical-report.pdf")
docs = loader.load()
print(f"Loaded {len(docs)} pages")

# 2. Split — chi tiết ở bài 22
splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,
    chunk_overlap=50,
)
chunks = splitter.split_documents(docs)
print(f"Split into {len(chunks)} chunks")

# 3. Embed + Index — đã có ở bài 14
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vector_store = Chroma.from_documents(
    documents=chunks,
    embedding=embeddings,
    persist_directory="./chroma-db",
)
print("Indexed successfully")

Metadata từ loader (source, page) được giữ nguyên qua split và vào vector store. Khi retrieve, bạn biết chunk nào đến từ page nào của file nào — quan trọng cho citation trong RAG response.

PyPDFLoader không parse được scanned PDF

pypdf chỉ extract text layer. File PDF là ảnh scan (không có text layer) trả về page_content="".

Giải pháp: dùng UnstructuredPDFLoader kết hợp OCR (chậm hơn 10–50x), hoặc tiền xử lý riêng với pytesseract. Kiểm tra sớm:

loader = PyPDFLoader("./scan.pdf")
docs = loader.load()
empty_pages = [i for i, d in enumerate(docs) if not d.page_content.strip()]
if empty_pages:
    print(f"WARNING: {len(empty_pages)} pages empty — possibly scanned PDF")

WebBaseLoader kéo quá nhiều noise

Mặc định parse toàn bộ text trong HTML, bao gồm nav, footer, sidebar, cookie notice. Kết quả là chunk embed về sau chứa toàn "Accept cookies", "Home About Contact" — retrieval kém chất lượng.

Luôn dùng bs_kwargs={"parse_only": SoupStrainer(...)} để giới hạn vùng parse, hoặc dùng thư viện trafilatura để extract main content:

import trafilatura
from langchain_core.documents import Document

def fetch_with_trafilatura(url: str) -> Document:
    downloaded = trafilatura.fetch_url(url)
    text = trafilatura.extract(downloaded) or ""
    return Document(page_content=text, metadata={"source": url})

CSVLoader granularity không phù hợp

1 row = 1 Document nghĩa là với file CSV có 10.000 row, bạn có 10.000 Document rất ngắn. Khi split tiếp, mỗi chunk có thể chỉ 1–2 dòng text — không đủ ngữ cảnh cho embedding có ý nghĩa.

Giải pháp: gom nhiều row trước khi tạo Document, hoặc dùng chunk_size nhỏ hơn bình thường và không overlap.

DirectoryLoader dừng khi 1 file lỗi

Mặc định silent_errors=False. Một file PDF corrupt hoặc encoding sai sẽ throw exception và dừng toàn bộ ingest. Với folder lớn, luôn bật:

loader = DirectoryLoader(
    "./docs",
    glob="**/*.pdf",
    loader_cls=PyPDFLoader,
    silent_errors=True,   # bỏ qua file lỗi, tiếp tục file tiếp theo
    show_progress=True,
)

Log ra file nào bị bỏ qua để review sau:

import logging
logging.getLogger("langchain_community.document_loaders").setLevel(logging.WARNING)

Quên lazy_load với dataset lớn

Gọi .load() trên DirectoryLoader với 5.000 PDF document, mỗi PDF 50 trang — kết quả là 250.000 Document trong RAM cùng lúc. Trên server 16GB RAM, điều này sẽ OOM kill process.

Quy tắc đơn giản: nếu tổng số Document ước tính > 10.000, dùng lazy_load() và stream qua pipeline.

JSONLoader cần cài jq binary

pip install jq thực ra wrap C library libjq. Trên một số Docker image tối giản hoặc Windows, cài đặt có thể fail. Kiểm tra trước khi deploy, hoặc dùng json stdlib và tự viết loader nhẹ hơn nếu cần.

• Document Loader chuyển data từ bất kỳ nguồn nào về Document(page_content, metadata) — interface đồng nhất cho toàn pipeline.
• File loader thông dụng: TextLoader, PyPDFLoader, CSVLoader, JSONLoader. Mỗi loader cần dependency phụ riêng.
• Web: WebBaseLoader (đơn URL/danh sách), RecursiveUrlLoader (crawl), SitemapLoader. Luôn filter noise HTML.
• Cloud: NotionDBLoader, GoogleDriveLoader, SlackDirectoryLoader, ... Cần credentials tương ứng.
• DirectoryLoader bulk load cả thư mục theo glob pattern. Dùng silent_errors=True cho ingest quy mô lớn.
• .lazy_load() trả về iterator, tránh OOM khi dataset lớn. Là chuẩn từ LangChain 0.1+.
• Codebase loader: GitLoader cho source file, GitHubIssuesLoader cho issue.
• Metadata theo document xuyên suốt pipeline — dùng cho filtering và citation khi retrieve.

Bài 22: Text Splitters — chia document thành chunk