Spaces:
Sleeping
Sleeping
File size: 4,268 Bytes
9cacfae | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 | #!/usr/bin/env python3
"""Accuracy check for the docqa service against corpus ground truth.
python bench/accuracy.py --corpus <dir> [--limit N] [--base URL]
Every number printed is measured from a real run against a running server.
Nothing here is estimated.
"""
from __future__ import annotations
import argparse
import json
import statistics
import subprocess
import time
from pathlib import Path
# Requested key -> ground-truth field in corpus/ground_truth.json
KEY_MAP = [
("INVOICE NO", "invoice_number"),
("Invoice Date", "invoice_date"),
("Due Date", "due_date"),
("PO Number", "po_number"),
("Payment Term", "payment_terms"),
("Currency", "currency"),
("Vendor Name", "vendor_name"),
("Vendor Address", "vendor_address"),
("Vendor Id", "vendor_tax_id"),
("GSTIN", "gstin"),
("Bill To", "bill_to"),
("Subtotal", "subtotal"),
("Tax Amount", "tax_amount"),
("Total Amount", "total"),
("Tax Label", "tax_label"),
]
def extract(base: str, pdf: Path, keys: list[str]) -> tuple[dict, float]:
started = time.perf_counter()
proc = subprocess.run(
["curl.exe", "-sS", "-X", "POST", f"{base}/v1/extract",
"-F", f"file=@{pdf}", "-F", "keys=" + ",".join(keys)],
capture_output=True, text=True,
)
wall = (time.perf_counter() - started) * 1000
if proc.returncode != 0:
raise RuntimeError(proc.stderr.strip() or "curl failed")
return json.loads(proc.stdout), wall
def main() -> int:
ap = argparse.ArgumentParser()
ap.add_argument("--base", default="http://127.0.0.1:7860")
ap.add_argument("--corpus", type=Path, required=True)
ap.add_argument("--limit", type=int, default=1)
args = ap.parse_args()
manifest = json.loads((args.corpus / "ground_truth.json").read_text())
records = manifest[: args.limit] if args.limit else manifest
keys = [k for k, _ in KEY_MAP]
print(f"target {args.base}")
print(f"model impira/layoutlm-invoices keys={len(keys)} "
f"documents={len(records)}\n")
per_key: dict[str, list[int]] = {k: [] for k in keys}
latencies: list[float] = []
for record in records:
body, wall = extract(args.base, Path(record["path"]), keys)
latencies.append(body.get("latency_ms", 0.0))
print(f"{record['document_id']} variant={record['variant']} "
f"source={body.get('source')} words={body.get('word_count')} "
f"pages={body.get('pages_processed')}")
print(f"server={body.get('latency_ms', 0):.0f} ms "
f"wall={wall:.0f} ms (includes model load on first call)\n")
by_key = {f["key"]: f for f in body.get("fields", [])}
print(f" {'KEY':<14} {'EXPECTED':<26} {'STATUS':<15} "
f"{'GOT':<28} CONF")
print(" " + "-" * 92)
for key, truth in KEY_MAP:
expected = str(record[truth])
field = by_key.get(key, {})
got = (field.get("value") or "").strip()
ok = got == expected.strip()
per_key[key].append(1 if ok else 0)
print(f" {key:<14} {expected[:26]:<26} "
f"{str(field.get('status')):<15} "
f"{(got[:26] + (' OK' if ok else ' BAD')):<28} "
f"{field.get('confidence', 0):.4f}")
print(" " + "-" * 92 + "\n")
total_hit = sum(sum(v) for v in per_key.values())
total_n = sum(len(v) for v in per_key.values())
print("PER-KEY EXACT MATCH")
print("-" * 46)
for key, _ in KEY_MAP:
scores = per_key[key]
if scores:
acc = 100 * sum(scores) / len(scores)
print(f" {key:<16} {acc:6.1f}% ({sum(scores)}/{len(scores)})")
if total_n:
print(f"\n {'OVERALL':<16} {100*total_hit/total_n:6.1f}% "
f"({total_hit}/{total_n})")
if latencies:
print("\nLATENCY (whole document, all keys)")
print("-" * 46)
print(f" mean {statistics.mean(latencies):8.0f} ms")
print(f" min {min(latencies):8.0f} ms")
print(f" max {max(latencies):8.0f} ms")
print(f" per key {statistics.mean(latencies)/len(keys):6.0f} ms")
return 0
if __name__ == "__main__":
raise SystemExit(main()) |