LangChain 教學：自訂 Document Loader

簡介

在使用 LangChain 建立聊天機器人或問答系統時，最常見的第一步就是把外部資料（PDF、Word、網頁、資料庫等）載入成 Document 物件，讓後續的向量化、檢索與生成流程得以運作。LangChain 已內建多種常用的 Loader（如 PyPDFLoader、CSVLoader、WebBaseLoader），但在實務專案中，我們常會遇到 自有格式、API 回傳的 JSON、或是 即時串流資料，這些情況都需要自行撰寫 Loader。

本篇文章將帶你從 概念、實作、測試、到 最佳實踐，一步步完成一個 自訂 Loader，讓你能夠靈活地把任何資料來源轉換為 LangChain 可辨識的 Document。即使你是剛接觸 LangChain 的新手，只要跟著範例走，也能在短時間內寫出自己的 Loader，並套用在實際的聊天機器人或企業知識庫中。

---

核心概念

1. Loader 的定位

在 LangChain 中，Loader 的唯一職責是 讀取原始資料，並回傳 List[Document]。每個 Document 只包含兩個屬性：

屬性	說明
page_content	真正的文字內容
metadata	任意的鍵值對，用於存放來源、檔案名稱、時間戳記等輔助資訊

只要遵守這個介面，就能與 LangChain 其他元件（如 TextSplitter、Embedding、VectorStore）無縫結合。

2. 自訂 Loader 必須實作的抽象類別

LangChain 在 langchain.document_loaders.base 中提供了抽象基底 BaseLoader，核心方法只有一個：

class BaseLoader(ABC):
    @abstractmethod
    def load(self) -> List[Document]:
        ...

因此，我們只需要繼承 BaseLoader，實作 load 方法，並在其中把資料轉成 Document 列表即可。

3. 常見的自訂情境

情境	為何需要自訂 Loader
API 回傳的 JSON	JSON 結構多變，內建 Loader 無法直接解析
自有檔案格式（.xyz）	企業內部系統產出的專屬檔案，需要自訂解析邏輯
即時串流文字	例如 Slack、Discord 訊息，需要邊接收邊產生 Document
多語言混雜	需要在載入時自動偵測語言並寫入 metadata

以下將以 三種實務範例 逐步說明如何實作自訂 Loader。

---

程式碼範例

範例 1️⃣：從 REST API 取得 JSON 並轉成 Document

假設我們有一個商品說明 API，回傳格式如下：

{
  "items": [
    {"id": "001", "title": "智慧手環", "description": "可測量心率與睡眠品質"},
    {"id": "002", "title": "藍牙耳機", "description": "降噪功能，支援多點連線"}
  ]
}

我們希望把每個商品的 description 當成 page_content，而 id、title 放入 metadata。

# file: custom_loaders/api_loader.py
import requests
from typing import List
from langchain.docstore.document import Document
from langchain.document_loaders.base import BaseLoader

class ProductAPILoader(BaseLoader):
    """從商品說明 API 取得資料的自訂 Loader"""

    def __init__(self, endpoint: str, api_key: str | None = None):
        self.endpoint = endpoint
        self.api_key = api_key

    def load(self) -> List[Document]:
        headers = {}
        if self.api_key:
            headers["Authorization"] = f"Bearer {self.api_key}"
        response = requests.get(self.endpoint, headers=headers)
        response.raise_for_status()
        data = response.json()

        docs: List[Document] = []
        for item in data.get("items", []):
            doc = Document(
                page_content=item["description"],
                metadata={
                    "product_id": item["id"],
                    "title": item["title"],
                    "source": self.endpoint,
                },
            )
            docs.append(doc)
        return docs

重點：在 metadata 中保留原始來源 URL (source) 有助於後續追蹤與除錯。

---

範例 2️⃣：自訂檔案格式（.xyz）

假設公司內部產生的 .xyz 檔案，每行都是「標題|內容」的組合，且檔名即為文件的唯一識別碼。

產品簡介|本公司推出的最新產品具備 AI 智慧分析功能。
使用說明|使用者只需要按下啟動鍵，即可開始自動偵測。

以下是一個簡易的 Loader，讀取檔案、拆解每行、產生 Document。

# file: custom_loaders/xyz_loader.py
import os
from typing import List
from langchain.docstore.document import Document
from langchain.document_loaders.base import BaseLoader

class XYZLoader(BaseLoader):
    """載入自訂 .xyz 檔案，每行「標題|內容」"""

    def __init__(self, file_path: str, encoding: str = "utf-8"):
        if not os.path.isfile(file_path):
            raise FileNotFoundError(f"找不到檔案: {file_path}")
        self.file_path = file_path
        self.encoding = encoding

    def load(self) -> List[Document]:
        docs: List[Document] = []
        with open(self.file_path, "r", encoding=self.encoding) as f:
            for idx, line in enumerate(f):
                line = line.strip()
                if not line:
                    continue
                try:
                    title, content = line.split("|", 1)
                except ValueError:
                    # 若沒有「|」則整行視為內容
                    title, content = f"line_{idx+1}", line
                doc = Document(
                    page_content=content,
                    metadata={
                        "title": title,
                        "source": self.file_path,
                        "line_no": idx + 1,
                    },
                )
                docs.append(doc)
        return docs

技巧：在 metadata 中加入 line_no，日後若要回溯到原始檔案的哪一行，會非常方便。

---

範例 3️⃣：即時串流文字（Slack Bot）

如果想把 Slack 頻道的訊息即時寫入向量資料庫，我們可以把每條訊息當成一個 Document，並在 metadata 中加入使用者、時間與頻道資訊。

# file: custom_loaders/slack_loader.py
import os
from typing import List
from slack_sdk import WebClient
from slack_sdk.errors import SlackApiError
from datetime import datetime
from langchain.docstore.document import Document
from langchain.document_loaders.base import BaseLoader

class SlackMessageLoader(BaseLoader):
    """從指定 Slack 頻道抓取最近 N 條訊息"""

    def __init__(self, token: str, channel_id: str, limit: int = 100):
        self.client = WebClient(token=token)
        self.channel_id = channel_id
        self.limit = limit

    def load(self) -> List[Document]:
        try:
            response = self.client.conversations_history(
                channel=self.channel_id, limit=self.limit
            )
        except SlackApiError as e:
            raise RuntimeError(f"Slack API error: {e.response['error']}")

        msgs = response["messages"]
        docs: List[Document] = []
        for msg in msgs:
            # 只處理純文字訊息
            if "text" not in msg:
                continue
            ts = float(msg["ts"])
            dt = datetime.fromtimestamp(ts).isoformat()
            doc = Document(
                page_content=msg["text"],
                metadata={
                    "user": msg.get("user", "unknown"),
                    "channel": self.channel_id,
                    "timestamp": dt,
                    "source": f"slack://{self.channel_id}",
                },
            )
            docs.append(doc)
        return docs

提醒：使用 Slack Bot Token 時，請確保已在 OAuth Scopes 中加入 channels:history、groups:history 等權限，否則會遭到 invalid_auth 錯誤。

---

範例 4️⃣：結合多種 Loader（Composite Loader）

實務上常會同時載入多種來源，LangChain 允許我們把多個 Loader 包裝成一個 CompositeLoader，只要把各自的 load() 結果合併即可。

# file: custom_loaders/composite_loader.py
from typing import List
from langchain.docstore.document import Document
from langchain.document_loaders.base import BaseLoader

class CompositeLoader(BaseLoader):
    """把多個 Loader 合併成一個，保持原始順序"""

    def __init__(self, loaders: List[BaseLoader]):
        self.loaders = loaders

    def load(self) -> List[Document]:
        all_docs: List[Document] = []
        for loader in self.loaders:
            docs = loader.load()
            all_docs.extend(docs)
        return all_docs

實務應用：可以把 ProductAPILoader、XYZLoader、SlackMessageLoader 放入同一個 CompositeLoader，一次性產出所有文件，之後直接送入向量化流程。

---

常見陷阱與最佳實踐

陷阱	可能的後果	解決方案 / 最佳實踐
忘記填寫 metadata["source"]	後續除錯或追蹤資料來源變得困難	必寫 source，統一使用 protocol://identifier 形式（如 file://, api://）
一次載入過多文件	記憶體爆炸、向量化速度下降	使用 分批載入（limit、offset）或 流式 Generator（自行實作 __iter__）
文字編碼不一致	產生亂碼或無法正確切分	在讀檔時明確指定 encoding，或使用 chardet 自動偵測
未處理例外（API 錯誤、檔案不存在）	程式直接崩潰，導致批次作業失敗	捕捉例外、加入 重試機制（tenacity）
metadata 欄位太多	向量搜尋結果過於雜訊，或儲存成本提升	只保留必要欄位，如 source、title、timestamp，其餘可放在外部資料庫
同步阻塞 I/O（大量網路請求）	效能低下	使用 asyncio + aiohttp 撰寫非同步 Loader，或使用 ThreadPoolExecutor 併行化

建議的程式碼結構

project_root/
│
├─ loaders/
│   ├─ __init__.py
│   ├─ api_loader.py
│   ├─ xyz_loader.py
│   ├─ slack_loader.py
│   └─ composite_loader.py
│
├─ pipelines/
│   └─ ingest.py          # 組合 Loader → Splitter → Embedding → VectorStore
│
└─ main.py                # 呼叫 pipeline，啟動服務或批次作業

如此分層能夠讓 測試、重用 與 維護 更加便利。

---

實際應用場景

1. 企業內部知識庫

   • 透過 CompositeLoader 同時抓取 Confluence API、本地 PDF、GitHub README，建立完整的技術文件向量庫，供員工透過聊天機器人快速搜尋。

2. 客服聊天機器人

   • 使用 SlackMessageLoader 把過去的客服對話即時寫入向量資料庫，讓模型能夠參考歷史對話，提供更具上下文的回覆。

3. 電商商品搜尋

   • ProductAPILoader 把商品說明與評價資料載入，結合 向量相似度搜尋，讓使用者以自然語言描述需求，即可找到最符合的商品。

4. 多語言文件管理

   • 在 XYZLoader 中加入語言偵測（如 langdetect），把偵測結果寫入 metadata["lang"]，之後可在向量化前針對不同語言使用不同的 Embedding 模型。

---

總結

• 自訂 Loader 只需要遵守 BaseLoader.load() -> List[Document] 的簡單介面，卻能讓任何資料來源無縫接入 LangChain。
• 透過 範例程式碼（API、檔案、即時串流、Composite），你可以快速構建符合實務需求的 Loader。
• 避免常見陷阱：記得填寫 source、分批載入、處理例外、保持 metadata 精簡。
• 把 Loader 與 Splitter → Embedding → VectorStore 組成完整的 資料管線，即可在聊天機器人、搜尋系統、知識庫等場景中發揮威力。

掌握了自訂 Loader，你就掌握了 資料入口 的主動權，未來再面對任何新型態的資料來源，都能輕鬆擴充、快速部署。祝你在 LangChain 的旅程中，開發出更加靈活、效能卓越的 AI 應用！