File size: 10,151 Bytes
767a051 | 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 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 | from __future__ import annotations
"""Independent exact oracle for the finite Learner 0.2 registry.
This module deliberately does not invoke :mod:`tomagi.formal` or reuse the
formal expression tree. It is a separately implemented falsification oracle
using :class:`fractions.Fraction` and ordinary finite enumeration.
"""
from dataclasses import dataclass
from fractions import Fraction
import copy
from typing import Any, Iterable, Mapping
from tomagi.canonical import attach_hash, content_hash, verify_hash
class OracleError(ValueError):
pass
def frac(value: Any) -> Fraction:
if isinstance(value, bool):
raise OracleError("boolean is not an exact rational")
if isinstance(value, int):
return Fraction(value, 1)
if isinstance(value, Mapping) and set(value) == {"num", "den"}:
n, d = value["num"], value["den"]
if isinstance(n, bool) or isinstance(d, bool) or not isinstance(n, int) or not isinstance(d, int):
raise OracleError("rational words must be integers")
if d <= 0:
raise OracleError("rational denominator must be positive")
f = Fraction(n, d)
if f.numerator != n or f.denominator != d:
raise OracleError("rational must be reduced")
return f
raise OracleError(f"not an exact rational: {value!r}")
def q(value: Fraction | int) -> dict[str, int]:
f = value if isinstance(value, Fraction) else Fraction(value, 1)
return {"num": f.numerator, "den": f.denominator}
def _eval_tree(node: Mapping[str, Any], x: Fraction, depth: int = 0) -> Fraction:
if depth > 2:
raise OracleError("expression tree exceeds depth 2")
op = node.get("op")
if op == "x":
return x
if op == "const":
return frac(node["value"])
if op in {"neg", "abs", "square"}:
value = _eval_tree(node["arg"], x, depth + 1)
if op == "neg":
return -value
if op == "abs":
return abs(value)
return value * value
if op in {"add", "sub", "mul"}:
left = _eval_tree(node["left"], x, depth + 1)
right = _eval_tree(node["right"], x, depth + 1)
if op == "add":
return left + right
if op == "sub":
return left - right
return left * right
raise OracleError(f"unknown expression-tree op {op!r}")
def predict(candidate: Mapping[str, Any], observation: Mapping[str, Any]) -> tuple[bool, Any]:
family = candidate.get("family_id")
x_value = observation.get("input")
if family == "family:polynomial:0.2":
x = frac(x_value)
coeffs = candidate["coefficients"]
return True, q(frac(coeffs[0]) + frac(coeffs[1]) * x + frac(coeffs[2]) * x * x)
if family == "family:piecewise-affine:0.2":
x = frac(x_value)
side = candidate["left"] if x <= frac(candidate["breakpoint"]) else candidate["right"]
return True, q(frac(side["slope"]) * x + frac(side["intercept"]))
if family == "family:transition-table:0.2":
matches = [entry for entry in candidate["entries"] if entry["input"] == x_value]
if len(matches) != 1:
return False, None
return True, copy.deepcopy(matches[0]["output"])
if family == "family:expression-tree:0.2":
return True, q(_eval_tree(candidate["tree"], frac(x_value)))
raise OracleError(f"unknown candidate family {family!r}")
def _matches(candidate: Mapping[str, Any], observation: Mapping[str, Any]) -> bool:
defined, value = predict(candidate, observation)
if not defined:
return False
target = observation["target"]
numeric = isinstance(target, int) or isinstance(target, Mapping)
if numeric:
try:
return frac(value) == frac(target)
except OracleError:
return False
return value == target
def _partition(dataset: Mapping[str, Any], name: str) -> list[Mapping[str, Any]]:
ids = dataset["partitions"][name]
by_id = {observation["id"]: observation for observation in dataset["observations"]}
return [by_id[item] for item in ids]
def _contradictions(observations: Iterable[Mapping[str, Any]]) -> list[dict[str, Any]]:
rows = list(observations)
out: list[dict[str, Any]] = []
for left_index, left in enumerate(rows):
for right in rows[left_index + 1:]:
if left["input"] == right["input"] and left["target"] != right["target"]:
out.append(attach_hash({
"schema": "TOM-LEARNER-0.2-CONTRADICTION-1.0",
"left_observation": left["content_hash"],
"right_observation": right["content_hash"],
"input": copy.deepcopy(left["input"]),
"left_target": copy.deepcopy(left["target"]),
"right_target": copy.deepcopy(right["target"]),
}))
return out
def _family_candidates(registry: Mapping[str, Any], dataset: Mapping[str, Any]) -> list[Mapping[str, Any]]:
eligible = set(dataset["eligible_families"])
out: list[Mapping[str, Any]] = []
for family in registry["families"]:
if family["id"] in eligible:
if len(family["candidates"]) > family["search_budget"]["max_candidates"]:
raise OracleError("family candidate registry exceeds declared search budget")
out.extend(family["candidates"])
return out
def _fit_hash(dataset: Mapping[str, Any], registry: Mapping[str, Any]) -> str:
train = _partition(dataset, "train")
return content_hash({
"schema": "TOM-LEARNER-0.2-TRAIN-FIT-INPUT-1.0",
"dataset_id": dataset["id"],
"family_registry_hash": registry["content_hash"],
"eligible_families": dataset["eligible_families"],
"train_observations": [item["content_hash"] for item in train],
})
def _regression_impact(selected: Mapping[str, Any] | None, dataset: Mapping[str, Any],
prior: Mapping[str, Any]) -> dict[str, Any]:
supersedes = dataset.get("supersedes")
found = supersedes is None
results: list[dict[str, Any]] = []
for definition in prior["definitions"]:
replacement = definition["model"]
replaced = supersedes == definition["content_hash"]
if replaced:
found = True
if selected is not None:
replacement = selected
case_results = []
for case in definition["regression_cases"]:
observation = {"input": case["input"], "target": case["expected"]}
passed = _matches(replacement, observation)
case_results.append({"case_id": case["id"], "passed": passed})
results.append({
"definition_hash": definition["content_hash"],
"replaced": replaced,
"passed": all(item["passed"] for item in case_results),
"cases": case_results,
})
all_pass = found and all(item["passed"] for item in results)
return attach_hash({
"schema": "TOM-LEARNER-0.2-REGRESSION-IMPACT-1.0",
"dataset_id": dataset["id"],
"supersedes": supersedes,
"superseded_definition_found": found,
"tested_definitions": len(results),
"all_pass": all_pass,
"results": results,
})
def evaluate_dataset(dataset: Mapping[str, Any], registry: Mapping[str, Any],
prior: Mapping[str, Any]) -> dict[str, Any]:
if not verify_hash(dict(dataset)) or not verify_hash(dict(registry)) or not verify_hash(dict(prior)):
raise OracleError("oracle input content hash mismatch")
train = _partition(dataset, "train")
validation = _partition(dataset, "validation")
holdout = _partition(dataset, "holdout")
candidates = _family_candidates(registry, dataset)
survivors = [candidate for candidate in candidates if all(_matches(candidate, row) for row in train)]
contradictions = _contradictions(dataset["observations"])
selected = survivors[0] if len(survivors) == 1 else None
validation_failures = [] if selected is None else [row for row in validation if not _matches(selected, row)]
holdout_failures = [] if selected is None else [row for row in holdout if not _matches(selected, row)]
regression = _regression_impact(selected, dataset, prior)
accepted = (
len(survivors) == 1
and not contradictions
and not validation_failures
and not holdout_failures
and regression["all_pass"]
)
if len(survivors) > 1:
reason = "ambiguous-train-survivors"
elif len(survivors) == 0:
reason = "no-exact-train-candidate"
elif contradictions:
reason = "contradiction"
elif validation_failures:
reason = "validation-counterexample"
elif holdout_failures:
reason = "holdout-counterexample"
elif not regression["all_pass"]:
reason = "regression-impact"
else:
reason = "accepted"
return {
"dataset_id": dataset["id"],
"accepted": accepted,
"reason": reason,
"candidate_count": len(candidates),
"survivor_hashes": [item["content_hash"] for item in survivors],
"selected_candidate_hash": None if selected is None else selected["content_hash"],
"fit_input_hash": _fit_hash(dataset, registry),
"validation_failures": [item["id"] for item in validation_failures],
"holdout_failures": [item["id"] for item in holdout_failures],
"contradiction_count": len(contradictions),
"regression_all_pass": regression["all_pass"],
"supersedes": dataset.get("supersedes"),
}
def evaluate_all(datasets: list[Mapping[str, Any]], registry: Mapping[str, Any],
prior: Mapping[str, Any]) -> dict[str, Any]:
rows = [evaluate_dataset(dataset, registry, prior) for dataset in datasets]
return {
"dataset_count": len(rows),
"accepted_count": sum(row["accepted"] for row in rows),
"rejected_count": sum(not row["accepted"] for row in rows),
"ambiguous_count": sum(row["reason"] == "ambiguous-train-survivors" for row in rows),
"rows": rows,
}
|