homelab/scripts/paperless-eval/eval.py

#!/usr/bin/env python3
"""Measure what Paperless suggests for documents whose labels a human already chose.

Runs both engines against the same documents:

    /api/documents/<id>/suggestions/      the built-in scikit-learn classifier
    /api/documents/<id>/ai_suggestions/   the LLM path, with RAG

WHAT THIS ANSWERS, AND WHAT AN EARLIER VERSION DID NOT

The two engines are not alternatives — the classifier costs 0.6 s per document
and keeps running whatever the LLM does. So the useful question is not "which
scores higher" but "what does the LLM add where the classifier is wrong", and
that cannot be recovered from two aggregate percentages: 74% and 72% could
overlap almost entirely or barely at all, and the difference between those two
worlds is the entire value of running an LLM.

So every document is written out individually, with both engines' verdicts, and
the summary is derived from that rather than counted along the way.

SCORING

Correspondent and document type are each scored twice:

    offered      the true label's id came back in `correspondents` /
                 `document_types` — what the UI actually presents
    recognised   also counting the true label's name appearing wordwise in
                 `suggested_correspondents` / `suggested_document_types`

The gap between them is one specific failure: the model names the right entity
but paperless_ai.matching cannot resolve it, because difflib runs at
MATCH_THRESHOLD = 0.8 and a correct but more verbose answer falls below it —
"Generali Deutschland Krankenversicherung AG" against the existing "Generali
AG". Reporting only the first blames the model for a matcher problem; only the
second claims a usefulness the UI does not deliver.

Document type is additionally split on whether the true type is "Sonstige",
which holds 1133 of ~3900 documents and is the one label with no semantic
content — a restaurant reservation lands there because nothing else fits. The
classifier can memorise that; a model reading the document has to infer "none
of the above". Scored separately rather than dropped, because the model does
sometimes get it right.

GROUND TRUTH

The current filing, which deserves care: over ten years conventions change, and
a disagreement with an old document is not automatically an error. Treat the
comparison between engines as the result and the absolute numbers as
indicative.

Env:
    PAPERLESS_URL     default http://127.0.0.1:8000
    PGHOST/PGDATABASE standard libpq, for ground truth and the API token
    EVAL_SAMPLE       documents to score, default 50
    EVAL_ENGINES      comma-separated subset of classic,ai — default both
    EVAL_LABEL        free-text name for this configuration, recorded in the
                      output so results from different jobs stay distinguishable
    EVAL_OUT          directory for results, default /results
"""

from __future__ import annotations

import hashlib
import json
import os
import re
import secrets
import sys
import time
import urllib.error
import urllib.request

import psycopg

PAPERLESS_URL = os.environ.get("PAPERLESS_URL", "http://127.0.0.1:8000")
SAMPLE = int(os.environ.get("EVAL_SAMPLE", "50"))
ENGINES = os.environ.get("EVAL_ENGINES", "classic,ai").split(",")
LABEL = os.environ.get("EVAL_LABEL", "unlabelled")
OUT_DIR = os.environ.get("EVAL_OUT", "/results")

# A backstop, not the real bound. Generation is capped server-side by
# --n-predict on the llama-server sidecar, which is what actually stops a model
# that will not terminate; this catches everything else — a wedged sidecar, a
# request that never reaches it.
#
# 300 s against the ~73 s a healthy document takes end to end, prompt included.
# It does work: llama-server logs `stop: cancel task` on the disconnect and
# starts the next request 0.4 s later.
REQUEST_TIMEOUT = 300

CATCHALL_TYPE = "Sonstige"


def log(msg: str) -> None:
    print(f"[eval] {msg}", flush=True)


def api_token(conn) -> str:
    """A token for this run, created in the database.

    The API needs one and this database is a disposable clone, so the run
    issues its own against a superuser rather than depending on a token that
    happened to come across in the restore.
    """
    token = secrets.token_hex(20)
    with conn.cursor() as cur:
        cur.execute(
            "select id from auth_user where is_superuser and is_active order by id limit 1",
        )
        row = cur.fetchone()
        if row is None:
            raise SystemExit("no active superuser in the restored database")
        user_id = row[0]
        cur.execute("delete from authtoken_token where user_id = %s", (user_id,))
        cur.execute(
            "insert into authtoken_token (key, created, user_id) values (%s, now(), %s)",
            (token, user_id),
        )
    conn.commit()
    return token


def sample(conn) -> list[dict]:
    """The documents to score, ordered by md5 of the id.

    Deterministic, so every engine and every configuration sees the same
    documents and results stay comparable across jobs. Only documents carrying
    both labels are usable as ground truth.
    """
    with conn.cursor() as cur:
        cur.execute(
            """
            select d.id, d.correspondent_id, d.document_type_id, c.name, t.name
              from documents_document d
              join documents_correspondent c on c.id = d.correspondent_id
              join documents_documenttype  t on t.id = d.document_type_id
             where d.correspondent_id is not null
               and d.document_type_id is not null
               and length(d.content) > 400
             order by md5(d.id::text)
             limit %s
            """,
            (SAMPLE,),
        )
        return [
            {
                "id": r[0],
                "correspondent_id": r[1],
                "document_type_id": r[2],
                "correspondent_name": r[3],
                "document_type_name": r[4],
            }
            for r in cur.fetchall()
        ]


def floor(conn) -> float:
    """Always guessing the most frequent correspondent.

    Without it a number like "70% correct" cannot be read: one correspondent
    holds several hundred of ~3900 documents.
    """
    with conn.cursor() as cur:
        cur.execute(
            """
            select max(c), sum(c) from (
                select count(*) c from documents_document
                 where correspondent_id is not null group by correspondent_id
            ) x
            """,
        )
        top, total = cur.fetchone()
    # float() on both: psycopg returns Postgres numerics as Decimal, and
    # mixing that with a float literal raises TypeError rather than coercing.
    return 100.0 * float(top) / float(total) if total else 0.0


def fetch(path: str, token: str) -> tuple[dict | None, str | None, float]:
    req = urllib.request.Request(
        f"{PAPERLESS_URL}{path}",
        headers={"Authorization": f"Token {token}"},
    )
    start = time.monotonic()
    try:
        with urllib.request.urlopen(req, timeout=REQUEST_TIMEOUT) as resp:
            body = resp.read()
        return json.loads(body), None, time.monotonic() - start
    except urllib.error.HTTPError as err:
        return None, f"HTTP {err.code}: {err.read()[:200]!r}", time.monotonic() - start
    except Exception as err:  # noqa: BLE001 — a failed document is data, not a crash
        return None, f"{type(err).__name__}: {err}", time.monotonic() - start


def normalise(text: str) -> set[str]:
    return {w for w in re.split(r"[^\w]+", text.lower()) if w}


def recognised(payload: dict, field: str, want_name: str) -> bool:
    """True when a suggested name contains every word of the true label's name.

    Handles the observed failure directly: the model answers with a superset of
    the stored name, which difflib rejects at 0.8 but a reader would accept.
    """
    want = normalise(want_name)
    return any(want <= normalise(name) for name in payload.get(field) or [])


def degenerate(payload: dict) -> str | None:
    """Whether this looks like broken inference rather than a wrong guess.

    Past roughly 14k prompt tokens this stack stops producing language — first
    empty completions, then runs of digits and CJK characters — and one such
    request poisons the llama.cpp slot until the pod restarts. Nothing in the
    HTTP response says so: status 200, finish_reason "stop", well-formed JSON.

    Scored as a wrong answer it would silently depress a candidate's numbers,
    and because the slot stays poisoned it would take the rest of the run with
    it. Flagged instead, so the failure is visible as itself.
    """
    title = payload.get("title") or ""
    if not title.strip():
        return "empty title"
    # Digits, punctuation and whitespace are normal in a document title —
    # "Rechnung 2025-5425541" is 8 letters in 21 characters and tripped an
    # earlier version of this check. What broken inference produces is
    # characters from other scripts, so look for those directly rather than
    # inferring them from a shortage of Latin ones.
    foreign = sum(c.isalpha() and not (c.isascii() or c in "äöüÄÖÜßáéèêíóúàçñ") for c in title)
    if foreign > 0.2 * len(title):
        return f"non-latin title: {title[:40]!r}"
    return None


def score_document(payload: dict, doc: dict) -> dict:
    offered_corr = doc["correspondent_id"] in (payload.get("correspondents") or [])
    offered_type = doc["document_type_id"] in (payload.get("document_types") or [])
    return {
        "correspondent_offered": offered_corr,
        "correspondent_recognised": offered_corr
        or recognised(payload, "suggested_correspondents", doc["correspondent_name"]),
        "type_offered": offered_type,
        "type_recognised": offered_type
        or recognised(payload, "suggested_document_types", doc["document_type_name"]),
        "suggested_correspondents": payload.get("suggested_correspondents") or [],
        "suggested_document_types": payload.get("suggested_document_types") or [],
        "title": payload.get("title"),
    }


def run(engine: str, suffix: str, docs: list[dict], token: str) -> list[dict]:
    log(f"=== {engine}: {len(docs)} documents")
    rows = []
    for n, doc in enumerate(docs, 1):
        payload, error, elapsed = fetch(
            f"/api/documents/{doc['id']}/{suffix}",
            token,
        )
        row = {
            "engine": engine,
            "document_id": doc["id"],
            "seconds": round(elapsed, 1),
            "truth_correspondent": doc["correspondent_name"],
            "truth_type": doc["document_type_name"],
            "catchall_type": doc["document_type_name"] == CATCHALL_TYPE,
        }
        if payload is None:
            row["error"] = error
            log(f"  {doc['id']}: {error}")
        else:
            row.update(score_document(payload, doc))
            # Only the AI path returns a title; the classifier answers with ids
            # alone, so an "empty title" there is normal rather than broken.
            if engine == "ai" and (broken := degenerate(payload)) is not None:
                row["degenerate"] = broken
                log(f"  {doc['id']}: DEGENERATE — {broken}")
        rows.append(row)
        if n % 10 == 0:
            log(f"  {n}/{len(docs)}")
    return rows


def pct(hits: int, total: int) -> float:
    return round(100.0 * hits / total, 1) if total else 0.0


def summarise(rows: list[dict], engine: str) -> dict:
    mine = [r for r in rows if r["engine"] == engine]
    # Degenerate rows are excluded from the accuracy figures: they measure the
    # backend giving up, not the model choosing wrongly, and leaving them in
    # would quietly punish whichever candidate happened to trip the slot.
    scored = [r for r in mine if "error" not in r and "degenerate" not in r]
    failed = [r for r in mine if "error" in r]
    broken = [r for r in mine if "degenerate" in r]
    specific = [r for r in scored if not r["catchall_type"]]
    catchall = [r for r in scored if r["catchall_type"]]
    return {
        "engine": engine,
        "scored": len(scored),
        "failed": len(failed),
        # Non-zero means the run is suspect from that point on, not that the
        # model is bad — the slot stays poisoned until the pod restarts.
        "degenerate": len(broken),
        "seconds_total": round(sum(r["seconds"] for r in mine)),
        "correspondent_offered": pct(
            sum(r["correspondent_offered"] for r in scored), len(scored)
        ),
        "correspondent_recognised": pct(
            sum(r["correspondent_recognised"] for r in scored), len(scored)
        ),
        "type_offered": pct(sum(r["type_offered"] for r in scored), len(scored)),
        "type_specific": pct(
            sum(r["type_offered"] for r in specific), len(specific)
        ),
        "type_catchall": pct(
            sum(r["type_offered"] for r in catchall), len(catchall)
        ),
    }


def complement(rows: list[dict]) -> dict | None:
    """What the LLM adds where the classifier is wrong, and what it costs.

    The reason this script writes per-document rows at all. Only computed when
    both engines ran, and only over documents both actually scored.
    """
    by_engine = {"classic": {}, "ai": {}}
    for row in rows:
        usable = "error" not in row and "degenerate" not in row
        if usable and row["engine"] in by_engine:
            by_engine[row["engine"]][row["document_id"]] = row
    both = set(by_engine["classic"]) & set(by_engine["ai"])
    if not both or not by_engine["ai"]:
        return None

    rescued = missed = broken = agreed = 0
    for doc_id in both:
        classic_hit = by_engine["classic"][doc_id]["correspondent_offered"]
        ai_hit = by_engine["ai"][doc_id]["correspondent_offered"]
        if classic_hit and ai_hit:
            agreed += 1
        elif not classic_hit and ai_hit:
            rescued += 1
        elif classic_hit and not ai_hit:
            broken += 1
        else:
            missed += 1
    return {
        "documents": len(both),
        "both_right": agreed,
        "llm_rescued": rescued,
        "llm_only_wrong": broken,
        "both_wrong": missed,
    }


def main() -> int:
    os.makedirs(OUT_DIR, exist_ok=True)
    with psycopg.connect() as conn:
        token = api_token(conn)
        docs = sample(conn)
        guess_floor = floor(conn)
    log(f"label={LABEL} documents={len(docs)} floor={guess_floor:.1f}%")

    rows: list[dict] = []
    for engine, suffix in (("classic", "suggestions/"), ("ai", "ai_suggestions/")):
        if engine in ENGINES:
            rows.extend(run(engine, suffix, docs, token))

    result = {
        "label": LABEL,
        "sample": len(docs),
        "floor": round(guess_floor, 1),
        "summaries": [summarise(rows, e) for e in ENGINES if e in ("classic", "ai")],
        "complement": complement(rows),
    }

    with open(f"{OUT_DIR}/{LABEL}.jsonl", "w") as handle:
        for row in rows:
            handle.write(json.dumps(row, ensure_ascii=False) + "\n")
    with open(f"{OUT_DIR}/{LABEL}.summary.json", "w") as handle:
        json.dump(result, handle, ensure_ascii=False, indent=2)

    # Also to stdout, so the numbers are in the pod log and reachable via Loki
    # without mounting the volume.
    log(json.dumps(result, ensure_ascii=False, indent=2))
    return 0


if __name__ == "__main__":
    sys.exit(main())