一. 前言：一張圖能回答，五百張圖呢？ #
把一張圖片交給本地 VLM，問「畫面裡有什麼」，很有趣。
但真正的資料工作通常長這樣：
- 替商品圖產生短 caption
- 從收據或表單擷取欄位
- 將資料集依內容分類
- 找出模型看不懂、需要人工複核的圖片
- 升級模型後重跑，確認結果沒有退步
這時候，漂亮回答不是終點。 你需要的是可重跑、可驗證、可續跑的批次管線。
今天拍拍君會用 mlx-vlm 的本機 server 做一條小型生產線：
manifest -> image request -> JSON Schema -> Pydantic
-> results.jsonl / quarantine.jsonl -> quality report
如果你還沒跑過單張圖片問答，先看 MLX-VLM 本地圖片問答。 如果你要處理純文字資料集，則看 MLX-LM 批次推論。
本篇不重講互動聊天，而是處理「很多張圖進來後，怎麼可靠地留下結構化結果」。
二. 先定義輸出契約，不要先寫迴圈 #
假設我們要整理一批一般物件照片。 每張圖都輸出：
| 欄位 | 意義 | 約束 |
|---|---|---|
caption |
一句客觀描述 | 1 到 160 字 |
category |
主要類別 | 固定 enum |
objects |
可見物件 | 最多 12 個 |
needs_review |
是否要人工看 | boolean |
review_reason |
複核原因 | 最多 120 字 |
這不是為了讓 JSON 看起來整齊。 契約會直接決定後面的驗證、統計與人工工作量。
先建立 Pydantic model：
from typing import Literal
from pydantic import BaseModel, ConfigDict, Field, model_validator
class ImageResult(BaseModel):
model_config = ConfigDict(extra="forbid")
caption: str = Field(min_length=1, max_length=160)
category: Literal["document", "food", "indoor", "outdoor", "other"]
objects: list[str] = Field(max_length=12)
needs_review: bool
review_reason: str = Field(max_length=120)
@model_validator(mode="after")
def review_has_reason(self) -> "ImageResult":
if self.needs_review and not self.review_reason.strip():
raise ValueError("needs_review=true 時必須提供原因")
return self
extra="forbid" 很重要。 模型若多生一個未定義欄位，我們寧可明確失敗，也不要讓下游默默猜意思。
三. 安裝與啟動本機 MLX-VLM Server #
本篇使用 Apple Silicon、Python 3.10 以上與 uv：
mkdir pypy-vlm-batch
cd pypy-vlm-batch
uv init --python 3.12
uv add mlx-vlm openai pydantic pillow
把模型 ID 放進環境變數：
export PYPY_VLM_MODEL="mlx-community/Qwen2-VL-2B-Instruct-4bit"
再啟動只監聽本機的 server：
uv run mlx_vlm.server \
--model "$PYPY_VLM_MODEL" \
--host 127.0.0.1 \
--port 8080
另一個 terminal 先確認服務可用：
curl --fail http://127.0.0.1:8080/health
curl --fail http://127.0.0.1:8080/v1/models
server 路徑的優點不只是介面熟悉。 目前 MLX-VLM server 會讓圖片請求各自完成 vision prefill，再加入共享 decoding batch。 批次 runner 不必直接碰內部 generation API，也比較容易替換 client。
四. 用 Manifest 管理輸入，不要相信檔名就是資料庫 #
建立 manifest.jsonl：
{"id":"img-001","path":"images/desk.jpg","source":"camera-a"}
{"id":"img-002","path":"images/lunch.png","source":"camera-a"}
{"id":"img-003","path":"images/receipt.jpg","source":"scanner-b"}
一行一筆的好處是容易 append、diff 與續跑。 每筆至少要有穩定 id 與相對路徑。
讀取時立刻檢查重複 ID 和不存在的圖片：
import json
from pathlib import Path
from typing import Any
def read_manifest(path: Path) -> list[dict[str, Any]]:
rows: list[dict[str, Any]] = []
seen: set[str] = set()
for line_no, line in enumerate(
path.read_text(encoding="utf-8").splitlines(), start=1
):
if not line.strip():
continue
row = json.loads(line)
item_id = str(row["id"])
image_path = Path(row["path"])
if item_id in seen:
raise ValueError(f"{path}:{line_no} duplicate id: {item_id}")
if not image_path.is_file():
raise FileNotFoundError(f"{path}:{line_no} missing: {image_path}")
seen.add(item_id)
rows.append(row)
return rows
批次工作最討厭跑到第 487 張才發現第 3 張的路徑早就錯了。 能在送模型前發現的錯，就不要留到推論階段。
五. 先做 Caption，再讓 JSON Schema 約束格式 #
建立 client：
import os
from openai import OpenAI
MODEL_ID = os.environ["PYPY_VLM_MODEL"]
client = OpenAI(
base_url="http://127.0.0.1:8080/v1",
api_key="not-needed",
timeout=120.0,
)
接著把 Pydantic schema 交給 response_format：
from pathlib import Path
TASK_PROMPT = """
只描述圖片中直接可見的內容，不猜人物身份、地點或畫面外資訊。
caption 使用繁體中文，簡短客觀。
objects 使用短名詞並去除重複。
若圖片模糊、資訊不足或類別不確定，needs_review 設為 true 並說明原因。
""".strip()
def analyze_image(image_path: Path) -> ImageResult:
response = client.chat.completions.create(
model=MODEL_ID,
messages=[
{
"role": "user",
"content": [
{"type": "text", "text": TASK_PROMPT},
{
"type": "image_url",
"image_url": {"url": str(image_path.resolve())},
},
],
}
],
response_format={
"type": "json_schema",
"json_schema": {
"name": "ImageResult",
"strict": True,
"schema": ImageResult.model_json_schema(),
},
},
temperature=0.0,
max_tokens=320,
)
content = response.choices[0].message.content
if not content:
raise ValueError("模型回傳空內容")
return ImageResult.model_validate_json(content)
這裡有兩層防線：
- server 依 JSON Schema 限制生成。
- client 再用 Pydantic 驗證語意規則。
Schema 能阻止欄位形狀亂掉，但不能保證 caption 與圖片相符。 格式正確與內容正確，是兩件不同的事。
六. 把成功與失敗都寫進 JSONL #
批次程式不能遇到一張壞圖就整批消失。 建立 append helper：
from datetime import UTC, datetime
def append_jsonl(path: Path, value: dict[str, Any]) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
with path.open("a", encoding="utf-8") as file:
file.write(json.dumps(value, ensure_ascii=False) + "\n")
def now_iso() -> str:
return datetime.now(UTC).isoformat()
主迴圈分成成功檔與隔離檔：
RESULTS = Path("outputs/results.jsonl")
QUARANTINE = Path("outputs/quarantine.jsonl")
def run_one(row: dict[str, Any]) -> None:
try:
result = analyze_image(Path(row["path"]))
append_jsonl(RESULTS, {
"id": row["id"],
"path": row["path"],
"source": row.get("source"),
"model": MODEL_ID,
"processed_at": now_iso(),
"result": result.model_dump(),
})
except Exception as exc:
append_jsonl(QUARANTINE, {
"id": row.get("id"),
"path": row.get("path"),
"failed_at": now_iso(),
"error_type": type(exc).__name__,
"error": str(exc),
})
不要只寫 failed。 保留錯誤型別與訊息，才能區分路徑錯誤、HTTP timeout、Schema 問題或模型空回應。
七. 續跑機制：已完成的 ID 不要再花一次時間 #
批次任務可能因為關機、更新或模型記憶體壓力中斷。 先讀出成功結果中的 ID：
def completed_ids(path: Path) -> set[str]:
if not path.exists():
return set()
done: set[str] = set()
for line in path.read_text(encoding="utf-8").splitlines():
if line.strip():
done.add(str(json.loads(line)["id"]))
return done
整合成可續跑的 main()：
def main() -> None:
rows = read_manifest(Path("manifest.jsonl"))
done = completed_ids(RESULTS)
pending = [row for row in rows if str(row["id"]) not in done]
print(f"total={len(rows)} done={len(done)} pending={len(pending)}")
for index, row in enumerate(pending, start=1):
print(f"[{index}/{len(pending)}] {row['id']}")
run_one(row)
if __name__ == "__main__":
main()
穩定 ID 是續跑的核心。 不要拿「第幾行」當 ID；manifest 一排序，身份就全變了。
若同一張圖片要用新模型重跑，結果 key 應升級成 (id, model, pipeline_version)，不能只看 id。
八. Retry 要挑錯誤，不是所有失敗都重送 #
連線暫斷與 timeout 可以重試；Pydantic 驗證失敗則應先保留輸出、檢查契約。
簡單的 bounded retry 可以這樣寫：
import time
from openai import APIConnectionError, APITimeoutError
RETRYABLE = (APIConnectionError, APITimeoutError)
def analyze_with_retry(image_path: Path, attempts: int = 3) -> ImageResult:
for attempt in range(1, attempts + 1):
try:
return analyze_image(image_path)
except RETRYABLE:
if attempt == attempts:
raise
time.sleep(2 ** (attempt - 1))
raise AssertionError("unreachable")
三個原則：
- 重試次數有上限。
- 只重試暫時性錯誤。
- 每次失敗最後仍要進 quarantine。
「再試一次」不是資料品質策略。 同一張模糊圖片送十次，只會得到十種自信的猜法。
九. 批次不等於無限併發 #
MLX-VLM server 支援 continuous batching，但 client 仍不該一次塞進全部圖片。
可以先從很小的 worker 數開始：
from concurrent.futures import ThreadPoolExecutor, as_completed
def run_pending(rows: list[dict[str, Any]], workers: int = 2) -> None:
with ThreadPoolExecutor(max_workers=workers) as pool:
futures = {pool.submit(run_one, row): row for row in rows}
for future in as_completed(futures):
row = futures[future]
try:
future.result()
except Exception as exc:
print(f"unexpected failure {row['id']}: {exc}")
workers=2 不是宇宙真理，只是安全起點。 實際值要用自己的模型、圖片尺寸、記憶體與輸出長度測量。
觀察：
- 每張圖的 p50 / p95 延遲
- 峰值記憶體
- server queue 長度
- 每分鐘完成數
- timeout 與 quarantine 比例
吞吐量上升但失敗率翻倍，不叫最佳化。
另外，官方文件目前明確指出：結構化輸出不能和 speculative decoding 同時使用。 這條管線優先要穩定 Schema，所以不要硬把兩者湊在一起。
十. 品質驗證：合法 JSON 也可能是在胡說 #
完成後先做機械式檢查：
from collections import Counter
def load_results(path: Path) -> list[dict[str, Any]]:
return [
json.loads(line)
for line in path.read_text(encoding="utf-8").splitlines()
if line.strip()
]
rows = load_results(RESULTS)
categories = Counter(row["result"]["category"] for row in rows)
review_count = sum(row["result"]["needs_review"] for row in rows)
print("categories:", categories)
print("review_rate:", review_count / max(len(rows), 1))
接著做人工抽查，而且不要只抽「看起來正常」的結果。
至少分層抽：
- 每個 category 各幾張。
needs_review=true的圖片。- caption 特別短或特別長的圖片。
- objects 數量為 0 或碰到上限的圖片。
- 新來源、新相機或新資料批次。
若人工抽查發現錯誤，要記錄錯誤類型：漏看、幻覺、分類錯、格式雖合法但語意矛盾。 只有這樣，下一版 prompt 或 model 才有可比較的基準。
十一. 用 Golden Set 做回歸測試 #
從資料中挑 20 到 50 張具代表性的圖片，建立 golden.jsonl：
{"id":"img-001","must_include":["筆電","杯子"],"category":"indoor"}
{"id":"img-003","must_include":["收據"],"category":"document"}
最小回歸檢查：
def check_golden(
result: ImageResult,
expected: dict[str, Any],
) -> list[str]:
problems: list[str] = []
if result.category != expected["category"]:
problems.append("category mismatch")
combined = " ".join([result.caption, *result.objects])
for keyword in expected.get("must_include", []):
if keyword not in combined:
problems.append(f"missing keyword: {keyword}")
return problems
這不是完整的視覺評測，但能抓到最明顯的退步。 升級 mlx-vlm、更換 checkpoint、改 prompt 或改 schema 前後，都跑同一批 golden set。
十二. 可重現性與隱私清單 #
結果檔至少保留：
input id
image relative path
image checksum
model id / revision
mlx-vlm version
prompt version
schema version
generation parameters
processed_at
圖片可能含個資時，再加上這些邊界：
- server 只綁
127.0.0.1 - 不在 log 印出完整 prompt 與敏感路徑
- quarantine 設定保存期限
- 暫存縮圖用完即清理
- 不把原圖與結果誤推到公開 repo
- 遠端圖片 URL 視為外部網路請求
本機推論降低了資料外送，但不會自動替你做好權限、日誌與檔案生命週期。
十三. 常見問題 #
- HTTP 連得到，結果卻是空的： 記錄 request ID 與 server log，並限制
max_tokens不要太小。 - Schema 一直失敗： 先縮小欄位、enum 與巢狀層級，確認模型能完成最小契約。
- 圖片很多時突然變慢： 降低 client worker、限制圖片尺寸，分開量 prefill 與 decode。
- 結果重複： 檢查 manifest ID、續跑 key 與 append 邏輯。
- 某一類全部進 quarantine： 這通常是資料分布或 prompt 問題，不要只加 retry。
- 想直接得到自由文字 Caption： 可以拿掉 Schema，但仍建議保存穩定 ID、模型版本與失敗紀錄。
結語 #
批次圖片推論的難點，不是寫出 for image in images。
真正要做的是把每一張圖片變成可追蹤的工作單位：有穩定 ID、有明確 Schema、有成功結果、有隔離失敗，也能在中斷後繼續。
mlx-vlm 讓 Apple Silicon 上的多模態推論變得容易；OpenAI-compatible server 與 JSON Schema 則讓它更容易接進一般 Python 工程流程。
先用十張圖片驗證契約，再擴到一百張。 先量品質與失敗率，再調高併發。
模型負責看圖，工程負責不讓結果失憶。這樣才是一條拍拍君敢明天再跑的管線。🔭