用 Jev 做文档分类：先拆分包，再判类型的两阶段管道

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: "把这份文档归入恰好一个类型",
    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% schema 错误。
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": "把这份文档归入恰好一个类型",
        "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:
    # 每页 ~100-500ms，仅输入计费 $0.042/M：一万包存量扫描
    # 不到一美元，完全不需要 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