Spaces:
Running
Running
File size: 4,520 Bytes
71b0992 | 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 | import json
import time
from datetime import datetime, timezone
from decimal import Decimal
from itertools import zip_longest
from pathlib import Path
from .extraction import extract, pages
from .runtime import FreeModel, Settings
from .validation import verify
def extraction_metrics(actual, expected):
numeric = {"subtotal", "tax", "total", "quantity", "unit_price", "amount"}
def same(key, left, right):
if left is None or right is None:
return left is right
return Decimal(str(left)) == Decimal(str(right)) if key in numeric else left == right
header = ("vendor", "number", "date", "currency", "subtotal", "tax", "total")
line_fields = ("sku", "description", "quantity", "unit_price", "amount")
scores = [
same(key, a.get(key), b.get(key))
for a, b in zip_longest(actual.get("lines", []), expected.get("lines", []), fillvalue={})
for key in line_fields
]
return {
"header_accuracy": sum(same(key, actual.get(key), expected.get(key)) for key in header)
/ len(header),
"line_accuracy": sum(scores) / len(scores) if scores else 1.0,
}
def evaluate(live=False, limit=None, resume=False, cases=None):
root = Path("data/samples")
labels = json.loads((root / "labels.json").read_text())
model = FreeModel(Settings()) if live else None
if model:
model.qualify()
output = Path("reports") / ("live.json" if live else "deterministic.json")
rows = []
if resume and output.exists():
previous = json.loads(output.read_text())
if previous.get("model") != (model.settings.model_id if model else None):
raise ValueError("Cannot resume an evaluation with a different model.")
rows = previous["cases"]
completed = {row["case"] for row in rows}
attempted = 0
def save():
output.parent.mkdir(exist_ok=True)
output.write_text(
json.dumps(
{
"mode": "live extraction"
if live
else "labeled structured inputs; does not measure extraction or OCR",
"dataset": "fictional-documents-v1",
"date_utc": datetime.now(timezone.utc).isoformat(),
"model": model.settings.model_id if model else None,
"unattempted": len(labels) - len(rows),
"cases": rows,
},
indent=2,
)
)
for case in labels:
if case["id"] in completed or (cases and case["id"] not in cases):
continue
if limit is not None and attempted >= limit:
break
attempted += 1
start = time.monotonic()
try:
if live:
inv_path, po_path = root / case["invoice_file"], root / case["purchase_file"]
inv = extract(model, pages(inv_path.read_bytes(), inv_path.name)).model_dump(
mode="json"
)
time.sleep(8)
po = extract(model, pages(po_path.read_bytes(), po_path.name)).model_dump(
mode="json"
)
else:
inv, po = case["invoice"], case["purchase"]
actual = set(f["code"] for f in verify(inv, po))
expected = set(case["expected_codes"])
rows.append(
{
"case": case["id"],
"split": case["split"],
"passed": actual == expected,
"expected": sorted(expected),
"actual": sorted(actual),
"invoice_extraction": extraction_metrics(inv, case["invoice"])
if live
else None,
"purchase_extraction": extraction_metrics(po, case["purchase"])
if live
else None,
"invoice": inv if live else None,
"purchase": po if live else None,
"seconds": time.monotonic() - start,
}
)
except (ValueError, RuntimeError) as error:
rows.append({"case": case["id"], "passed": False, "error": str(error)})
if "RateLimitError" in str(error):
save()
break
save()
if live:
time.sleep(8)
save()
print(f"{sum(row['passed'] for row in rows)}/{len(rows)} scenarios passed; {output}")
|