Sync Pi agent Space: Added options to add eval dependencies to main app Dockerfile / compose file
4545172 | #!/usr/bin/env python3 | |
| """Recompute gold_geometry.json for mirrored runs and optionally rebuild eval_run_summary.csv. | |
| Use after changing core-coverage scoring or updating ideal_review_file.csv, without | |
| re-running the agent. | |
| Example:: | |
| python agent-redact/eval/refresh_eval_geometry.py | |
| python agent-redact/eval/refresh_eval_geometry.py --run-id gradio_8s2ysucu_20261104_160453 | |
| python agent-redact/eval/refresh_eval_geometry.py --no-rebuild-csv | |
| """ | |
| from __future__ import annotations | |
| import argparse | |
| import json | |
| import sys | |
| from dataclasses import dataclass | |
| from pathlib import Path | |
| def _repo_root() -> Path: | |
| return Path(__file__).resolve().parents[2] | |
| def _resolve_gold_case_dir(repo_root: Path, gold_case: str) -> Path | None: | |
| raw = (gold_case or "").replace("\\", "/").strip() | |
| if not raw: | |
| return None | |
| if "agent-redact/" in raw: | |
| rel = raw.split("agent-redact/", 1)[1] | |
| candidate = (repo_root / "agent-redact" / rel).resolve() | |
| return candidate if candidate.is_dir() else None | |
| path = Path(raw) | |
| if path.is_dir(): | |
| return path.resolve() | |
| if not path.is_absolute(): | |
| candidate = (repo_root / raw).resolve() | |
| return candidate if candidate.is_dir() else None | |
| return None | |
| def _gold_case_from_run(run_dir: Path) -> str: | |
| for name in ("eval_summary.json", "gradio_run_meta.json"): | |
| path = run_dir / name | |
| if not path.is_file(): | |
| continue | |
| try: | |
| data = json.loads(path.read_text(encoding="utf-8")) | |
| except (OSError, json.JSONDecodeError): | |
| continue | |
| if isinstance(data, dict): | |
| gc = str(data.get("gold_case") or "").strip() | |
| if gc: | |
| return gc | |
| return "" | |
| class RefreshResult: | |
| run_id: str | |
| scoring: str | |
| f1: float | |
| recall: float | |
| precision: float | |
| core_gold_count: int | |
| skipped_reason: str = "" | |
| def refresh_run_geometry( | |
| run_dir: Path, | |
| *, | |
| repo_root: Path, | |
| geo_opts: dict, | |
| ) -> RefreshResult | None: | |
| from compare_review_csvs import compare_review_csvs | |
| run_dir = run_dir.resolve() | |
| run_id = run_dir.name | |
| agent_csv = run_dir / "agent_review_file.csv" | |
| if not agent_csv.is_file(): | |
| return RefreshResult(run_id, "", 0.0, 0.0, 0.0, 0, "no agent_review_file.csv") | |
| gold_dir = _resolve_gold_case_dir(repo_root, _gold_case_from_run(run_dir)) | |
| if gold_dir is None: | |
| return RefreshResult( | |
| run_id, "", 0.0, 0.0, 0.0, 0, "no resolvable gold_case dir" | |
| ) | |
| ideal = gold_dir / "ideal_review_file.csv" | |
| if not ideal.is_file(): | |
| return RefreshResult( | |
| run_id, "", 0.0, 0.0, 0.0, 0, "gold case has no ideal_review_file.csv" | |
| ) | |
| report = compare_review_csvs( | |
| ideal, | |
| agent_csv, | |
| threshold=geo_opts["threshold"], | |
| mode=geo_opts["mode"], | |
| many_to_one=geo_opts["many_to_one"], | |
| scoring=geo_opts["scoring"], | |
| ) | |
| (run_dir / "gold_geometry.json").write_text( | |
| json.dumps(report.to_dict(), indent=2), encoding="utf-8" | |
| ) | |
| summary_path = run_dir / "eval_summary.json" | |
| if summary_path.is_file(): | |
| try: | |
| summary = json.loads(summary_path.read_text(encoding="utf-8")) | |
| except (OSError, json.JSONDecodeError): | |
| summary = None | |
| if isinstance(summary, dict): | |
| metrics = summary.get("metrics") | |
| if isinstance(metrics, list): | |
| for m in metrics: | |
| if isinstance(m, dict) and m.get("name") == "gold_geometry": | |
| m["score"] = report.f1 | |
| m["reason"] = ( | |
| f"precision={report.precision} recall={report.recall} " | |
| f"scoring={report.scoring} core_gold={report.core_gold_count}" | |
| ) | |
| summary_path.write_text(json.dumps(summary, indent=2), encoding="utf-8") | |
| return RefreshResult( | |
| run_id, | |
| report.scoring, | |
| float(report.f1), | |
| float(report.recall), | |
| float(report.precision), | |
| int(report.core_gold_count), | |
| ) | |
| def refresh_runs( | |
| *, | |
| runs_root: Path, | |
| repo_root: Path, | |
| run_id: str | None = None, | |
| ) -> list[RefreshResult]: | |
| from run_agent_regression import gold_geometry_compare_options | |
| geo_opts = gold_geometry_compare_options() | |
| results: list[RefreshResult] = [] | |
| if run_id: | |
| run_dir = runs_root / run_id | |
| if not run_dir.is_dir(): | |
| results.append( | |
| RefreshResult(run_id, "", 0.0, 0.0, 0.0, 0, "run directory not found") | |
| ) | |
| return results | |
| out = refresh_run_geometry(run_dir, repo_root=repo_root, geo_opts=geo_opts) | |
| if out is not None: | |
| results.append(out) | |
| return results | |
| if not runs_root.is_dir(): | |
| return results | |
| for run_dir in sorted(runs_root.iterdir()): | |
| if not run_dir.is_dir(): | |
| continue | |
| out = refresh_run_geometry(run_dir, repo_root=repo_root, geo_opts=geo_opts) | |
| if out is not None: | |
| results.append(out) | |
| return results | |
| def _build_parser() -> argparse.ArgumentParser: | |
| repo = _repo_root() | |
| p = argparse.ArgumentParser( | |
| description="Refresh gold_geometry.json under eval/runs and rebuild CSV." | |
| ) | |
| p.add_argument( | |
| "--runs-root", | |
| type=Path, | |
| default=repo / "agent-redact" / "eval" / "runs", | |
| help="Directory containing mirrored gradio_* run folders.", | |
| ) | |
| p.add_argument( | |
| "--run-id", | |
| type=str, | |
| default="", | |
| help="Refresh a single run folder name (default: all under --runs-root).", | |
| ) | |
| p.add_argument( | |
| "--csv-out", | |
| type=Path, | |
| default=repo / "agent-redact" / "eval" / "outputs" / "eval_run_summary.csv", | |
| help="Rebuild eval_run_summary.csv to this path.", | |
| ) | |
| p.add_argument( | |
| "--no-rebuild-csv", | |
| action="store_true", | |
| help="Only update gold_geometry.json / eval_summary.json per run.", | |
| ) | |
| return p | |
| def main(argv: list[str] | None = None) -> int: | |
| args = _build_parser().parse_args(argv) | |
| repo_root = _repo_root() | |
| eval_dir = repo_root / "agent-redact" / "eval" | |
| if str(eval_dir) not in sys.path: | |
| sys.path.insert(0, str(eval_dir)) | |
| run_id = (args.run_id or "").strip() or None | |
| results = refresh_runs( | |
| runs_root=args.runs_root.resolve(), | |
| repo_root=repo_root, | |
| run_id=run_id, | |
| ) | |
| updated = [r for r in results if not r.skipped_reason] | |
| skipped = [r for r in results if r.skipped_reason] | |
| for r in updated: | |
| print( | |
| f"{r.run_id}: scoring={r.scoring} f1={r.f1:.4f} " | |
| f"recall={r.recall:.4f} precision={r.precision:.4f} " | |
| f"core_gold={r.core_gold_count}" | |
| ) | |
| for r in skipped: | |
| print(f"{r.run_id}: skipped ({r.skipped_reason})", file=sys.stderr) | |
| if not args.no_rebuild_csv: | |
| from eval_run_summary_csv import rebuild_eval_run_summary_csv | |
| count = rebuild_eval_run_summary_csv( | |
| search_root=args.runs_root.resolve(), | |
| dest=args.csv_out.resolve(), | |
| ) | |
| print(f"Rebuilt {args.csv_out} ({count} rows from eval_summary.json).") | |
| return 0 if updated or skipped else 1 | |
| if __name__ == "__main__": | |
| raise SystemExit(main()) | |