Buckets:
| """Analyze zero-shot results: how much of the high CER is ref/image misalignment | |
| vs real OCR error. For each page, find the reference window best matching the | |
| hypothesis (rapidfuzz partial alignment) and compute aligned CER on that window. | |
| Also strip model preamble/postamble. Uploads analysis.md + analysis.jsonl to bucket. | |
| """ | |
| import json | |
| import re | |
| from statistics import mean, median | |
| from huggingface_hub import batch_bucket_files, download_bucket_files | |
| from rapidfuzz import fuzz | |
| import jiwer | |
| BUCKET = "ktrapeznikov/russian-hw-ocr-eval" | |
| RUN_PREFIX = "zeroshot-comparison" | |
| RESULTS = f"{RUN_PREFIX}/results_qwen3-vl-235b.jsonl" | |
| def normalize(t: str) -> str: | |
| t = t.replace("[НОВЫЙ АБЗАЦ]", " ") | |
| t = t.lower() | |
| t = re.sub(r"\s+", " ", t) | |
| return t.strip() | |
| def strip_chatter(hyp: str) -> str: | |
| # drop leading "Вот распознанный текст..." line(s) and trailing notes/--- sections | |
| lines = hyp.split("\n") | |
| while lines and (not lines[0].strip() or "распознан" in lines[0].lower() or lines[0].strip() == "---"): | |
| lines.pop(0) | |
| out = [] | |
| for ln in lines: | |
| s = ln.strip() | |
| if s == "---" or s.startswith("**") or s.lower().startswith("примечан"): | |
| break | |
| out.append(ln) | |
| return "\n".join(out).strip() | |
| def main(): | |
| download_bucket_files(BUCKET, files=[(RESULTS, "results.jsonl")]) | |
| rows = [json.loads(l) for l in open("results.jsonl", encoding="utf-8")] | |
| ok = [r for r in rows if "hypothesis" in r] | |
| print(f"{len(ok)} pages with hypotheses") | |
| out_rows = [] | |
| for r in ok: | |
| ref_n = normalize(r["reference"]) | |
| hyp_full = normalize(r["hypothesis"]) | |
| hyp_clean = normalize(strip_chatter(r["hypothesis"])) | |
| # best-matching ref window for the cleaned hypothesis | |
| al = fuzz.partial_ratio_alignment(ref_n, hyp_clean) | |
| window = ref_n[al.src_start:al.src_end] if al else ref_n | |
| aligned_cer = round(jiwer.cer(window, hyp_clean), 4) | |
| aligned_wer = round(jiwer.wer(window, hyp_clean), 4) | |
| # how much of the reference is NOT visible in the image (approx) | |
| ref_coverage = round((al.src_end - al.src_start) / max(len(ref_n), 1), 3) if al else None | |
| chatter_frac = round(1 - len(hyp_clean) / max(len(hyp_full), 1), 3) | |
| out_rows.append({ | |
| "idx": r["idx"], "source_folder": r["source_folder"], | |
| "dataset_name": r["dataset_name"], "image_type": r["image_type"], | |
| "cer_raw_reported": r["cer_norm"], | |
| "cer_aligned": aligned_cer, "wer_aligned": aligned_wer, | |
| "ref_coverage_by_image": ref_coverage, | |
| "chatter_frac_of_hyp": chatter_frac, | |
| }) | |
| print(f"idx {r['idx']:2d} [{r['source_folder']}/{r['dataset_name']}] " | |
| f"reported={r['cer_norm']:.3f} aligned={aligned_cer:.3f} " | |
| f"ref_visible={ref_coverage:.2f} chatter={chatter_frac:.2f}", flush=True) | |
| def agg(rows, k): | |
| v = [r[k] for r in rows if r[k] is not None] | |
| return {"mean": round(mean(v), 4), "median": round(median(v), 4)} if v else None | |
| by_folder = {} | |
| for r in out_rows: | |
| by_folder.setdefault(r["source_folder"], []).append(r) | |
| folder_stats = {f: {"n": len(rs), "cer_aligned": agg(rs, "cer_aligned"), | |
| "ref_visible": agg(rs, "ref_coverage_by_image")} | |
| for f, rs in sorted(by_folder.items())} | |
| summary = { | |
| "n": len(out_rows), | |
| "cer_norm_reported": agg(out_rows, "cer_raw_reported"), | |
| "cer_aligned": agg(out_rows, "cer_aligned"), | |
| "wer_aligned": agg(out_rows, "wer_aligned"), | |
| "ref_coverage_by_image": agg(out_rows, "ref_coverage_by_image"), | |
| "chatter_frac": agg(out_rows, "chatter_frac_of_hyp"), | |
| "by_source_folder": folder_stats, | |
| } | |
| print(json.dumps(summary, ensure_ascii=False, indent=2)) | |
| with open("analysis.jsonl", "w", encoding="utf-8") as f: | |
| for r in out_rows: | |
| f.write(json.dumps(r, ensure_ascii=False) + "\n") | |
| md = [ | |
| "# Alignment analysis: zero-shot qwen3-vl-235b vs dataset references", | |
| "", | |
| f"Pages: {len(out_rows)}. Method: strip model preamble/notes from hypothesis, then find the", | |
| "best-matching reference window (rapidfuzz partial alignment) and compute CER against that", | |
| "window only. This separates **real OCR error** from **ref/image misalignment** (references", | |
| "contain more text than the page shows).", | |
| "", | |
| "| Metric | Mean | Median |", | |
| "|---|---|---|", | |
| f"| CER as reported (full ref) | {summary['cer_norm_reported']['mean']} | {summary['cer_norm_reported']['median']} |", | |
| f"| CER aligned (visible window only) | {summary['cer_aligned']['mean']} | {summary['cer_aligned']['median']} |", | |
| f"| WER aligned | {summary['wer_aligned']['mean']} | {summary['wer_aligned']['median']} |", | |
| f"| Fraction of reference visible in image | {summary['ref_coverage_by_image']['mean']} | — |", | |
| f"| Hypothesis fraction that is chatter | {summary['chatter_frac']['mean']} | — |", | |
| "", | |
| "## By source_folder", | |
| "", | |
| "| source_folder | n | aligned CER mean | ref visible mean |", | |
| "|---|---|---|---|", | |
| ] | |
| for f, s in folder_stats.items(): | |
| md.append(f"| {f} | {s['n']} | {s['cer_aligned']['mean']} | {s['ref_visible']['mean']} |") | |
| md += ["", "Per-page: analysis.jsonl"] | |
| with open("analysis.md", "w", encoding="utf-8") as f: | |
| f.write("\n".join(md)) | |
| batch_bucket_files(BUCKET, add=[ | |
| ("analysis.jsonl", f"{RUN_PREFIX}/analysis.jsonl"), | |
| ("analysis.md", f"{RUN_PREFIX}/analysis.md"), | |
| ]) | |
| print(f"Uploaded analysis to hf://buckets/{BUCKET}/{RUN_PREFIX}/") | |
| if __name__ == "__main__": | |
| main() | |
Xet Storage Details
- Size:
- 5.82 kB
- Xet hash:
- badbe6236082225c27da19a120cdd99f9ee267246ab32046b8bbe0cda9712f5e
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.