File size: 3,708 Bytes
fbc9e3f 6fda4b7 fbc9e3f | 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 | import argparse
import json
from pathlib import Path
from datetime import datetime
from collections import Counter
def latest_report():
reports = sorted(Path("reports").glob("bench_14b_champion*.json"), key=lambda p: p.stat().st_mtime, reverse=True)
if not reports:
node-7.example.invalid FileNotFoundError("reports/bench_14b_champion*.json not found")
return reports[0]
def main():
ap = argparse.ArgumentParser()
ap.add_argument("--report", type=str, default=None)
ap.add_argument("--latest", action="store_true")
ap.add_argument("--threshold", type=int, default=85)
args = ap.parse_args()
report_path = latest_report() if args.latest or not args.report else Path(args.report)
obj = json.loads(report_path.read_text(encoding="utf-8"))
results = obj.get("results", [])
failed = [r for r in results if int(r.get("score", 0)) < args.threshold or r.get("issues")]
ts = datetime.now().strftime("%Y%m%d_%H%M%S")
out_json = Path("reports") / f"failed_items_{ts}.json"
out_md = Path("reports") / f"failed_items_{ts}.md"
out_prompts = Path("data") / f"failed_items_prompts_{ts}.jsonl"
issue_counter = Counter()
cat_counter = Counter()
for r in failed:
cat_counter[r.get("category", "unknown")] += 1
for issue in r.get("issues", []):
issue_counter[issue] += 1
payload = {
"source_report": str(report_path),
"threshold": args.threshold,
"num_results": len(results),
"num_failed_or_issued": len(failed),
"category_counts": dict(cat_counter),
"issue_counts": dict(issue_counter),
"failed_items": failed,
}
out_json.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
lines = ["# Failed Items Report", "", f"- source_report: `{report_path}`", f"- threshold: `{args.threshold}`",
f"- total results: {len(results)}", f"- failed or issued: {len(failed)}", "", "## Issue Counts", ""]
for k, v in issue_counter.most_common():
lines.append(f"- {k}: {v}")
lines += ["", "## Failed Items", ""]
for r in failed:
lines.append(f"### {r.get('id')} / {r.get('category')} / score={r.get('score')}")
lines.append("")
lines.append(f"- issues: `{r.get('issues')}`")
lines.append(f"- finish_reason: `{r.get('finish_reason')}`")
lines.append(f"- latency_sec: `{r.get('latency_sec')}`")
lines.append("")
lines.append("Prompt:")
lines.append("```text")
lines.append(r.get("prompt", ""))
lines.append("```")
lines.append("")
lines.append("Content preview:")
lines.append("```text")
lines.append((r.get("content", "") or "")[:1200])
lines.append("```")
lines.append("")
out_md.write_text("\n".join(lines), encoding="utf-8")
with out_prompts.open("w", encoding="utf-8") as f:
for r in failed:
f.write(json.dumps({
"id": r.get("id"),
"category": r.get("category"),
"score": r.get("score"),
"issues": r.get("issues", []),
"prompt": r.get("prompt", ""),
"bad_content": r.get("content", ""),
}, ensure_ascii=False) + "\n")
print("==== FAILED ITEMS EXTRACTED ====")
print("SOURCE:", report_path)
print("THRESHOLD:", args.threshold)
print("FAILED_OR_ISSUED:", len(failed), "/", len(results))
print("CATEGORY_COUNTS:", dict(cat_counter))
print("ISSUE_COUNTS:", dict(issue_counter))
print("JSON:", out_json)
print("MD:", out_md)
print("PROMPTS:", out_prompts)
if __name__ == "__main__":
main()
|