dgx-harness-engineering / scripts /04_print_decision_summary.py

Upload anonymized DGX harness engineering export V1.4 (part 2)

fbc9e3f verified about 1 month ago

2.47 kB

import argparse
import json
from pathlib import Path

def latest(pattern):
    files = sorted(Path(".").glob(pattern), key=lambda p: p.stat().st_mtime, reverse=True)
    return files[0] if files else None

def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--latest", action="store_true")
    args = ap.parse_args()

    failed_json = latest("reports/failed_items_*.json")
    repair_jsonl = latest("data/repair_dataset_from_failed_items_*.jsonl")
    bench_json = latest("reports/bench_14b_champion*.json")

    print("==== DECISION SUMMARY ====")

    if bench_json:
        obj = json.loads(bench_json.read_text(encoding="utf-8"))
        print("LATEST_BENCH:", bench_json)
        print("SUMMARY:", json.dumps(obj.get("summary", {}), ensure_ascii=False))
    else:
        print("LATEST_BENCH: not found")

    print()

    if failed_json:
        obj = json.loads(failed_json.read_text(encoding="utf-8"))
        n = obj.get("num_failed_or_issued", 0)
        print("FAILED_JSON:", failed_json)
        print("FAILED_COUNT:", n)
        print("CATEGORY_COUNTS:", json.dumps(obj.get("category_counts", {}), ensure_ascii=False))
        print("ISSUE_COUNTS:", json.dumps(obj.get("issue_counts", {}), ensure_ascii=False))
        print()

        issue_counts = obj.get("issue_counts", {})
        if n == 0:
            print("RECOMMENDATION: 실패 항목이 없으므로 재학습하지 말고 현재 모델을 유지하셔도 됩니다.")
        elif any("possible_chinese_leakage" in k for k in issue_counts):
            print("RECOMMENDATION: 중국어 leakage가 있으므로 safety repair를 우선해야 합니다.")
        elif set(obj.get("category_counts", {}).keys()).issubset({"systemd", "jsonl", "cuda"}):
            print("RECOMMENDATION: 전체 재학습보다 systemd/jsonl/cuda low-item repair만 고려하시면 됩니다.")
        else:
            print("RECOMMENDATION: 실패 범위가 넓습니다. 실패 항목 내용을 먼저 검토한 뒤 repair 범위를 정하시는 게 좋습니다.")
    else:
        print("FAILED_JSON: not found")
        print("RECOMMENDATION: 먼저 02_extract_failed_items.py를 실행해 주세요.")

    print()

    if repair_jsonl:
        rows = sum(1 for _ in repair_jsonl.open("r", encoding="utf-8"))
        print("REPAIR_DATASET:", repair_jsonl)
        print("REPAIR_ROWS:", rows)
    else:
        print("REPAIR_DATASET: not created yet")

if __name__ == "__main__":
    main()

	import argparse
	import json
	from pathlib import Path

	def latest(pattern):
	files = sorted(Path(".").glob(pattern), key=lambda p: p.stat().st_mtime, reverse=True)
	return files[0] if files else None

	def main():
	ap = argparse.ArgumentParser()
	ap.add_argument("--latest", action="store_true")
	args = ap.parse_args()

	failed_json = latest("reports/failed_items_*.json")
	repair_jsonl = latest("data/repair_dataset_from_failed_items_*.jsonl")
	bench_json = latest("reports/bench_14b_champion*.json")

	print("==== DECISION SUMMARY ====")

	if bench_json:
	obj = json.loads(bench_json.read_text(encoding="utf-8"))
	print("LATEST_BENCH:", bench_json)
	print("SUMMARY:", json.dumps(obj.get("summary", {}), ensure_ascii=False))
	else:
	print("LATEST_BENCH: not found")

	print()

	if failed_json:
	obj = json.loads(failed_json.read_text(encoding="utf-8"))
	n = obj.get("num_failed_or_issued", 0)
	print("FAILED_JSON:", failed_json)
	print("FAILED_COUNT:", n)
	print("CATEGORY_COUNTS:", json.dumps(obj.get("category_counts", {}), ensure_ascii=False))
	print("ISSUE_COUNTS:", json.dumps(obj.get("issue_counts", {}), ensure_ascii=False))
	print()

	issue_counts = obj.get("issue_counts", {})
	if n == 0:
	print("RECOMMENDATION: 실패 항목이 없으므로 재학습하지 말고 현재 모델을 유지하셔도 됩니다.")
	elif any("possible_chinese_leakage" in k for k in issue_counts):
	print("RECOMMENDATION: 중국어 leakage가 있으므로 safety repair를 우선해야 합니다.")
	elif set(obj.get("category_counts", {}).keys()).issubset({"systemd", "jsonl", "cuda"}):
	print("RECOMMENDATION: 전체 재학습보다 systemd/jsonl/cuda low-item repair만 고려하시면 됩니다.")
	else:
	print("RECOMMENDATION: 실패 범위가 넓습니다. 실패 항목 내용을 먼저 검토한 뒤 repair 범위를 정하시는 게 좋습니다.")
	else:
	print("FAILED_JSON: not found")
	print("RECOMMENDATION: 먼저 02_extract_failed_items.py를 실행해 주세요.")

	print()

	if repair_jsonl:
	rows = sum(1 for _ in repair_jsonl.open("r", encoding="utf-8"))
	print("REPAIR_DATASET:", repair_jsonl)
	print("REPAIR_ROWS:", rows)
	else:
	print("REPAIR_DATASET: not created yet")

	if __name__ == "__main__":
	main()