Buckets:
| """Merge fine-tuned full-rollout and direct baseline results.""" | |
| from __future__ import annotations | |
| import argparse | |
| import json | |
| from collections import defaultdict | |
| from pathlib import Path | |
| def _read_jsonl(path: Path) -> list[dict]: | |
| with path.open(encoding="utf-8") as handle: | |
| return [json.loads(line) for line in handle if line.strip()] | |
| def _short(text: str, limit: int = 80) -> str: | |
| text = " ".join(text.replace("<|im_end|>", "").split()) | |
| return text if len(text) <= limit else text[: limit - 1] + "…" | |
| def summarize(root: Path) -> dict: | |
| fine = _read_jsonl(root / "finetuned_interleaved" / "predictions.jsonl") | |
| direct = _read_jsonl(root / "baseline_direct512" / "predictions.jsonl") | |
| native = _read_jsonl(root / "baseline_native512" / "predictions.jsonl") | |
| direct_by_id = {row["sample_id"]: row for row in direct} | |
| native_by_id = {row["sample_id"]: row for row in native} | |
| rows = [] | |
| by_step = defaultdict(list) | |
| for row in fine: | |
| sample_id = row["sample_id"] | |
| direct_row = direct_by_id[sample_id] | |
| native_row = native_by_id[sample_id] | |
| for step in row["visual_steps"]: | |
| by_step[step["step"]].append(step) | |
| rows.append( | |
| { | |
| "sample_id": sample_id, | |
| "target": row["score"]["target"], | |
| "direct_answer": direct_row["score"]["candidate"], | |
| "direct_correct": direct_row["score"]["correct"], | |
| "native_answer": native_row["score"]["candidate"], | |
| "native_correct": native_row["score"]["correct"], | |
| "interleaved_answer": row["score"]["candidate"], | |
| "interleaved_correct": row["score"]["correct"], | |
| "protocol_ok": row["protocol_ok"], | |
| "images": f"{row['generated_images']}/{row['expected_images']}", | |
| "mean_bit_accuracy": ( | |
| sum(step["bit_accuracy"] for step in row["visual_steps"]) | |
| / len(row["visual_steps"]) | |
| if row["visual_steps"] | |
| else None | |
| ), | |
| } | |
| ) | |
| count = len(rows) | |
| direct_correct = sum(row["direct_correct"] for row in rows) | |
| native_correct = sum(row["native_correct"] for row in rows) | |
| interleaved_correct = sum(row["interleaved_correct"] for row in rows) | |
| result = { | |
| "samples": count, | |
| "direct": {"correct": direct_correct, "accuracy": direct_correct / count}, | |
| "native": {"correct": native_correct, "accuracy": native_correct / count}, | |
| "interleaved": { | |
| "correct": interleaved_correct, | |
| "accuracy": interleaved_correct / count, | |
| "protocol_complete": sum(row["protocol_ok"] for row in rows), | |
| }, | |
| "gain_vs_best_baseline": ( | |
| interleaved_correct - max(direct_correct, native_correct) | |
| ) | |
| / count, | |
| "visual_by_step": { | |
| str(step): { | |
| "count": len(values), | |
| "mean_bit_accuracy": sum(v["bit_accuracy"] for v in values) / len(values), | |
| "mean_pixel_psnr": sum(v["pixel_psnr"] for v in values) / len(values), | |
| } | |
| for step, values in sorted(by_step.items()) | |
| }, | |
| "rows": rows, | |
| } | |
| (root / "comparison.json").write_text( | |
| json.dumps(result, indent=2, ensure_ascii=False) + "\n", encoding="utf-8" | |
| ) | |
| lines = [ | |
| "# Zebra-CoT overfit16: full interleaved vs direct baseline", | |
| "", | |
| "| Mode | Correct | Accuracy |", | |
| "|---|---:|---:|", | |
| f"| Base direct-only | {direct_correct}/{count} | {direct_correct / count:.2%} |", | |
| f"| Base native free-form | {native_correct}/{count} | {native_correct / count:.2%} |", | |
| f"| Fine-tuned full interleaved | {interleaved_correct}/{count} | {interleaved_correct / count:.2%} |", | |
| "", | |
| "| Sample | GT | Direct | Native | Interleaved | Protocol | Images | Visual bits |", | |
| "|---|---|---|---|---|---:|---:|---:|", | |
| ] | |
| for row in rows: | |
| lines.append( | |
| f"| {row['sample_id']} | {_short(row['target'])} | " | |
| f"{'✓' if row['direct_correct'] else '✗'} {_short(row['direct_answer'])} | " | |
| f"{'✓' if row['native_correct'] else '✗'} {_short(row['native_answer'])} | " | |
| f"{'✓' if row['interleaved_correct'] else '✗'} {_short(row['interleaved_answer'])} | " | |
| f"{'✓' if row['protocol_ok'] else '✗'} | {row['images']} | " | |
| f"{row['mean_bit_accuracy']:.2%} |" | |
| ) | |
| lines.extend( | |
| [ | |
| "", | |
| "This is training-set overfit evaluation, not held-out evidence of transfer.", | |
| "Visual bit agreement and contact sheets must be read alongside answer accuracy.", | |
| ] | |
| ) | |
| (root / "summary.md").write_text("\n".join(lines) + "\n", encoding="utf-8") | |
| return result | |
| def main() -> None: | |
| parser = argparse.ArgumentParser(description=__doc__) | |
| parser.add_argument("--root", type=Path, required=True) | |
| result = summarize(parser.parse_args().root) | |
| print(json.dumps({key: value for key, value in result.items() if key != "rows"}, indent=2)) | |
| if __name__ == "__main__": | |
| main() | |
Xet Storage Details
- Size:
- 5.24 kB
- Xet hash:
- 17d6412680385ca9ed23b2d1c552519b3ca9e670d3c279e2d5d63ed308880829
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.