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):了解如何在程式化擷取前正確檢查網站的存取規則。
留言
張貼留言