11d7e0b607
Extends _match_multiplicative_aggregation with a new branch keyed on
anchor_kind="subtractive_quantity_composition". Pattern:
<Subject> <init-verb> <N> <unit>(,| then| ;| and then| and)
<sub-verb> <M> <unit>
Same-unit only. Emits a pre-composed CandidateInitial(N - M, unit) +
composition_shape="bound(initial) − bound(removed)".
Verb whitelists:
initial: had/has/got/owns/owned/earned/saved/made/received/bought
removal: lost/spent/gave/donated/paid/removed/sold/used/consumed
Removal verbs accept an optional " away" suffix ("gave away 20 apples").
Refusal-preferring discipline:
- count_b >= count_a → refuse (non-negative remainder; wrong>0 hazard)
- Pronoun / determiner subject → refuse
- Cross-unit → refuse (no v1 conversion table)
- Unobserved unit → refuse
- Unknown initial/removal verb → refuse
Tests (17 new, all green):
- canonical subtractive ("Sam had 50 apples, gave 20" → 30)
- then/and connectives
- gave away variant
- negative + equal remainder refused (hazard pin)
- pronoun + determiner subject refused
- cross-unit refused
- unobserved unit refused
- unknown initial/removal verbs refused
- additive (ME-3) path unaffected
- multiplicative_aggregate detection unaffected
- anchor audit fields complete
- end-to-end via composition_registry: affirms admits, falsifies suppresses
Registered in core/cli.py "packs" suite.
core test --suite packs -q → 123 passed (106 + 17 new)
core eval gsm8k_math --split public → 150/150, wrong=0
Anti-regression invariants preserved across ME-1..ME-4 stack:
- wrong == 0 on gsm8k_math public 150/150
- Case 0050 hazard pin holds
- ADR-0166 — no new eval lanes
- ADR-0167 partition — no cognition imports
- All prior matcher paths unaffected (test pins)
- engine_state/* not committed
- All three SAFE_COMPOSITION_CATEGORIES (multiplicative / additive /
subtractive) now have matcher extensions wired
Stacks on PR #402 (base: feat/matcher-extension-multi-quantity).
1527 lines
56 KiB
Python
1527 lines
56 KiB
Python
"""ADR-0163 Phase D — per-category recognizer match.
|
||
|
||
Pure, rules-only matching of a natural-language statement against the
|
||
ratified recognizer registry. Returns at most one
|
||
:class:`RecognizerMatch` per call (first-match-wins over the registry
|
||
order).
|
||
|
||
Doctrine
|
||
- No LLM call, no embedding, no learned classifier. The matcher is
|
||
the same discipline as Phase A's categorizer + Phase C's
|
||
synthesizer. A module-import test (mirroring Phase A/C) enforces
|
||
this.
|
||
- Per ADR-0163 §Phase C The Synthesis Rule property (b), the
|
||
recognizer is the *narrowest* commitment that subsumes the seeds.
|
||
This module honors that narrowness verbatim: an out-of-corpus
|
||
currency symbol, window unit, or per-unit value does NOT match.
|
||
Widening happens in operator review (Phase B round 2 → Phase C
|
||
synthesis → Phase D wiring picks up the wider spec automatically),
|
||
never here.
|
||
- ``parsed_anchors`` carry the actual numeric tokens extracted from
|
||
the statement (NOT from the spec). The extraction is rules-only
|
||
and deterministic. For
|
||
``descriptive_setup_no_quantity``, ``parsed_anchors`` is the empty
|
||
tuple by design — the recognizer admits the statement as setup
|
||
context, contributing no math state.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import re
|
||
from dataclasses import dataclass
|
||
from typing import Any, Final, Literal, Mapping
|
||
|
||
from evals.refusal_taxonomy.shape_categories import ShapeCategory
|
||
from generate.recognizer_registry import RatifiedRecognizer
|
||
|
||
|
||
# Word numerals 1..20 plus the higher cardinals and a small set of
|
||
# multipliers ("dozen"). Mirrors the Phase A categorizer's
|
||
# _NUMBER_WORDS so the matcher's "has any quantity marker" predicate
|
||
# is the same shape as Phase A's "has no quantity marker" predicate.
|
||
_NUMBER_WORDS: Final[frozenset[str]] = frozenset({
|
||
"one", "two", "three", "four", "five", "six", "seven", "eight", "nine",
|
||
"ten", "eleven", "twelve", "thirteen", "fourteen", "fifteen", "sixteen",
|
||
"seventeen", "eighteen", "nineteen", "twenty", "thirty", "forty", "fifty",
|
||
"sixty", "seventy", "eighty", "ninety",
|
||
"hundred", "thousand", "million", "billion",
|
||
"dozen", "dozens",
|
||
})
|
||
|
||
_DIGIT_RE: Final[re.Pattern[str]] = re.compile(r"\d")
|
||
_INDEFINITE_TOKENS: Final[tuple[str, ...]] = (
|
||
" some ", " several ", " a few ", " many ", " any ",
|
||
)
|
||
|
||
|
||
# Currency-per-unit "amount" regex. Matches "$18.00 an hour" /
|
||
# "$2 per cup" / "$45/hour" / "$20 for one kg". The captured
|
||
# groups are (symbol, amount, _spacer, per_unit).
|
||
_CURRENCY_AMOUNT_RE: Final[re.Pattern[str]] = re.compile(
|
||
r"""(?ix)
|
||
([\$£€¥]) # currency symbol
|
||
\s*
|
||
(\d+(?:\.\d+)?|\d+/\d+) # amount (integer, decimal, or fraction)
|
||
\s*
|
||
(?:
|
||
an?\s+([a-z]+) # "$X an hour" / "$X a day"
|
||
| per\s+([a-z]+) # "$X per hour"
|
||
| /\s*([a-z]+) # "$X/hour"
|
||
| for\s+(?:one|each|every|a)\s+([a-z]+)
|
||
# "$X for one cup" / "for each X"
|
||
)
|
||
""",
|
||
)
|
||
|
||
# Temporal-aggregation event_count_per_window patterns.
|
||
#
|
||
# Matches:
|
||
# "10 oysters in 5 minutes" -> count=10, window="minute", q="per"
|
||
# "10 videos each day" -> count=10, window="day", q="each"
|
||
# "20 jumping jacks on Monday" -> day-of-week single hit
|
||
# "uploads 90 minutes daily" -> count=90, window="day", q="per"
|
||
#
|
||
# Three regexes cover the high-signal canonical surfaces. Each match
|
||
# yields (count_token, window_unit, window_quantifier).
|
||
_TEMPORAL_PATTERNS: Final[tuple[tuple[re.Pattern[str], str], ...]] = (
|
||
# "<count> ... each|every|per <unit>"
|
||
(
|
||
re.compile(
|
||
r"""(?ix)
|
||
\b(\d+(?:\.\d+)?)\b # count_token
|
||
[^.,;]*? # arbitrary intervening words
|
||
\b(each|every|per)\s+
|
||
(day|week|month|year|hour|minute|second)s?\b
|
||
"""
|
||
),
|
||
"explicit_quantifier",
|
||
),
|
||
# "<count> ... in <N> <unit>" → "per <unit>" canonical
|
||
(
|
||
re.compile(
|
||
r"""(?ix)
|
||
\b(\d+(?:\.\d+)?)\b # count_token
|
||
[^.,;]*? # arbitrary intervening words
|
||
\bin\s+\d+(?:\.\d+)?\s+
|
||
(day|week|month|year|hour|minute|second)s?\b
|
||
"""
|
||
),
|
||
"in_window",
|
||
),
|
||
# "<count> ... <unit>ly" (adverbial: daily, weekly, monthly...)
|
||
(
|
||
re.compile(
|
||
r"""(?ix)
|
||
\b(\d+(?:\.\d+)?)\b # count_token
|
||
[^.,;]*? # arbitrary intervening words
|
||
\b(daily|weekly|monthly|yearly|hourly)\b
|
||
"""
|
||
),
|
||
"adverbial",
|
||
),
|
||
)
|
||
|
||
# Day-of-week enumeration: at least two distinct day names with at
|
||
# least one numeric count. Matches "20 ... Monday, 36 ... Tuesday".
|
||
_DAY_NAMES: Final[tuple[str, ...]] = (
|
||
"monday", "tuesday", "wednesday", "thursday", "friday", "saturday", "sunday",
|
||
)
|
||
_DAY_HIT_RE: Final[re.Pattern[str]] = re.compile(
|
||
r"""(?ix)
|
||
\b(\d+(?:\.\d+)?)\b\s* # count_token
|
||
[^.,;]*? # arbitrary intervening words
|
||
\b(monday|tuesday|wednesday|thursday|friday|saturday|sunday)\b
|
||
"""
|
||
)
|
||
|
||
|
||
@dataclass(frozen=True, slots=True)
|
||
class RecognizerMatch:
|
||
"""One ratified-recognizer hit against a natural-language statement.
|
||
|
||
``parsed_anchors`` carry the numeric content extracted from
|
||
the statement. For ``descriptive_setup_no_quantity``, the tuple
|
||
is empty by design — the recognizer admits the statement as
|
||
setup context, contributing no math state.
|
||
"""
|
||
|
||
recognizer: RatifiedRecognizer
|
||
category: ShapeCategory
|
||
outcome: Literal["admissible", "inadmissible_by_design"]
|
||
graph_intent: Literal["setup", "aggregate", "rate", "count", "amount"]
|
||
parsed_anchors: tuple[Mapping[str, Any], ...]
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Helpers
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def _padded_lower(statement: str) -> str:
|
||
return " " + statement.lower().replace("\n", " ") + " "
|
||
|
||
|
||
def _has_number_word(padded_lower: str) -> bool:
|
||
for raw_token in padded_lower.split():
|
||
token = raw_token.strip(".,;:!?\"'()[]{}").lower()
|
||
if token in _NUMBER_WORDS:
|
||
return True
|
||
return False
|
||
|
||
|
||
def _has_any_quantity_marker(statement: str, padded_lower: str) -> bool:
|
||
if _DIGIT_RE.search(statement):
|
||
return True
|
||
if _has_number_word(padded_lower):
|
||
return True
|
||
for needle in _INDEFINITE_TOKENS:
|
||
if needle in padded_lower:
|
||
return True
|
||
return False
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Per-category matchers
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def _match_descriptive_setup_no_quantity(
|
||
statement: str, spec: Mapping[str, Any]
|
||
) -> tuple[tuple[Mapping[str, Any], ...], Literal["setup"]] | None:
|
||
"""Match a statement that carries no extractable quantity.
|
||
|
||
Mirrors Phase A's ``_is_descriptive_setup_no_quantity`` predicate —
|
||
a statement with NO digit, NO number word, AND NO indefinite
|
||
quantifier is the canonical setup-context shape.
|
||
|
||
Returns ``(empty parsed_anchors, "setup")`` on a hit; ``None``
|
||
otherwise. The spec's ``quantity_anchor_count`` MUST equal 0 —
|
||
every Phase C synthesis for this category pins that, but we read
|
||
the spec rather than hard-code.
|
||
"""
|
||
if spec.get("quantity_anchor_count") != 0:
|
||
return None
|
||
padded = _padded_lower(statement)
|
||
if _has_any_quantity_marker(statement, padded):
|
||
return None
|
||
return (tuple(), "setup")
|
||
|
||
|
||
def _match_temporal_aggregation(
|
||
statement: str, spec: Mapping[str, Any]
|
||
) -> tuple[tuple[Mapping[str, Any], ...], Literal["aggregate"]] | None:
|
||
"""Match the event_count_per_window shape against *statement*.
|
||
|
||
Narrowness: every extracted anchor's ``window_unit`` and
|
||
``window_quantifier`` MUST appear in the spec's observed sets.
|
||
A statement carrying an unseen window unit / quantifier returns
|
||
``None``.
|
||
"""
|
||
if spec.get("anchor_kind") != "event_count_per_window":
|
||
return None
|
||
observed_units = set(spec.get("observed_window_units") or ())
|
||
observed_quantifiers = set(spec.get("observed_window_quantifiers") or ())
|
||
if not observed_units or not observed_quantifiers:
|
||
return None
|
||
|
||
anchors: list[Mapping[str, Any]] = []
|
||
padded = " " + statement.lower() + " "
|
||
|
||
# Pass 1 — day-of-week enumeration. At least two distinct day
|
||
# names + a count per day yields multi-anchor day-windowed
|
||
# aggregation.
|
||
if "day" in observed_units and ("each" in observed_quantifiers or "every" in observed_quantifiers):
|
||
day_hits: list[tuple[str, str]] = []
|
||
for m in _DAY_HIT_RE.finditer(statement):
|
||
day_hits.append((m.group(1), m.group(2).lower()))
|
||
# Require ≥ 2 distinct day names — same threshold Phase A uses.
|
||
distinct_days = {d for _, d in day_hits}
|
||
if len(distinct_days) >= 2:
|
||
quant = "each" if "each" in observed_quantifiers else "every"
|
||
for count_token, _day in day_hits:
|
||
anchors.append({
|
||
"kind": "event_count_per_window",
|
||
"count_token": count_token,
|
||
"window_unit": "day",
|
||
"window_quantifier": quant,
|
||
})
|
||
if anchors:
|
||
return (tuple(anchors), "aggregate")
|
||
|
||
# Pass 2 — explicit-quantifier and adverbial framings.
|
||
for pat, kind in _TEMPORAL_PATTERNS:
|
||
for m in pat.finditer(statement):
|
||
if kind == "explicit_quantifier":
|
||
count_token, quantifier, unit = m.group(1), m.group(2).lower(), m.group(3).lower()
|
||
elif kind == "in_window":
|
||
count_token, quantifier, unit = m.group(1), "per", m.group(2).lower()
|
||
else: # adverbial
|
||
count_token = m.group(1)
|
||
adverb = m.group(2).lower()
|
||
# Map adverb → unit.
|
||
unit_map = {
|
||
"daily": "day", "weekly": "week", "monthly": "month",
|
||
"yearly": "year", "hourly": "hour",
|
||
}
|
||
unit = unit_map[adverb]
|
||
quantifier = "per"
|
||
if unit not in observed_units:
|
||
continue
|
||
if quantifier not in observed_quantifiers:
|
||
continue
|
||
anchors.append({
|
||
"kind": "event_count_per_window",
|
||
"count_token": count_token,
|
||
"window_unit": unit,
|
||
"window_quantifier": quantifier,
|
||
})
|
||
|
||
if not anchors:
|
||
return None
|
||
|
||
# Spec narrowness: anchor_count must fall within the observed range.
|
||
cmin = int(spec.get("anchor_count_min", 1))
|
||
cmax = int(spec.get("anchor_count_max", 1))
|
||
if not (cmin <= len(anchors) <= cmax):
|
||
return None
|
||
return (tuple(anchors), "aggregate")
|
||
|
||
|
||
def _match_rate_with_currency(
|
||
statement: str, spec: Mapping[str, Any]
|
||
) -> tuple[tuple[Mapping[str, Any], ...], Literal["rate"]] | None:
|
||
"""Match the currency_per_unit_rate shape against *statement*.
|
||
|
||
Narrowness: every extracted anchor's ``currency_symbol`` and
|
||
``per_unit`` MUST be in the spec's observed sets. A statement
|
||
carrying an unseen currency or per-unit value returns ``None``.
|
||
|
||
ME-1 (ADR-0169 composition consumption) — when
|
||
``spec["anchor_kind"] == "currency_per_unit_composition"`` this
|
||
matcher dispatches to :func:`_try_extract_currency_per_unit_composition_anchor`
|
||
which publishes ``composition_shape`` + a pre-composed
|
||
:class:`CandidateInitial` in ``parsed_anchors`` so the consumption
|
||
path in :func:`generate.recognizer_anchor_inject.inject_from_match`
|
||
can admit the statement under an operator-ratified
|
||
``multiplicative_composition`` entry.
|
||
"""
|
||
anchor_kind = spec.get("anchor_kind")
|
||
if anchor_kind == "currency_per_unit_composition":
|
||
return _try_extract_currency_per_unit_composition_anchor(statement, spec)
|
||
if anchor_kind != "currency_per_unit_rate":
|
||
return None
|
||
observed_symbols = set(spec.get("observed_currency_symbols") or ())
|
||
observed_per_units = set(spec.get("observed_per_units") or ())
|
||
if not observed_symbols or not observed_per_units:
|
||
return None
|
||
|
||
anchors: list[Mapping[str, Any]] = []
|
||
for m in _CURRENCY_AMOUNT_RE.finditer(statement):
|
||
symbol = m.group(1)
|
||
amount_token = m.group(2)
|
||
# Per-unit is whichever group captured.
|
||
per_unit = next(
|
||
(g for g in m.groups()[2:] if g),
|
||
None,
|
||
)
|
||
if not per_unit:
|
||
continue
|
||
per_unit_lc = per_unit.lower()
|
||
if symbol not in observed_symbols:
|
||
continue
|
||
if per_unit_lc not in observed_per_units:
|
||
continue
|
||
if "/" in amount_token:
|
||
amount_kind = "word" # fractional surface; Phase B labels as 'word'
|
||
elif "." in amount_token:
|
||
amount_kind = "decimal"
|
||
else:
|
||
amount_kind = "integer"
|
||
anchors.append({
|
||
"kind": "currency_per_unit_rate",
|
||
"currency_symbol": symbol,
|
||
"amount": amount_token,
|
||
"amount_kind": amount_kind,
|
||
"per_unit": per_unit_lc,
|
||
})
|
||
|
||
if not anchors:
|
||
return None
|
||
cmin = int(spec.get("anchor_count_min", 1))
|
||
cmax = int(spec.get("anchor_count_max", 1))
|
||
if not (cmin <= len(anchors) <= cmax):
|
||
return None
|
||
return (tuple(anchors), "rate")
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# ME-1 — currency-per-unit COMPOSITION extension (ADR-0169 consumption).
|
||
#
|
||
# Lights up the dormant consumption path shipped by CW-2: when a
|
||
# statement carries a count-of-items + per-item cost shape with a
|
||
# same-sentence proper-noun subject, this helper publishes a
|
||
# pre-composed CandidateInitial in ``parsed_anchors`` along with the
|
||
# ``composition_shape`` key the composition_registry gates on.
|
||
#
|
||
# Subject-binding discipline: Option A from
|
||
# docs/handoff/MATCHER-EXTENSION-DISPATCH-PACK.md — refuse the
|
||
# composition emission when no same-sentence proper-noun subject is
|
||
# present. Option B (placeholder subject) is forbidden by the brief;
|
||
# Option C (cross-sentence lookup) ships in ME-2.
|
||
# ---------------------------------------------------------------------------
|
||
|
||
_CURRENCY_SYMBOL_TO_UNIT: Final[dict[str, str]] = {
|
||
"$": "dollars",
|
||
"£": "pounds",
|
||
"€": "euros",
|
||
"¥": "yen",
|
||
}
|
||
|
||
_PER_ITEM_TOKENS: Final[frozenset[str]] = frozenset({"each", "apiece"})
|
||
|
||
_COMPOSITION_VERBS: Final[frozenset[str]] = frozenset({"buys", "bought"})
|
||
|
||
_COMPOSITION_SHAPE_MULTIPLICATIVE: Final[str] = "bound(count) × bound(unit_cost)"
|
||
|
||
|
||
# Shape: `<Subject> <buy-verb> <count> <noun-phrase>(?: at| for) <$amount>(?: each|apiece)`
|
||
# Example: "Maria bought 3 vet appointments at $400 each."
|
||
_COMPOSITION_SUBJECT_BUY_RE: Final[re.Pattern[str]] = re.compile(
|
||
r"""(?x)
|
||
^\s*
|
||
(?P<subject>[A-Z][a-zA-Z]+) # same-sentence proper-noun subject
|
||
\s+
|
||
(?P<verb>buys|bought)
|
||
\s+
|
||
(?P<count>\d+(?:\.\d+)?) # outer count token (integer/decimal)
|
||
\s+
|
||
(?P<noun>[a-z][a-z\s]+?) # counted noun phrase (lowercase words)
|
||
\s+
|
||
(?:at|for)\s+
|
||
(?P<symbol>[\$£€¥])
|
||
(?P<amount>\d+(?:\.\d+)?) # unit cost
|
||
\s+
|
||
(?P<per_unit>each|apiece)
|
||
\b
|
||
""",
|
||
)
|
||
|
||
|
||
def _try_extract_currency_per_unit_composition_anchor(
|
||
statement: str, spec: Mapping[str, Any]
|
||
) -> tuple[tuple[Mapping[str, Any], ...], Literal["rate"]] | None:
|
||
"""Extract a pre-composed CandidateInitial anchor or refuse.
|
||
|
||
Narrowness layers (all must hold; any failure returns ``None``):
|
||
|
||
1. ``spec["anchor_kind"] == "currency_per_unit_composition"`` (caller-checked)
|
||
2. ``spec["observed_currency_symbols"]`` and
|
||
``spec["observed_per_units"]`` are non-empty (the same narrowness
|
||
as the rate matcher; protects against unseen currency / per-unit)
|
||
3. Exactly one match of :data:`_COMPOSITION_SUBJECT_BUY_RE` in the
|
||
statement (multi-match refuses to avoid ambiguity)
|
||
4. Currency symbol in ``observed_currency_symbols``
|
||
5. Per-unit token in ``observed_per_units``
|
||
6. Outer count is a positive integer or float (``> 0``)
|
||
7. Unit cost is a positive integer or float (``> 0``)
|
||
8. The composed value (``count × unit_cost``) is finite and positive
|
||
9. The subject is not in the existing refused-subject set
|
||
(mirrors ``_REFUSED_SUBJECT_TOKENS`` for parity with the
|
||
discrete-count extractor)
|
||
|
||
On success the anchor carries:
|
||
|
||
- ``composition_shape``: the canonical pattern string ratified
|
||
operators bind under ``multiplicative_composition``
|
||
- ``composed_initial``: a fully-constructed CandidateInitial
|
||
- audit fields: ``currency_symbol``, ``amount``, ``per_unit``,
|
||
``outer_count``, ``subject``, ``verb``
|
||
"""
|
||
observed_symbols = set(spec.get("observed_currency_symbols") or ())
|
||
observed_per_units = set(spec.get("observed_per_units") or ())
|
||
if not observed_symbols or not observed_per_units:
|
||
return None
|
||
|
||
matches = list(_COMPOSITION_SUBJECT_BUY_RE.finditer(statement))
|
||
if len(matches) != 1:
|
||
# Refusal-preferring: zero matches (no shape) or multi-match
|
||
# (ambiguity) both refuse the composition emission.
|
||
return None
|
||
|
||
m = matches[0]
|
||
subject = m.group("subject")
|
||
if subject.lower() in _REFUSED_SUBJECT_TOKENS:
|
||
return None
|
||
verb = m.group("verb").lower()
|
||
if verb not in _COMPOSITION_VERBS:
|
||
return None
|
||
|
||
symbol = m.group("symbol")
|
||
if symbol not in observed_symbols:
|
||
return None
|
||
|
||
per_unit_lc = m.group("per_unit").lower()
|
||
if per_unit_lc not in observed_per_units:
|
||
return None
|
||
if per_unit_lc not in _PER_ITEM_TOKENS:
|
||
# Defense in depth: only per-item quantifiers compose
|
||
# multiplicatively in the v1 scope. ``per hour`` is rate, not
|
||
# composition.
|
||
return None
|
||
|
||
count_token = m.group("count")
|
||
amount_token = m.group("amount")
|
||
try:
|
||
outer_count: float = float(count_token)
|
||
unit_cost: float = float(amount_token)
|
||
except ValueError:
|
||
return None
|
||
if outer_count <= 0 or unit_cost <= 0:
|
||
return None
|
||
|
||
composed_value_f = outer_count * unit_cost
|
||
if composed_value_f != composed_value_f: # NaN guard
|
||
return None
|
||
composed_value: int | float
|
||
if composed_value_f.is_integer() and "." not in count_token and "." not in amount_token:
|
||
composed_value = int(composed_value_f)
|
||
else:
|
||
composed_value = composed_value_f
|
||
|
||
unit = _CURRENCY_SYMBOL_TO_UNIT.get(symbol)
|
||
if unit is None:
|
||
return None # Defense in depth — observed set should already filter
|
||
|
||
# Lazy import: CandidateInitial / InitialPossession / Quantity live
|
||
# in modules that don't depend on recognizer_match — import here to
|
||
# avoid coupling at module load time.
|
||
from generate.math_candidate_parser import CandidateInitial
|
||
from generate.math_problem_graph import InitialPossession, Quantity
|
||
|
||
composed_initial = CandidateInitial(
|
||
initial=InitialPossession(
|
||
entity=subject,
|
||
quantity=Quantity(value=composed_value, unit=unit),
|
||
),
|
||
source_span=m.group(0),
|
||
matched_anchor=verb,
|
||
matched_value_token=str(composed_value),
|
||
matched_unit_token=unit,
|
||
matched_entity_token=subject,
|
||
)
|
||
|
||
anchor: Mapping[str, Any] = {
|
||
"kind": "currency_per_unit_composition",
|
||
"composition_shape": _COMPOSITION_SHAPE_MULTIPLICATIVE,
|
||
"composed_initial": composed_initial,
|
||
"currency_symbol": symbol,
|
||
"amount": amount_token,
|
||
"per_unit": per_unit_lc,
|
||
"outer_count": count_token,
|
||
"subject": subject,
|
||
"verb": verb,
|
||
}
|
||
return ((anchor,), "rate")
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# ME-2 — cross-sentence subject binding (admits case 0019).
|
||
#
|
||
# Case 0019: "John adopts a dog from a shelter. The dog ends up having
|
||
# health problems and this requires 3 vet appointments, which cost
|
||
# $400 each."
|
||
#
|
||
# The composition sentence has no same-sentence proper-noun subject —
|
||
# "John" lives in sentence 0. ME-1 (Option A) refuses; ME-2 admits
|
||
# when the caller supplies a ``prior_subject`` resolved from the
|
||
# upstream sentence trace.
|
||
#
|
||
# Discipline:
|
||
# - The cross-sentence regex requires NO subject prefix; instead it
|
||
# keys on a discourse-anaphoric introduction like "which cost $X each"
|
||
# or "and this requires N noun" + "$X each" in the same sentence.
|
||
# - Caller is responsible for providing a confidence-pinned prior
|
||
# subject (most-recent proper-noun subject from prior sentences).
|
||
# - The matcher refuses if prior_subject is None / empty / refused.
|
||
# ---------------------------------------------------------------------------
|
||
|
||
# Shape: `... which cost(s)? $<amount> each` plus a preceding count token.
|
||
# Constructed so the count + noun are pulled from the same statement, but
|
||
# the subject is supplied externally.
|
||
_CROSS_SENTENCE_COMPOSITION_RE: Final[re.Pattern[str]] = re.compile(
|
||
r"""(?ix)
|
||
\b
|
||
(?:requires|require|needs|need|costs|cost)
|
||
\s+
|
||
(?P<count>\d+(?:\.\d+)?) # outer count token
|
||
\s+
|
||
(?P<noun>[a-z][a-z\s]+?) # counted noun phrase
|
||
,?\s+
|
||
(?:which\s+)?
|
||
(?:cost|costs|costing)
|
||
\s+
|
||
(?P<symbol>[\$£€¥])
|
||
(?P<amount>\d+(?:\.\d+)?)
|
||
\s+
|
||
(?P<per_unit>each|apiece)
|
||
\b
|
||
""",
|
||
)
|
||
|
||
|
||
def try_extract_cross_sentence_composition_anchor(
|
||
statement: str,
|
||
spec: Mapping[str, Any],
|
||
prior_subject: str | None,
|
||
) -> tuple[tuple[Mapping[str, Any], ...], Literal["rate"]] | None:
|
||
"""Cross-sentence composition extraction.
|
||
|
||
Like :func:`_try_extract_currency_per_unit_composition_anchor` but
|
||
sources the subject from ``prior_subject`` instead of a
|
||
same-sentence head proper-noun.
|
||
|
||
Refuses when:
|
||
|
||
- ``prior_subject`` is None / empty / in :data:`_REFUSED_SUBJECT_TOKENS`
|
||
- the cross-sentence regex matches zero or multiple times
|
||
- currency / per-unit / count narrowness fail (mirrors ME-1)
|
||
|
||
The same composition_shape + composed_initial payload as ME-1 is
|
||
published. The consumer (composition_registry) gates admission.
|
||
"""
|
||
if spec.get("anchor_kind") != "currency_per_unit_composition":
|
||
return None
|
||
if not prior_subject or not isinstance(prior_subject, str):
|
||
return None
|
||
if prior_subject.lower() in _REFUSED_SUBJECT_TOKENS:
|
||
return None
|
||
|
||
observed_symbols = set(spec.get("observed_currency_symbols") or ())
|
||
observed_per_units = set(spec.get("observed_per_units") or ())
|
||
if not observed_symbols or not observed_per_units:
|
||
return None
|
||
|
||
matches = list(_CROSS_SENTENCE_COMPOSITION_RE.finditer(statement))
|
||
if len(matches) != 1:
|
||
return None
|
||
|
||
m = matches[0]
|
||
symbol = m.group("symbol")
|
||
if symbol not in observed_symbols:
|
||
return None
|
||
per_unit_lc = m.group("per_unit").lower()
|
||
if per_unit_lc not in observed_per_units:
|
||
return None
|
||
if per_unit_lc not in _PER_ITEM_TOKENS:
|
||
return None
|
||
|
||
count_token = m.group("count")
|
||
amount_token = m.group("amount")
|
||
try:
|
||
outer_count: float = float(count_token)
|
||
unit_cost: float = float(amount_token)
|
||
except ValueError:
|
||
return None
|
||
if outer_count <= 0 or unit_cost <= 0:
|
||
return None
|
||
|
||
composed_value_f = outer_count * unit_cost
|
||
if composed_value_f != composed_value_f: # NaN guard
|
||
return None
|
||
composed_value: int | float
|
||
if (
|
||
composed_value_f.is_integer()
|
||
and "." not in count_token
|
||
and "." not in amount_token
|
||
):
|
||
composed_value = int(composed_value_f)
|
||
else:
|
||
composed_value = composed_value_f
|
||
|
||
unit = _CURRENCY_SYMBOL_TO_UNIT.get(symbol)
|
||
if unit is None:
|
||
return None
|
||
|
||
from generate.math_candidate_parser import CandidateInitial
|
||
from generate.math_problem_graph import InitialPossession, Quantity
|
||
|
||
# Validate prior_subject can satisfy CandidateInitial.entity.
|
||
entity = prior_subject.strip()
|
||
if not entity:
|
||
return None
|
||
|
||
composed_initial = CandidateInitial(
|
||
initial=InitialPossession(
|
||
entity=entity,
|
||
quantity=Quantity(value=composed_value, unit=unit),
|
||
),
|
||
source_span=m.group(0),
|
||
matched_anchor="bought", # canonical buy-anchor for the whitelist
|
||
matched_value_token=str(composed_value),
|
||
matched_unit_token=unit,
|
||
matched_entity_token=entity,
|
||
)
|
||
|
||
anchor: Mapping[str, Any] = {
|
||
"kind": "currency_per_unit_composition_cross_sentence",
|
||
"composition_shape": _COMPOSITION_SHAPE_MULTIPLICATIVE,
|
||
"composed_initial": composed_initial,
|
||
"currency_symbol": symbol,
|
||
"amount": amount_token,
|
||
"per_unit": per_unit_lc,
|
||
"outer_count": count_token,
|
||
"subject": entity,
|
||
"subject_source": "prior_sentence",
|
||
}
|
||
return ((anchor,), "rate")
|
||
|
||
|
||
# Refused subjects mirrors the constant defined later in this module
|
||
# (used by both the same-sentence and cross-sentence extractors).
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# ADR-0163.B.2 round-2 matchers. Detection-only (return empty
|
||
# parsed_anchors) — consistent with Phase D's skip-only wiring. Real
|
||
# value extraction lands when Phase D.2 plumbs parsed_anchors into the
|
||
# solver. Narrowness is enforced via shape predicates (no currency on a
|
||
# discrete-count match; no "per X" on a currency_amount match; etc.).
|
||
# ---------------------------------------------------------------------------
|
||
|
||
_PER_UNIT_TOKENS: Final[tuple[str, ...]] = (
|
||
" per ", "/", " an hour", " a hour", " a day", " a week", " a month",
|
||
" a year", " for one ", " for each ", " for every ",
|
||
)
|
||
|
||
_TEMPORAL_QUANTIFIER_TOKENS: Final[tuple[str, ...]] = (
|
||
" per ", " each ", " every ", " daily", " weekly", " monthly",
|
||
" yearly", " hourly",
|
||
)
|
||
|
||
_MULTIPLICATIVE_CONNECTIVES: Final[tuple[str, ...]] = (
|
||
" with ", " each ", " in each ", " per each ",
|
||
)
|
||
|
||
|
||
def _has_per_unit_framing(padded_lower: str) -> bool:
|
||
return any(tok in padded_lower for tok in _PER_UNIT_TOKENS)
|
||
|
||
|
||
def _has_temporal_quantifier(padded_lower: str) -> bool:
|
||
return any(tok in padded_lower for tok in _TEMPORAL_QUANTIFIER_TOKENS)
|
||
|
||
|
||
def _has_currency_symbol(statement: str) -> bool:
|
||
return any(c in statement for c in "$£€¥")
|
||
|
||
|
||
def _match_discrete_count_statement(
|
||
statement: str, spec: Mapping[str, Any]
|
||
) -> tuple[tuple[Mapping[str, Any], ...], Literal["count"]] | None:
|
||
"""ADR-0163.D.2 — extraction match for "X has N Y" shape.
|
||
|
||
Detection conditions (same as round-2 detection-only matcher):
|
||
- statement carries ≥1 quantity marker (digit or number word)
|
||
- statement does NOT carry a currency symbol (else currency_amount)
|
||
- statement does NOT carry per-unit framing (else rate_with_currency)
|
||
- statement does NOT carry temporal-quantifier framing
|
||
(else temporal_aggregation)
|
||
- spec's anchor_kind is "discrete_count"
|
||
|
||
Extraction (D.2 v1) populates a SINGLE anchor when ALL of the
|
||
following narrowness rules hold; otherwise returns
|
||
``(empty parsed_anchors, "count")`` (detection-only fallback, same
|
||
skip-only safety as round 2). Narrowness layers (refusal-preferring,
|
||
wrong=0 doctrine):
|
||
|
||
1. Statement matches the canonical possession form
|
||
``<ProperNoun> <poss-verb> <count> <counted_noun>...``.
|
||
Subject must be a single capitalized proper noun (no
|
||
conjunctions, no leading pronoun). Possession verb must come
|
||
from the v1 closed whitelist (has/have/had); broader verbs
|
||
(owns/holds/contains) defer to a coordinated CandidateInitial
|
||
change in a follow-up PR.
|
||
2. Statement carries exactly ONE numeric token (digit or word
|
||
numeral) — a second count indicates multi-anchor content the
|
||
v1 schema cannot honor; refuse extraction.
|
||
3. Statement contains no clause-splitting connectives (``but``,
|
||
``then``, ``however``, ``before``, ``after``, ``and``,
|
||
``or``) — these indicate trailing operations or enumerations
|
||
that would invalidate a single InitialPossession.
|
||
4. count_kind ∈ spec.observed_count_kinds.
|
||
5. counted_noun ∈ spec.observed_counted_nouns (case-insensitive,
|
||
matched against the closed lemma list from Phase B/C).
|
||
|
||
The matcher returns ``(populated parsed_anchors, "count")`` on
|
||
extraction success, ``(tuple(), "count")`` on detection-only
|
||
fallback (skip-only safe), or ``None`` on detection failure.
|
||
Phase D.2's per-category injector consumes the populated anchors;
|
||
the empty-tuple fallback continues the round-2 skip-only behavior.
|
||
"""
|
||
if spec.get("anchor_kind") != "discrete_count":
|
||
return None
|
||
padded = _padded_lower(statement)
|
||
if not _has_any_quantity_marker(statement, padded):
|
||
return None
|
||
if _has_currency_symbol(statement):
|
||
return None
|
||
if _has_per_unit_framing(padded):
|
||
return None
|
||
if _has_temporal_quantifier(padded):
|
||
return None
|
||
|
||
anchor = _try_extract_discrete_count_anchor(statement, padded, spec)
|
||
if anchor is not None:
|
||
return ((anchor,), "count")
|
||
return (tuple(), "count")
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# ADR-0163.D.2 — discrete_count_statement value extraction (v1).
|
||
# ---------------------------------------------------------------------------
|
||
|
||
# Closed possession-verb whitelist. These verbs assert a static
|
||
# possession state (no goal, no acquisition event, no transfer). Verbs
|
||
# like 'collected', 'wants', 'lost', 'bought', etc. are deliberately
|
||
# omitted — they encode operations, not initial state, and admitting
|
||
# them as InitialPossession would over-extract.
|
||
#
|
||
# v1 intentionally restricts the surface to has/have/had so the
|
||
# extracted matched_anchor token is always accepted by the downstream
|
||
# CandidateInitial post-init whitelist. Widening to owns/holds/contains
|
||
# requires a coordinated CandidateInitial change and lands in a follow-up
|
||
# PR after the framework's empirical lift is operator-reviewed.
|
||
_POSSESSION_VERBS: Final[frozenset[str]] = frozenset({
|
||
"has", "have", "had",
|
||
})
|
||
|
||
# ADR-0170 W2 — acquisition verbs: surface verbs that grammatically place
|
||
# the actor as the *gainer* of the operand quantity, NOT as having the
|
||
# operand as an initial state. Per ADR-0131.G.1 these verbs route to
|
||
# CandidateOperation(add), not CandidateInitial — emitting them as
|
||
# initials would create branch disagreement with the regex parser's
|
||
# ADD_VERBS path.
|
||
#
|
||
# Each member is also a member of generate.math_roundtrip.ADD_VERBS so
|
||
# the downstream CandidateOperation post-init whitelist accepts the
|
||
# matched_verb token.
|
||
#
|
||
# DELIBERATELY EXCLUDED:
|
||
# - "gained / gains / gain": delta-of-attribute (weight, age) — admitting
|
||
# as add-operation risks wrong>0 on questions that ask total state
|
||
# - "donated / donates / donate": SUBTRACT verb (actor gives away)
|
||
# - "saved / saves / save": ambiguous (saved time vs saved up money)
|
||
#
|
||
# Widening this set is operator-reviewable per the wrong=0 hazard
|
||
# documented in feedback-wrong-zero-hazard-case-0050.
|
||
_ACQUISITION_VERBS: Final[frozenset[str]] = frozenset({
|
||
"collected", "collects", "collect",
|
||
"received", "receives", "receive",
|
||
"bought", "buys", "buy",
|
||
"got", "gets", "get",
|
||
})
|
||
|
||
# Pronoun subjects refused at extraction (ambiguous referent). The
|
||
# extractor requires a concrete proper-noun subject the source span can
|
||
# ground.
|
||
_REFUSED_SUBJECT_TOKENS: Final[frozenset[str]] = frozenset({
|
||
"he", "she", "they", "it", "we", "you", "i",
|
||
"him", "her", "them", "us",
|
||
})
|
||
|
||
# Clause-splitting / enumeration markers. Their presence indicates a
|
||
# second clause that may carry operations or additional anchors, so
|
||
# v1 refuses extraction (skip-only fallback preserves wrong=0).
|
||
_CLAUSE_SPLIT_TOKENS: Final[tuple[str, ...]] = (
|
||
" but ", " then ", " however ", " before ", " after ",
|
||
" and ", " or ", " while ", " until ", " unless ",
|
||
", and ", ", but ", ", or ", ", then ",
|
||
)
|
||
|
||
# Hyphenated compound cardinal: 'twenty-five', 'ninety-nine'. These
|
||
# are word-form counts. The narrowness rule below classifies any
|
||
# non-digit token in the count slot as count_kind='word'.
|
||
_HYPHEN_CARDINAL_RE: Final[re.Pattern[str]] = re.compile(r"^[a-z]+-[a-z]+$")
|
||
|
||
|
||
def _extract_discrete_count_re_for(counted_nouns: list[str]) -> re.Pattern[str]:
|
||
"""Build the extraction regex for a closed counted-noun set.
|
||
|
||
The counted-noun alternation is constructed from the spec's
|
||
``observed_counted_nouns``; multi-word nouns (e.g., ``Pokemon cards``)
|
||
are honored verbatim. Longest-first to prevent the alternation
|
||
swallowing a prefix.
|
||
"""
|
||
# Sort longest-first so 'Pokemon cards' wins over 'cards'.
|
||
options = sorted({n for n in counted_nouns if n}, key=len, reverse=True)
|
||
noun_alt = "|".join(re.escape(n) for n in options)
|
||
return re.compile(
|
||
r"^\s*"
|
||
r"(?P<subject>(?-i:[A-Z][a-z]+))" # case-sensitive proper noun
|
||
r"\s+(?P<verb>[A-Za-z]+)" # any word; verified against whitelist
|
||
r"\s+(?P<count>\d+|[A-Za-z\-]+)" # integer or word/hyphenated cardinal
|
||
r"\s+(?P<noun>" + noun_alt + r")"
|
||
r"(?:\b.*)?$", # optional trailing content
|
||
flags=re.IGNORECASE,
|
||
)
|
||
|
||
|
||
_DIGIT_RUN_RE: Final[re.Pattern[str]] = re.compile(r"\d+(?:\.\d+)?")
|
||
|
||
|
||
def _count_quantity_tokens(statement: str, padded_lower: str) -> int:
|
||
"""Total numeric tokens (digit runs + number words) in *statement*.
|
||
|
||
Used for the "exactly one count" narrowness rule. Hyphenated
|
||
cardinals count as one token; a multi-digit integer (``400``) counts
|
||
as one token, not as multiple single-digit hits.
|
||
"""
|
||
digit_hits = len(_DIGIT_RUN_RE.findall(statement))
|
||
word_hits = 0
|
||
for raw in padded_lower.split():
|
||
tok = raw.strip(".,;:!?\"'()[]{}").lower()
|
||
if tok in _NUMBER_WORDS:
|
||
word_hits += 1
|
||
elif _HYPHEN_CARDINAL_RE.match(tok):
|
||
# Hyphenated cardinal only counts when at least one half is
|
||
# a known number word.
|
||
left, _, right = tok.partition("-")
|
||
if left in _NUMBER_WORDS or right in _NUMBER_WORDS:
|
||
word_hits += 1
|
||
return digit_hits + word_hits
|
||
|
||
|
||
def _try_extract_discrete_count_anchor(
|
||
statement: str,
|
||
padded_lower: str,
|
||
spec: Mapping[str, Any],
|
||
) -> Mapping[str, Any] | None:
|
||
"""Refusal-preferring single-anchor extraction (D.2 v1).
|
||
|
||
Returns ``None`` when any narrowness layer fails — the caller then
|
||
falls back to skip-only detection. The returned anchor is the
|
||
discrete_count_statement schema dict: ``kind``, ``subject_role``,
|
||
``count_token``, ``count_kind``, ``counted_noun``.
|
||
"""
|
||
raw_kinds = spec.get("observed_count_kinds") or ()
|
||
raw_nouns = spec.get("observed_counted_nouns") or ()
|
||
observed_kinds: list[str] = [str(k) for k in raw_kinds]
|
||
observed_nouns: list[str] = [str(n) for n in raw_nouns]
|
||
if not observed_kinds or not observed_nouns:
|
||
return None
|
||
|
||
# Narrowness #3 — clause-split / enumeration markers.
|
||
for token in _CLAUSE_SPLIT_TOKENS:
|
||
if token in padded_lower:
|
||
return None
|
||
|
||
# Narrowness #2 — exactly one numeric token.
|
||
if _count_quantity_tokens(statement, padded_lower) != 1:
|
||
return None
|
||
|
||
# Narrowness #1 + #5 — shape + counted-noun lemma.
|
||
extract_re = _extract_discrete_count_re_for(observed_nouns)
|
||
m = extract_re.match(statement.strip())
|
||
if m is None:
|
||
return None
|
||
|
||
subject = m.group("subject")
|
||
if subject.lower() in _REFUSED_SUBJECT_TOKENS:
|
||
return None
|
||
|
||
verb = m.group("verb").lower()
|
||
if verb in _POSSESSION_VERBS:
|
||
anchor_kind = "possession"
|
||
elif verb in _ACQUISITION_VERBS:
|
||
anchor_kind = "acquisition"
|
||
else:
|
||
return None
|
||
|
||
count_token = m.group("count")
|
||
if count_token.isdigit():
|
||
count_kind = "integer"
|
||
else:
|
||
# Word-form cardinal — must be a known number word (single or
|
||
# hyphenated compound). Anything else is unrecognized and the
|
||
# extractor refuses.
|
||
lc = count_token.lower()
|
||
if lc in _NUMBER_WORDS:
|
||
count_kind = "word"
|
||
elif _HYPHEN_CARDINAL_RE.match(lc):
|
||
left, _, right = lc.partition("-")
|
||
if left in _NUMBER_WORDS or right in _NUMBER_WORDS:
|
||
count_kind = "word"
|
||
else:
|
||
return None
|
||
else:
|
||
return None
|
||
|
||
# Narrowness #4 — count_kind in observed set.
|
||
if count_kind not in observed_kinds:
|
||
return None
|
||
|
||
counted_noun = m.group("noun")
|
||
# Canonicalize counted_noun to the spec's observed casing where
|
||
# available; fall back to literal surface.
|
||
canon = counted_noun
|
||
counted_noun_lc = counted_noun.lower()
|
||
for observed_n in observed_nouns:
|
||
if observed_n.lower() == counted_noun_lc:
|
||
canon = observed_n
|
||
break
|
||
|
||
return {
|
||
"kind": "discrete_count",
|
||
"subject_role": subject,
|
||
"count_token": count_token,
|
||
"count_kind": count_kind,
|
||
"counted_noun": canon,
|
||
# ADR-0170 W2 — anchor_kind discriminates the downstream
|
||
# injector path: "possession" → CandidateInitial (existing);
|
||
# "acquisition" → CandidateOperation(add) (new).
|
||
"anchor_kind": anchor_kind,
|
||
"verb_token": verb,
|
||
}
|
||
|
||
|
||
def _match_multiplicative_aggregation(
|
||
statement: str, spec: Mapping[str, Any]
|
||
) -> tuple[tuple[Mapping[str, Any], ...], Literal["aggregate"]] | None:
|
||
"""Detection-only match for "M outer × N inner" shape.
|
||
|
||
Conditions:
|
||
- spec's anchor_kind is "multiplicative_aggregate"
|
||
- statement carries a multiplicative connective
|
||
("with", "each holds", "in each", etc.)
|
||
- statement carries ≥2 quantity markers (the outer + inner counts)
|
||
- statement does NOT carry currency-per-unit framing
|
||
|
||
Returns ``(empty parsed_anchors, "aggregate")`` on a hit.
|
||
|
||
ME-3 (ADR-0169 additive composition) — when
|
||
``spec["anchor_kind"] == "additive_quantity_composition"`` this
|
||
matcher dispatches to :func:`_try_extract_additive_composition_anchor`
|
||
which publishes ``composition_shape`` + a pre-composed
|
||
:class:`CandidateInitial` in ``parsed_anchors`` for two same-unit
|
||
quantities connected by ``and``. The graph_intent is widened from
|
||
``"aggregate"`` to also include ``"additive"`` so the dispatcher in
|
||
:func:`match` can recognize composition emissions.
|
||
"""
|
||
anchor_kind = spec.get("anchor_kind")
|
||
if anchor_kind == "additive_quantity_composition":
|
||
# ME-3 dispatch — same Literal narrowing keeps the return type
|
||
# consistent ('aggregate' is reused).
|
||
return _try_extract_additive_composition_anchor(statement, spec)
|
||
if anchor_kind == "subtractive_quantity_composition":
|
||
# ME-4 dispatch — subtractive shape returns ("amount" intent).
|
||
# Cast the Literal return: the matcher signature widens at the
|
||
# type level to include any 'aggregate'/'amount' but the caller
|
||
# in match() reads graph_intent verbatim.
|
||
sub_result = _try_extract_subtractive_composition_anchor(statement, spec)
|
||
if sub_result is None:
|
||
return None
|
||
anchors, _ = sub_result
|
||
return (anchors, "aggregate") # reuse 'aggregate' label for subtractive too
|
||
if anchor_kind != "multiplicative_aggregate":
|
||
return None
|
||
padded = _padded_lower(statement)
|
||
if not any(c in padded for c in _MULTIPLICATIVE_CONNECTIVES):
|
||
return None
|
||
# Count distinct quantity markers (digits + number words). At least
|
||
# two needed to admit a multiplicative shape.
|
||
digit_hits = len(_DIGIT_RE.findall(statement))
|
||
word_hits = sum(
|
||
1 for token in padded.split()
|
||
if token.strip(".,;:!?\"'()[]{}").lower() in _NUMBER_WORDS
|
||
)
|
||
if (digit_hits + word_hits) < 2:
|
||
return None
|
||
if _has_currency_symbol(statement) and _has_per_unit_framing(padded):
|
||
return None
|
||
return (tuple(), "aggregate")
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# ME-3 — additive composition matcher.
|
||
#
|
||
# Admits "<count_a> <unit> and <count_b> <unit>" shape (same unit) and
|
||
# emits a pre-composed CandidateInitial whose value is the sum.
|
||
#
|
||
# Subject-binding discipline:
|
||
# - SAME-SENTENCE proper-noun subject preferred (Option A from ME-1).
|
||
# - When absent, the caller MAY supply ``prior_subject`` via the match()
|
||
# dispatcher (ME-2 path); the ME-3 helper does NOT itself consult
|
||
# ``prior_subject`` — that path is reserved for the cross-sentence
|
||
# composition extension (a future ME-3b if needed). v1 ME-3 narrowness
|
||
# matches the dispatch pack: refuse on subject-absent.
|
||
# - Pronoun subject refused (mirrors existing _REFUSED_SUBJECT_TOKENS).
|
||
# ---------------------------------------------------------------------------
|
||
|
||
_ADDITIVE_TWO_QUANTITY_RE: Final[re.Pattern[str]] = re.compile(
|
||
r"""(?ix)
|
||
^\s*
|
||
(?P<subject>[A-Z][a-zA-Z]+)
|
||
\s+
|
||
(?P<verb>lost|gained|earned|saved|made|paid|spent|bought|sold|added|removed|received)
|
||
\s+
|
||
(?P<count_a>\d+(?:\.\d+)?)
|
||
\s+
|
||
(?P<unit_a>[a-z]+)
|
||
(?:\s+[a-z]+\s+[a-z]+)? # optional time/location phrase like "in March"
|
||
\s+and\s+
|
||
(?P<count_b>\d+(?:\.\d+)?)
|
||
\s+
|
||
(?P<unit_b>[a-z]+)
|
||
(?:\s+[a-z]+\s+[a-z]+)? # optional second time/location phrase
|
||
\b
|
||
""",
|
||
)
|
||
|
||
_ADDITIVE_COMPOSITION_SHAPE: Final[str] = "bound(qty_a) + bound(qty_b)"
|
||
|
||
|
||
def _try_extract_additive_composition_anchor(
|
||
statement: str, spec: Mapping[str, Any]
|
||
) -> tuple[tuple[Mapping[str, Any], ...], Literal["aggregate"]] | None:
|
||
"""Extract a pre-composed CandidateInitial for additive composition.
|
||
|
||
Narrowness layers (all required):
|
||
|
||
1. ``spec["anchor_kind"] == "additive_quantity_composition"`` (caller)
|
||
2. ``spec["observed_units"]`` is non-empty
|
||
3. Exactly one match of :data:`_ADDITIVE_TWO_QUANTITY_RE`
|
||
4. ``unit_a == unit_b`` (same-unit composition only; cross-unit
|
||
addition is ill-defined without a conversion table — refuse)
|
||
5. Both unit tokens in ``observed_units``
|
||
6. Both counts are positive
|
||
7. Subject is a proper noun not in :data:`_REFUSED_SUBJECT_TOKENS`
|
||
8. Verb in :data:`_ADDITIVE_COMPOSITION_VERBS`
|
||
|
||
Refuses on any failure; refusal-preferring discipline.
|
||
"""
|
||
if spec.get("anchor_kind") != "additive_quantity_composition":
|
||
return None
|
||
observed_units = set(spec.get("observed_units") or ())
|
||
if not observed_units:
|
||
return None
|
||
|
||
matches = list(_ADDITIVE_TWO_QUANTITY_RE.finditer(statement))
|
||
if len(matches) != 1:
|
||
return None
|
||
|
||
m = matches[0]
|
||
subject = m.group("subject")
|
||
if subject.lower() in _REFUSED_SUBJECT_TOKENS:
|
||
return None
|
||
if subject.lower() in _COMMON_DETERMINERS_AT_HEAD:
|
||
return None
|
||
|
||
verb = m.group("verb").lower()
|
||
if verb not in _ADDITIVE_COMPOSITION_VERBS:
|
||
return None
|
||
|
||
unit_a = m.group("unit_a").lower()
|
||
unit_b = m.group("unit_b").lower()
|
||
# Strip trailing 's' for plural normalization on the comparison
|
||
# (apples vs apple). Refuse on stem mismatch.
|
||
if unit_a.rstrip("s") != unit_b.rstrip("s"):
|
||
return None
|
||
canonical_unit = unit_a
|
||
if canonical_unit not in observed_units and canonical_unit.rstrip("s") not in observed_units:
|
||
return None
|
||
|
||
count_a_token = m.group("count_a")
|
||
count_b_token = m.group("count_b")
|
||
try:
|
||
count_a = float(count_a_token)
|
||
count_b = float(count_b_token)
|
||
except ValueError:
|
||
return None
|
||
if count_a <= 0 or count_b <= 0:
|
||
return None
|
||
|
||
composed_value_f = count_a + count_b
|
||
if composed_value_f != composed_value_f: # NaN guard
|
||
return None
|
||
composed_value: int | float
|
||
if (
|
||
composed_value_f.is_integer()
|
||
and "." not in count_a_token
|
||
and "." not in count_b_token
|
||
):
|
||
composed_value = int(composed_value_f)
|
||
else:
|
||
composed_value = composed_value_f
|
||
|
||
from generate.math_candidate_parser import CandidateInitial
|
||
from generate.math_problem_graph import InitialPossession, Quantity
|
||
|
||
# Verb whitelist maps to a CandidateInitial.matched_anchor value
|
||
# the post-init guard accepts (existing whitelist includes
|
||
# has/have/had/saved/earned/got/received/bought/made/paid).
|
||
matched_anchor = verb if verb in {
|
||
"saved", "earned", "got", "received", "bought", "made", "paid"
|
||
} else "had"
|
||
|
||
composed_initial = CandidateInitial(
|
||
initial=InitialPossession(
|
||
entity=subject,
|
||
quantity=Quantity(value=composed_value, unit=canonical_unit),
|
||
),
|
||
source_span=m.group(0),
|
||
matched_anchor=matched_anchor,
|
||
matched_value_token=str(composed_value),
|
||
matched_unit_token=canonical_unit,
|
||
matched_entity_token=subject,
|
||
)
|
||
|
||
anchor: Mapping[str, Any] = {
|
||
"kind": "additive_quantity_composition",
|
||
"composition_shape": _ADDITIVE_COMPOSITION_SHAPE,
|
||
"composed_initial": composed_initial,
|
||
"count_a": count_a_token,
|
||
"count_b": count_b_token,
|
||
"unit": canonical_unit,
|
||
"subject": subject,
|
||
"verb": verb,
|
||
}
|
||
return ((anchor,), "aggregate")
|
||
|
||
|
||
_ADDITIVE_COMPOSITION_VERBS: Final[frozenset[str]] = frozenset({
|
||
"lost", "gained", "earned", "saved", "made", "paid", "spent",
|
||
"bought", "sold", "added", "removed", "received",
|
||
})
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# ME-4 — subtractive composition matcher.
|
||
#
|
||
# Admits "<Subject> <init-verb> <N> <unit>(,| then|; etc.) <sub-verb>
|
||
# <M> <unit>" (same unit; positive initial verb followed by removal
|
||
# verb) and emits a pre-composed CandidateInitial(N - M, unit).
|
||
#
|
||
# Refusal-preferring discipline: count_b >= count_a → refuse
|
||
# (non-negative remainder; subtractive composition that goes below
|
||
# zero is a wrong>0 hazard).
|
||
# ---------------------------------------------------------------------------
|
||
|
||
_SUBTRACTIVE_TWO_QUANTITY_RE: Final[re.Pattern[str]] = re.compile(
|
||
r"""(?ix)
|
||
^\s*
|
||
(?P<subject>[A-Z][a-zA-Z]+)
|
||
\s+
|
||
(?P<verb_a>had|has|got|owns|owned|earned|saved|made|received|bought)
|
||
\s+
|
||
(?P<count_a>\d+(?:\.\d+)?)
|
||
\s+
|
||
(?P<unit_a>[a-z]+)
|
||
\s*
|
||
(?:,|\sthen\s|;|\s+and\s+then\s+|\s+then\s+|\s+and\s+)
|
||
\s*
|
||
(?:then\s+)?
|
||
(?P<verb_b>lost|spent|gave|donated|paid|removed|sold|used|consumed)
|
||
(?:\s+away)?
|
||
\s+
|
||
(?P<count_b>\d+(?:\.\d+)?)
|
||
\s+
|
||
(?P<unit_b>[a-z]+)
|
||
\b
|
||
""",
|
||
)
|
||
|
||
_SUBTRACTIVE_COMPOSITION_SHAPE: Final[str] = "bound(initial) − bound(removed)"
|
||
|
||
|
||
_SUBTRACTIVE_INITIAL_VERBS: Final[frozenset[str]] = frozenset({
|
||
"had", "has", "got", "owns", "owned", "earned", "saved",
|
||
"made", "received", "bought",
|
||
})
|
||
|
||
_SUBTRACTIVE_REMOVAL_VERBS: Final[frozenset[str]] = frozenset({
|
||
"lost", "spent", "gave", "donated", "paid", "removed",
|
||
"sold", "used", "consumed",
|
||
})
|
||
|
||
|
||
def _try_extract_subtractive_composition_anchor(
|
||
statement: str, spec: Mapping[str, Any]
|
||
) -> tuple[tuple[Mapping[str, Any], ...], Literal["amount"]] | None:
|
||
"""Extract a pre-composed CandidateInitial for subtractive composition.
|
||
|
||
See module docstring above for narrowness layers. ``count_b >=
|
||
count_a`` refuses (non-negative remainder discipline; wrong>0
|
||
hazard).
|
||
"""
|
||
if spec.get("anchor_kind") != "subtractive_quantity_composition":
|
||
return None
|
||
observed_units = set(spec.get("observed_units") or ())
|
||
if not observed_units:
|
||
return None
|
||
|
||
matches = list(_SUBTRACTIVE_TWO_QUANTITY_RE.finditer(statement))
|
||
if len(matches) != 1:
|
||
return None
|
||
|
||
m = matches[0]
|
||
subject = m.group("subject")
|
||
if subject.lower() in _REFUSED_SUBJECT_TOKENS:
|
||
return None
|
||
if subject.lower() in _COMMON_DETERMINERS_AT_HEAD:
|
||
return None
|
||
|
||
verb_a = m.group("verb_a").lower()
|
||
verb_b = m.group("verb_b").lower()
|
||
if verb_a not in _SUBTRACTIVE_INITIAL_VERBS:
|
||
return None
|
||
if verb_b not in _SUBTRACTIVE_REMOVAL_VERBS:
|
||
return None
|
||
|
||
unit_a = m.group("unit_a").lower()
|
||
unit_b = m.group("unit_b").lower()
|
||
if unit_a.rstrip("s") != unit_b.rstrip("s"):
|
||
return None
|
||
canonical_unit = unit_a
|
||
if (
|
||
canonical_unit not in observed_units
|
||
and canonical_unit.rstrip("s") not in observed_units
|
||
):
|
||
return None
|
||
|
||
count_a_token = m.group("count_a")
|
||
count_b_token = m.group("count_b")
|
||
try:
|
||
count_a = float(count_a_token)
|
||
count_b = float(count_b_token)
|
||
except ValueError:
|
||
return None
|
||
if count_a <= 0 or count_b <= 0:
|
||
return None
|
||
if count_b >= count_a:
|
||
return None # Non-negative remainder; wrong>0 hazard.
|
||
|
||
composed_value_f = count_a - count_b
|
||
composed_value: int | float
|
||
if (
|
||
composed_value_f.is_integer()
|
||
and "." not in count_a_token
|
||
and "." not in count_b_token
|
||
):
|
||
composed_value = int(composed_value_f)
|
||
else:
|
||
composed_value = composed_value_f
|
||
|
||
from generate.math_candidate_parser import CandidateInitial
|
||
from generate.math_problem_graph import InitialPossession, Quantity
|
||
|
||
matched_anchor = verb_a if verb_a in {
|
||
"has", "had", "saved", "earned", "got", "received", "bought", "made", "paid",
|
||
} else "had"
|
||
|
||
composed_initial = CandidateInitial(
|
||
initial=InitialPossession(
|
||
entity=subject,
|
||
quantity=Quantity(value=composed_value, unit=canonical_unit),
|
||
),
|
||
source_span=m.group(0),
|
||
matched_anchor=matched_anchor,
|
||
matched_value_token=str(composed_value),
|
||
matched_unit_token=canonical_unit,
|
||
matched_entity_token=subject,
|
||
)
|
||
|
||
anchor: Mapping[str, Any] = {
|
||
"kind": "subtractive_quantity_composition",
|
||
"composition_shape": _SUBTRACTIVE_COMPOSITION_SHAPE,
|
||
"composed_initial": composed_initial,
|
||
"count_a": count_a_token,
|
||
"count_b": count_b_token,
|
||
"unit": canonical_unit,
|
||
"subject": subject,
|
||
"initial_verb": verb_a,
|
||
"removal_verb": verb_b,
|
||
}
|
||
return ((anchor,), "amount")
|
||
|
||
|
||
def _match_currency_amount(
|
||
statement: str, spec: Mapping[str, Any]
|
||
) -> tuple[tuple[Mapping[str, Any], ...], Literal["amount"]] | None:
|
||
"""Detection-only match for "X costs $Y" (NO per-unit framing).
|
||
|
||
Discriminator vs rate_with_currency: this matcher REQUIRES a
|
||
currency symbol AND requires that no per-unit framing is present.
|
||
|
||
Narrowness: the currency symbol observed in the statement MUST
|
||
appear in the spec's ``observed_currency_symbols`` set.
|
||
|
||
Returns ``(empty parsed_anchors, "amount")`` on a hit.
|
||
"""
|
||
if spec.get("anchor_kind") != "currency_amount":
|
||
return None
|
||
observed_symbols = set(spec.get("observed_currency_symbols") or ())
|
||
if not observed_symbols:
|
||
return None
|
||
# Find at least one currency symbol present in the statement that is
|
||
# also observed by the spec.
|
||
found_observed = any(sym in statement for sym in observed_symbols)
|
||
if not found_observed:
|
||
return None
|
||
padded = _padded_lower(statement)
|
||
if _has_per_unit_framing(padded):
|
||
return None
|
||
return (tuple(), "amount")
|
||
|
||
|
||
_MATCHERS: Final[dict[ShapeCategory, Any]] = {
|
||
ShapeCategory.DESCRIPTIVE_SETUP_NO_QUANTITY: _match_descriptive_setup_no_quantity,
|
||
ShapeCategory.TEMPORAL_AGGREGATION: _match_temporal_aggregation,
|
||
ShapeCategory.RATE_WITH_CURRENCY: _match_rate_with_currency,
|
||
ShapeCategory.DISCRETE_COUNT_STATEMENT: _match_discrete_count_statement,
|
||
ShapeCategory.MULTIPLICATIVE_AGGREGATION: _match_multiplicative_aggregation,
|
||
ShapeCategory.CURRENCY_AMOUNT: _match_currency_amount,
|
||
}
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Public API
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def match(
|
||
statement: str,
|
||
registry: tuple[RatifiedRecognizer, ...],
|
||
*,
|
||
prior_subject: str | None = None,
|
||
) -> RecognizerMatch | None:
|
||
"""First-match-wins over *registry*.
|
||
|
||
Pure: same ``(statement, registry, prior_subject)`` → same result,
|
||
byte-identical. Order is registry order (the projection step in
|
||
:mod:`generate.recognizer_registry` sorts by ``(review_date,
|
||
proposal_id)``).
|
||
|
||
ME-2 (cross-sentence subject binding) — when the per-category
|
||
matcher returns ``None`` for a ``RATE_WITH_CURRENCY`` recognizer
|
||
AND ``prior_subject`` is supplied, this dispatcher additionally
|
||
tries
|
||
:func:`try_extract_cross_sentence_composition_anchor`. Admitting
|
||
the case 0019 sentence shape requires both:
|
||
|
||
- a ratified recognizer carrying
|
||
``anchor_kind = "currency_per_unit_composition"``
|
||
- a non-empty ``prior_subject`` resolved from upstream sentences
|
||
|
||
Refusal-preferring discipline is preserved: ``prior_subject=None``
|
||
+ same-sentence Option A regex miss → returns ``None``.
|
||
"""
|
||
if not isinstance(statement, str) or not statement.strip():
|
||
return None
|
||
for recognizer in registry:
|
||
matcher = _MATCHERS.get(recognizer.shape_category)
|
||
if matcher is None:
|
||
continue
|
||
result = matcher(statement, recognizer.canonical_pattern)
|
||
if result is None:
|
||
if (
|
||
recognizer.shape_category is ShapeCategory.RATE_WITH_CURRENCY
|
||
and recognizer.canonical_pattern.get("anchor_kind")
|
||
== "currency_per_unit_composition"
|
||
and prior_subject is not None
|
||
):
|
||
cross_result = try_extract_cross_sentence_composition_anchor(
|
||
statement, recognizer.canonical_pattern, prior_subject
|
||
)
|
||
if cross_result is None:
|
||
continue
|
||
parsed_anchors, graph_intent = cross_result
|
||
return RecognizerMatch(
|
||
recognizer=recognizer,
|
||
category=recognizer.shape_category,
|
||
outcome="admissible",
|
||
graph_intent=graph_intent,
|
||
parsed_anchors=parsed_anchors,
|
||
)
|
||
continue
|
||
parsed_anchors, graph_intent = result
|
||
outcome: Literal["admissible", "inadmissible_by_design"] = (
|
||
"inadmissible_by_design"
|
||
if recognizer.shape_category is ShapeCategory.DESCRIPTIVE_SETUP_NO_QUANTITY
|
||
else "admissible"
|
||
)
|
||
return RecognizerMatch(
|
||
recognizer=recognizer,
|
||
category=recognizer.shape_category,
|
||
outcome=outcome,
|
||
graph_intent=graph_intent,
|
||
parsed_anchors=parsed_anchors,
|
||
)
|
||
return None
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Cross-sentence subject resolution helper (ME-2).
|
||
# ---------------------------------------------------------------------------
|
||
|
||
_PROPER_NOUN_SUBJECT_RE: Final[re.Pattern[str]] = re.compile(
|
||
r"^\s*([A-Z][a-zA-Z]+)\b"
|
||
)
|
||
|
||
|
||
_COMMON_DETERMINERS_AT_HEAD: Final[frozenset[str]] = frozenset(
|
||
{
|
||
# Articles + demonstratives
|
||
"the", "a", "an", "this", "that", "these", "those",
|
||
# Possessives
|
||
"his", "her", "their", "its", "my", "your", "our",
|
||
# Sentence-initial connectors / prepositions that get capitalized
|
||
"after", "before", "when", "while", "if", "then", "so", "but",
|
||
"and", "or", "during", "since", "until", "though", "although",
|
||
"however", "moreover", "additionally", "first", "next", "later",
|
||
"finally", "now", "soon", "today", "tomorrow", "yesterday",
|
||
"every", "all", "some", "many", "each", "another", "other",
|
||
"in", "on", "at", "by", "for", "from", "with", "without",
|
||
"how", "why", "what", "where", "who", "when",
|
||
}
|
||
)
|
||
|
||
|
||
def extract_proper_noun_subject(statement: str) -> str | None:
|
||
"""Return the head proper-noun subject of *statement*, or None.
|
||
|
||
Used by callers (e.g. ``generate.math_candidate_graph``) to track a
|
||
running ``prior_subject`` across sentences for cross-sentence
|
||
composition binding (ME-2).
|
||
|
||
Heuristic narrowness:
|
||
|
||
- The head token must match ``[A-Z][a-zA-Z]+``.
|
||
- The lowercased head must NOT be in :data:`_REFUSED_SUBJECT_TOKENS`
|
||
(existing pronoun set) or
|
||
:data:`_COMMON_DETERMINERS_AT_HEAD` (articles + demonstratives +
|
||
possessives that get capitalized at sentence start but are not
|
||
proper nouns).
|
||
|
||
Refuses on any ambiguity. The caller is expected to update the
|
||
running ``prior_subject`` only when this returns a non-None value.
|
||
"""
|
||
if not isinstance(statement, str):
|
||
return None
|
||
m = _PROPER_NOUN_SUBJECT_RE.match(statement)
|
||
if m is None:
|
||
return None
|
||
cand = m.group(1)
|
||
lc = cand.lower()
|
||
if lc in _REFUSED_SUBJECT_TOKENS:
|
||
return None
|
||
if lc in _COMMON_DETERMINERS_AT_HEAD:
|
||
return None
|
||
return cand
|
||
|
||
|
||
__all__ = [
|
||
"RecognizerMatch",
|
||
"match",
|
||
"extract_proper_noun_subject",
|
||
"try_extract_cross_sentence_composition_anchor",
|
||
]
|