__author__ = "Bill MacCartney" __copyright__ = "Copyright 2015, Bill MacCartney" __credits__ = [] __license__ = "GNU General Public License, version 2.0" __version__ = "0.9" __maintainer__ = "Bill MacCartney" __email__ = "See the author's website" from collections import defaultdict import random from metrics import SemanticsAccuracyMetric, NumParsesMetric, standard_metrics from example import Example from learning import latent_sgd from parsing import is_cat, parse_to_pretty_string, print_grammar from scoring import Model, rule_features # TODO: comment def print_sample_outcomes(model=None, examples=[], name=None, metric=None, metric_test=None, k=10): candidates = [] for example in examples: parses = model.parse_input(example.input) metric_value = metric.evaluate(example, parses) if metric_test(metric_value): candidates.append(example) k = min(k, len(candidates)) samples = random.sample(candidates, k) print('%d of %d %s on %s:\n' % (k, len(candidates), name, metric.name())) inputs = [example.input for example in samples] for input in sorted(inputs): print(' ', input) print() return samples # TODO: comment def sample_wins_and_losses(domain=None, model=None, metric=None, seed=None): if seed: random.seed(seed) metric = metric or domain.training_metric() model = model or domain.model() train_examples=domain.train_examples() evaluate_model(model=model, examples=train_examples, metrics=domain.metrics(), print_examples=False) def my_print_sample_outcomes(name=None, metric_test=None, k=10): examples = print_sample_outcomes(model=model, examples=train_examples, name=name, metric=metric, metric_test=metric_test, k=k) my_print_sample_outcomes(name='wins', metric_test=(lambda n: n > 0), k=5) my_print_sample_outcomes(name='losses', metric_test=(lambda n: n == 0), k=10) # TODO: comment def print_parses(example, parses, metrics=[NumParsesMetric()], max_parses=10000, show_syntax=False): print('%-34s %s' % ('input', example.input)) if example.semantics != None: print('%-34s %s' % ('target semantics', str(example.semantics))) if example.denotation != None: print('%-34s %s' % ('target denotation', str(example.denotation))) print() for metric in metrics: metric_value = metric.evaluate(example, parses) print('%-34s %.3g' % (metric.name(), metric_value)) print() for idx, parse in enumerate(parses[:max_parses]): def outcome_marker(target, prediction): return ('+' if prediction == target else '-') if target != None else ' ' parse_outcome = outcome_marker(example.parse, parse) semantics_outcome = outcome_marker(example.semantics, parse.semantics) if example.denotation != None: denotation_outcome = outcome_marker(example.denotation, parse.denotation) else: denotation_outcome = ' ' lines = [] if show_syntax: lines.append('%-15s %s \n%s' % ('parse', parse_outcome, parse_to_pretty_string(parse))) lines.append('%-15s %s %s' % ('semantics', semantics_outcome, parse.semantics)) if example.denotation != None or parse.denotation != None: lines.append('%-15s %s %s' % ('denotation', denotation_outcome, parse.denotation)) for l, line in enumerate(lines): if l == 0: print('%-3s %8.3f %s' % (idx, parse.score, line)) else: print('%-3s %8s %s' % ('', '', line)) if len(parses) > max_parses: print('(additional parses truncated)') print('\n' + '-' * 80) # TODO: comment def evaluate_model(model=None, examples=[], examples_label=None, metrics=standard_metrics(), print_examples=True): print('=' * 80) print('Evaluating on %d %sexamples\n' % ( len(examples), examples_label + ' ' if examples_label else '')) print('-' * 80) metric_values = defaultdict(int) for example in examples: parses = model.parse_input(example.input) for metric in metrics: metric_value = metric.evaluate(example, parses) metric_values[metric.name()] += metric_value if print_examples: print_parses(example, parses, metrics=metrics) print('Over %d examples:' % len(examples)) print() for metric in metrics: print('%-34s %.3f' % (metric.name(), 1.0 * metric_values[metric.name()] / len(examples))) print() # TODO: comment def evaluate_grammar(grammar=None, executor=None, examples=[], examples_label=None, metrics=standard_metrics(), print_examples=True): evaluate_model(model=Model(grammar=grammar, executor=executor), examples=examples, metrics=metrics, print_examples=print_examples) def test_executor(domain): print('=' * 80) print('Test Executor\n') for example in domain.examples(): print('%-24s %s' % ('semantics', example.semantics)) print('%-24s %s' % ('target denotation', example.denotation)) actual_denotation = domain.execute(example.semantics) match = (actual_denotation == example.denotation) print('%-24s %s' % ('actual denotation', '[match]' if match else actual_denotation)) print() def evaluate_for_domain(domain, print_examples=True): print('#' * 80) print('Standard evaluation for domain: %s\n' % domain.__class__.__name__) # grammar = domain.grammar() # print_grammar(grammar) # print model = domain.model() evaluate_model(model=model, examples=domain.train_examples(), examples_label='train', metrics=domain.metrics(), print_examples=print_examples) evaluate_model(model=model, examples=domain.test_examples(), examples_label='test', metrics=domain.metrics(), print_examples=print_examples) def evaluate_dev_examples_for_domain(domain): evaluate_model(model=domain.model(), examples=domain.dev_examples(), examples_label='dev', metrics=domain.metrics(), print_examples=True) def train_test(model=None, train_examples=[], test_examples=[], metrics=standard_metrics(), training_metric=SemanticsAccuracyMetric(), seed=None, print_examples=False): # print_grammar(model.grammar) # print print('%d training examples, %d test examples' % (len(train_examples), len(test_examples))) # 'Before' test model.weights = defaultdict(float) # no weights evaluate_model(model=model, examples=train_examples, examples_label='train', metrics=metrics, print_examples=print_examples) evaluate_model(model=model, examples=test_examples, examples_label='test', metrics=metrics, print_examples=print_examples) # Train model = latent_sgd(model, train_examples, training_metric=training_metric, seed=seed) # 'After' test evaluate_model(model=model, examples=train_examples, examples_label='train', metrics=metrics, print_examples=print_examples) evaluate_model(model=model, examples=test_examples, examples_label='test', metrics=metrics, print_examples=print_examples) def train_test_for_domain(domain, seed=None, print_examples=False): print('#' * 80) print('Train/test experiment for domain: %s\n' % domain.__class__.__name__) train_test(model=domain.model(), train_examples=domain.train_examples(), test_examples=domain.test_examples(), metrics=domain.metrics(), training_metric=domain.training_metric(), seed=seed, print_examples=print_examples) def cartesian_product_of_lexical_rules(rules, restrict_by_lhs=True): """ Expands the given collection of rules by iterating through all possible pairs of existing lexical rules and adding a new rule which combines the RHS of the first rule with the semantics of the second. If restrict_by_lhs is true, we only consider pairs which have the same LHS, which helps to avoid constructing malformed semantics. """ from itertools import product from parsing import Rule, is_lexical lexical_rules = [rule for rule in rules if is_lexical(rule)] expanded_rules = [rule for rule in rules if not is_lexical(rule)] # Partition rules by lhs. lexical_rules_by_lhs = defaultdict(list) for rule in lexical_rules: lhs = rule.lhs if restrict_by_lhs else 'dummy' lexical_rules_by_lhs[lhs].append(rule) # In each partition, iterate through Cartesian product of lexical rules. for lhs, rules in list(lexical_rules_by_lhs.items()): sems = set([rule.sem for rule in rules]) for rule, sem in product(rules, sems): expanded_rules.append(Rule(rule.lhs, rule.rhs, sem)) return expanded_rules def learn_lexical_semantics(domain, seed=None): from parsing import Grammar print('#' * 80) print('Learn lexical semantics experiment for domain: %s\n' % domain.__class__.__name__) original_grammar = domain.grammar() expanded_rules = cartesian_product_of_lexical_rules(domain.rules()) grammar = Grammar(rules=expanded_rules, annotators=original_grammar.annotators, start_symbol=original_grammar.start_symbol) model = Model(grammar=grammar, feature_fn=domain.features, weights=domain.weights, executor=domain.execute) train_test(model=model, train_examples=domain.train_examples(), test_examples=domain.test_examples(), metrics=domain.metrics(), training_metric=domain.training_metric(), seed=seed, print_examples=False) def interact(domain, example_input=None, T=10): import readline model = domain.model() model = latent_sgd(model=model, examples=domain.train_examples(), training_metric=domain.training_metric(), T=T) print('\nHello! Enter a query%s:' % (', such as "%s"' % example_input if example_input else '')) while True: try: query = input('>>> ') except EOFError: print('\nBye!') return example = Example(input=query) parses = model.parse_input(query) if parses: print_parses(example, parses) else: print('No parse!') def generate(rules, start_symbol='$ROOT', n=100, min_tokens=3, max_tokens=10): rules_by_lhs = defaultdict(list) for rule in rules: rules_by_lhs[rule.lhs].append(rule) def sample_phrase(label): if label.startswith('?'): label = label[1:] if not is_cat(label): return label if label not in rules_by_lhs: return label rule = random.choice(rules_by_lhs[label]) phrases = [sample_phrase(label) for label in rule.rhs] return ' '.join(phrases) inputs = set() while len(inputs) < n: input = sample_phrase(start_symbol) print(input) num_tokens = len(input.split()) if num_tokens >= min_tokens and num_tokens <= max_tokens: inputs.add(input) print('-' * 80) for input in inputs: print(input) def find_best_rules(domain): model = domain.model() examples = domain.train_examples() metric = domain.training_metric() rule_counts = defaultdict(int) for example in examples: parses = model.parse_input(example.input) good_parses = [p for p in parses if metric.evaluate(example, [p])] if good_parses: best_parse = good_parses[0] features = rule_features(best_parse) for rule, count in list(features.items()): rule_counts[rule] = rule_counts[rule] + count counts = [(count, rule) for rule, count in list(rule_counts.items())] counts = sorted(counts, reverse=True) for count, rule in counts: print('%d\t%s' % (count, rule))