File size: 8,242 Bytes
7845694 | 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 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 | """Recompute the published TypeSafe-subset and Every metrics."""
from __future__ import annotations
import argparse
from collections import defaultdict
import json
import math
from pathlib import Path
import statistics
def read(path):
return [json.loads(line) for line in path.read_text().splitlines() if line.strip()]
def prediction_map(rows):
result = {}
for row in rows:
if row["id"] in result:
raise ValueError(f"Duplicate prediction ID {row['id']}")
probabilities = row.get("probabilities")
option_ids = row.get("option_ids")
if (
not isinstance(probabilities, list)
or not isinstance(option_ids, list)
or len(probabilities) != len(option_ids)
or len(probabilities) < 2
or any(not isinstance(value, (int, float)) or not math.isfinite(value) for value in probabilities)
or abs(sum(probabilities) - 1) > 1e-4
):
raise ValueError(f"Invalid distribution for {row['id']}")
result[row["id"]] = row
return result
def aligned_distribution(row, prediction):
option_ids = [option["id"] for option in row["options"]]
if prediction["option_ids"] != option_ids:
raise ValueError(f"Option IDs/order differ for {row['id']}")
return prediction["probabilities"]
def type_safe(gold, direct, reranker):
systems = {"direct": prediction_map(direct), "reranker": prediction_map(reranker)}
expected = {row["id"] for row in gold}
if any(set(predictions) != expected for predictions in systems.values()):
raise ValueError("TypeSafe gold and prediction IDs differ")
records = defaultdict(lambda: defaultdict(list))
for row in gold:
target = row["target_distribution"]
systems_for_row = {
name: aligned_distribution(row, predictions[row["id"]])
for name, predictions in systems.items()
}
systems_for_row["published_jev"] = row["published_models"]["typesafe"]["distribution"]
for name, distribution in systems_for_row.items():
if len(distribution) != len(target):
raise ValueError(f"Distribution width differs for {row['id']}")
predicted = max(range(len(distribution)), key=distribution.__getitem__)
records[name][row["group_id"]].append(
{
"agreement": float(predicted == row["label"]),
"total_variation": sum(abs(left - right) for left, right in zip(distribution, target)) / 2,
}
)
result = {}
for name, groups in records.items():
result[name] = {
"rows": len(gold),
"cases": len(groups),
"equal_case_modal_agreement": statistics.mean(
statistics.mean(item["agreement"] for item in rows) for rows in groups.values()
),
"equal_case_total_variation": statistics.mean(
statistics.mean(item["total_variation"] for item in rows) for rows in groups.values()
),
}
return result
def yes_probability(row, predictions):
prediction = predictions[row["id"]]
aligned_distribution(row, prediction)
return prediction["probabilities"][prediction["option_ids"].index("yes")]
def hard_label(rows, predictions):
correct = 0
for row in rows:
probability = yes_probability(row, predictions)
correct += (probability >= 0.5) == (row["options"][row["label"]]["id"] == "yes")
return {"rows": len(rows), "correct": correct, "accuracy": correct / len(rows)}
def retrieval(rows, predictions):
queries = defaultdict(list)
for row in rows:
queries[row["provenance"]["question_id"]].append(row)
reciprocal_ranks, recall_one, recall_three = [], [], []
for candidates in queries.values():
ranked = sorted(candidates, key=lambda row: -yes_probability(row, predictions))
relevant_ids = {
row["id"] for row in ranked if row["options"][row["label"]]["id"] == "yes"
}
if not relevant_ids:
raise ValueError("Retrieval query has no relevant document")
first = next(index + 1 for index, row in enumerate(ranked) if row["id"] in relevant_ids)
reciprocal_ranks.append(1 / first)
recall_one.append(sum(row["id"] in relevant_ids for row in ranked[:1]) / len(relevant_ids))
recall_three.append(sum(row["id"] in relevant_ids for row in ranked[:3]) / len(relevant_ids))
return {
**hard_label(rows, predictions),
"queries": len(queries),
"recall_at_1": statistics.mean(recall_one),
"recall_at_3": statistics.mean(recall_three),
"mrr": statistics.mean(reciprocal_ranks),
}
def firewall_gate(signals):
if (
(signals["destructive"] > 0.72 and signals["reversible"] < 0.35)
or signals["exceeds_scope"] > 0.78
or signals["shares_sensitive"] > 0.78
):
return "block"
if (
signals["needs_confirmation"] > 0.55
or signals["destructive"] > 0.45
or signals["shares_sensitive"] > 0.65
):
return "confirm"
return "allow"
def firewall(rows, predictions, actions):
by_item = defaultdict(dict)
for row in rows:
by_item[row["provenance"]["source_item"]][row["provenance"]["question_id"]] = yes_probability(
row, predictions
)
expected = actions["expected_actions"]
if set(by_item) != set(expected):
raise ValueError("Firewall action IDs differ")
decisions = {item: firewall_gate(signals) for item, signals in by_item.items()}
correct = sum(decisions[item] == action for item, action in expected.items())
return {"rows": len(rows), "actions": len(expected), "correct": correct, "accuracy": correct / len(expected)}
def every(gold, inference, direct, reranker, actions):
systems = {"direct": prediction_map(direct), "reranker": prediction_map(reranker)}
gold_ids = {row["id"] for row in gold}
inference_ids = {row["id"] for row in inference}
if len(inference_ids) != len(inference) or any(set(predictions) != inference_ids for predictions in systems.values()):
raise ValueError("Every inference and prediction IDs differ")
if not gold_ids <= inference_ids:
raise ValueError("Every inference rows omit author-labeled rows")
families = defaultdict(list)
for row in gold:
families[row["provenance"]["experiment"]].append(row)
result = {}
for name, predictions in systems.items():
firewall_rows = [row for row in inference if row["id"].startswith("every/action-firewall/")]
result[name] = {
"judge-grid": hard_label(families["judge-grid"], predictions),
"code-rag": retrieval(families["code-rag"], predictions),
"company-brain": retrieval(families["company-brain"], predictions),
"action-firewall": firewall(firewall_rows, predictions, actions),
}
return result
def main() -> None:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--source", choices=("typesafe", "every"), required=True)
parser.add_argument("--gold", type=Path, required=True)
parser.add_argument("--direct", type=Path, required=True)
parser.add_argument("--reranker", type=Path, required=True)
parser.add_argument("--inference", type=Path)
parser.add_argument("--firewall-actions", type=Path)
args = parser.parse_args()
gold, direct, reranker = read(args.gold), read(args.direct), read(args.reranker)
if args.source == "typesafe":
if args.firewall_actions or args.inference:
parser.error("--inference and --firewall-actions apply only to Every")
report = type_safe(gold, direct, reranker)
else:
if not args.firewall_actions or not args.inference:
parser.error("Every requires --inference and --firewall-actions from build_every.py")
report = every(
gold,
read(args.inference),
direct,
reranker,
json.loads(args.firewall_actions.read_text()),
)
print(json.dumps(report, indent=2))
if __name__ == "__main__":
main()
|