iT邦幫忙

2026 iThome 鐵人賽

DAY 18
0
AI Engineering

從 LLM 到 Harness: 打造隱私與可信任的繁中進階 OCR Agent系列 第 18 篇

Day 18 - Pipeline 組裝:PyMuPDF、Docling、Region Routing

  • 分享至 

  • xImage
  •  

零件散落在桌上已經好幾天了。

Day 5 說 PyMuPDF 抽文字層幾乎不花錢。Day 10 說合併表格一起丟,覆蓋率從單張的 0.98 掉到 0.9124,diff 塊從 7、9 跳到 35。Day 14 說整頁台積電年報塞給 gemma-4-26b,四頁全部撞 3000 token 上限。Day 15 說把段落緊裁出來、DPI 拉到 400,page 61 那段裁罰揭露從 0/6 變 6/6。

每一條結論都指向同一件事:送進模型之前,頁面要先切好。

問題是 Day 15 那個裁切座標 (40, 55, 305, 260) 是我自己一個 block 一個 block 看出來的。一頁可以,92 頁不行。今天要把「怎麼切、切完送去哪」寫成程式。

整條線長這樣:

PDF
 └─ ① PyMuPDF:文字層 + 品質訊號
     └─ ② Docling:版面偵測,得到帶座標的區塊
         └─ ③ Region Routing:每個區塊決定走哪條路
             ├─ 文字層可信 → 直接採用,不碰 GPU
             ├─ 一般區塊   → 渲染成圖,送 VLM
             ├─ 太密       → 再切小、提高 DPI
             └─ 圖表       → 送 VLM,標記「無外部校驗來源」
         └─ ④ 組裝:每個區塊的結果帶著定位寫成 JSONL

第一段:PyMuPDF,先問文字層能不能信

文字層是最便宜的資料來源,但它不是答案卷。Day 6 抓到過 page_02 的底紋、page_19 的圖表標籤,文字層完全沒抽到;Day 7 則發現有些 PDF 的文字層用了筆畫碼位,看起來像字,其實不是標準碼位。

所以這一段的工作不是「抽文字」,是「抽文字,順便判斷它有多可疑」。

台積電這份年報還給了我一個意外。Day 14 做切半的時候,page.get_text("words") 跟 get_text("dict") 對這份 PDF 內嵌的中文字型吐出來的是亂碼,get_text("text") 跟 get_text("blocks") 卻正常。同一頁、同一個函式庫,換一個參數結果天差地遠。我到現在還不知道確切原因,只知道這件事記在 step4_split_and_retest.py 的註解裡,而且它決定了我後面全部用 blocks。

from __future__ import annotations

import unicodedata
from dataclasses import dataclass

import pymupdf


@dataclass
class TextBlock:
    page_index: int                              # PyMuPDF 從 0 開始
    bbox: tuple[float, float, float, float]      # x0, y0, x1, y1,左上原點,單位 pt
    text: str


@dataclass
class TextLayerQuality:
    n_chars: int
    pua_ratio: float          # 私有區字元比例(Wingdings 類符號、怪字型)
    replacement_ratio: float  # U+FFFD 比例
    cjk_ratio: float

    @property
    def trustworthy(self) -> bool:
        # 門檻是起始值,不是量測結果;要靠後面的稽核資料校準
        return self.n_chars >= 30 and self.pua_ratio < 0.02 and self.replacement_ratio == 0


def _quality(text: str) -> TextLayerQuality:
    chars = [c for c in text if not c.isspace()]
    n = len(chars) or 1
    pua = sum(1 for c in chars if 0xE000 <= ord(c) <= 0xF8FF)
    rep = chars.count("�")
    cjk = sum(1 for c in chars if unicodedata.name(c, "").startswith("CJK"))
    return TextLayerQuality(len(chars), pua / n, rep / n, cjk / n)


def extract_text_layer(doc: pymupdf.Document, page_index: int) -> tuple[list[TextBlock], TextLayerQuality]:
    page = doc[page_index]
    blocks: list[TextBlock] = []
    # blocks 的每個元素是 (x0, y0, x1, y1, text, block_no, block_type),block_type 0 是文字
    for x0, y0, x1, y1, text, _no, btype in page.get_text("blocks"):
        if btype == 0 and text.strip():
            blocks.append(TextBlock(page_index, (x0, y0, x1, y1), text))
    full = "".join(b.text for b in blocks)
    return blocks, _quality(full)

trustworthy 那幾個門檻,30 字是沿用 Day 6 排除「沒得比」頁面的那條線,其他兩個是我拍腦袋的。寫在這裡不是因為它們對,是因為它們得有個起點,之後才有東西可以改。

Tip:文字層可信的時候,這一頁可以完全不送 GPU。台積電年報是原生文字層 PDF,理論上大部分頁面都能走這條路。但「理論上」三個字我已經被咬過太多次,所以就算文字層看起來很乾淨,我還是會抽一部分頁面送 OCR 做對照。不做對照,你永遠不會知道文字層哪天開始騙你。

第二段:Docling,把一頁拆成區塊

PyMuPDF 的 block 是 PDF 內部的文字物件,它不知道「這是表格」「這是標題」。Day 15 裁切時踩到的坑就是這個:page 61 左半頁內部還分兩欄,block 本身沒告訴我,是我把第一次裁出來的 GT 打開,看到隔壁欄的「展望未來,台積公司將持續以員工體驗為核心…」混進來,才發現的。

Docling 做的是版面分析,它會告訴你這一塊是 text、table、picture、section_header 還是 page_header,連同頁碼跟 bbox。

from docling.document_converter import DocumentConverter
from docling_core.types.doc import TableItem, PictureItem, TextItem


@dataclass
class Region:
    region_id: str
    page_index: int                              # 統一換成 0 起算
    kind: str                                    # "text" / "table" / "picture" / "header" / ...
    bbox: tuple[float, float, float, float]      # 左上原點,pt
    docling_text: str | None = None


def detect_regions(pdf_path: str) -> list[Region]:
    result = DocumentConverter().convert(pdf_path)
    doc = result.document
    regions: list[Region] = []
    for i, (item, _level) in enumerate(doc.iterate_items()):
        if not getattr(item, "prov", None):
            continue
        prov = item.prov[0]
        page_h = doc.pages[prov.page_no].size.height
        bb = prov.bbox.to_top_left_origin(page_height=page_h)
        kind = str(getattr(item, "label", "text")).split(".")[-1].lower()
        text = None
        if isinstance(item, TableItem):
            text = item.export_to_markdown(doc=doc)
        elif isinstance(item, TextItem):
            text = item.text
        regions.append(Region(
            region_id=f"p{prov.page_no - 1:03d}_r{i:03d}",
            page_index=prov.page_no - 1,         # Docling 的 page_no 從 1 開始
            kind=kind,
            bbox=(bb.l, bb.t, bb.r, bb.b),
            docling_text=text,
        ))
    return regions

兩個座標系的坑,寫出來省得你再踩一次:

Docling 的 page_no 從 1 開始,PyMuPDF 的 index 從 0 開始。差一頁這種錯最陰險,因為相鄰兩頁的版面常常很像,你會在錯的頁面上裁出一塊看起來很合理的區域。

Docling 的 bbox 預設是左下原點(PDF 的原生座標),PyMuPDF 的 clip 吃的是左上原點。to_top_left_origin 轉一次就好,但忘了轉的話,你會裁到頁面上下翻轉後的位置。

還有一個跟這份年報有關的:它每一個 PDF page 其實是兩個實體頁拼成的對開,渲染出來 3386×2205。Docling 看到的是一張寬頁,兩頁中間那條空白的裝訂線它不一定會當成分界。Day 14 切半沒出事是因為那條 gutter 夠寬,沒有字跨過中線,但我不會假設每份年報都這樣。

Docling 我目前只當版面偵測用,它自己也能做 OCR 跟表格結構辨識,但那部分我沒打算交給它。原因很單純:OCR 的主角在 Day 9 已經選好了,我不想在同一條線上出現兩個 OCR 來源,卻沒有規則說聽誰的。

第三段:Region Routing

這是今天真正新的東西。每個區塊進來,決定一條路。

決策依據有三個:區塊類型、文字層品質、文字密度。

密度是 Day 14 最重要的教訓。整頁崩、切半還是崩、緊裁才好,病因是「文字密度對上固定的視覺 token 預算」。page 61 左半頁塞了 1238 個字,照樣 0/6;緊裁出來那一段的 GT 是 296 字,配 400 DPI,CER 0.0845、覆蓋率 0.9966。

所以我需要一個「這塊送進去會不會太擠」的判斷:

def char_density(region: Region, blocks: list[TextBlock]) -> tuple[int, float]:
    """回傳 (區塊內文字層字數, 每千平方 pt 的字數)。文字層不可信時只當參考。"""
    x0, y0, x1, y1 = region.bbox
    inside = [b for b in blocks
              if b.bbox[0] >= x0 - 2 and b.bbox[1] >= y0 - 2
              and b.bbox[2] <= x1 + 2 and b.bbox[3] <= y1 + 2]
    n = sum(len(b.text.replace("\n", "").replace(" ", "")) for b in inside)
    area = max((x1 - x0) * (y1 - y0), 1.0)
    return n, n / area * 1000

門檻怎麼訂?我手上只有兩個點:1238 字崩了,296 字沒崩。中間在哪裡斷,我不知道。所以一開始我用最保守的做法:單次送進模型的字數上限先壓在 400 左右,超過就切。這個 400 是從 296 往上抓一點餘裕的猜測,不是量出來的門檻。

from enum import Enum


class Route(str, Enum):
    TEXT_LAYER = "text_layer"        # 直接採用文字層
    VLM = "vlm"                      # 渲染後送 gemma-4-26b
    SPLIT = "split"                  # 太密,先切小再送
    VLM_UNVERIFIABLE = "vlm_unverifiable"   # 圖表:送 VLM,但沒有外部校驗來源
    SKIP = "skip"                    # 頁首頁尾、頁碼


MAX_CHARS_PER_CALL = 400   # 起始值,依 Day 14/15 兩個觀測點保守抓的


def route(region: Region, quality: TextLayerQuality, n_chars: int) -> Route:
    if region.kind in {"page_header", "page_footer"}:
        return Route.SKIP
    if region.kind == "picture":
        return Route.VLM_UNVERIFIABLE
    if region.kind == "table":
        # 表格不直接信文字層:文字層會把表格拉平,欄位對應會丟
        return Route.SPLIT if n_chars > MAX_CHARS_PER_CALL else Route.VLM
    if quality.trustworthy:
        return Route.TEXT_LAYER
    return Route.SPLIT if n_chars > MAX_CHARS_PER_CALL else Route.VLM

表格那一條我猶豫最久。表格區的文字層其實常常是完整的,字都在,只是順序被拉成一長串。Day 10 算過,覆蓋率看不出順序錯。如果我讓表格也走文字層,速度會快很多,但 Day 17 那個 verifier 需要知道「這個數字在哪一列、哪一欄」,拉平的文字層給不了這個。所以表格一律送 VLM,這條我先不妥協。

SPLIT 怎麼切,我沿用 Day 15 學到的:沿著內容的實際版面切,不是沿頁面幾何中心切。做法是用落在區塊內的 PyMuPDF block,照 y 座標往下累加字數,累到上限就斷一刀:

def split_region(region: Region, blocks: list[TextBlock],
                 limit: int = MAX_CHARS_PER_CALL) -> list[Region]:
    x0, y0, x1, y1 = region.bbox
    inside = sorted(
        (b for b in blocks if b.bbox[1] >= y0 - 2 and b.bbox[3] <= y1 + 2
         and b.bbox[0] >= x0 - 2 and b.bbox[2] <= x1 + 2),
        key=lambda b: b.bbox[1],
    )
    parts, cur, count = [], [], 0
    for b in inside:
        n = len(b.text.replace("\n", ""))
        if cur and count + n > limit:
            parts.append(cur)
            cur, count = [], 0
        cur.append(b)
        count += n
    if cur:
        parts.append(cur)
    return [
        Region(f"{region.region_id}_s{k}", region.page_index, region.kind,
               (x0, min(b.bbox[1] for b in p) - 4, x1, max(b.bbox[3] for b in p) + 4))
        for k, p in enumerate(parts)
    ]

一個 block 在台積電這份檔案裡通常就是一行,所以切點會落在行與行之間,不會把一行字從中間剖開。上下各留 4pt 的邊,是怕把字的上緣或下緣切掉。Day 15 手動裁的時候留了大約 15pt,這裡縮小是因為切出來的片段彼此相鄰,留太多會讓兩片重疊,同一行字被讀兩次。

第四段:送出去,收回來,帶著定位

import base64
import json
from pathlib import Path


def render_region(doc: pymupdf.Document, region: Region, dpi: int) -> bytes:
    page = doc[region.page_index]
    pix = page.get_pixmap(dpi=dpi, clip=pymupdf.Rect(*region.bbox))
    return pix.tobytes("png")


def run_pipeline(pdf_path: str, out_jsonl: Path, call_vision) -> None:
    """call_vision(png_bytes, prompt) -> dict,沿用 experiments 裡的呼叫方式。"""
    doc = pymupdf.open(pdf_path)
    regions = detect_regions(pdf_path)
    layer_cache: dict[int, tuple[list[TextBlock], TextLayerQuality]] = {}

    with out_jsonl.open("w", encoding="utf-8") as out:
        for region in regions:
            if region.page_index not in layer_cache:
                layer_cache[region.page_index] = extract_text_layer(doc, region.page_index)
            blocks, quality = layer_cache[region.page_index]
            n_chars, _ = char_density(region, blocks)
            decision = route(region, quality, n_chars)

            targets = split_region(region, blocks) if decision == Route.SPLIT else [region]
            for r in targets:
                record = {
                    "region_id": r.region_id,
                    "location": f"page_{r.page_index:02d} / {r.kind} / {r.region_id}",
                    "bbox": r.bbox,
                    "route": decision.value,
                    "text_layer_chars": n_chars,
                }
                if decision == Route.SKIP:
                    continue
                if decision == Route.TEXT_LAYER:
                    record["text"] = "".join(
                        b.text for b in blocks if b.bbox[1] >= r.bbox[1] - 2 and b.bbox[3] <= r.bbox[3] + 2
                        and b.bbox[0] >= r.bbox[0] - 2 and b.bbox[2] <= r.bbox[2] + 2)
                    record["source"] = "pymupdf"
                else:
                    dpi = 400 if decision == Route.SPLIT else 200
                    resp = call_vision(render_region(doc, r, dpi), prompt_for(r.kind))
                    record.update(text=resp["text"], source="gemma-4-26b", dpi=dpi,
                                  finish_reason=resp["finish_reason"],
                                  completion_tokens=resp["completion_tokens"])
                    if decision == Route.VLM_UNVERIFIABLE:
                        record["external_check"] = "none"
                out.write(json.dumps(record, ensure_ascii=False) + "\n")

prompt_for(kind) 我沒展開,表格跟一般段落用不同的 prompt 而已。Day 10 講過,表格區將近六成的 token 花在畫 Markdown 框線,所以表格 prompt 我會要求輸出一格一行的簡單格式,這部分等 Day 22 用 Pydantic 定 schema 時再定案。

每一筆紀錄都有 location,格式刻意跟 Day 17 的 OcrField.location、Day 11 的 VerifierIssue.location 對齊。這不是巧合,是我回頭改的。昨天寫 XBRL verifier 的時候才意識到,所有下游都要靠這個字串回到「PDF 上的哪一塊」,要是每支程式各寫各的格式,到時候光是對齊定位就會耗掉一整天。

source 欄位也一樣重要。同一份輸出裡,有的字是 PyMuPDF 抽的,有的是模型讀的。它們的錯誤型態完全不同:文字層的錯是漏抽、碼位怪,模型的錯是幻覺、形近字。混在一起的話,事後要追錯都不知道該怪誰。

各條路線的成本,用手上有的數字估

路線 單次成本依據 狀態
TEXT_LAYER PyMuPDF 抽取,毫秒級、不用 GPU(Day 5) 實際使用過
VLM(200 DPI) 單頁級請求 3.4–15.2 秒(Day 10 那五頁) 實測
SPLIT(400 DPI 緊裁) page 61 緊裁段落 5.561 秒、277 completion tokens 實測,只有一個樣本
整頁硬送(對照組) 台積電年報四頁,每頁約 58–60 秒,全撞 3000 token 實測

看最後兩列就懂為什麼要切。page 61 整頁送進去花了 58.191 秒,換來一段編出來的文字加 700 多行空表格列;緊裁一段 5.561 秒,拿到幾乎逐字正確的內容。切小不只是比較準,在這個例子裡還比較快,因為模型沒機會陷入複讀一路跑到上限。

一頁切成十塊,請求數就是十倍。這筆帳在密度低的頁面上是虧的,所以 TEXT_LAYER 這條路才那麼重要,它把能省的都省掉,GPU 只花在真的需要看圖的地方。

整條 Pipeline 還沒有在 92 頁上完整跑過一次,上表之外的數字我都不會寫。我現在比較想知道的是 route() 在整份年報上的分布:多少頁能走文字層、多少塊會被切。這個數字出來之前,任何「整份年報要跑幾分鐘」的估計都是空話。

寫完這支程式的晚上,我盯著 run_pipeline 的 for 迴圈看了很久。它很整齊,每個區塊進去、出來、寫一行 JSONL。可是如果某一塊回來的 finish_reason 是 length,它會怎麼做?

它會把那行寫進檔案,然後處理下一塊。


上一篇
Day 17 - 領域語意 Verifier 設計:用 MOPS 既有結構化資料當校驗來源
系列文
從 LLM 到 Harness: 打造隱私與可信任的繁中進階 OCR Agent 共 18 篇
圖片
  熱門推薦
圖片
{{ item.channelVendor }} | {{ item.webinarstarted }} |
{{ formatDate(item.duration) }}
直播中

尚未有邦友留言

立即登入留言