Document Classification with Jev: A Two-Stage Packet Pipeline

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

// Policy lives in application code — the model only reports facts.
const AUTO_FILE_CONFIDENCE = 0.85; // >= 0.85 -> file automatically
const REVIEW_MIN_CONFIDENCE = 0.6; // 0.60-0.85 -> review, below -> human

const QUESTIONS = {
  is_single_document: {
    type: "noul",
    instructions:
      "Does this page start a new document, or continue the document on the previous page?",
  },
  needs_ocr: {
    type: "noul",
    instructions:
      "Is this page's text missing or too garbled to decide on, such that OCR must run first?",
  },
  doc_type: {
    type: "choice",
    instructions: "Classify this document into exactly one type",
    criteria: {
      invoice: "Itemized billing: line items, quantities, totals, tax, payment terms",
      contract: "Binding agreement: parties, clauses, term dates, signature blocks",
      receipt: "Proof of completed payment: paid marker, transaction reference, nothing due",
      letter: "Human correspondence: salutation, body prose, signature block",
      other: "None of the above matches with confidence",
    },
  },
} 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,
  };
}

// Boundary and type answers come back in the same call: pages where
// starts_document = true open a new segment; the rest accumulate.
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;
}

// Pages are independent — fan them out concurrently. A 9-page packet
// stays under ~500ms wall time, input tokens only, 0% schema errors.
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 -> file automatically
REVIEW_MIN_CONFIDENCE = 0.60  # 0.60-0.85 -> review queue, below -> human

QUESTIONS = {
    "is_single_document": {
        "type": "noul",
        "instructions": "Does this page start a new document, or continue the document on the previous page?",
    },
    "needs_ocr": {
        "type": "noul",
        "instructions": "Is this page's text missing or too garbled to decide on, such that OCR must run first?",
    },
    "doc_type": {
        "type": "choice",
        "instructions": "Classify this document into exactly one type",
        "criteria": {
            "invoice": "Itemized billing: line items, quantities, totals, tax, payment terms",
            "contract": "Binding agreement: parties, clauses, term dates, signature blocks",
            "receipt": "Proof of completed payment: paid marker, transaction reference, nothing due",
            "letter": "Human correspondence: salutation, body prose, signature block",
            "other": "None of the above matches with confidence",
        },
    },
}

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]]:
    # Pages are independent — fan out concurrently in production.
    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 per page, input-only pricing at $0.042/M: a
    # 10,000-packet back-catalog scan costs about a dollar, no GPUs.
    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