301 lines
11 KiB
Python
301 lines
11 KiB
Python
"""NPC Reason — mechanical step verifier (THE core artifact).
|
||
|
||
PURE CODE. No model, no LLM judgment anywhere. This is what makes the
|
||
"verifiable-rate" metric un-fakeable, and it is reused verbatim as the RL reward
|
||
signal in a later dispatch — so it is frozen (VERIFIER.lock) the moment its tests pass.
|
||
|
||
DEFINITION (committed; see reports/PREREG.md and VERIFIER.lock):
|
||
- A reasoning chain is a sequence of steps ending in a final answer.
|
||
- A load-bearing numeric step carries an inline checkable assertion in the canonical
|
||
form <<EXPR = RESULT>> where EXPR is an arithmetic/algebraic expression over
|
||
numbers and earlier-BOUND variables, and RESULT is the claimed value.
|
||
- A step is VERIFIED if a SymPy evaluation of EXPR equals RESULT within tolerance
|
||
(exact for integers/rationals; 1e-6 relative for floats).
|
||
- A chain is VERIFIABLE iff:
|
||
(a) it has >=1 load-bearing <<...>> assertion (no bare asserted numbers drive it),
|
||
(b) every such assertion VERIFIES, and
|
||
(c) the final answer equals the RESULT of the last load-bearing step
|
||
(the chain COMPOSES to its conclusion).
|
||
- A chain is CORRECT iff its final answer equals the gold answer. CORRECT and
|
||
VERIFIABLE are INDEPENDENT axes.
|
||
|
||
TOLERANCE POLICY (explicit, frozen):
|
||
- Parse EXPR and RESULT with sympy.sympify (after a small, fixed normalization).
|
||
- Exact branch: if simplify(EXPR_value - RESULT_value) == 0 -> VERIFIED.
|
||
- Float branch: else, if both are finite numbers and
|
||
|EXPR_value - RESULT_value| <= 1e-6 * max(1, |RESULT_value|) -> VERIFIED.
|
||
- Otherwise NOT verified.
|
||
|
||
FAIL-CLOSED (conservative by design):
|
||
- Unparseable EXPR or RESULT -> NOT verified (reason recorded).
|
||
- EXPR references an UNBOUND symbol -> NOT verified (cannot confirm).
|
||
- Any exception during evaluation -> NOT verified.
|
||
A chain only counts VERIFIABLE if the checker can ACTUALLY confirm every load-bearing step.
|
||
|
||
VARIABLE BINDING (v1, documented):
|
||
- A step may bind a name: `let total = <<3*8 = 24>>` or `total = <<3*8 = 24>>`.
|
||
The name binds to the (parsed) RESULT value and is usable by later EXPRs.
|
||
- Binding tests internal consistency: each later step is checked against the values
|
||
the chain itself previously stated.
|
||
- V1 SCOPE: assertions are concrete-valued (EXPR evaluates to a number once prior
|
||
bindings are substituted). Algebra with FREE variables / equation-solving such as
|
||
`<<x**2 - 4 = 0>>` with x unbound is OUT OF SCOPE for v1 and FAILS CLOSED
|
||
(recorded reason: "unbound symbol"). It is noted as a verifier-v2 extension.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import re
|
||
from dataclasses import dataclass, field
|
||
from typing import Optional
|
||
|
||
import sympy
|
||
from sympy import Rational, simplify, sympify
|
||
|
||
# ----------------------------------------------------------------------------- #
|
||
# Patterns
|
||
# ----------------------------------------------------------------------------- #
|
||
|
||
# Optional binding name, then << EXPR = RESULT >>. Non-greedy EXPR stops at the
|
||
# FIRST '=' inside the brackets. Binding name must start with a letter/underscore,
|
||
# so numeric tokens to the left (e.g. "2 + 2 =") are never captured as a name.
|
||
ASSERTION = re.compile(
|
||
r"(?:(?:let\s+)?([A-Za-z_]\w*)\s*=\s*)?<<\s*(.+?)\s*=\s*(.+?)\s*>>"
|
||
)
|
||
|
||
# Final-answer extractors, tried in priority order.
|
||
_BOXED = re.compile(r"\\boxed\{\s*([^{}]+?)\s*\}")
|
||
_GSM = re.compile(r"####\s*([^\n]+)")
|
||
_ANSWER_IS = re.compile(
|
||
r"(?:final answer|the answer is|answer\s*[:=])\s*\$?\\?\(?\s*"
|
||
r"([+-]?[\d.,/eE^*+\-() ]*\d)",
|
||
re.IGNORECASE,
|
||
)
|
||
|
||
# Characters/sequences to normalize before sympify.
|
||
_THOUSANDS = re.compile(r"(?<=\d),(?=\d{3}\b)")
|
||
_NORM_REPLACE = (
|
||
(r"\times", "*"),
|
||
(r"\cdot", "*"),
|
||
(r"\div", "/"),
|
||
(r"\left", ""),
|
||
(r"\right", ""),
|
||
("^", "**"),
|
||
("%", "/100"),
|
||
("$", ""),
|
||
("×", "*"),
|
||
("÷", "/"),
|
||
("−", "-"), # unicode minus
|
||
)
|
||
|
||
|
||
@dataclass
|
||
class StepResult:
|
||
expr: str
|
||
claimed: str
|
||
ok: bool
|
||
reason: str
|
||
binding: Optional[str] = None
|
||
|
||
|
||
@dataclass
|
||
class ChainRecord:
|
||
n_assertions: int = 0
|
||
n_verified: int = 0
|
||
has_loadbearing_assertions: bool = False
|
||
all_assertions_verified: bool = False
|
||
final_answer: Optional[str] = None
|
||
final_answer_value: Optional[object] = None
|
||
composes_to_final: bool = False
|
||
verifiable: bool = False
|
||
correct: Optional[bool] = None
|
||
verified_and_correct: Optional[bool] = None
|
||
steps: list = field(default_factory=list)
|
||
failures: list = field(default_factory=list)
|
||
|
||
def as_dict(self) -> dict:
|
||
return {
|
||
"n_assertions": self.n_assertions,
|
||
"n_verified": self.n_verified,
|
||
"has_loadbearing_assertions": self.has_loadbearing_assertions,
|
||
"all_assertions_verified": self.all_assertions_verified,
|
||
"final_answer": self.final_answer,
|
||
"composes_to_final": self.composes_to_final,
|
||
"verifiable": self.verifiable,
|
||
"correct": self.correct,
|
||
"verified_and_correct": self.verified_and_correct,
|
||
"failures": self.failures,
|
||
"steps": [
|
||
{"expr": s.expr, "claimed": s.claimed, "ok": s.ok,
|
||
"reason": s.reason, "binding": s.binding}
|
||
for s in self.steps
|
||
],
|
||
}
|
||
|
||
|
||
# ----------------------------------------------------------------------------- #
|
||
# Numeric core
|
||
# ----------------------------------------------------------------------------- #
|
||
|
||
def _normalize(raw: str) -> str:
|
||
s = raw.strip()
|
||
s = _THOUSANDS.sub("", s) # 1,234 -> 1234 (drop thousands separators)
|
||
for a, b in _NORM_REPLACE:
|
||
s = s.replace(a, b)
|
||
# \frac{a}{b} -> ((a)/(b))
|
||
s = re.sub(r"\\d?frac\{([^{}]+)\}\{([^{}]+)\}", r"((\1)/(\2))", s)
|
||
s = s.replace("\\", " ")
|
||
return s.strip()
|
||
|
||
|
||
def _to_sympy(raw: str, bindings: dict):
|
||
"""sympify a normalized token; raises on failure (caller fails closed)."""
|
||
expr = sympify(_normalize(raw), locals=bindings, rational=True)
|
||
return expr
|
||
|
||
|
||
def _values_match(a, b) -> bool:
|
||
"""Exact for rationals/integers; 1e-6 relative for floats."""
|
||
diff = simplify(a - b)
|
||
if diff == 0:
|
||
return True
|
||
try:
|
||
if diff.free_symbols:
|
||
return False
|
||
fa, fb = float(a), float(b)
|
||
return abs(fa - fb) <= 1e-6 * max(1.0, abs(fb))
|
||
except (TypeError, ValueError):
|
||
return False
|
||
|
||
|
||
def verify_assertion(expr: str, claimed: str, bindings: dict) -> StepResult:
|
||
"""Evaluate one <<EXPR = RESULT>> against current bindings. Fail-closed."""
|
||
try:
|
||
e = _to_sympy(expr, bindings)
|
||
except Exception as ex: # noqa: BLE001 — fail closed on any parse error
|
||
return StepResult(expr, claimed, False, f"unparseable expr: {ex}")
|
||
try:
|
||
c = _to_sympy(claimed, bindings)
|
||
except Exception as ex: # noqa: BLE001
|
||
return StepResult(expr, claimed, False, f"unparseable result: {ex}")
|
||
|
||
# EXPR must reduce to a concrete value (v1 scope: no free variables).
|
||
if getattr(e, "free_symbols", set()):
|
||
unbound = ", ".join(sorted(str(s) for s in e.free_symbols))
|
||
return StepResult(expr, claimed, False, f"unbound symbol(s): {unbound}")
|
||
|
||
try:
|
||
ok = _values_match(e, c)
|
||
except Exception as ex: # noqa: BLE001
|
||
return StepResult(expr, claimed, False, f"comparison error: {ex}")
|
||
reason = "verified" if ok else f"mismatch: {expr} -> {e} != {claimed}"
|
||
return StepResult(expr, claimed, ok, reason)
|
||
|
||
|
||
# ----------------------------------------------------------------------------- #
|
||
# Final answer
|
||
# ----------------------------------------------------------------------------- #
|
||
|
||
def extract_final_answer(text: str) -> Optional[str]:
|
||
"""Last \\boxed{}, else last ####, else 'the answer is X'. None if absent."""
|
||
boxed = _BOXED.findall(text)
|
||
if boxed:
|
||
return boxed[-1].strip()
|
||
gsm = _GSM.findall(text)
|
||
if gsm:
|
||
return gsm[-1].strip()
|
||
m = list(_ANSWER_IS.finditer(text))
|
||
if m:
|
||
return m[-1].group(1).strip().rstrip(".")
|
||
return None
|
||
|
||
|
||
def _safe_value(raw: str, bindings: dict):
|
||
try:
|
||
v = _to_sympy(raw, bindings)
|
||
return None if getattr(v, "free_symbols", set()) else v
|
||
except Exception: # noqa: BLE001
|
||
return None
|
||
|
||
|
||
# ----------------------------------------------------------------------------- #
|
||
# Chain
|
||
# ----------------------------------------------------------------------------- #
|
||
|
||
def verify_chain(text: str, gold_answer=None) -> dict:
|
||
"""Mechanically derive every field. No judgment. Returns ChainRecord.as_dict()."""
|
||
rec = ChainRecord()
|
||
bindings: dict = {}
|
||
|
||
for m in ASSERTION.finditer(text):
|
||
name, expr, claimed = m.group(1), m.group(2), m.group(3)
|
||
res = verify_assertion(expr, claimed, bindings)
|
||
res.binding = name
|
||
rec.steps.append(res)
|
||
rec.n_assertions += 1
|
||
if res.ok:
|
||
rec.n_verified += 1
|
||
else:
|
||
rec.failures.append({"expr": expr, "claimed": claimed, "reason": res.reason})
|
||
# Bind the name to the CLAIMED result value (internal-consistency semantics),
|
||
# whenever the result parses to a concrete value — even if the step failed,
|
||
# so downstream reasons are about the downstream step, not a cascade.
|
||
if name:
|
||
v = _safe_value(claimed, bindings)
|
||
if v is not None:
|
||
bindings[name] = v
|
||
|
||
rec.has_loadbearing_assertions = rec.n_assertions > 0
|
||
rec.all_assertions_verified = (
|
||
rec.has_loadbearing_assertions and rec.n_verified == rec.n_assertions
|
||
)
|
||
|
||
# Final answer + composition.
|
||
fa = extract_final_answer(text)
|
||
rec.final_answer = fa
|
||
fa_val = _safe_value(fa, bindings) if fa is not None else None
|
||
rec.final_answer_value = fa_val
|
||
|
||
if rec.has_loadbearing_assertions and fa_val is not None:
|
||
last_claimed = rec.steps[-1].claimed
|
||
last_val = _safe_value(last_claimed, bindings)
|
||
if last_val is not None:
|
||
try:
|
||
rec.composes_to_final = _values_match(fa_val, last_val)
|
||
except Exception: # noqa: BLE001
|
||
rec.composes_to_final = False
|
||
if not rec.composes_to_final and rec.has_loadbearing_assertions:
|
||
rec.failures.append({"reason": "final answer does not compose from last step"})
|
||
|
||
rec.verifiable = (
|
||
rec.has_loadbearing_assertions
|
||
and rec.all_assertions_verified
|
||
and rec.composes_to_final
|
||
)
|
||
|
||
# Correctness (independent axis).
|
||
if gold_answer is not None:
|
||
gold_val = _safe_value(str(gold_answer), {})
|
||
if fa_val is not None and gold_val is not None:
|
||
try:
|
||
rec.correct = _values_match(fa_val, gold_val)
|
||
except Exception: # noqa: BLE001
|
||
rec.correct = False
|
||
else:
|
||
# Fall back to normalized string compare (handles non-numeric MATH answers).
|
||
rec.correct = (
|
||
fa is not None
|
||
and _normalize(fa).replace(" ", "") == _normalize(str(gold_answer)).replace(" ", "")
|
||
)
|
||
rec.verified_and_correct = bool(rec.verifiable and rec.correct)
|
||
|
||
return rec.as_dict()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
import json
|
||
import sys
|
||
|
||
blob = sys.stdin.read()
|
||
print(json.dumps(verify_chain(blob), indent=2, default=str))
|