iT邦幫忙

2026 iThome 鐵人賽

DAY 21
0
AI Engineering

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

Day 21 - Subagent 分工:各司其職的邊界怎麼畫

  • 分享至 

  • xImage
  •  

昨天的 Commander 有一行程式我故意沒解釋:

result = self.dispatch(task)

dispatch 背後是誰?今天把這一行打開。

先回答一個一定會被問的問題:為什麼不乾脆丟給一個大模型全包?讓它看圖、讀字、自己檢查、自己查 XBRL,一次搞定。

我有三個很具體的理由,全部來自前面踩過的坑。

Day 14:gemma-4-26b 看一張圖固定用 256 個視覺 token。page 61 整頁塞進去,它連字都看不清楚,你再叫它「順便驗證一下」,它驗證的是自己編的內容。模型再大,這個輸入預算的問題不會消失,只會換個門檻出現。

Day 5 跟 Day 11:同源盲點。把「飽和聚酯樹脂」讀成「酚醛樹脂」的那組權重,回頭檢查時也會覺得「酚醛樹脂」很通順。

第三個理由比較少人講:責任。一個全包的模型出錯時,你只知道「模型錯了」。拆開之後,你可以說「OCR 讀錯了,text verifier 沒抓到,XBRL 那邊這個欄位不在範圍內」。三句話指向三個不同的改進方向。

角色表

角色 負責的事 實作 用模型嗎
Extractor 文字層、版面區塊、依版面切小 PyMuPDF + Docling(Day 18) 不用
Reader 把區塊影像讀成文字 gemma-4-26b(Day 9) 用
Text Verifier 獨立檢查 Reader 的文字 gpt-oss:20b(Day 11,尚無機器可跑) 用,且必須不同源
Field Checker 金額、日期、簽約對象的格式與一致性 semantic_field_validator.py(Day 16) 不用
Citation Extractor 公告字號與關聯金額 citation_extractor.py(Day 15) 不用
MOPS Domain Verifier 財報欄位對 XBRL fact mops_xbrl_verifier.py(Day 17) 不用
Human Auditor 看所有被送上來的疑點 人 不適用

七個角色裡只有兩個用模型。這不是我刻意壓低,是一個一個問「這件事需要語言理解嗎」問出來的。千分位對不對、字號格式對不對、XBRL 裡有沒有這個值,都不需要。

Day 17 的 MOPS 領域語意 Verifier 在這裡的位置很明確:它是驗證者,跟 Text Verifier 平行,但看的東西完全不同。Text Verifier 看「這段字讀得對不對」,MOPS Verifier 看「這個數字是不是那家公司那一期那個科目的值」。一個管字形,一個管語意。

三條邊界規則

規則不多,但每條都有人(主要是我自己)差點違反過。

一、只能讀別人的輸出,不能改。Reader 的文字是原始證據,任何角色都不准覆寫它。Field Checker 發現 NT$1,23元 格式不對,它能說的就是「這裡千分位不合法」,不能說「應該是 1,230」。

二、只回報自己看得懂的範圍。MOPS Verifier 碰到 page 5「若以美元計,營收年增35.9%」這種欄位,正確回應是「不在我的範圍」,不是 PASS,也不是 FLAG。

三、每個結果都要標明是誰做的。模型角色要寫 model_id,規則角色要寫程式版本。

第二條逼我改了一個東西。Day 11 的 VerifierVerdict 有三個值:PASS、FLAG、FAIL。Day 17 的 mops_xbrl_verifier.Verdict 只有 PASS、FLAG。兩個都沒辦法表達「我沒檢查」。

這個缺口很危險。如果 MOPS Verifier 對它看不懂的欄位回 PASS,下游會以為這個數字被外部資料確認過;如果回 FLAG,人工稽核的佇列會被一堆根本不是錯誤的東西淹掉。Day 8 預估過 Tesseract 在小字上可能製造大量假陽性,同樣的道理,一個回報太多 FLAG 的驗證器,很快就會被人忽略。

所以在 subagent 層多一個狀態:

from enum import Enum


class Outcome(str, Enum):
    PASS = "PASS"
    FLAG = "FLAG"
    FAIL = "FAIL"                  # 角色本身沒完成工作(例外、逾時、撞上限)
    NOT_APPLICABLE = "N/A"         # 不在這個角色的檢查範圍內

FAIL 跟 FLAG 我也分得比 Day 11 更硬一點:FLAG 是「做完了,發現疑點」,FAIL 是「沒做完」。Commander 對這兩種的反應不一樣,前者送人工,後者照 Day 20 的規則表重試或換方法。

介面定義

所有 subagent 吃同一種請求、吐同一種結果:

from __future__ import annotations

from typing import Protocol

from pydantic import BaseModel, Field

from verifier_protocol import VerifierIssue


class Evidence(BaseModel):
    """subagent 產出的原始證據。一旦寫入就不再修改。"""
    location: str
    source: str                    # "pymupdf" / "docling" / "gemma-4-26b" / "rule:semantic_field_validator"
    text: str | None = None
    data: dict = Field(default_factory=dict)


class SubagentRequest(BaseModel):
    task_id: str
    doc_id: str
    location: str
    inputs: list[Evidence]         # 上游證據,唯讀
    params: dict = Field(default_factory=dict)


class SubagentResult(BaseModel):
    task_id: str
    role: str
    outcome: Outcome
    produced: list[Evidence] = Field(default_factory=list)   # 只有 Extractor、Reader 會產生新證據
    issues: list[VerifierIssue] = Field(default_factory=list)
    signals: dict = Field(default_factory=dict)
    model_id: str | None = None
    impl_version: str


class Subagent(Protocol):
    role: str
    uses_model: bool
    model_name: str | None         # 規則角色填 None

    def run(self, req: SubagentRequest) -> SubagentResult: ...

issues 直接沿用 Day 11 的 VerifierIssue。它有 location、original_text、concern、confidence,四個欄位剛好夠。original_text 那個欄位名稱本身就在提醒:你只能引用原文,不能給修正版。

produced 跟 issues 分成兩個欄位,是為了第一條邊界規則。驗證類角色的 produced 永遠應該是空的。我在註冊表那邊加了一道檢查:

PRODUCERS = {"extractor", "reader"}


def enforce_boundaries(agent: Subagent, req: SubagentRequest, res: SubagentResult) -> SubagentResult:
    if res.produced and agent.role not in PRODUCERS:
        raise PermissionError(f"{agent.role} 不能產生新證據,只能回報 issues")
    before = [e.model_dump() for e in req.inputs]
    # 驗證角色跑完之後,上游證據必須一字不差
    if any(e.model_dump() != b for e, b in zip(req.inputs, before)):
        raise PermissionError(f"{agent.role} 修改了上游證據")
    if agent.uses_model and not res.model_id:
        raise ValueError(f"{agent.role} 用了模型卻沒有回報 model_id")
    return res

老實說,第二個檢查在這個寫法下抓不到什麼,因為 before 是在 run() 之後才拍的。真正要防的話,要在呼叫 run() 之前先深拷貝一份。我把它留在這裡當作提醒,註冊表的實際版本會在呼叫前後各拍一次:

def call(agent: Subagent, req: SubagentRequest) -> SubagentResult:
    snapshot = [e.model_dump() for e in req.inputs]
    res = agent.run(req)
    if [e.model_dump() for e in req.inputs] != snapshot:
        raise PermissionError(f"{agent.role} 修改了上游證據")
    if res.produced and agent.role not in PRODUCERS:
        raise PermissionError(f"{agent.role} 不能產生新證據,只能回報 issues")
    if agent.uses_model and not res.model_id:
        raise ValueError(f"{agent.role} 用了模型卻沒有回報 model_id")
    return res

我會把上面那個錯的版本留在文章裡,是因為我第一次真的寫成那樣。檢查寫在錯的時間點,看起來有檢查,其實什麼都沒擋。這種 bug 的可怕在於它永遠不會失敗。

把 Day 17 的 verifier 包進來

mops_xbrl_verifier.py 原本的介面是 compare_ocr_fields(facts, fields)。包成 subagent 只需要一層轉接,原本的程式一行都不用改:

from mops_xbrl_verifier import OcrField, Verdict, XbrlFact, compare_ocr_fields
from verifier_protocol import VerifierIssue

XBRL_COVERED_CONCEPTS = {"Revenue", "Assets"}   # 只收已確認對應的科目,示範用


class MopsDomainVerifier:
    role = "mops_domain_verifier"
    uses_model = False
    model_name = None

    def __init__(self, facts: list[XbrlFact]):
        self.facts = facts

    def run(self, req: SubagentRequest) -> SubagentResult:
        fields = [OcrField(**e.data) for e in req.inputs if e.data.get("concept")]
        in_scope = [f for f in fields if f.concept in XBRL_COVERED_CONCEPTS]
        if not in_scope:
            return SubagentResult(task_id=req.task_id, role=self.role,
                                  outcome=Outcome.NOT_APPLICABLE,
                                  impl_version="mops_xbrl_verifier@day17")
        results = compare_ocr_fields(self.facts, in_scope)
        issues = [
            VerifierIssue(location=r.field.location, original_text=r.field.original_value,
                          concern=r.reason or "", confidence=0.95)
            for r in results if r.verdict is Verdict.FLAG
        ]
        return SubagentResult(task_id=req.task_id, role=self.role,
                              outcome=Outcome.FLAG if issues else Outcome.PASS,
                              issues=issues, impl_version="mops_xbrl_verifier@day17",
                              signals={"checked": len(in_scope), "skipped": len(fields) - len(in_scope)})

confidence=0.95 沿用 Day 16 semantic_field_validator.py 的慣例。規則式的檢查其實沒有「信心」這回事,比對結果不是對就是錯;但 VerifierIssue 要求這個欄位,我填一個固定值,同時在 impl_version 標明這是規則角色,下游看到就知道這個數字不是模型的機率。

signals 裡的 skipped 很重要。一份年報丟進來,如果 checked 是 12、skipped 是 300,這就是 XBRL 在這份文件上的真實覆蓋範圍。我寧可每次都看到這個難看的比例,也不要一個只算被檢查過的欄位、看起來很漂亮的通過率。

Field Checker 的包法幾乎一樣,把 Evidence 轉成 Day 16 的 FieldValue,呼叫 validate_semantic_fields,把 issues 直接帶出來。它回傳的本來就是 VerifierIssue,連轉換都省了。

邊界案例

規則寫起來都很乾淨,麻煩在它們相撞的時候。

案例一:Text Verifier 說某個金額可疑,MOPS Verifier 說 PASS。

聽起來 XBRL 贏了,外部資料比模型可靠。但要小心 PASS 的範圍:MOPS Verifier 只確認了「數值」跟 XBRL 一致。如果 Text Verifier 的疑點是科目名稱(Day 2 那種「一年內到期長期負債」被讀成「短期負債」),數字對了不代表科目對了。反過來說,要是 OCR 把科目讀錯,轉接層根本會把它對到另一個 concept 上去比。所以兩個結果我都保留,交給 Commander。怎麼合併是 Day 22 的事。

案例二:MOPS Verifier 回 N/A,其他驗證全部 PASS。

page 44 的「已發行限制員工權利新股股數 2,110,000股」就是這種。格式規則過了,Text Verifier 也可能說沒問題,但沒有任何外部資料確認過這個數字。這種欄位在最後的輸出裡要帶一個標記:它的可信度來自模型之間的一致,不是來自外部答案。

案例三:Reader 跟 Text Verifier 的 model_id 來自同一家。

這種情況 call() 本身不會擋,因為每個角色只看得到自己。我把檢查放在註冊表建立的時候:

def check_independence(reader: Subagent, verifier: Subagent, vendor_of: dict[str, str]) -> None:
    if vendor_of[reader.model_name] == vendor_of[verifier.model_name]:
        raise ValueError("Reader 與 Text Verifier 同源,違反 Day 11 的獨立性原則")

vendor_of 是一張手寫的對照表:gemma 對應 Google,gpt-oss 對應 OpenAI,Nemotron 對應 NVIDIA。Day 13 的決策樹在這裡變成一行程式。在啟動時就擋,比跑完一整份文件才發現便宜得多。

案例四:Citation Extractor 在 OCR 輸出上抽到的東西跟在文字層上抽到的不一樣。

Day 15 那次運氣好,GT 和緊裁 OCR 都抽到 3/3,而且數字集合一致。但如果不一致,誰對?Citation Extractor 不該判斷,它只是抽取器,它對兩份輸入各自回報抽到什麼。「兩個來源不一致」這件事本身就是一條證據,交給 Commander。

四個案例有一個共同點:每一次我都想讓某個角色「順便」多做一點判斷,最後都把那一點判斷推回了 Commander 或人工。角色本身保持窄,合併的邏輯集中在一個地方。

還沒解決的地方

這張分工表現在有一個很明顯的洞:Text Verifier 沒有機器可以跑。整個架構裡唯一一個「不同源的模型」位置,現在是空的,所有 Reader 的輸出只能靠規則跟人工把關。

另一個洞是這些驗證之間沒有順序。Field Checker 很便宜,幾毫秒;MOPS Verifier 需要先有 XBRL facts;Text Verifier 要一次模型呼叫。它們全部對同一個區塊跑一遍,還是先跑便宜的、有問題才跑貴的?如果便宜的已經 FLAG 了,貴的還要不要跑?

Day 8 的規劃裡,驗證是四層:Checksum、Logprob、Tesseract、MOPS 領域語意 Verifier。今天定義好的角色裡,MOPS 那層已經有了位置,另外三層要怎麼跟它排在一起、誰先誰後、誰的結果能蓋過誰,是下一篇的題目。


上一篇
Day 20 - Commander 設計:分派邏輯與任務拆解
下一篇
Day 22 - 四重驗證:Checksum、Logprob、Tesseract、MOPS 領域語意 Verifier
系列文
從 LLM 到 Harness: 打造隱私與可信任的繁中進階 OCR Agent 共 24 篇
圖片
  熱門推薦
圖片
{{ item.channelVendor }} | {{ item.webinarstarted }} |
{{ formatDate(item.duration) }}
直播中

尚未有邦友留言

立即登入留言