custom_matchers.py

"""Purpose: custom matching logic as a property of grounded atoms, built
entirely on the public surface. An object whose class defines match_ owns
its matching inside (unify ...) with no registration; a scored matcher is
an ordinary operation answering candidates with the degree as each
answer's annotation. Fuzzy, regex, semantic: all outside the library,
a few lines each, composing with structural match through evaluation.
Guarantees:
  - the unannotated ranked operation makes no synthetic type claim [tested:
    test_example_runs_and_verifies_itself; commit=3cfbe0d7417b1c453c2dc12d47e2e47e7de461f7]
Open Obligations:
  To Do: None
  Hacks: None
  Future Enhancements: None
"""

import difflib
import re

from _common import check, done, skip

try:
    # An embedding store lives in metta-arrays, and metta-numpy is the row
    # that makes NumPy its default library: `pip install 'pymetta[arrays]'`.
    import metta_numpy  # noqa: F401  -- the row that makes numpy the default
    import numpy
    from metta_arrays import EmbeddingStore
except ImportError:
    skip("metta-arrays and numpy are needed")

from metta import Answer, Bindings, Expression, MeTTa, S, V
from metta._atoms.factories import Grounded

m = MeTTa().space()

# Crisp lexical closeness: a regex value that matches inside unify. The
# pattern IS the value; matching succeeds exactly when the operand's
# printed form matches it.
class Regex:
    def __init__(self, pattern):
        self.pattern = re.compile(pattern)

    def match_(self, other):
        text = other.value if isinstance(other, Grounded) else str(other)
        if self.pattern.search(str(text)):
            yield other

starts_with_a = Grounded(Regex("^a"))
(hit,) = m.eval(S.unify(starts_with_a, S.abbey, S.hit, S.miss))
check("regex value matches", hit, S.hit)
(miss,) = m.eval(S.unify(starts_with_a, S.zebra, S.hit, S.miss))
check("regex value refuses", miss, S.miss)

# Composition with structural match, no new syntax: the matchable gates
# candidates the pattern produced.
m.add(S.person(S.ada), S.person(S.alan), S.person(S.grace))
(gated,) = m.eval(
    S.collapse(
        S.match(
            S["context-space"](),
            S.person(V.p),
            S.unify(starts_with_a, V.p, V.p, S.superpose(Expression())),
        )
    )
)
check("composes with structural match", sorted(str(x) for x in gated), ["ada", "alan"])

# Scored matching is an annotated operation: candidates are the values,
# degrees ride as annotations, top orders, (annotation) reads.
lexicon = ["class", "clause", "close"]

def fuzzy(query, candidate=None):
    for word in lexicon:
        degree = difflib.SequenceMatcher(None, str(query), word).ratio()
        yield Answer(value=word, k=round(degree, 6))

# reads the lexicon and never writes; the generator lifts the class itself
m.reads(fuzzy, name="fuzmatch")
m.annotations("fuzmatch", "ranked")
(best,) = m.eval(S.collapse(S.top(1, S.fuzmatch("clase", V.w))))
check("fuzzy best is difflib's own ranking", str(best.children[0]), '"clause"')
(weighted,) = m.eval(
    S.collapse(S.let(V.w, S.fuzmatch("clase", V.c), S.pair(S.annotation(), V.w)))
)
check("degrees read through (annotation)", len(weighted.children), 3)

# Semantic closeness: the embedding store's retrieval as matching logic.
# The store binds the variable it was handed to the nearest key.
store = EmbeddingStore(m, name="vec", mirror=False)
store.add(S.espresso, numpy.array([0.9, 0.1, 0.0]))
store.add(S.latte, numpy.array([0.8, 0.3, 0.0]))
store.add(S.granite, numpy.array([0.0, 0.1, 0.9]))
check(
    "stored embedding lookup",
    tuple(round(float(value), 1) for value in store.vector_for(S.espresso)),
    (0.9, 0.1, 0.0),
)

class Nearest:
    def match_(self, other):
        query, out = other.children[0], other.children[1]
        key, _score = next(iter(store.ranked(query, 1)))
        yield Bindings({out: key})

(nearest,) = m.eval(
    S.unify(Grounded(Nearest()), S.espresso(V.k), V.k, S.none)
)
check("nearest neighbour", nearest, S.espresso)
done("custom_matchers")