Day 20 - Commander 設計：分派邏輯與任務拆解 - iT 邦幫忙::一起幫忙解決難題，拯救 IT 人的一天

2026 iThome 鐵人賽

DAY 20

AI Engineering

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

Day 20 - Commander 設計：分派邏輯與任務拆解

18th鐵人賽

onedream

團隊nutc imac

2026-10-04 23:12:14

75 瀏覽

分享至

先講一個可能不太受歡迎的決定：我的 Commander 不是 LLM。

至少主幹不是。它是一段普通的 Python，裡面是一張規則表、一個佇列、一本決策紀錄簿。

我知道現在流行的做法是找一顆最強的模型當 orchestrator，讓它讀任務、自己規劃、自己呼叫工具。我也試著想像過這個版本用在財報 OCR 上會怎樣。想到 page 61 就停了：一顆模型負責決定「這塊區域撞上限是因為太密還是太長」，它的依據是什麼？它看不到圖（看得到的話，那它自己就是 OCR 了），它能看的是文字層字數、輸出重複率、之前試過幾次。這些都是數字，數字的判斷用 if 寫比較便宜，也比較好稽核。

LLM 可以留在一個很小的位置：規則表完全對不上的情況。那種時候讓它提一個建議，建議本身也要被記下來、被人看。

Commander 管什麼、不管什麼

做	不做
把一份文件拆成任務	讀圖、辨識文字
決定每個任務派給誰	判定某段文字對不對
收回結果，決定下一步	修改任何 subagent 的輸出
控制重試次數與預算	自己發明新的處理方式
每個決定都寫下規則編號與理由	在沒有證據的情況下放行

右邊那一欄比左邊重要。Day 11 講過同源盲點：產生答案的人不能自己驗證。Commander 如果手癢去「順便」修一個看起來明顯錯的數字，它就同時當了產生者和驗證者，而且還是那個最沒有領域知識的。

任務長什麼樣

所有派出去的東西都是 Task。欄位是從 Day 19 那張失敗表倒推的：每一次我手動修正，都需要知道「哪一塊、試過什麼、上一個是誰的結果」。

from __future__ import annotations

from datetime import datetime, timezone
from enum import Enum
from uuid import uuid4

from pydantic import BaseModel, Field


class TaskKind(str, Enum):
    EXTRACT_LAYER = "extract_layer"      # PyMuPDF 文字層 + 品質
    DETECT_LAYOUT = "detect_layout"      # Docling 版面
    OCR_REGION = "ocr_region"            # 送 VLM
    RECROP = "recrop"                    # 依版面切小
    VERIFY_TEXT = "verify_text"          # 獨立文字 verifier
    VERIFY_FIELDS = "verify_fields"      # 欄位格式與一致性規則（Day 16）
    VERIFY_XBRL = "verify_xbrl"          # MOPS XBRL 對照（Day 17）
    EXTRACT_CITATIONS = "extract_citations"  # 公告字號（Day 15）
    HUMAN_REVIEW = "human_review"


class Task(BaseModel):
    task_id: str = Field(default_factory=lambda: uuid4().hex[:12])
    doc_id: str
    region_id: str | None = None
    location: str | None = None              # 跟 Day 18 的 location 同格式
    kind: TaskKind
    params: dict = Field(default_factory=dict)   # 例如 {"dpi": 400, "bbox": [...]}
    attempt: int = 1
    parent_task_id: str | None = None        # 由哪個任務的結果衍生出來
    reason: str = ""                         # 為什麼會有這個任務（規則編號 + 說明）
    created_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))


class TaskStatus(str, Enum):
    OK = "ok"
    NEEDS_ATTENTION = "needs_attention"      # 做完了，但結果有疑點
    FAILED = "failed"                        # 沒做完（逾時、例外、撞上限）


class TaskResult(BaseModel):
    task_id: str
    status: TaskStatus
    payload: dict = Field(default_factory=dict)  # 各 subagent 的實際輸出，Day 21 定型
    signals: dict = Field(default_factory=dict)  # finish_reason、重複率、字數等給 Commander 看的訊號
    model_id: str | None = None

parent_task_id 是我最想要的欄位。Day 14 到 Day 15 那段「整頁 → 切半 → 緊裁」，在這個結構裡會是三代任務，一路指回去。三個月後有人問「這三筆罰鍰是怎麼讀出來的」，順著 parent 往上爬就有答案，不用翻我的聊天紀錄。

reason 是字串，但內容不是自由發揮。它一定以規則編號開頭，例如 R3: 撞上限且文字層字數 1238 > 400，依版面切小。自然語言的理由很好讀，但不好統計；有了規則編號，Day 26 設計記錄系統時才能算「R3 觸發了幾次、之後成功幾次」。

拆解規則：一份文件變成哪些任務

拆解分兩層。文件層的第一批任務是固定的：

每一頁 → EXTRACT_LAYER
整份   → DETECT_LAYOUT

這兩個完成之後，每個區塊依 Day 18 的 route() 產生下一批。這裡 Commander 等於把昨天 Pipeline 裡的 if 搬過來，差別是它產生的是任務，不是直接呼叫：

區塊情況	產生的任務
頁首、頁尾	無
文字層可信的一般段落	`VERIFY_FIELDS`（文字層本身當輸出，仍然要過欄位規則）
文字層不可信、字數在上限內	`OCR_REGION`（200 DPI）
字數超過上限	`RECROP`，完成後每一片各一個 `OCR_REGION`（400 DPI）
表格	`OCR_REGION`，完成後加 `VERIFY_FIELDS`、`VERIFY_XBRL`
圖表	`OCR_REGION`，結果標記沒有外部校驗來源
文字含「字第…號」樣式	追加 `EXTRACT_CITATIONS`

最後一列是用文字層先掃一次決定的。正則跟 Day 15 的 citation_extractor.py 用同一條：字\s*第\s*(\d{4,12})\s*號。先用便宜的方法決定要不要派一個比較貴的任務，這種模式在 Commander 裡會一直出現。

分派與回收：規則表

結果回來之後，Commander 看 status 和 signals 決定下一步。這是整篇的核心，我直接給程式：

from dataclasses import dataclass
from typing import Callable

MAX_ATTEMPTS = {TaskKind.OCR_REGION: 3, TaskKind.RECROP: 2}
MAX_CHARS_PER_CALL = 400      # 跟 Day 18 同一個起始值


@dataclass
class Rule:
    rule_id: str
    when: Callable[[Task, TaskResult], bool]
    then: Callable[[Task, TaskResult], list[Task]]
    note: str


def _child(task: Task, kind: TaskKind, reason: str, **params) -> Task:
    return Task(doc_id=task.doc_id, region_id=task.region_id, location=task.location,
                kind=kind, params={**task.params, **params},
                attempt=task.attempt + 1 if kind == task.kind else 1,
                parent_task_id=task.task_id, reason=reason)


RULES: list[Rule] = [
    Rule("R1",
         lambda t, r: t.kind == TaskKind.OCR_REGION and r.signals.get("finish_reason") == "length"
                      and r.signals.get("text_layer_chars", 0) > MAX_CHARS_PER_CALL,
         lambda t, r: [_child(t, TaskKind.RECROP,
                              f"R1: 撞上限且文字層字數 {r.signals['text_layer_chars']} > {MAX_CHARS_PER_CALL}，依版面切小")],
         "輸入太密：切小，不重送原圖"),
    Rule("R2",
         lambda t, r: t.kind == TaskKind.OCR_REGION and r.signals.get("finish_reason") == "length"
                      and r.signals.get("text_layer_chars", 0) <= MAX_CHARS_PER_CALL
                      and t.params.get("dpi", 200) < 400,
         lambda t, r: [_child(t, TaskKind.OCR_REGION, "R2: 字數不多仍撞上限，提高 DPI 重送", dpi=400)],
         "字不多卻讀不好：多半是看不清楚"),
    Rule("R3",
         lambda t, r: t.kind == TaskKind.OCR_REGION and r.signals.get("repeat_ratio", 0) > 0.3,
         lambda t, r: [_child(t, TaskKind.HUMAN_REVIEW, "R3: 輸出大段重複，疑似退化，先交人工")],
         "重複退化時不盲目重跑"),
    Rule("R4",
         lambda t, r: t.kind == TaskKind.OCR_REGION and r.status == TaskStatus.OK,
         lambda t, r: [_child(t, TaskKind.VERIFY_TEXT, "R4: OCR 完成，交獨立 verifier")],
         "所有模型輸出都要過一個不同源的檢查"),
    Rule("R5",
         lambda t, r: t.kind in {TaskKind.VERIFY_TEXT, TaskKind.VERIFY_FIELDS, TaskKind.VERIFY_XBRL}
                      and r.status == TaskStatus.NEEDS_ATTENTION,
         lambda t, r: [_child(t, TaskKind.HUMAN_REVIEW, f"R5: {t.kind.value} 回報疑點，交人工，不自動修正")],
         "任何驗證的 FLAG 都不由 Commander 自己吞掉"),
]

repeat_ratio 的 0.3 是我隨手訂的，沒有量過。Day 8 提過「最長重複子串的出現次數超過門檻」這種檢查幾行就能寫，門檻要等有一批真實輸出再調。

R1 跟 R2 的分界就是 Day 19 那個「同一個 length，兩種病」。字多的撞上限，切；字少的撞上限，看清楚一點再試。它們各自對應 page 61 左半（1238 字）跟 page_19 那種情況。這兩條規則沒辦法保證一定對，但至少它們會做不一樣的事，而且會寫下為什麼。

R5 我寫得最硬。驗證回報疑點，Commander 唯一被允許的動作是送人工。它不能說「XBRL 說對了，所以 verifier 的疑點可以忽略」，至少現在不行。不同驗證結果之間怎麼取捨，是 Day 22 的題目，在那之前我寧可多送一些給人看。

然後是主迴圈：

import json
from collections import deque
from pathlib import Path


class Commander:
    def __init__(self, dispatch: Callable[[Task], TaskResult], log_path: Path,
                 rules: list[Rule] = RULES):
        self.dispatch = dispatch          # Day 21 的 subagent 註冊表會提供這個
        self.rules = rules
        self.queue: deque[Task] = deque()
        self.log = log_path.open("a", encoding="utf-8")

    def submit(self, tasks: list[Task]) -> None:
        self.queue.extend(tasks)

    def _over_budget(self, task: Task) -> bool:
        return task.attempt > MAX_ATTEMPTS.get(task.kind, 1)

    def _record(self, event: str, task: Task, **extra) -> None:
        self.log.write(json.dumps({"event": event, "task_id": task.task_id,
                                   "parent": task.parent_task_id, "kind": task.kind.value,
                                   "location": task.location, "attempt": task.attempt,
                                   "reason": task.reason, **extra},
                                  ensure_ascii=False, default=str) + "\n")

    def run(self) -> None:
        while self.queue:
            task = self.queue.popleft()
            if self._over_budget(task):
                escalated = _child(task, TaskKind.HUMAN_REVIEW,
                                   f"R0: {task.kind.value} 已試 {task.attempt - 1} 次，超出預算")
                self._record("escalate", task)
                self.queue.append(escalated)
                continue

            self._record("dispatch", task)
            result = self.dispatch(task)
            self._record("result", task, status=result.status.value,
                         signals=result.signals, model_id=result.model_id)

            fired = [rule for rule in self.rules if rule.when(task, result)]
            if not fired:
                self._record("no_rule", task, status=result.status.value)
                continue
            for rule in fired:
                children = rule.then(task, result)
                self._record("decide", task, rule_id=rule.rule_id,
                             children=[c.task_id for c in children])
                self.queue.extend(children)

有幾個地方是刻意的：

_record 在 dispatch 之前就寫。如果 subagent 當掉、整個程式崩了，至少 log 裡看得到「最後派出去的是哪個任務」。Day 3 那次 GB10 上 Ollama 跟 vLLM 搶記憶體直接 segfault，事後只能翻 journalctl 一點一點拼，那種感覺我不想再有一次。

no_rule 也會被記錄。規則表對不上的情況，就是 Commander 的盲區。這些紀錄累積起來，就是下一版規則表的草稿，也是將來那顆 LLM 顧問唯一該出場的地方。

多條規則同時觸發時全部執行。例如一個 OCR 結果可能同時符合 R3（重複）和 R4（完成、交 verifier）。我沒有設優先順序，因為這兩個任務不衝突，都做比較安全。真的會衝突的組合目前還沒遇到；遇到了再加，不先為想像中的情況設計。

拿 page 61 走一遍

用 Day 14、15 實際發生過的事當輸入，Commander 的決策紀錄大概會長這樣。訊號值是真實落檔裡的數字，但這份 trace 本身是我依規則手推的，不是程式跑出來的：

dispatch  t01  ocr_region    page_61 / 整頁          attempt=1  dpi=200
result    t01  failed        finish_reason=length, completion_tokens=3000, text_layer_chars=1822
decide    t01  R1 → t02      撞上限且文字層字數 1822 > 400，依版面切小
dispatch  t02  recrop        page_61
result    t02  ok            切出 N 片；裁罰段落那一片文字層約 296 字
dispatch  t03  ocr_region    page_61 / 文字 / 裁罰段落   attempt=1  dpi=400
result    t03  ok            finish_reason=stop, completion_tokens=277
decide    t03  R4 → t04      OCR 完成，交獨立 verifier
dispatch  t04  verify_text   page_61 / 文字 / 裁罰段落
result    t04  needs_attention  verifier 尚未接上真實模型
decide    t04  R5 → t06      verify_text 回報疑點，交人工，不自動修正
dispatch  t05  extract_citations  page_61 / 文字 / 裁罰段落
result    t05  ok            3/3 字號，各自關聯罰鍰金額

1822 是 page 61 整頁文字層的字數（pages_manifest.json 的 gt_char_count），296 是緊裁段落 GT 的字數，277 個 token 跟 stop 出自 step5_tight_crop_results.jsonl。

對照 Day 14 實際的路徑：整頁、切半、切半還是失敗、再手動抓座標緊裁。Commander 的版本跳過了「切半」，因為 R1 切的依據是版面上的 block 跟字數，不是頁面的幾何中線。Day 14 的切半是我當時想得不夠，規則表把這個教訓直接寫進去了。

t05 其實在拆解階段、文字層一掃到「字第…號」時就能排進佇列，跟 OCR 那條線平行，這裡為了好讀把它放在後面。t04 那個 verifier 任務，目前會落到一個還沒接上真實模型的 subagent。Day 11 算過 gpt-oss:20b 暫時沒有機器跑，所以它現在的實際行為是回 NEEDS_ATTENTION，然後觸發 R5 送人工。聽起來很蠢，但它是誠實的：在 verifier 真的能跑之前，所有模型輸出都該有人看過。