Skills › Scientific Agent Skills › hypogenic › inspect_outputs.py
inspect_outputs.py
166 lines · skills/hypogenic/scripts/inspect_outputs.py at 5f2bc03
1#!/usr/bin/env python3
2"""Inspect local HypoGeniC hypothesis/result JSON without echoing raw content."""
4from __future__ import annotations
6import argparse
7from collections import Counter
8from collections.abc import Sequence
10if __package__:
11 from ._common import (
12 MAX_JSON_BYTES,
13 MISSING_PREDICTION,
14 CliError,
15 checked_input_file,
16 count_values,
17 emit_error,
18 emit_json,
19 load_json_document,
20 sha256_file,
21 summarize_numbers,
22 validate_hypothesis_bank,
23 validate_result_document,
24 )
25else:
26 from _common import ( # type: ignore
27 MAX_JSON_BYTES,
28 MISSING_PREDICTION,
29 CliError,
30 checked_input_file,
31 count_values,
32 emit_error,
33 emit_json,
34 load_json_document,
35 sha256_file,
36 summarize_numbers,
37 validate_hypothesis_bank,
38 validate_result_document,
39 )
42def build_parser() -> argparse.ArgumentParser:
43 parser = argparse.ArgumentParser(
44 description=(
45 "Inspect bounded local output JSON. Raw hypotheses, prompts, responses, "
46 "record IDs, labels, and predictions are never printed."
47 )
48 )
49 subparsers = parser.add_subparsers(dest="command", required=True)
50 for name, help_text in (
51 ("hypotheses", "Inspect an upstream hypothesis-bank JSON object"),
52 ("results", "Inspect a strict local prediction-result JSON object"),
53 ):
54 child = subparsers.add_parser(name, help=help_text)
55 child.add_argument("--input", required=True, help="Local strict JSON file")
56 child.add_argument("--root", default=".", help="Existing local I/O boundary")
57 return parser
60def inspect_hypotheses(document: object, *, file_sha256: str) -> dict:
61 entries = validate_hypothesis_bank(document)
62 normalized_counts = Counter(entry["normalized_sha256"] for entry in entries)
63 duplicate_hashes = sorted(
64 digest for digest, count in normalized_counts.items() if count > 1
65 )
66 return {
67 "ok": True,
68 "inspection_kind": "hypogenic_hypothesis_bank",
69 "file_sha256": file_sha256,
70 "hypothesis_count": len(entries),
71 "unique_normalized_hypothesis_count": len(normalized_counts),
72 "normalized_duplicate_group_count": len(duplicate_hashes),
73 "normalized_duplicate_groups": [
74 {
75 "normalized_sha256": digest,
76 "count": normalized_counts[digest],
77 }
78 for digest in duplicate_hashes[:20]
79 ],
80 "duplicate_groups_truncated": len(duplicate_hashes) > 20,
81 "character_length": summarize_numbers(
82 [entry["characters"] for entry in entries]
83 ),
84 "word_count": summarize_numbers([entry["words"] for entry in entries]),
85 "accuracy": summarize_numbers([entry["accuracy"] for entry in entries]),
86 "reward": summarize_numbers([entry["reward"] for entry in entries]),
87 "num_visits": summarize_numbers([entry["num_visits"] for entry in entries]),
88 "correct_example_count": summarize_numbers(
89 [entry["correct_example_count"] for entry in entries]
90 ),
91 "hypothesis_sha256_sample": sorted(
92 entry["text_sha256"] for entry in entries
93 )[:20],
94 "hypothesis_sample_truncated": len(entries) > 20,
95 "candidate_text_is_scientific_evidence": False,
96 "raw_hypothesis_text_included": False,
97 "text_interpreted_as_instructions": False,
98 "network_access": False,
99 "model_called": False,
100 }
103def inspect_results(document: object, *, file_sha256: str) -> dict:
104 result = validate_result_document(document)
105 records = result["records"]
106 predictions = [
107 (
108 str(record["prediction"])
109 if record["prediction"] is not None
110 else MISSING_PREDICTION
111 )
112 for record in records
113 ]
114 return {
115 "ok": True,
116 "inspection_kind": "hypogenic_local_results",
117 "file_sha256": file_sha256,
118 "schema_version": result["schema_version"],
119 "dataset_manifest_sha256": result["dataset_manifest_sha256"],
120 "hypothesis_bank_sha256": result["hypothesis_bank_sha256"],
121 "split": result["split"],
122 "record_count": len(records),
123 "unique_record_id_count": len({record["id"] for record in records}),
124 "missing_prediction_count": sum(
125 record["prediction"] is None for record in records
126 ),
127 "label_counts_redacted": count_values(
128 str(record["label"]) for record in records
129 ),
130 "prediction_counts_redacted": count_values(predictions),
131 "raw_records_included": False,
132 "record_text_interpreted_as_instructions": False,
133 "candidate_metrics_are_scientific_validation": False,
134 "network_access": False,
135 "model_called": False,
136 }
139def main(argv: Sequence[str] | None = None) -> int:
140 args = build_parser().parse_args(argv)
141 try:
142 path = checked_input_file(
143 args.input,
144 root=args.root,
145 suffixes={".json"},
146 max_bytes=MAX_JSON_BYTES,
147 )
148 document = load_json_document(
149 args.input,
150 root=args.root,
151 max_bytes=MAX_JSON_BYTES,
152 )
153 digest = sha256_file(path, max_bytes=MAX_JSON_BYTES)
154 report = (
155 inspect_hypotheses(document, file_sha256=digest)
156 if args.command == "hypotheses"
157 else inspect_results(document, file_sha256=digest)
158 )
159 emit_json(report)
160 return 0
161 except (CliError, OSError) as error:
162 return emit_error(error)
165if __name__ == "__main__":
166 raise SystemExit(main())