Jev でドキュメント分類：分割してから判定する 2 段階パイプライン

const JEV_ENDPOINT = "https://api.typesafe.ai/v1/jev/evaluate";

// ポリシーはアプリケーションコードに置く。モデルは事実を報告するだけ。
const AUTO_FILE_CONFIDENCE = 0.85; // 0.85 以上 → 自動ファイリング
const REVIEW_MIN_CONFIDENCE = 0.6; // 0.60〜0.85 → レビュー、未満 → 人手

const QUESTIONS = {
  is_single_document: {
    type: "noul",
    instructions:
      "このページは新しい文書の始まりですか、前のページの文書の続きですか？",
  },
  needs_ocr: {
    type: "noul",
    instructions:
      "このページのテキストは欠落または判読不能で、信頼できる判定の前に OCR が必要ですか？",
  },
  doc_type: {
    type: "choice",
    instructions: "この文書をちょうど 1 つのタイプに分類する",
    criteria: {
      invoice: "明細請求：行項目、数量、合計、税、支払条件",
      contract: "拘束力のある合意：当事者、条項、期間、署名欄",
      receipt: "支払完了の証憑：支払済みマーク、取引参照番号、未払いなし",
      letter: "人同士の連絡文書：頭語、本文、署名ブロック",
      other: "上記のいずれも確信を持って当てはまらない",
    },
  },
} as const;

export type DocLane = "auto_file" | "review" | "human_review";

export async function evaluatePage(page: {
  text: string;
  page_index: number;
  source: string;
  has_ocr: boolean;
}) {
  const response = await fetch(JEV_ENDPOINT, {
    method: "POST",
    headers: { "Content-Type": "application/json" },
    body: JSON.stringify({ state: page, questions: QUESTIONS }),
  });
  if (!response.ok) throw new Error("Jev evaluate failed");
  const data = await response.json();

  const { is_single_document, needs_ocr, doc_type } = data;
  const lane: DocLane =
    needs_ocr.answer || doc_type.answer === "other"
      ? "human_review"
      : doc_type.confidence >= AUTO_FILE_CONFIDENCE
        ? "auto_file"
        : doc_type.confidence >= REVIEW_MIN_CONFIDENCE
          ? "review"
          : "human_review";

  return {
    page_index: page.page_index,
    starts_document: is_single_document.answer,
    boundary_confidence: is_single_document.confidence,
    needs_ocr: needs_ocr.answer,
    doc_type: doc_type.answer,
    doc_type_confidence: doc_type.confidence,
    lane,
  };
}

// 境界とタイプの答えは同じ呼び出しで返る。starts_document = true の
// ページが新しいセグメントを開き、残りは現在のセグメントに蓄積。
export function splitPacket(pages: Awaited<ReturnType<typeof evaluatePage>>[]) {
  const segments: (typeof pages)[] = [];
  for (const page of pages) {
    if (page.starts_document || segments.length === 0) segments.push([page]);
    else segments[segments.length - 1].push(page);
  }
  return segments;
}

// ページは互いに独立——本番では並列にファンアウト。9 ページのパケット
// は約 500ms 以内、入力トークンのみ課金、スキーマエラー 0%。
const results = await Promise.all(packetPages.map(evaluatePage));
const segments = splitPacket(results);
const reviewQueue = results.filter((r) => r.lane !== "auto_file");

import requests

JEV_ENDPOINT = "https://api.typesafe.ai/v1/jev/evaluate"
AUTO_FILE_CONFIDENCE = 0.85   # 0.85 以上 → 自動ファイリング
REVIEW_MIN_CONFIDENCE = 0.60  # 0.60〜0.85 → レビューキュー、未満 → 人手

QUESTIONS = {
    "is_single_document": {
        "type": "noul",
        "instructions": "このページは新しい文書の始まりですか、前のページの文書の続きですか？",
    },
    "needs_ocr": {
        "type": "noul",
        "instructions": "このページのテキストは欠落または判読不能で、信頼できる判定の前に OCR が必要ですか？",
    },
    "doc_type": {
        "type": "choice",
        "instructions": "この文書をちょうど 1 つのタイプに分類する",
        "criteria": {
            "invoice": "明細請求：行項目、数量、合計、税、支払条件",
            "contract": "拘束力のある合意：当事者、条項、期間、署名欄",
            "receipt": "支払完了の証憑：支払済みマーク、取引参照番号、未払いなし",
            "letter": "人同士の連絡文書：頭語、本文、署名ブロック",
            "other": "上記のいずれも確信を持って当てはまらない",
        },
    },
}

def evaluate_page(page: dict) -> dict:
    resp = requests.post(
        JEV_ENDPOINT,
        json={"state": page, "questions": QUESTIONS},
        timeout=5,
    )
    resp.raise_for_status()
    data = resp.json()
    doc = data["doc_type"]

    if data["needs_ocr"]["answer"] or doc["answer"] == "other":
        lane = "human_review"
    elif doc["confidence"] >= AUTO_FILE_CONFIDENCE:
        lane = "auto_file"
    elif doc["confidence"] >= REVIEW_MIN_CONFIDENCE:
        lane = "review"
    else:
        lane = "human_review"

    return {
        "page_index": page["page_index"],
        "starts_document": data["is_single_document"]["answer"],
        "doc_type": doc["answer"],
        "confidence": doc["confidence"],
        "lane": lane,
    }

def classify_packet(pages: list[dict]) -> list[list[dict]]:
    # ページは互いに独立——本番では並列にファンアウト。
    results = [evaluate_page(p) for p in pages]
    segments: list[list[dict]] = []
    for page in results:
        if page["starts_document"] or not segments:
            segments.append([page])
        else:
            segments[-1].append(page)
    return segments

def scan_inbound_queue(packets: list[list[dict]]) -> dict:
    # 1 ページ約 100〜500ms、入力のみ課金 $0.042/M：1 万パケットの
    # 既存アーカイブスキャンは 1 ドル未満、GPU 不要。
    lanes: dict[str, int] = {}
    for packet in packets:
        for segment in classify_packet(packet):
            lane = segment[0]["lane"]
            lanes[lane] = lanes.get(lane, 0) + 1
    return lanes