7b0c0ae8ce
- anti_unifier: remove arbitrary root fallback when no agent_node_id (only apply root-form when nid known; primary pipeline path unaffected) - depth_canonical: filter build_node_depths to he/grc (or nodes with root) so English nodes no longer pollute node_depths with language-only entries Tests re-run green. Matches review recommendations for the 3-lang depth feature.
706 lines
27 KiB
Python
706 lines
27 KiB
Python
"""Deterministic Phase 1 anti-unification over taught token sequences."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
from dataclasses import dataclass
|
|
from typing import Any, Iterable, Mapping, Sequence
|
|
|
|
from recognition.outcome import (
|
|
CONTRADICTED,
|
|
EVIDENCED,
|
|
UNDETERMINED,
|
|
BoundFeature,
|
|
EvidenceSpan,
|
|
FeatureBundle,
|
|
FeatureConsistencyRefusal,
|
|
FeatureEvidence,
|
|
FeatureEvidenceRefusal,
|
|
NegativeEvidence,
|
|
RecognitionOutcome,
|
|
RecognitionProvenance,
|
|
ShapeRefusal,
|
|
)
|
|
from recognition.depth_canonical import canonicalize_agent_slot, canonicalize_token
|
|
|
|
TokenSequence = Sequence[str]
|
|
Scalar = str | int | float
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class Constant:
|
|
token: str
|
|
|
|
def as_dict(self) -> dict[str, str]:
|
|
return {"kind": "constant", "token": self.token}
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class TypedSlot:
|
|
feature_name: str
|
|
slot_type: str
|
|
min_width: int = 1
|
|
max_width: int = 1
|
|
ignored_prefix_tokens: tuple[str, ...] = ()
|
|
|
|
def as_dict(self) -> dict[str, Any]:
|
|
return {
|
|
"feature_name": self.feature_name,
|
|
"ignored_prefix_tokens": list(self.ignored_prefix_tokens),
|
|
"kind": "typed_slot",
|
|
"max_width": self.max_width,
|
|
"min_width": self.min_width,
|
|
"slot_type": self.slot_type,
|
|
}
|
|
|
|
|
|
PatternElement = Constant | TypedSlot
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class DerivedRecognizer:
|
|
pattern: tuple[PatternElement, ...]
|
|
teaching_set_id: str
|
|
constant_features: Mapping[str, Scalar]
|
|
absent_features: Mapping[str, Scalar]
|
|
|
|
def as_dict(self) -> dict[str, Any]:
|
|
return {
|
|
"absent_features": dict(sorted(self.absent_features.items())),
|
|
"constant_features": dict(sorted(self.constant_features.items())),
|
|
"pattern": [_pattern_element_as_dict(element) for element in self.pattern],
|
|
"teaching_set_id": self.teaching_set_id,
|
|
}
|
|
|
|
def to_json(self) -> str:
|
|
return json.dumps(self.as_dict(), ensure_ascii=False, separators=(",", ":"), sort_keys=True)
|
|
|
|
@classmethod
|
|
def from_json(cls, payload: str) -> "DerivedRecognizer":
|
|
raw = json.loads(payload)
|
|
# L10 engine_state migration discipline (step-2): the v1 keys below are
|
|
# required. Any field ADDED in a later schema_version must be read via
|
|
# raw.get(name, default) and omitted-when-default in as_dict(), so old
|
|
# checkpoints load without migration and un-evolved records stay
|
|
# byte-identical (cf. DiscoveryCandidate's C1 fields for the pattern).
|
|
return cls(
|
|
pattern=tuple(_pattern_element_from_dict(element) for element in raw["pattern"]),
|
|
teaching_set_id=str(raw["teaching_set_id"]),
|
|
constant_features=dict(raw["constant_features"]),
|
|
absent_features=dict(raw["absent_features"]),
|
|
)
|
|
|
|
|
|
def derive_recognizer(examples: Sequence[tuple[TokenSequence, FeatureBundle]], depths: dict[str, dict] | None = None, *, agent_node_id: str | None = None) -> DerivedRecognizer:
|
|
"""Derive recognizer, optionally using node_depths to apply root canonicalization
|
|
for Hebrew/Greek. This makes root-equivalent teaching examples produce
|
|
equivalent patterns (exact, no approx).
|
|
depths keyed by node_id or feature name; values have 'language', 'root'.
|
|
"""
|
|
if not examples:
|
|
raise ValueError("derive_recognizer requires at least one teaching example")
|
|
|
|
normalized = tuple((tuple(tokens), bundle) for tokens, bundle in examples)
|
|
# Use canonicalize_agent_slot for precise, node keyed root norm (from depth_canonical)
|
|
if depths:
|
|
normed = []
|
|
for toks, bdl in normalized:
|
|
new_toks = canonicalize_agent_slot(toks, bdl, depths, agent_node_id=agent_node_id)
|
|
normed.append((new_toks, bdl))
|
|
normalized = tuple(normed)
|
|
# Compute id after canonical to make root-equiv produce same teaching_set_id
|
|
teaching_set_id = _teaching_set_id(tokens for tokens, _bundle in normalized)
|
|
feature_names = _feature_names(normalized)
|
|
|
|
# Check if we have verb/auxiliary phrase variations
|
|
unique_vps = set()
|
|
for tokens, bundle in normalized:
|
|
agent_feat = bundle.get("agent")
|
|
count_feat = bundle.get("count")
|
|
if agent_feat is not None and count_feat is not None:
|
|
agent_ev = agent_feat.evidence
|
|
count_ev = count_feat.evidence
|
|
if isinstance(agent_ev, EvidenceSpan) and isinstance(count_ev, EvidenceSpan):
|
|
vp_tokens = tokens[agent_ev.end : count_ev.start]
|
|
unique_vps.add(" ".join(vp_tokens))
|
|
|
|
if len(unique_vps) <= 1:
|
|
# Phase 1 logic
|
|
slot_names = _slot_feature_names(normalized, feature_names)
|
|
absent_features = _absent_uniform_features(normalized, feature_names, slot_names)
|
|
|
|
relation = _uniform_feature_value(normalized, "relation")
|
|
relation_token = str(relation)
|
|
anchors = tuple(_single_token_index(tokens, relation_token) for tokens, _bundle in normalized)
|
|
|
|
prefix_widths = tuple(anchor for anchor in anchors)
|
|
suffix_widths = tuple(len(tokens) - anchor - 1 for (tokens, _bundle), anchor in zip(normalized, anchors))
|
|
if min(prefix_widths) < 1:
|
|
raise ValueError("agent slot must contain at least one token")
|
|
if set(suffix_widths) != {2}:
|
|
raise ValueError("Phase 1 expects count and unit slots after the relation anchor")
|
|
|
|
constant_features = {"relation": relation}
|
|
ignored_prefix_tokens = _ignored_prefix_tokens(normalized, "agent")
|
|
pattern: tuple[PatternElement, ...] = (
|
|
TypedSlot(
|
|
feature_name="agent",
|
|
slot_type=_slot_type(normalized, "agent"),
|
|
min_width=min(prefix_widths),
|
|
max_width=max(prefix_widths),
|
|
ignored_prefix_tokens=ignored_prefix_tokens,
|
|
),
|
|
Constant(relation_token),
|
|
TypedSlot(feature_name="count", slot_type=_slot_type(normalized, "count")),
|
|
TypedSlot(feature_name="unit", slot_type=_slot_type(normalized, "unit")),
|
|
)
|
|
return DerivedRecognizer(
|
|
pattern=pattern,
|
|
teaching_set_id=teaching_set_id,
|
|
constant_features=constant_features,
|
|
absent_features=absent_features,
|
|
)
|
|
else:
|
|
# Phase 2 logic
|
|
slot_names = frozenset({"agent", "relation", "count", "unit"})
|
|
absent_features = _absent_uniform_features(normalized, feature_names, slot_names)
|
|
|
|
vp_list = sorted(list(unique_vps))
|
|
constant_features = {"__allowed_verbs": "|".join(vp_list)}
|
|
|
|
prefix_widths = []
|
|
for tokens, bundle in normalized:
|
|
agent_feat = bundle.get("agent")
|
|
if agent_feat is not None and isinstance(agent_feat.evidence, EvidenceSpan):
|
|
prefix_widths.append(agent_feat.evidence.end)
|
|
|
|
ignored_prefix_tokens = _ignored_prefix_tokens(normalized, "agent")
|
|
max_verb_width = max(len(vp.split()) for vp in vp_list)
|
|
|
|
pattern = (
|
|
TypedSlot(
|
|
feature_name="agent",
|
|
slot_type="str",
|
|
min_width=1,
|
|
max_width=max(prefix_widths),
|
|
ignored_prefix_tokens=ignored_prefix_tokens,
|
|
),
|
|
TypedSlot(
|
|
feature_name="relation",
|
|
slot_type="str",
|
|
min_width=1,
|
|
max_width=max_verb_width,
|
|
),
|
|
TypedSlot(
|
|
feature_name="count",
|
|
slot_type="int",
|
|
min_width=1,
|
|
max_width=1,
|
|
),
|
|
TypedSlot(
|
|
feature_name="unit",
|
|
slot_type="str",
|
|
min_width=1,
|
|
max_width=1,
|
|
),
|
|
)
|
|
|
|
return DerivedRecognizer(
|
|
pattern=pattern,
|
|
teaching_set_id=teaching_set_id,
|
|
constant_features=constant_features,
|
|
absent_features=absent_features,
|
|
)
|
|
|
|
|
|
def recognize(
|
|
recognizer: DerivedRecognizer,
|
|
token_sequence: TokenSequence,
|
|
depths: dict[str, dict] | None = None,
|
|
*, agent_node_id: str | None = None,
|
|
) -> RecognitionOutcome:
|
|
"""Recognize using optional node_depths for 3-lang root normalization.
|
|
|
|
depths: node_id -> {"language": , "root": , ...} from PropositionGraph / OOV context.
|
|
When present for he/grc, root_normalize is used on relevant tokens for exact
|
|
canonical matching (no surface variance for roots).
|
|
Normalization for matching only; lifting uses original tokens to keep reported values (surface).
|
|
"""
|
|
original_tokens = tuple(token_sequence)
|
|
tokens = original_tokens
|
|
# Use canonicalize with start_idx=0 for agent in token-seq recognize path + agent_node_id for nid keying. No synth bundle.
|
|
match_tokens = canonicalize_agent_slot(tokens, None, depths, agent_node_id=agent_node_id, start_idx=0) if depths else tokens
|
|
match_tokens_t = tuple(match_tokens)
|
|
|
|
# If this is Phase 1 (no __allowed_verbs in constant_features), run Phase 1 logic
|
|
if "__allowed_verbs" not in recognizer.constant_features:
|
|
provenance = RecognitionProvenance(
|
|
mechanism="anti_unification",
|
|
teaching_set_id=recognizer.teaching_set_id,
|
|
resolution_level="chunk",
|
|
replay_seed="",
|
|
)
|
|
matches = _match_pattern(recognizer.pattern, match_tokens_t)
|
|
if matches is None:
|
|
return RecognitionOutcome(
|
|
state=UNDETERMINED,
|
|
provenance=RecognitionProvenance(
|
|
mechanism="anti_unification",
|
|
teaching_set_id=recognizer.teaching_set_id,
|
|
resolution_level="shape",
|
|
replay_seed="",
|
|
),
|
|
refusal_reason=ShapeRefusal(
|
|
reason=f"shape_mismatch:{_shape_description(recognizer.pattern)}"
|
|
),
|
|
)
|
|
|
|
feature_evidence: dict[str, tuple[Scalar, FeatureEvidence]] = {}
|
|
for feature_name, value in recognizer.constant_features.items():
|
|
feature_evidence[feature_name] = (value, _constant_evidence(str(value), original_tokens))
|
|
for feature_name, value in recognizer.absent_features.items():
|
|
feature_evidence[feature_name] = (
|
|
value,
|
|
NegativeEvidence(
|
|
scope_start=0,
|
|
scope_end=len(original_tokens),
|
|
description=f"{feature_name}={value!r} evidenced by absence of taught counter-marker",
|
|
),
|
|
)
|
|
for slot, span in matches.items():
|
|
value, evidence = _lift_slot(slot, original_tokens, span)
|
|
feature_evidence[slot.feature_name] = (value, evidence)
|
|
|
|
# 3-lang root form for agent in proposition when depth (nid-keyed via agent_node_id).
|
|
# Only applied when explicit nid is provided (primary pipeline path always does).
|
|
# Without nid we do not guess a root (avoids unrelated root in proposition).
|
|
if depths and 'agent' in feature_evidence and agent_node_id and agent_node_id in depths:
|
|
val, ev = feature_evidence['agent']
|
|
val = canonicalize_token(str(val), agent_node_id, depths)
|
|
feature_evidence['agent'] = (val, ev)
|
|
|
|
proposition = FeatureBundle.from_mapping(feature_evidence)
|
|
return RecognitionOutcome(
|
|
state=EVIDENCED,
|
|
provenance=provenance,
|
|
proposition=proposition,
|
|
refusal_reason=None,
|
|
)
|
|
|
|
# Phase 2 logic
|
|
provenance = RecognitionProvenance(
|
|
mechanism="anti_unification",
|
|
teaching_set_id=recognizer.teaching_set_id,
|
|
resolution_level="chunk",
|
|
replay_seed="",
|
|
)
|
|
|
|
allowed_vps = recognizer.constant_features["__allowed_verbs"].split("|")
|
|
# Sort by token length descending, then alphabetically for deterministic precedence
|
|
allowed_vps_sorted = sorted(allowed_vps, key=lambda x: (-len(x.split()), x))
|
|
|
|
verb_match = None
|
|
for vp_str in allowed_vps_sorted:
|
|
vp_tokens = tuple(vp_str.split())
|
|
n_vp = len(vp_tokens)
|
|
# Search using match_tokens for root-normed matching
|
|
for i in range(1, len(match_tokens_t) - n_vp + 1):
|
|
if match_tokens_t[i : i + n_vp] == vp_tokens:
|
|
verb_match = (i, i + n_vp, vp_str)
|
|
break
|
|
if verb_match is not None:
|
|
break
|
|
|
|
if verb_match is None:
|
|
return RecognitionOutcome(
|
|
state=UNDETERMINED,
|
|
provenance=RecognitionProvenance(
|
|
mechanism="anti_unification",
|
|
teaching_set_id=recognizer.teaching_set_id,
|
|
resolution_level="shape",
|
|
replay_seed="",
|
|
),
|
|
refusal_reason=ShapeRefusal(
|
|
reason=f"shape_mismatch:{_shape_description(recognizer.pattern)}"
|
|
),
|
|
)
|
|
|
|
verb_start, verb_end, vp_str = verb_match
|
|
|
|
# Agent validation using ORIGINAL for reported values
|
|
agent_start = 0
|
|
if verb_start > 1 and original_tokens[0].lower() in {"a", "the"}:
|
|
agent_start = 1
|
|
|
|
agent_tokens = original_tokens[agent_start:verb_start]
|
|
if not agent_tokens:
|
|
return RecognitionOutcome(
|
|
state=UNDETERMINED,
|
|
provenance=RecognitionProvenance(
|
|
mechanism="anti_unification",
|
|
teaching_set_id=recognizer.teaching_set_id,
|
|
resolution_level="shape",
|
|
replay_seed="",
|
|
),
|
|
refusal_reason=ShapeRefusal(
|
|
reason=f"shape_mismatch:{_shape_description(recognizer.pattern)}"
|
|
),
|
|
)
|
|
|
|
agent_value = " ".join(agent_tokens)
|
|
agent_span = EvidenceSpan(agent_start, verb_start, agent_value)
|
|
|
|
# Scan for digit/number tokens in the suffix -- ORIGINAL
|
|
digits: list[int] = []
|
|
digit_spans: list[EvidenceSpan] = []
|
|
for i in range(verb_end, len(original_tokens)):
|
|
if original_tokens[i].isdigit():
|
|
digits.append(int(original_tokens[i]))
|
|
digit_spans.append(EvidenceSpan(i, i + 1, original_tokens[i]))
|
|
|
|
# Layer 2 refusal: missing count
|
|
if not digits:
|
|
return RecognitionOutcome(
|
|
state=UNDETERMINED,
|
|
provenance=RecognitionProvenance(
|
|
mechanism="anti_unification",
|
|
teaching_set_id=recognizer.teaching_set_id,
|
|
resolution_level="word",
|
|
replay_seed="",
|
|
),
|
|
refusal_reason=FeatureEvidenceRefusal(
|
|
missing_feature="count",
|
|
reason="missing count feature evidence span",
|
|
),
|
|
)
|
|
|
|
# Layer 3 refusal: count contradiction
|
|
if len(set(digits)) > 1:
|
|
return RecognitionOutcome(
|
|
state=CONTRADICTED,
|
|
provenance=provenance,
|
|
refusal_reason=FeatureConsistencyRefusal(
|
|
feature="count",
|
|
reason="contradictory values for count",
|
|
spans=tuple(digit_spans),
|
|
),
|
|
)
|
|
|
|
# Validate remaining tokens structure (must be [Count, Unit] or [Count, Unit, "and", Count, Unit])
|
|
count_index = digit_spans[0].start
|
|
is_valid_structure = False
|
|
if len(original_tokens) == count_index + 2:
|
|
is_valid_structure = True
|
|
elif len(original_tokens) == count_index + 5 and original_tokens[count_index + 2] == "and":
|
|
if original_tokens[count_index + 3].isdigit():
|
|
is_valid_structure = True
|
|
|
|
if not is_valid_structure:
|
|
return RecognitionOutcome(
|
|
state=UNDETERMINED,
|
|
provenance=RecognitionProvenance(
|
|
mechanism="anti_unification",
|
|
teaching_set_id=recognizer.teaching_set_id,
|
|
resolution_level="shape",
|
|
replay_seed="",
|
|
),
|
|
refusal_reason=ShapeRefusal(
|
|
reason=f"shape_mismatch:{_shape_description(recognizer.pattern)}"
|
|
),
|
|
)
|
|
|
|
# Lift features -- use ORIGINAL for reported surface values
|
|
count_val = digits[0]
|
|
count_span = digit_spans[0]
|
|
|
|
unit_token = original_tokens[count_index + 1]
|
|
unit_val = _singularize(unit_token)
|
|
unit_span = EvidenceSpan(count_index + 1, count_index + 2, unit_token)
|
|
|
|
relation_val = original_tokens[verb_end - 1]
|
|
relation_span = EvidenceSpan(verb_end - 1, verb_end, relation_val)
|
|
|
|
# Tense
|
|
first_verb_token = original_tokens[verb_start]
|
|
if first_verb_token == "had":
|
|
tense_val = "past"
|
|
elif first_verb_token == "will":
|
|
tense_val = "future"
|
|
else:
|
|
tense_val = "present"
|
|
tense_span = EvidenceSpan(verb_start, verb_start + 1, first_verb_token)
|
|
|
|
# Polarity
|
|
if "not" in original_tokens[verb_start:verb_end]:
|
|
polarity_val = "-"
|
|
not_idx = original_tokens[verb_start:verb_end].index("not") + verb_start
|
|
polarity_span = EvidenceSpan(not_idx, not_idx + 1, "not")
|
|
else:
|
|
polarity_val = "+"
|
|
polarity_span = NegativeEvidence(0, len(original_tokens), "no negator present")
|
|
|
|
# Modality
|
|
if "may" in original_tokens[verb_start:verb_end]:
|
|
modality_val = "possibility"
|
|
may_idx = original_tokens[verb_start:verb_end].index("may") + verb_start
|
|
modality_span = EvidenceSpan(may_idx, may_idx + 1, "may")
|
|
else:
|
|
modality_val = "actual"
|
|
modality_span = NegativeEvidence(0, len(original_tokens), "no modal counter-marker present")
|
|
|
|
# Intentionality
|
|
intentionality_val = "possession"
|
|
intentionality_text = " ".join(original_tokens[agent_start:verb_end])
|
|
intentionality_span = EvidenceSpan(agent_start, verb_end, intentionality_text)
|
|
|
|
feature_evidence: dict[str, tuple[Scalar, FeatureEvidence]] = {
|
|
"agent": (agent_value, agent_span),
|
|
"count": (count_val, count_span),
|
|
"unit": (unit_val, unit_span),
|
|
"relation": (relation_val, relation_span),
|
|
"tense": (tense_val, tense_span),
|
|
"polarity": (polarity_val, polarity_span),
|
|
"modality": (modality_val, modality_span),
|
|
"intentionality": (intentionality_val, intentionality_span),
|
|
}
|
|
|
|
# Add other absent uniform features if they are not overridden
|
|
for feature_name, value in recognizer.absent_features.items():
|
|
if feature_name not in feature_evidence:
|
|
feature_evidence[feature_name] = (
|
|
value,
|
|
NegativeEvidence(
|
|
scope_start=0,
|
|
scope_end=len(original_tokens),
|
|
description=f"{feature_name}={value!r} evidenced by absence of taught counter-marker",
|
|
),
|
|
)
|
|
|
|
proposition = FeatureBundle.from_mapping(feature_evidence)
|
|
return RecognitionOutcome(
|
|
state=EVIDENCED,
|
|
provenance=provenance,
|
|
proposition=proposition,
|
|
refusal_reason=None,
|
|
)
|
|
|
|
|
|
def _feature_names(examples: Sequence[tuple[tuple[str, ...], FeatureBundle]]) -> tuple[str, ...]:
|
|
names = tuple(feature.name for feature in examples[0][1].features)
|
|
for _tokens, bundle in examples[1:]:
|
|
if tuple(feature.name for feature in bundle.features) != names:
|
|
raise ValueError("all teaching bundles must expose the same feature set")
|
|
return names
|
|
|
|
|
|
def _slot_feature_names(
|
|
examples: Sequence[tuple[tuple[str, ...], FeatureBundle]], feature_names: tuple[str, ...]
|
|
) -> frozenset[str]:
|
|
slots = []
|
|
for name in feature_names:
|
|
evidences = tuple(_feature(bundle, name).evidence for _tokens, bundle in examples)
|
|
if all(isinstance(evidence, EvidenceSpan) for evidence in evidences):
|
|
values = tuple(_feature(bundle, name).value for _tokens, bundle in examples)
|
|
if len(set(values)) > 1:
|
|
slots.append(name)
|
|
return frozenset(slots)
|
|
|
|
|
|
def _absent_uniform_features(
|
|
examples: Sequence[tuple[tuple[str, ...], FeatureBundle]],
|
|
feature_names: tuple[str, ...],
|
|
slot_names: frozenset[str],
|
|
) -> dict[str, Scalar]:
|
|
absent: dict[str, Scalar] = {}
|
|
for name in feature_names:
|
|
if name in slot_names or name == "relation":
|
|
continue
|
|
feature_values = tuple(_feature(bundle, name).value for _tokens, bundle in examples)
|
|
if len(set(feature_values)) == 1:
|
|
absent[name] = feature_values[0]
|
|
return absent
|
|
|
|
|
|
def _uniform_feature_value(examples: Sequence[tuple[tuple[str, ...], FeatureBundle]], name: str) -> Scalar:
|
|
values = tuple(_feature(bundle, name).value for _tokens, bundle in examples)
|
|
if len(set(values)) != 1:
|
|
raise ValueError(f"feature must be uniform in Phase 1: {name}")
|
|
return values[0]
|
|
|
|
|
|
def _feature(bundle: FeatureBundle, name: str) -> BoundFeature:
|
|
feature = bundle.get(name)
|
|
if feature is None:
|
|
raise ValueError(f"missing feature in teaching bundle: {name}")
|
|
return feature
|
|
|
|
|
|
def _slot_type(examples: Sequence[tuple[tuple[str, ...], FeatureBundle]], name: str) -> str:
|
|
values = tuple(_feature(bundle, name).value for _tokens, bundle in examples)
|
|
if all(isinstance(value, int) for value in values):
|
|
return "int"
|
|
if all(isinstance(value, float) for value in values):
|
|
return "float"
|
|
return "str"
|
|
|
|
|
|
def _ignored_prefix_tokens(examples: Sequence[tuple[tuple[str, ...], FeatureBundle]], name: str) -> tuple[str, ...]:
|
|
ignored = set()
|
|
for tokens, bundle in examples:
|
|
evidence = _feature(bundle, name).evidence
|
|
if isinstance(evidence, EvidenceSpan):
|
|
ignored.update(token.lower() for token in tokens[: evidence.start])
|
|
return tuple(sorted(ignored))
|
|
|
|
|
|
def _single_token_index(tokens: tuple[str, ...], token: str) -> int:
|
|
indexes = [index for index, candidate in enumerate(tokens) if candidate == token]
|
|
if len(indexes) != 1:
|
|
raise ValueError(f"constant anchor must occur exactly once: {token!r}")
|
|
return indexes[0]
|
|
|
|
|
|
def _teaching_set_id(token_sequences: Iterable[tuple[str, ...]]) -> str:
|
|
canonical = json.dumps(sorted(token_sequences), ensure_ascii=False, separators=(",", ":"))
|
|
return hashlib.sha256(canonical.encode("utf-8")).hexdigest()
|
|
|
|
|
|
def _match_pattern(pattern: tuple[PatternElement, ...], tokens: tuple[str, ...]) -> dict[TypedSlot, tuple[int, int]] | None:
|
|
def walk(index: int, cursor: int, spans: dict[TypedSlot, tuple[int, int]]) -> dict[TypedSlot, tuple[int, int]] | None:
|
|
if index == len(pattern):
|
|
return spans if cursor == len(tokens) else None
|
|
element = pattern[index]
|
|
if isinstance(element, Constant):
|
|
if cursor < len(tokens) and tokens[cursor] == element.token:
|
|
return walk(index + 1, cursor + 1, spans)
|
|
return None
|
|
|
|
remaining_min = _minimum_width(pattern[index + 1 :])
|
|
max_end = min(cursor + element.max_width, len(tokens) - remaining_min)
|
|
for end in range(cursor + element.min_width, max_end + 1):
|
|
next_spans = dict(spans)
|
|
next_spans[element] = (cursor, end)
|
|
matched = walk(index + 1, end, next_spans)
|
|
if matched is not None:
|
|
return matched
|
|
return None
|
|
|
|
return walk(0, 0, {})
|
|
|
|
|
|
def _minimum_width(pattern: Sequence[PatternElement]) -> int:
|
|
return sum(1 if isinstance(element, Constant) else element.min_width for element in pattern)
|
|
|
|
|
|
def _lift_slot(slot: TypedSlot, tokens: tuple[str, ...], span: tuple[int, int]) -> tuple[Scalar, EvidenceSpan]:
|
|
raw_tokens = tokens[span[0] : span[1]]
|
|
start = span[0]
|
|
if slot.ignored_prefix_tokens and raw_tokens and raw_tokens[0].lower() in slot.ignored_prefix_tokens:
|
|
raw_tokens = raw_tokens[1:]
|
|
start += 1
|
|
if not raw_tokens:
|
|
raise ValueError(f"slot {slot.feature_name!r} had no evidence tokens after prefix removal")
|
|
|
|
text = " ".join(raw_tokens)
|
|
if slot.slot_type == "int":
|
|
if len(raw_tokens) != 1 or not raw_tokens[0].isdigit():
|
|
raise ValueError(f"slot {slot.feature_name!r} expected one integer token")
|
|
return int(raw_tokens[0]), EvidenceSpan(start=start, end=start + 1, text=raw_tokens[0])
|
|
if slot.slot_type == "float":
|
|
if len(raw_tokens) != 1:
|
|
raise ValueError(f"slot {slot.feature_name!r} expected one float token")
|
|
return float(raw_tokens[0]), EvidenceSpan(start=start, end=start + 1, text=raw_tokens[0])
|
|
if slot.feature_name == "unit":
|
|
return _singularize(raw_tokens[-1]), EvidenceSpan(start=span[1] - 1, end=span[1], text=raw_tokens[-1])
|
|
return text, EvidenceSpan(start=start, end=span[1], text=text)
|
|
|
|
|
|
def _constant_evidence(value: str, tokens: tuple[str, ...]) -> EvidenceSpan:
|
|
for index, token in enumerate(tokens):
|
|
if token == value:
|
|
return EvidenceSpan(start=index, end=index + 1, text=token)
|
|
raise ValueError(f"constant feature had no evidence in matched token sequence: {value!r}")
|
|
|
|
|
|
def _singularize(token: str) -> str:
|
|
lowered = token.lower()
|
|
if lowered.endswith("ves") and len(lowered) > 3:
|
|
return lowered[:-3] + "f"
|
|
if lowered.endswith("ies") and len(lowered) > 3:
|
|
return lowered[:-3] + "y"
|
|
if lowered.endswith("s") and len(lowered) > 1:
|
|
return lowered[:-1]
|
|
return lowered
|
|
|
|
|
|
def _shape_description(pattern: tuple[PatternElement, ...]) -> str:
|
|
pieces = []
|
|
for element in pattern:
|
|
if isinstance(element, Constant):
|
|
pieces.append(repr(element.token))
|
|
else:
|
|
pieces.append(f"<{element.feature_name}:{element.slot_type}[{element.min_width},{element.max_width}]>")
|
|
return " ".join(pieces)
|
|
|
|
|
|
def _pattern_element_as_dict(element: PatternElement) -> dict[str, Any]:
|
|
return element.as_dict()
|
|
|
|
|
|
def _pattern_element_from_dict(raw: Mapping[str, Any]) -> PatternElement:
|
|
if raw["kind"] == "constant":
|
|
return Constant(token=str(raw["token"]))
|
|
if raw["kind"] == "typed_slot":
|
|
return TypedSlot(
|
|
feature_name=str(raw["feature_name"]),
|
|
slot_type=str(raw["slot_type"]),
|
|
min_width=int(raw["min_width"]),
|
|
max_width=int(raw["max_width"]),
|
|
ignored_prefix_tokens=tuple(str(token) for token in raw.get("ignored_prefix_tokens", ())),
|
|
)
|
|
raise ValueError(f"unknown pattern element kind: {raw['kind']!r}")
|
|
|
|
|
|
def root_normalize(token: str, language: str | None = None, root: str | None = None) -> str:
|
|
"""Canonical form for exact anti-unification using 3-lang depth.
|
|
|
|
For Hebrew (root density) and Koine Greek (Logos precision), prefer the
|
|
root from pack morphology over surface token. English uses surface.
|
|
This is pure data lookup from resolved depth (LexicalResolution / GraphNode);
|
|
preserves exactness, no similarity, no ANN. Enables cross-lang unification
|
|
in OOV / recognition paths via node_depths from PropositionGraph.
|
|
|
|
Invariant protected: exact recall + depth as semantic (not repair).
|
|
"""
|
|
if language in ("he", "grc") and root:
|
|
return root
|
|
return token
|
|
|
|
|
|
def graph_anti_unify(topology: tuple, depths: dict | None = None) -> dict:
|
|
"""Minimal graph-level anti-unification using unresolved topology + node_depths.
|
|
Keys on root where present for 3-lang (exact structural match).
|
|
Returns dict with matched roots or empty.
|
|
Pure, for extension point in OOV geometric context.
|
|
"""
|
|
result = {"matched_roots": [], "topology": topology}
|
|
if not depths:
|
|
return result
|
|
roots = []
|
|
for nid, d in depths.items():
|
|
if d.get("root"):
|
|
roots.append((nid, d["root"]))
|
|
result["matched_roots"] = roots
|
|
return result
|
|
|
|
|
|
__all__ = [
|
|
"Constant",
|
|
"DerivedRecognizer",
|
|
"TypedSlot",
|
|
"derive_recognizer",
|
|
"recognize",
|
|
"root_normalize",
|
|
"graph_anti_unify",
|
|
]
|