跳到主要內容

AG Day 21 文件擷取與分塊:從網頁到知識庫

AG Day 21 文件擷取與分塊:從網頁到知識庫

執行需求:CPU 可跑。在昨天 AG Day 20(原文連結)中,我們讓 research-agent 學會把文字轉成向量、存進 Chroma,並驗證了語意檢索確實比關鍵字比對更能理解使用者的問題。但昨天用來測試的句子都是我們自己手打的乾淨文字,真實世界的網頁完全不是這樣:一個網頁裡通常混雜著導覽列、廣告、留言區、版權宣告等大量與主題無關的雜訊,而且往往長達數千字,不能整篇直接拿去向量化。今天我們要把這整段「從一個網址,變成一批乾淨、大小適中、可以送進向量庫的文字分塊」的流程做到完整,正式把 AG Day 2 建立的 documents、chunks 資料表跟昨天的向量庫真正串連在一起,補齊知識庫的最後一塊拼圖。

引言

如果把 RAG 系統比喻成一座圖書館,昨天我們做的是「設計一套依照書籍內容查找的目錄系統」,今天要做的則是「把書真正搬進圖書館、拆成適合閱讀的章節、貼上分類標籤」。這兩件事同樣重要,但今天的工作往往更繁瑣:真實網頁的結構千變萬化,同一個擷取邏輯在一個網站上運作良好,換一個網站可能整段抓到的都是導覽選單文字;而分塊策略如果切得不好,可能把一個完整的論點硬生生切成兩半,導致檢索到片段時完全看不懂上下文。

今天的實作分成三個明確的階段:擷取(extraction)、分塊(chunking)、落地(landing)。擷取階段要從原始 HTML 裡找出「真正的正文」,去掉導覽列、廣告等雜訊;分塊階段要把擷取出來的長文字切成大小適中、語意盡量完整的小段落;落地階段則要把原始全文與各個分塊,分別寫進我們在 AG Day 2 設計好的 documents 與 chunks 資料表,並把分塊的向量寫進 Chroma。做完這三步,research-agent 就有了一條完整的「網頁進、知識庫出」的擷取管線。

原理與觀念

為什麼不能直接用 BeautifulSoup 取全部文字

很多人第一次做網頁擷取時,會直接用 BeautifulSoup 之類的函式庫把整個網頁的 <body> 文字取出來,這樣做在乾淨的靜態頁面上或許堪用,但大多數真實網站會摻雜大量「樣板文字」:頁首選單、側邊欄推薦文章、頁尾的公司資訊與社群連結、留言區的使用者互動內容等。這些內容雖然也是「文字」,卻跟頁面真正想傳達的主題無關,如果照單全收送進向量庫,會嚴重稀釋檢索結果的品質。今天我們選用 trafilatura 這個專門做正文擷取的函式庫,它內部用一系列啟發式規則與版面特徵,判斷 HTML 裡哪些區塊最可能是「正文」,比單純取全部文字準確得多,也省去我們自己針對每個網站寫客製化擷取規則的麻煩。

分塊策略:固定長度、重疊區間與語意邊界

分塊最簡單的做法是「每 N 個字元切一段」,實作起來最容易,但完全不管語意邊界,經常會把一個句子從中間切斷。稍微進階一點的做法,是盡量沿著自然的段落或句子邊界切,讓每個分塊盡量是完整的語意單位;如果一段話特別長,超過我們設定的長度上限,才在句子邊界內部再細分。另一個常見技巧是「重疊區間」(overlap):讓相鄰兩個分塊之間保留一小段重複的文字,確保原本橫跨在兩個分塊交界處的語意,至少有一個分塊能完整涵蓋,不會因為切點位置不巧而讓某個重要的句子被攔腰截斷、兩邊讀起來都不完整。

分塊大小該怎麼決定

分塊切得太小(例如只有幾十個字),檢索時雖然精準,卻容易缺乏足夠的上下文,模型看到片段可能難以理解完整意思;分塊切得太大(例如整篇文章當作一整塊),又會讓相似度計算被稀釋,因為一大段文字裡可能只有一小部分跟查詢真正相關,計算出來的整體相似度反而不突出。實務上分塊大小需要依照使用情境與 embedding 模型的建議輸入長度做取捨,沒有放諸四海皆準的答案;今天我們會用一個可調整的參數化設計,讓分塊大小與重疊長度都能依情境調整,而不是寫死一個固定數字。

完整實作

先安裝今天需要的套件:

uv pip install trafilatura

第一步:在 src/research_agent/ingest.py 建立擷取函式,用 trafilatura 從網址或原始 HTML 取出乾淨正文與標題:

# research-agent/src/research_agent/ingest.py
import hashlib
import trafilatura


def extract_from_url(url: str) -> dict | None:
    """從指定網址下載並擷取正文,失敗時回傳 None,由呼叫端決定如何處理。"""
    downloaded = trafilatura.fetch_url(url)
    if downloaded is None:
        return None

    text = trafilatura.extract(downloaded, include_comments=False, include_tables=False)
    if not text or not text.strip():
        return None

    metadata = trafilatura.extract_metadata(downloaded)
    title = metadata.title if metadata and metadata.title else url

    return {
        "url": url,
        "title": title,
        "text": text.strip(),
        "content_hash": hashlib.sha256(text.strip().encode("utf-8")).hexdigest(),
    }

這裡刻意關掉留言區(include_comments=False)與表格(include_tables=False),因為留言區的雜訊比例通常很高,而表格結構化資料若原樣塞進純文字分塊裡,容易變成一堆看不懂的符號,這兩者實際要不要保留,可以依你的目標網站型態調整。content_hash 是內文的雜湊值,用來判斷同一個網址的內容有沒有變動過,避免重複擷取完全相同的內容。

第二步:實作分塊函式,採用「先沿句子邊界累積、超過長度上限才切段、相鄰分塊保留重疊」的策略:

# research-agent/src/research_agent/ingest.py(分塊邏輯)
import re

SENTENCE_SPLIT_PATTERN = re.compile(r"(?<=[。!?\n])")


def chunk_text(text: str, max_chars: int = 500, overlap_chars: int = 80) -> list[str]:
    """把長文字切成語意盡量完整的分塊,相鄰分塊之間保留重疊區間。"""
    sentences = [s for s in SENTENCE_SPLIT_PATTERN.split(text) if s.strip()]
    chunks: list[str] = []
    current = ""

    for sentence in sentences:
        if len(current) + len(sentence) <= max_chars:
            current += sentence
        else:
            if current:
                chunks.append(current.strip())
            # 用上一塊的結尾當作這一塊的重疊開頭,避免語意斷裂
            overlap = current[-overlap_chars:] if current else ""
            current = overlap + sentence

    if current.strip():
        chunks.append(current.strip())

    return chunks

第三步:把擷取到的原始文件與分塊結果,寫進 AG Day 2 設計好的 documents、chunks 資料表,並同步把每個分塊的向量寫進昨天建立的 Chroma 集合,讓「結構化紀錄」與「向量檢索」使用同一份分塊資料,不會有兩邊不同步的問題:

# research-agent/src/research_agent/ingest.py(落地邏輯)
import sqlite3
from research_agent.config import load_settings
from research_agent.retrieval import add_documents


def ingest_url(url: str) -> int | None:
    """完整跑一次擷取、分塊、落地的流程,回傳寫入的分塊數量,失敗回傳 None。"""
    extracted = extract_from_url(url)
    if extracted is None:
        return None

    settings = load_settings()
    with sqlite3.connect(settings.database_path) as conn:
        cursor = conn.execute(
            "INSERT OR IGNORE INTO documents (source_url, title, raw_content, content_hash) "
            "VALUES (?, ?, ?, ?)",
            (extracted["url"], extracted["title"], extracted["text"], extracted["content_hash"]),
        )
        conn.commit()
        row = conn.execute(
            "SELECT id FROM documents WHERE source_url = ?", (extracted["url"],)
        ).fetchone()
        document_id = row[0]

        chunks = chunk_text(extracted["text"])
        chunk_ids, chunk_texts, metadatas = [], [], []
        for index, chunk in enumerate(chunks):
            conn.execute(
                "INSERT INTO chunks (document_id, chunk_index, chunk_text, char_length) "
                "VALUES (?, ?, ?, ?)",
                (document_id, index, chunk, len(chunk)),
            )
            chunk_ids.append(f"doc{document_id}-chunk{index}")
            chunk_texts.append(chunk)
            metadatas.append({"source_url": extracted["url"], "title": extracted["title"]})
        conn.commit()

    add_documents(ids=chunk_ids, texts=chunk_texts, metadatas=metadatas)
    return len(chunks)

第四步:驗證整條管線,用一個實際網址跑一次完整流程(示範環境若無法連外,可改用本機儲存的靜態 HTML 檔案傳入 trafilatura.extract 測試分塊與落地邏輯):

# research-agent/verify_ingest.py
from research_agent.ingest import ingest_url
from research_agent.retrieval import search_similar

count = ingest_url("https://zh.wikipedia.org/wiki/離岸風力發電")
if count is None:
    print("擷取失敗,請確認網址可連線或改用本機 HTML 測試。")
else:
    print(f"已擷取並寫入 {count} 個分塊。")
    for hit in search_similar("離岸風電對海洋生態的影響", top_k=2):
        print(f"- {hit['text'][:60]}...")

示範輸出(實際分塊數量與內容依網頁當下版本而定):

已擷取並寫入 14 個分塊。
- 離岸風力發電對海洋生態的影響主要包括對海洋哺乳動物聲學干擾...
- 環境影響評估報告指出,風機基座可能形成人工魚礁效應...

第五步:補一段針對 chunk_text 的單元測試,確保重疊邏輯與長度上限確實生效,這部分完全不需要網路連線:

# research-agent/tests/test_chunking.py
from research_agent.ingest import chunk_text


def test_chunk_respects_max_length():
    long_text = "這是一句測試句子。" * 100
    chunks = chunk_text(long_text, max_chars=50, overlap_chars=10)
    assert all(len(c) <= 60 for c in chunks)  # 允許重疊區間造成的些微超額
    assert len(chunks) > 1


def test_chunk_keeps_overlap_between_segments():
    text = "第一段內容說明背景。第二段內容承接上文並展開細節。第三段內容做出結論。"
    chunks = chunk_text(text, max_chars=20, overlap_chars=6)
    assert len(chunks) >= 2
    assert chunks[1][:6] in chunks[0]

第六步:真實情境下我們通常要一次處理一批網址,而不是單一網址。這裡補一個批次擷取函式,在每次請求之間加入禮貌延遲,並把成功、失敗的網址分別彙整,方便事後檢視哪些來源需要額外處理:

# research-agent/src/research_agent/ingest.py(批次擷取)
import time


def ingest_urls(urls: list[str], delay_seconds: float = 1.5) -> dict:
    """依序擷取多個網址,請求之間加入延遲,避免對單一網站造成過大負擔。"""
    succeeded: list[tuple[str, int]] = []
    failed: list[str] = []

    for index, url in enumerate(urls):
        count = ingest_url(url)
        if count is None:
            failed.append(url)
        else:
            succeeded.append((url, count))

        if index < len(urls) - 1:
            time.sleep(delay_seconds)

    return {"succeeded": succeeded, "failed": failed}

常見錯誤與踩雷

第一個常見錯誤,是遇到大量依賴 JavaScript 動態渲染內容的網站(例如單頁應用程式),trafilatura.fetch_url 拿到的只是初始 HTML,實際內容要等瀏覽器執行 JavaScript 後才會出現,導致擷取結果是空的或極度不完整。這種情況需要改用瀏覽器自動化工具(例如 Playwright)先把頁面完整渲染出來,再把渲染後的 HTML 交給 trafilatura.extract 處理,單純的 HTTP 請求無法解決這個問題。

第二個是忽略重複擷取的問題。如果沒有先檢查 content_hash 或網址是否已經存在,同一個網頁被重複擷取好幾次,會在 chunks 表跟向量庫裡累積大量重複資料,不只浪費儲存空間,檢索結果也會被同樣的內容洗版。我們在 ingest_url 裡用 INSERT OR IGNORE 搭配 documents.source_url 的唯一性約束(在 AG Day 2 建表時已經設定 UNIQUE)來避免這個問題,但要注意這只能防止「同一個網址重複寫入」,如果同樣的內容出現在兩個不同網址下,仍然會被當成兩份獨立資料。

第三個雷是分塊時用固定字元數硬切,完全不管標點符號位置,導致分塊中間出現斷詞斷句的情況,例如一句話被硬生生切成「離岸風電開發需要考量對海」與「洋生態系統的長期衝擊」兩塊,兩塊分開看都語意不完整。我們今天的 chunk_text 先用正規表示式沿標點符號與換行切成句子,再依句子累積,就是為了避免這種硬切造成的語意破碎。

第四個常見疏漏,是完全沒有處理擷取失敗或內容過短的情況。有些網址可能是死連結、被反爬蟲機制擋下、或頁面本身只有一兩句話,如果沒有像我們在 extract_from_url 裡做 None 與空字串的檢查,後續分塊與落地邏輯可能對著空字串或 None 值操作,拋出難以理解的例外,讓整條批次擷取流程意外中斷。

效能與實務提醒

擷取網頁時務必尊重目標網站的存取禮節:檢查 robots.txt 是否允許程式化存取、在連續請求之間加入合理的延遲、並在 User-Agent 標頭裡誠實標明自己是研究用途的擷取程式。短時間內對同一個網站發送大量請求,很容易觸發對方的防護機制,導致 IP 被暫時封鎖,甚至影響到其他共用網路環境的使用者。

分塊參數(max_chars、overlap_chars)沒有一組放諸四海皆準的數字,實務上建議先用一小批具代表性的資料手動檢視分塊結果,確認語意完整度可以接受,再套用到大量資料上。我們把這兩個參數設計成函式參數而不是寫死的常數,就是方便之後依不同類型的來源(新聞文章、技術文件、法規條文)調整不同的分塊策略。

批次擷取大量網址時,落地與向量化這兩步的效能瓶頸點不同:落地(寫入 SQLite)通常很快,向量化(呼叫 embedding 模型)相對耗時。建議把「擷取+分塊+寫入 SQLite」與「向量化+寫入 Chroma」分成兩個階段分開執行,中間可以視需要批次處理、甚至平行化向量化這一步,而不是把整個流程綁成一條完全同步、逐筆處理的長串列。今天寫的 ingest_urls 也刻意把成功與失敗的結果分開回傳,而不是遇到第一個失敗就整批中斷,這在批次處理外部資源時是相對穩健的設計:少數幾個網址擷取失敗,不應該連累其他原本可以順利完成的來源。

小結

今天我們把 research-agent 從「只能查詢我們手打的示範句子」,推進到「能從真實網頁擷取正文、做語意完整的分塊、寫進結構化資料表與向量庫」的完整擷取管線。trafilatura 負責從雜訊裡找出正文,我們自己實作的分塊函式負責在長度限制與語意完整度之間取得平衡,ingest_url 則把整條流程串起來,確保 documents、chunks 與 Chroma 三邊的資料保持一致。

今天新增的關鍵詞:擷取(extraction)——從原始 HTML 中找出真正正文、去除雜訊的過程;分塊(chunking)——把長文字切成大小適中、語意盡量完整的小段落;重疊區間(overlap)——相鄰分塊之間保留的重複文字,避免語意在切點處被截斷。這三個詞合起來,構成了任何一個 RAG 系統在「檢索」之前,最容易被低估、卻也最直接決定檢索品質上限的前置工程。

結語

有了完整的擷取管線,research-agent 已經能建立一份真正屬於自己、持續累積的知識庫。但目前的檢索方式還很直接:使用者問什麼,就直接拿那句話去查向量庫,查詢字串本身的用詞如果跟知識庫裡的內容有落差,檢索效果依然會打折扣。

明天,我們會進入「AG Day 22 檢索品質:查詢改寫與 rerank」,介紹如何在查詢送進向量庫之前先做改寫、以及在拿到候選結果之後用重排序模型做二次篩選,讓最終送給模型的檢索結果更精準、更貼近使用者心裡真正想問的那個問題,而不只是字面上打出來的那句話。

延伸資源

  • trafilatura 官方文件:https://trafilatura.readthedocs.io/。正文擷取演算法、參數選項與已知限制的完整說明。
  • Python 官方文件:re 模組,正規表示式語法與 lookbehind 斷言的使用方式。
  • robots.txt 規範說明(The Robots Exclusion Protocol):了解如何在程式化擷取前正確檢查網站的存取規則。

留言

這個網誌中的熱門文章

Day 2 變數與資料型別

Day 2 變數與資料型別 引言 寫程式的過程中,變數與資料型別是處理資料的基礎。變數是存放資料的容器,資料型別則決定這筆資料有哪些特性、可以進行哪些操作。學會定義變數、認識各種資料型別,是學好 Python 的關鍵一步。 這篇文章會帶你了解 Python 中變數的觀念、如何定義變數,以及常見的資料型別,包括整數、浮點數、字串、布林值,還有串列、元組、字典與集合等容器型別。我們也會介紹變數的命名規則與撰寫風格建議,以及如何用 type() 檢查資料型別。 什麼是變數?如何在 Python 中定義變數 變數是在程式執行時用來存放資料的名稱。透過定義變數,我們可以給一筆資料一個名字,並在程式的其他地方用這個名字取用該筆資料。在 Python 中,變數不需要事先宣告型別,因為 Python 是動態型別語言,變數的型別由指定給它的值決定。 定義變數的基本語法 在 Python 中定義變數非常簡單,只要用賦值符號 = 把值指定給變數即可。例如: x = 5 # 定義變數 x,並把整數 5 賦值給它 name = "Alice" # 定義變數 name,並把字串 "Alice" 賦值給它 在這裡,x 是一個變數,被賦予整數 5;name 是另一個變數,被賦予字串 "Alice"。 變數的更新與覆寫 變數的值可以修改,也就是說,我們可以在程式的不同地方給同一個變數新的值。例如: x = 10 # x 最初被賦予 10 x = 15 # x 的值現在被更新為 15 這樣就能依照需求,在程式執行過程中靈活調整變數的值。 Python 的動態型別系統 Python 和某些靜態型別語言不同,定義變數時不需要宣告型別。賦值時,Python 會根據值自動判斷變數的型別。例如: x = 5 # x 是整數 x = 3.14 # x 變成浮點數 x = "Hi" # x 變成字串 同一個變數在程式執行過程中可以存放不同型別的值,這是 Python 的彈性之一。 常見資料型別 在 Python 中,資料型別決定我們可以對變數進行哪些操作...

Day 1 Python 簡介與環境設定

Day 1 Python 簡介與環境設定 引言 在現在的科技環境裡,程式設計已經是一項重要技能。無論你是對資料科學有興趣、想成為開發者,或是想踏入人工智慧(AI)領域,學會寫程式都能明顯提升你的競爭力。在眾多程式語言中,Python 因為語法簡單、功能強大、應用範圍廣泛,成為許多人進入程式世界的第一選擇。這篇文章會帶你認識 Python 的背景與優勢,並一步步教你在不同系統上安裝與設定 Python 開發環境,最後寫出第一支 Python 程式。 為什麼選擇 Python? Python 是一種高階程式語言,由 Guido van Rossum 在 1991 年發布。Python 的設計哲學強調程式碼的可讀性,並用縮排來定義程式區塊,這點和許多使用大括號的語言不同。簡潔的語法讓它成為初學者的理想選擇;就算是經驗豐富的開發者,也能用它完成複雜的專案。 Python 的優勢如下: 簡單易學 :Python 的語法清楚、結構簡潔,初學者很快就能上手。和其他語言相比,學習曲線相對平緩,不需要先弄懂一堆複雜觀念,就能開始寫程式。 應用範圍廣泛 :從資料科學、網頁開發、人工智慧、機器學習、自動化測試到網路爬蟲,Python 都有大量開源函式庫與工具支援,而且在這些領域都扮演關鍵角色。 豐富的函式庫與框架 :Python 的函式庫生態系非常龐大。做資料分析有 NumPy、Pandas;開發網站有 Django、Flask;做深度學習有 TensorFlow、PyTorch。各種需求幾乎都能找到對應的套件,讓開發更有效率。 跨平台支援 :Python 支援 Windows、macOS、Linux 等作業系統,程式通常不需要太多修改就能跨平台執行,讓開發與部署更有彈性。 活躍的社群 :Python 擁有龐大的開發者社群。學習或開發上遇到問題,幾乎都能在社群與論壇(例如 Stack Overflow)找到答案,對初學者來說是很強的後盾,也能減少卡關時的挫折感。 Python 的應用領域 Python 的流行與強大功能,讓許多領域都開始大量使用它。以下是幾個常見的應用方向: 資料科學 :隨著大數據與人工智慧興起,資料科學大量使用 Python。NumPy、Pandas 與 Matplotlib 等工具能處理和分析龐...

Python 從入門到 PyTorch 深度學習:開啟 AI 世界的大門

Python 從入門到 PyTorch 深度學習:開啟 AI 世界的大門 隨著人工智慧(AI)與深度學習(Deep Learning)快速發展,越來越多人對這些技術產生興趣。不論你是想踏入 AI 領域的初學者,還是已經有程式基礎的開發者,學好 Python 與深度學習框架(例如 PyTorch),都能為你打開更多可能。 為什麼選擇 Python? Python 已經是資料科學與人工智慧領域的首選語言。它的語法簡潔、容易上手,而且擁有龐大的生態系與大量開源函式庫。無論是資料處理、資料視覺化,還是建立機器學習與深度學習模型,Python 都能勝任。對想進入 AI 或資料科學領域的人來說,它幾乎是必備工具。 PyTorch 是什麼? PyTorch 是由 Meta(原 Facebook)AI 研究團隊開發的開源深度學習框架,以易用、靈活和動態計算圖著稱,是許多 AI 研究人員與開發者的首選。相較於其他框架,PyTorch 的寫法更貼近原生 Python,對初學者相對友善。無論是簡單的實驗,還是複雜的深度學習模型,PyTorch 都能提供強大的支援。 這個系列能帶給你什麼? 這個系列會從 Python 的基礎開始,帶你一步一步學習,最後能自己用 PyTorch 建立深度學習模型。即使你完全沒有寫過程式,也能跟著文章的節奏累積技能,理解 AI 與深度學習的核心觀念。 本系列涵蓋的主題 Python 基礎:從變數、條件判斷到函式與模組。 資料處理工具:用 NumPy 與 Pandas 有效率地操作資料。 資料視覺化:用 Matplotlib 與 Seaborn 把資料畫成圖表。 深度學習的數學基礎:線性代數、微積分與機率。 PyTorch 入門:理解張量、模型建構與 GPU 加速。 基礎深度學習模型:CNN 與 RNN 的實作應用。 深度學習專案實戰:從資料前處理到模型部署的端到端流程。 誰適合這個系列? 程式初學者 :如果你對 AI 充滿好奇,卻還沒寫過程式,系列的第一部分會帶你快速上手 Python,並幫助你理解深度學習的基本觀念。 資料科學愛好者 :如果你已經熟悉一些資料處理方法,進階部分會教你如何用 PyTorch 建構深度學習模型。 開發者與研究人員 :想更深入了...