Spaces:
Running
Running
| 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}") | |