MLX-LM 結構化 JSON：Schema、Pydantic 驗證與 Repair/Retry

科技觀點 - 本文屬於一個選集。

§ 31: 本文

一. 前言：能 `json.loads()`，不代表資料可以用
#

請本地模型從客服訊息擷取工單，畫面上得到這段：

{"category":"billing","priority":9,"summary":"重複扣款"}

它是合法 JSON，卻不一定是合法工單。 priority 也許只允許 1 到 5；category 也許只能來自固定集合；更麻煩的是，模型可能把「尚未確認」猜成肯定答案。所以拍拍君不把 structured output 當成單一功能，而是四層防線：

生成約束：解碼時只允許符合 JSON Schema 的 token。
資料驗證：Pydantic 檢查型別、範圍與跨欄位規則。
失敗政策：分清楚哪些錯可 repair、哪些該 retry、哪些必須隔離。
批次 QA：記錄每次嘗試、失敗原因與品質抽查結果。本文使用 MLX-LM + Outlines + Pydantic v2，在 Apple Silicon 上建立純文字擷取管線。如果你還不熟悉本機模型載入，可先讀MLX-LM 本地模型推論；若要處理圖片，請看另一篇 MLX-VLM 批次結構化擷取。

二. 先畫清楚資料邊界
#

我們要處理的是客服訊息，不是聊天回覆：

input text
   │
   ▼
constrained generation ──► raw JSON text
   │
   ▼
strict Pydantic validation
   ├── success ──► accepted.jsonl
   └── failure
        ├── safe repair
        ├── bounded retry
        └── quarantine.jsonl

這張圖有兩個重要觀念：

結構正確不等於內容正確。
失敗也是輸出，不能只印一行 error 就消失。生成約束負責讓括號、引號、欄位形狀正確； Pydantic 與 QA 則負責應用程式真正關心的契約。

三. 安裝與版本快照
#

建立一個乾淨專案：

uv init mlx-json-pipeline
cd mlx-json-pipeline
uv add mlx-lm "outlines[mlxlm]" pydantic rich

正式執行時請另外保存 mlx-lm、Outlines 與 Pydantic 版本；結構化生成牽涉 tokenizer、推論引擎與 grammar backend，版本是重要除錯證據。本文範例模型使用一個 instruct checkpoint：

MODEL_ID = "mlx-community/Qwen3-4B-Instruct-2507-4bit"

模型名稱只是範例。正式工作應固定 revision，並先用小型 golden set 驗證。

四. 用 Pydantic 寫出真正的輸出契約
#

先不要碰 prompt，先定義下游究竟接受什麼：

from typing import Literal
from pydantic import BaseModel, ConfigDict, Field, model_validator

class Ticket(BaseModel):
    model_config = ConfigDict(extra="forbid", strict=True)
    category: Literal["billing", "account", "bug", "other"]
    priority: int = Field(ge=1, le=5)
    summary: str = Field(min_length=5, max_length=120)
    needs_human: bool
    evidence: list[str] = Field(min_length=1, max_length=3)
    @model_validator(mode="after")
    def urgent_ticket_needs_human(self) -> "Ticket":
        if self.priority >= 4 and not self.needs_human:
            raise ValueError("priority >= 4 時 needs_human 必須為 true")
        return self

這裡故意開啟兩個嚴格選項：

extra="forbid"：模型多生一個看似合理的欄位，也要報錯。
strict=True：不把字串 "5" 悄悄轉成整數 5。欄位 description 也可說明 evidence 只能逐字摘錄；但它不是安全保證，「證據真的來自原文」仍要另外檢查。

五. 用 Outlines 約束 MLX-LM 解碼
#

Outlines 可以包住 MLX-LM 模型，將 Pydantic model 轉成結構化輸出限制：

import mlx_lm
import outlines

def load_generator(model_id: str):
    model, tokenizer = mlx_lm.load(model_id)
    return outlines.from_mlxlm(model, tokenizer)

generator = load_generator(MODEL_ID)

建立只做擷取、不做自由聊天的 prompt：

def build_prompt(text: str) -> str:
    return f"""
你是客服工單資料擷取器。
只使用 INPUT 明確提供的資訊，不可補寫不存在的事實。
priority 規則：
- 5：安全、帳務重大損失或帳號完全無法使用
- 4：重複扣款或核心功能中斷
- 3：明顯影響但有替代方式
- 2：一般問題
- 1：詢問或建議
evidence 必須逐字摘錄 INPUT，不能改寫。
INPUT:
<input>{text}</input>
""".strip()

呼叫時傳入 output_type=Ticket：

raw = generator(
    build_prompt("我昨天升級後被扣款兩次，訂單編號 A123，請協助。"),
    output_type=Ticket,
    max_tokens=240,
)
print(raw)

這不是「在 prompt 裡拜託模型回 JSON」。 Outlines 會在每一步解碼遮蔽不符合結構的 token，讓輸出維持 schema 形狀。但請注意目前的重要限制：MLX-LM 的 constrained generation 不支援 Outlines 的 batch 呼叫。需要大量處理時，可以逐筆執行或由工作佇列控制並行，不能直接假設 model.batch(..., output_type=Ticket) 可用。

六. 生成後仍要做嚴格驗證
#

拿到文字後，再由 Pydantic 驗證一次：

from pydantic import ValidationError

def validate_ticket(raw: str) -> Ticket:
    return Ticket.model_validate_json(raw, strict=True)

try:
    ticket = validate_ticket(raw)
except ValidationError as exc:
    print(exc.errors(include_url=False))
else:
    print(ticket.model_dump())

為什麼解碼已受約束，還要驗證？因為兩層關心的事情不同：

層級	能擋下	不保證
constrained decoding	壞掉的 JSON、錯誤欄位形狀、部分型別錯誤	摘要正確、證據存在、分類合理
Pydantic	範圍、enum、額外欄位、跨欄位規則	模型沒有幻想、業務判斷正確
QA	evidence 是否逐字存在、標註一致性、分布漂移	每一筆都由人閱讀
再補上 evidence 檢查：

def verify_evidence(ticket: Ticket, source: str) -> None:
    missing = [quote for quote in ticket.evidence if quote not in source]
    if missing:
        raise ValueError(f"evidence 不在原文：{missing!r}")

這個檢查很便宜，卻能阻止「看起來很像原文」的模型改寫。

七. Repair、Retry、Quarantine 要分家
#

很多管線把所有失敗都丟給「JSON repair」。這很危險。拍拍君建議用以下界線：

可以 deterministic repair
#

移除外層 Markdown code fence。
移除模型在 JSON 前後多說的固定前綴。
將已知舊版欄位名稱映射到新版名稱，而且有明確 migration 規格。

不該自動 repair
#

把 priority: 9 偷偷改成 5。
自動補一段原文不存在的 evidence。
猜測遺失的 category。
把 needs_human 從 false 改成 true，只為了通過 validator。前者是格式正規化；後者是在竄改模型判斷。在 constrained decoding 正常工作時，code fence repair 通常不該被觸發；它比較適合相容舊資料或非約束式 fallback 的輸出。

八. Retry 只帶必要錯誤，而且要有限次數
#

Retry 的目標是讓模型重新判斷，不是無限碰碰運氣。先把 Pydantic error 壓縮成不含內部堆疊的提示：

from pydantic import ValidationError

def compact_errors(exc: ValidationError) -> list[dict[str, str]]:
    rows = []
    for error in exc.errors(include_url=False, include_input=False):
        location = ".".join(str(part) for part in error["loc"])
        rows.append({"field": location, "message": error["msg"]})
    return rows

實作有上限的重試：

import json
from collections.abc import Callable

def extract_ticket(
    source: str,
    generate: Callable[..., str],
    max_attempts: int = 2,
) -> tuple[Ticket, list[dict]]:
    attempts: list[dict] = []
    feedback = ""
    for attempt_no in range(1, max_attempts + 1):
        prompt = build_prompt(source) + feedback
        raw = generate(prompt, output_type=Ticket, max_tokens=240)
        try:
            ticket = Ticket.model_validate_json(raw, strict=True)
            verify_evidence(ticket, source)
        except (ValidationError, ValueError) as exc:
            if isinstance(exc, ValidationError):
                errors = compact_errors(exc)
            else:
                errors = [{"field": "evidence", "message": str(exc)}]
            attempts.append(
                {"attempt": attempt_no, "status": "rejected", "errors": errors}
            )
            feedback = (
                "\n\n上一次輸出未通過驗證。請重新閱讀原文後完整重做；"
                "不要只修改錯誤欄位。錯誤摘要："
                + json.dumps(errors, ensure_ascii=False)
            )
            continue
        attempts.append({"attempt": attempt_no, "status": "accepted"})
        return ticket, attempts
    raise RuntimeError(json.dumps(attempts, ensure_ascii=False))

這裡刻意要求「完整重做」。若模型只局部修補，舊答案裡的錯誤推論很容易被保留下來。實務上通常 2 次已足夠；連續失敗應進 quarantine，交給人工或離線分析。

九. 批次處理：成功與失敗都寫入 JSONL
#

定義可追蹤的輸入：

from pydantic import BaseModel, ConfigDict

class InputRow(BaseModel):
    model_config = ConfigDict(extra="forbid", strict=True)
    id: str
    text: str

不要在 failure record 保存完整敏感內容；可保留雜湊與必要摘要：

import hashlib
import json
from pathlib import Path

def text_sha256(text: str) -> str:
    return hashlib.sha256(text.encode("utf-8")).hexdigest()

def append_jsonl(path: Path, record: dict) -> None:
    with path.open("a", encoding="utf-8") as handle:
        handle.write(json.dumps(record, ensure_ascii=False) + "\n")

逐筆跑，並把例外隔離：

def run_batch(rows: list[InputRow], generator) -> None:
    accepted_path = Path("accepted.jsonl")
    rejected_path = Path("quarantine.jsonl")
    for row in rows:
        try:
            ticket, attempts = extract_ticket(row.text, generator)
        except Exception as exc:
            append_jsonl(
                rejected_path,
                {
                    "id": row.id,
                    "source_sha256": text_sha256(row.text),
                    "status": "quarantined",
                    "error_type": type(exc).__name__,
                    "error": str(exc)[:1000],
                },
            )
            continue
        append_jsonl(
            accepted_path,
            {
                "id": row.id,
                "source_sha256": text_sha256(row.text),
                "status": "accepted",
                "attempts": attempts,
                "ticket": ticket.model_dump(mode="json"),
            },
        )

正式環境還應增加：

run_id、模型 ID、模型 revision。
mlx-lm、Outlines、Pydantic 版本。
prompt template 版本與 schema 版本。
latency、生成 token 數與 peak memory。
原始輸出的受限保存政策。沒有這些欄位，換模型後品質下降時只會剩下猜測。

十. QA 不只看「成功率」
#

至少追蹤這五組指標：

指標	意義	警訊
first-pass acceptance	第一次就通過的比例	突然下降，可能模型或 prompt 漂移
retry recovery	第二次救回的比例	太高代表第一版流程不穩
quarantine rate	最終隔離比例	直接影響人工成本
category distribution	類別分布	某類突然消失或暴增
evidence pass rate	引文逐字存在比例	下降代表模型開始改寫或幻想
統計之外，仍要固定抽樣人工複核：

每個 category 至少抽幾筆，而不是只做全體隨機抽樣。
priority 4、5 全數或高比例檢查。
retry 後才成功的資料提高抽樣率。
新模型上線前，先跑同一份 golden set。 Schema 只能保證輸出「長得對」；人工抽查才知道模型「理解得對不對」。

十一. 用 Fake Generator 測試失敗路徑
#

測試不必每次真的載入模型。先用可預測的 fake generator 驗證 retry 邊界：

class FakeGenerator:
    def __init__(self, outputs: list[str]) -> None:
        self.outputs = iter(outputs)
    def __call__(self, prompt: str, **kwargs) -> str:
        return next(self.outputs)

第一次 priority 超界，第二次成功：

def test_retry_then_accept() -> None:
    source = "我被重複扣款兩次，請協助退款。"
    fake = FakeGenerator(
        [
            '{"category":"billing","priority":9,'
            '"summary":"發生重複扣款","needs_human":true,'
            '"evidence":["重複扣款兩次"]}',
            '{"category":"billing","priority":4,'
            '"summary":"使用者回報重複扣款","needs_human":true,'
            '"evidence":["重複扣款兩次"]}',
        ]
    )
    ticket, attempts = extract_ticket(source, fake, max_attempts=2)
    assert ticket.priority == 4
    assert [item["status"] for item in attempts] == ["rejected", "accepted"]