前言

過去幾天分別討論了文件解析、Chunking、Metadata 設計，今天要把這些環節串接成一個完整、可重複執行的前處理 Pipeline，為明天開始的 Embedding 與向量資料庫建置做準備。

Pipeline 整體流程

原始文件
   │
   ▼
[1] 文件載入（依格式選擇對應 Loader）
   │
   ▼
[2] 清洗（移除雜訊、統一編碼）
   │
   ▼
[3] Chunking（依文件結構切分）
   │
   ▼
[4] Metadata 附加（來源、時間、章節資訊）
   │
   ▼
輸出：待 Embedding 的 chunk 清單

實作：整合成一個 Pipeline 函式

def preprocess_pipeline(file_path):
    # Step 1: 依副檔名選擇 Loader
    ext = file_path.suffix
    if ext == ".pdf":
        raw_text = load_pdf(file_path)
    elif ext == ".html":
        raw_text = load_html(file_path.read_text())
    elif ext == ".md":
        raw_text = file_path.read_text(encoding="utf-8")
    else:
        raise ValueError(f"不支援的格式:{ext}")

    # Step 2: 清洗
    cleaned_text = clean_text(raw_text)

    # Step 3: Chunking
    chunks = structured_split(cleaned_text)

    # Step 4: 附加 Metadata
    result = []
    for i, chunk in enumerate(chunks):
        result.append({
            "content": chunk,
            "metadata": {
                "source": file_path.name,
                "chunk_index": i,
                "created_date": get_file_date(file_path),
            }
        })
    return result

自動化考量

如果知識庫會持續更新（例如每週新增技術文件），建議把 Pipeline 設計成可以：

1. 增量處理：只處理新增/修改過的文件，避免每次都重跑全部資料
2. 失敗重試機制：某份文件解析失敗時，記錄下來但不中斷整個流程
3. 版本控管：文件更新時，能找到舊版本的 chunk 並替換，而不是產生重複資料
4. 排程執行：可以搭配排程工具（cron、Airflow）定期執行

def incremental_update(doc_dir, last_run_time):
    for file_path in doc_dir.glob("**/*"):
        if file_path.stat().st_mtime > last_run_time:
            chunks = preprocess_pipeline(file_path)
            upsert_to_vector_db(chunks)  # 有則更新，無則新增

小結

把前處理步驟整合成自動化 Pipeline，不只是為了方便，更是為了讓知識庫能長期維護、持續反映最新資訊——這也是 RAG 相較於微調的一大優勢：知識庫更新不需要重新訓練模型，只需要重跑這個 Pipeline。到這裡，Indexing 階段的前半段（資料前處理）就告一段落了。明天我們要進入 Embedding，把文字正式轉換成向量。