diff --git a/src/examples/indefinites/grammar.py b/src/examples/indefinites/grammar.py index 4aeb5d14..f2daeacd 100644 --- a/src/examples/indefinites/grammar.py +++ b/src/examples/indefinites/grammar.py @@ -1,4 +1,4 @@ -from ultk.language.grammar import Grammar, Rule +from ultk.language.grammar.grammar import Grammar, Rule # indefinites_grammar = Grammar.from_yaml("indefinites/grammar.yml") indefinites_grammar = Grammar.from_module("indefinites.grammar_functions") diff --git a/src/examples/indefinites/measures.py b/src/examples/indefinites/measures.py index 1a517d83..3f7904e9 100644 --- a/src/examples/indefinites/measures.py +++ b/src/examples/indefinites/measures.py @@ -1,5 +1,5 @@ from ultk.effcomm.informativity import informativity -from ultk.language.grammar import GrammaticalExpression +from ultk.language.grammar.grammar import GrammaticalExpression from ultk.language.language import Language, aggregate_expression_complexity from ultk.language.semantics import Meaning diff --git a/src/examples/indefinites/scripts/generate_expressions.py b/src/examples/indefinites/scripts/generate_expressions.py index b5dcc6c0..8c265f41 100644 --- a/src/examples/indefinites/scripts/generate_expressions.py +++ b/src/examples/indefinites/scripts/generate_expressions.py @@ -1,7 +1,7 @@ from ultk.util.io import write_expressions from ultk.language.semantics import Meaning -from ultk.language.grammar import GrammaticalExpression +from ultk.language.grammar.grammar import GrammaticalExpression from ..grammar import indefinites_grammar from ..meaning import universe as indefinites_universe diff --git a/src/examples/modals/grammar.py b/src/examples/modals/grammar.py index 89e37ac4..097fc040 100644 --- a/src/examples/modals/grammar.py +++ b/src/examples/modals/grammar.py @@ -1,3 +1,3 @@ -from ultk.language.grammar import Grammar +from ultk.language.grammar.grammar import Grammar modals_grammar = Grammar.from_yaml("modals/data/grammar.yaml") diff --git a/src/examples/modals/measures.py b/src/examples/modals/measures.py index 8ebcbf65..036d2c40 100644 --- a/src/examples/modals/measures.py +++ b/src/examples/modals/measures.py @@ -5,7 +5,7 @@ Referent, aggregate_expression_complexity, ) -from ultk.language.grammar import GrammaticalExpression +from ultk.language.grammar.grammar import GrammaticalExpression from ultk.effcomm.informativity import informativity, build_pairwise_matrix from .meaning import universe as modals_universe diff --git a/src/examples/modals/scripts/generate_expressions.py b/src/examples/modals/scripts/generate_expressions.py index 03ee977a..cefb629a 100644 --- a/src/examples/modals/scripts/generate_expressions.py +++ b/src/examples/modals/scripts/generate_expressions.py @@ -1,7 +1,7 @@ from ultk.util.io import write_expressions from ultk.language.semantics import Meaning -from ultk.language.grammar import GrammaticalExpression +from ultk.language.grammar.grammar import GrammaticalExpression from ..grammar import modals_grammar from ..meaning import universe as modals_universe diff --git a/src/tests/test_grammar.py b/src/tests/test_grammar.py index e9be3b23..ef9906ed 100644 --- a/src/tests/test_grammar.py +++ b/src/tests/test_grammar.py @@ -1,4 +1,4 @@ -from ultk.language.grammar import Grammar, GrammaticalExpression, Rule +from ultk.language.grammar.grammar import Grammar, GrammaticalExpression, Rule from ultk.language.semantics import Meaning, Referent, Universe diff --git a/src/tests/test_likelihood.py b/src/tests/test_likelihood.py new file mode 100644 index 00000000..968e37d4 --- /dev/null +++ b/src/tests/test_likelihood.py @@ -0,0 +1,57 @@ +from ultk.language.grammar.likelihood import ( + all_or_nothing, + percent_match_unique, + percent_match, + noise_match, +) +from math import log + + +# The expression can simply act as a function for this purpose +def expression(_): + return True + + +def even(i): + return i % 2 == 0 + + +class TestLikelihood: + all_true = [(i, True) for i in range(10)] + all_false = [(i, False) for i in range(10)] + half = [(i, i % 2 == 0) for i in range(10)] + + def test_all_or_nothing(self): + assert all_or_nothing(self.all_true, expression) == 1 + assert all_or_nothing(self.all_false, expression) == 0 + assert all_or_nothing(self.half, expression) == 0 + + def test_percent_match(self): + assert percent_match(self.all_true, expression) == 1 + assert percent_match(self.all_false, expression) == 0 + assert percent_match(self.half, expression) == 0.5 + + def test_percent_match_unique(self): + assert percent_match_unique(self.all_true, expression) == 0 + assert percent_match_unique(self.all_false, expression) == 0 + assert percent_match_unique(self.half, expression) == 0 + assert percent_match_unique(self.all_true, even) == 0.5 + assert percent_match_unique(self.all_false, even) == 0.5 + assert percent_match_unique(self.half, even) == 1 + + def test_noise_match(self): + noise_match_func = noise_match(2) + assert ( + abs(noise_match_func(self.all_true, expression) - log(0.995) * 10) < 0.00001 + ) + assert ( + abs(noise_match_func(self.all_false, expression) - log(0.005) * 10) + < 0.00001 + ) + assert ( + abs( + noise_match_func(self.half, expression) + - (log(0.995) * 5 + log(0.005) * 5) + ) + < 0.00001 + ) diff --git a/src/ultk/language/grammar/__init__.py b/src/ultk/language/grammar/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/src/ultk/language/grammar.py b/src/ultk/language/grammar/grammar.py similarity index 87% rename from src/ultk/language/grammar.py rename to src/ultk/language/grammar/grammar.py index 9911fb3c..5943365d 100644 --- a/src/ultk/language/grammar.py +++ b/src/ultk/language/grammar/grammar.py @@ -8,6 +8,8 @@ from itertools import product from typing import Any, Callable, Generator, TypedDict, TypeVar from yaml import load +from functools import cache +from math import log try: from yaml import CLoader as Loader @@ -90,7 +92,7 @@ def _and(p1: bool, p2: bool) -> bool: del args["weight"] # allow custon names too rule_name = func.__name__ - if "name" in args: + if "name" in args and args["name"].default is not inspect._empty: rule_name = args["name"].default del args["name"] # parameters = {'name': Parameter} ordereddict, so we want the values @@ -178,6 +180,26 @@ def count_atoms(self): return 1 return sum(child.count_atoms() for child in self.children) + def replace_children(self, children) -> None: + self.children = children + + @cache + def node_count(self) -> int: + """Count the node of a GrammaticalExpression + + Returns: + int: node count + """ + counter = 1 + stack = [self] + while stack: + current_node = stack.pop() + children = current_node.children if current_node.children else () + for child in children: + stack.append(child) + counter += 1 + return counter + @classmethod def from_dict(cls, the_dict: dict, grammar: "Grammar") -> "GrammaticalExpression": children = the_dict.get("children") @@ -258,6 +280,46 @@ def add_rule(self, rule: Rule): ) self._rules_by_name[rule.name] = rule + # @cache, unhashable, or embed it as a property (change every time Grammar is changed) + def probability(self, rule: Rule) -> float: + return float(rule.weight) / sum([r.weight for r in self._rules[rule.lhs]]) + + # @cache, unhashable, or embed it as a property (change every time Grammar is changed) + def log_probability(self, rule: Rule) -> float: + return log(float(rule.weight)) - log( + sum([r.weight for r in self._rules[rule.lhs]]) + ) + + def prior(self, expr: GrammaticalExpression) -> float: + """Prior of a GrammaticalExpression + + Args: + expr (GrammaticalExpression): the GrammaticalExpression for compuation + + Returns: + float: prior + """ + probability = self.probability(self._rules_by_name[expr.rule_name]) + children = expr.children if expr.children else () + for child in children: + probability = probability * (self.prior(child)) + return probability + + def log_prior(self, expr: GrammaticalExpression) -> float: + """Prior of a GrammaticalExpression in log probability + + Args: + expr (GrammaticalExpression): the GrammaticalExpression for compuation + + Returns: + float: log prior + """ + probability = self.log_probability(self._rules_by_name[expr.rule_name]) + children = expr.children if expr.children else () + for child in children: + probability = probability + (self.log_prior(child)) + return probability + def parse( self, expression: str, @@ -321,21 +383,29 @@ def parse( ) ) if len(stack) != 1: - raise ValueError("Could not parse string {expression}") + raise ValueError(f"Could not parse string {expression}") return stack[0] - def generate(self, lhs: Any = None) -> GrammaticalExpression: + def generate(self, lhs: Any = None, max_depth=3, depth=0) -> GrammaticalExpression: """Generate an expression from a given lhs.""" if lhs is None: lhs = self._start rules = self._rules[lhs] + # Stop there from being a high chance of infinite recusion + if depth > max_depth: + filtered_rules = list(filter(lambda rule: rule.rhs is None, rules)) + if len(filtered_rules) != 0: + rules = filtered_rules the_rule = random.choices(rules, weights=[rule.weight for rule in rules], k=1)[ 0 ] children = ( None if the_rule.rhs is None - else tuple([self.generate(child_lhs) for child_lhs in the_rule.rhs]) + else tuple( + self.generate(child_lhs, max_depth=max_depth, depth=depth + 1) + for child_lhs in the_rule.rhs + ) ) # if the rule is terminal, rhs will be empty, so no recursive calls to generate will be made in this comprehension return GrammaticalExpression( @@ -550,6 +620,9 @@ def from_module(cls, module_name: str) -> "Grammar": The module should have a list of type-annotated method definitions, each of which will correspond to one Rule in the new Grammar. See the docstring for `Rule.from_callable` for more information on how that step works. + The function will normally attempt to convert all functions (including imported functions) into Rules. However, if a tuple of + functions called `grammar_rules` is defined in the module, it will only try to convert the functions contained in the tuple. + The start symbol of the grammar can either be specified by `start = XXX` somewhere in the module, or will default to the LHS of the first rule in the module (aka the return type annotation of the first method definition). @@ -558,7 +631,11 @@ def from_module(cls, module_name: str) -> "Grammar": """ module = import_module(module_name) grammar = cls(None) - for name, value in inspect.getmembers(module): + if hasattr(module, "grammar_rules") and type(module.grammar_rules) == tuple: + possible_rules = module.grammar_rules + else: + possible_rules = tuple(value for _, value in inspect.getmembers(module)) + for value in possible_rules: # functions become rules if inspect.isfunction(value): grammar.add_rule(Rule.from_callable(value)) diff --git a/src/ultk/language/grammar/inference.py b/src/ultk/language/grammar/inference.py new file mode 100644 index 00000000..d4c7173a --- /dev/null +++ b/src/ultk/language/grammar/inference.py @@ -0,0 +1,162 @@ +from typing import TypeVar, Iterable, Callable +from ultk.language.grammar.grammar import Grammar, GrammaticalExpression +from ultk.language.grammar.likelihood import Dataset, all_or_nothing +from math import isnan, isinf, exp, log +import copy +import random + + +def log_mh_sample( + expr: GrammaticalExpression, + grammar: Grammar, + data: Dataset, + likelihood_func: Callable[[Dataset, GrammaticalExpression], float] = all_or_nothing, + likelihood_weight: float = 1, + subtree_weight: float = 1, +) -> GrammaticalExpression: + """Sample a new GrammaticalExpression from an exsiting one and data using Metropolis Hastings using log probabilities + + Args: + expr (GrammaticalExpression): the exsiting GrammaticalExpression + grammar (Grammar): the grammar for generation + data (Dataset): data used for calculation of acceptance probability + likelihood_func (Callable[[Dataset, GrammaticalExpression], float], optional): _description_. Defaults to all_or_nothing. + likelihood_weight (float, optional): Weight for the likelihood/prior of the full tree. Defaults to 1. + subtree_weight (float, optional): Weight for the subtree prior and length. Defaults to 1. + + Returns: + GrammaticalExpression: newly sampled GrammaticalExpression + """ + old_tree_prior = grammar.log_prior(expr) + old_node_count = log(expr.node_count()) + old_tree_likelihood = likelihood_func(data, expr) + while True: + old_tree = copy.deepcopy(expr) + current_node, parent_node = mh_select(old_tree) + old_subtree_prior = grammar.log_prior(current_node) + new_tree, new_node = mh_generate(old_tree, current_node, parent_node, grammar) + new_tree_prior = grammar.log_prior(new_tree) + new_node_count = log(new_tree.node_count()) + new_subtree_prior = grammar.log_prior(new_node) + mh_accept = likelihood_weight * ( + (new_tree_prior + likelihood_func(data, new_tree)) + - (old_tree_prior + old_tree_likelihood) + ) + subtree_weight * ( + (old_subtree_prior - new_node_count) - (new_subtree_prior - old_node_count) + ) + if not (isnan(mh_accept) or (isinf(mh_accept) and mh_accept < 0)) and ( + mh_accept >= 0 + or random.random() + < exp(mh_accept) # Perhaps change later if rounding errors occur + ): + return new_tree + + +def mh_sample( + expr: GrammaticalExpression, + grammar: Grammar, + data: Dataset, + likelihood_func: Callable[[Dataset, GrammaticalExpression], float] = all_or_nothing, +) -> GrammaticalExpression: + """Sample a new GrammaticalExpression from an exsiting one and data using Metropolis Hastings + + Args: + expr (GrammaticalExpression): the exsiting GrammaticalExpression + grammar (Grammar): the grammar for generation + data (Dataset): data used for calculation of acceptance probability + likelihood_func (Callable[[Dataset, GrammaticalExpression], float], optional): _description_. Defaults to all_or_nothing. + + Returns: + GrammaticalExpression: newly sampled GrammaticalExpression + """ + old_tree_prior = grammar.prior(expr) + old_node_count = expr.node_count() + old_tree_likelihood = likelihood_func(data, expr) + while True: + old_tree = copy.deepcopy(expr) + current_node, parent_node = mh_select(old_tree) + old_subtree_prior = grammar.prior(current_node) + new_tree, new_node = mh_generate(old_tree, current_node, parent_node, grammar) + new_tree_prior = grammar.prior(new_tree) + new_node_count = new_tree.node_count() + new_subtree_prior = grammar.prior(new_node) + try: + mh_accept = min( + 1, + ( + (new_tree_prior * likelihood_func(data, new_tree)) + / (old_tree_prior * old_tree_likelihood) + ) + * ( + (old_subtree_prior / new_node_count) + / (new_subtree_prior / old_node_count) + ), + ) + except ZeroDivisionError: + mh_accept = 0 + if random.random() < mh_accept: + return new_tree + + +def mh_select( + old_tree: GrammaticalExpression, +) -> tuple[GrammaticalExpression, GrammaticalExpression]: + """Select a node for futher change from a GrammaticalExpression + + Args: + old_tree (GrammaticalExpression): input GrammaticalExpression + + Returns: + tuple[GrammaticalExpression, GrammaticalExpression]: the node selected for change and its parent node + """ + linearized_self = [] + parents = [] + stack = [(old_tree, -1)] + while stack: + current_node, parent_index = stack.pop() + linearized_self.append(current_node) + parents.append(parent_index) + current_index = len(linearized_self) - 1 + children = current_node.children if current_node.children else [] + for child in children: + stack.append((child, current_index)) + changing_node = random.choice(range(len(linearized_self))) + current_node = linearized_self[changing_node] + parent_node = linearized_self[parents[changing_node]] + return (current_node, parent_node) + + +def mh_generate( + old_tree: GrammaticalExpression, + current_node: GrammaticalExpression, + parent_node: GrammaticalExpression, + grammar: Grammar, +) -> tuple[GrammaticalExpression, GrammaticalExpression]: + """Generate a new GrammaticalExpression + + Args: + old_tree (GrammaticalExpression): the original full GrammaticalExpression + current_node (GrammaticalExpression): the node selected for change + parent_node (GrammaticalExpression): the parent node for the chaging node + grammar (Grammar): grammar used for generation + + Returns: + tuple[GrammaticalExpression, GrammaticalExpression]: the new full GrammaticalExpression and the changed node + """ + if current_node != old_tree: + new_children = [] + children = parent_node.children if parent_node.children else () + for child in children: + if child is current_node: + new_node = grammar.generate( + grammar._rules_by_name[current_node.rule_name].lhs + ) + new_children.append(new_node) + else: + new_children.append(child) + parent_node.replace_children(tuple(new_children)) + new_tree = old_tree + else: + new_node = grammar.generate(grammar._rules_by_name[old_tree.rule_name].lhs) + new_tree = new_node + return (new_tree, new_node) diff --git a/src/ultk/language/grammar/likelihood.py b/src/ultk/language/grammar/likelihood.py new file mode 100644 index 00000000..fd4eb7d3 --- /dev/null +++ b/src/ultk/language/grammar/likelihood.py @@ -0,0 +1,132 @@ +from typing import Callable, TypeVar, Iterable +from ultk.language.grammar.grammar import GrammaticalExpression +from ultk.language.semantics import Referent +from math import log + +T = TypeVar("T") +Datum = tuple[Referent, T] +Dataset = Iterable[Datum] + + +def all_or_nothing(data: Dataset, tree: GrammaticalExpression) -> float: + """Basic all or nothing likelihood, return 1 if all data are correctly predicted, 0 otherwise + + Args: + data (Dataset): data for likelihood calculation + tree (GrammaticalExpression): GrammaticalExpression for likelihood calculation + + Returns: + float: likelihood + """ + return float(all(tree(datum[0]) == datum[1] for datum in data)) + + +def percent_match(data: Dataset, tree: GrammaticalExpression) -> float: + """Basic percentage-based likelihood, returns the percent of matches across the output from the tree + and the expected output from the data + + Args: + data (Dataset): data for likelihood calculation + tree (GrammaticalExpression): GrammaticalExpression for likelihood calculation + + Returns: + float: likelihood + """ + return sum([tree(datum[0]) == datum[1] for datum in data]) / len(data) + + +def percent_match_unique(data: Dataset, tree: GrammaticalExpression) -> float: + """Basic percentage-based likelihood, returns the percent of matches across the output from the tree + and the expected output from the data. However, if all of the outputs of the tree are the same returns 0. + + Args: + data (Dataset): data for likelihood calculation + tree (GrammaticalExpression): GrammaticalExpression for likelihood calculation + + Returns: + float: likelihood + """ + first_value = None + same = True + total_matches = 0 + for datum in data: + val = tree(datum[0]) + if first_value is None: + first_value = val + elif same and val != first_value: + same = False + total_matches += int(val == datum[1]) + if same: + return 0 + return total_matches / len(data) + + +def noise_match( + possible_outputs: int, alpha: float = 0.01 +) -> Callable[[Dataset, GrammaticalExpression], float]: + """Taken from Piantadosi et al. Attempts to discern the probability by believing that the output is correct + and was passed through a noise function which has an `alpha` chance to corrupt each item in the output list. + + Takes in the number of possible values the output can be and the percent chance of a corruption and returns a + probability function which `mh_sample` is able to use. + + Specifically for log_mh_sample only. + + See also: https://github.com/piantado/LOTlib3/blob/master/Hypotheses/Likelihoods/BinaryLikelihood.py + + Args: + possible_ouputs (int): The number of possible values an output is able to be + alpha (float): The percentage chance that a value will be mutated + + Returns: + Callable likelihood function: + Args: + data (Dataset): Data for likelihood calculation + tree (GrammaticalExpression): GrammaticalExpression for likelihood calculation + Returns: + float: Likelihood in log probability + """ + # If the item is correct then it was either correct or was mutated to from an incorrect option + # It could also have been the correct option originally and still mutated + correct_chance = log(1 - alpha + alpha / possible_outputs) + # If the item is incorrect then it could've been mutated from a correct option + incorrect_chance = log(alpha / possible_outputs) + + def noise_match_probability(datum: Datum, tree: GrammaticalExpression) -> float: + return correct_chance if tree(datum[0]) == datum[1] else incorrect_chance + + return aggregate_individual_likelihoods(noise_match_probability) + + +def aggregate_individual_likelihoods( + likelihood_function: Callable[[Datum, GrammaticalExpression], float], +) -> Callable[[Dataset, GrammaticalExpression], float]: + """Takes in a likelihood function for an individual datum (in log probability) returns a likelihood function which calls the + individual probability function and calls it across the dataset, summing it to get the final probability. + + Specifically for log_mh_sample only. + + Args: + Callable individual likelihood function: + Args: + datum (Datum): An individual element from the dataset, the first element is the input, the second the output. + tree (GrammarticalExpression): GrammaticalExpression for likelihood calculation + Returns: + float: Likelihood in log probability. + + Returns: + Callable likelihood function: + Args: + data (Dataset): Data for likelihood calculation + tree (GrammaticalExpression): GrammaticalExpression for likelihood calculation + Returns: + float: Likelihood in log probability + """ + + def output_func(data: Dataset, tree: GrammaticalExpression) -> float: + output = 0 + for datum in data: + output += likelihood_function(datum, tree) + return output + + return output_func diff --git a/src/ultk/util/io.py b/src/ultk/util/io.py index eb9da213..e61de061 100644 --- a/src/ultk/util/io.py +++ b/src/ultk/util/io.py @@ -1,7 +1,7 @@ import pickle from ultk.language.language import Expression from ultk.language.semantics import Meaning, Universe -from ultk.language.grammar import Grammar, GrammaticalExpression +from ultk.language.grammar.grammar import Grammar, GrammaticalExpression from typing import Iterable from yaml import dump, Dumper, load, Loader