def PluralityLearner(dataset):\n",
- " """\n",
- " A very dumb algorithm: always pick the result that was most popular\n",
- " in the training data. Makes a baseline for comparison.\n",
- " """\n",
- " most_popular = mode([e[dataset.target] for e in dataset.examples])\n",
- "\n",
- " def predict(example):\n",
- " """Always return same result: the most popular from the training set."""\n",
- " return most_popular\n",
- "\n",
- " return predict\n",
- "def NearestNeighborLearner(dataset, k=1):\n",
- " """k-NearestNeighbor: the k nearest neighbors vote."""\n",
- "\n",
- " def predict(example):\n",
- " """Find the k closest items, and have them vote for the best."""\n",
- " best = heapq.nsmallest(k, ((dataset.distance(e, example), e) for e in dataset.examples))\n",
- " return mode(e[dataset.target] for (d, e) in best)\n",
- "\n",
- " return predict\n",
- "class DecisionFork:\n",
- " """\n",
- " A fork of a decision tree holds an attribute to test, and a dict\n",
- " of branches, one for each of the attribute's values.\n",
- " """\n",
- "\n",
- " def __init__(self, attr, attr_name=None, default_child=None, branches=None):\n",
- " """Initialize by saying what attribute this node tests."""\n",
- " self.attr = attr\n",
- " self.attr_name = attr_name or attr\n",
- " self.default_child = default_child\n",
- " self.branches = branches or {}\n",
- "\n",
- " def __call__(self, example):\n",
- " """Given an example, classify it using the attribute and the branches."""\n",
- " attr_val = example[self.attr]\n",
- " if attr_val in self.branches:\n",
- " return self.branches[attr_val](example)\n",
- " else:\n",
- " # return default class when attribute is unknown\n",
- " return self.default_child(example)\n",
- "\n",
- " def add(self, val, subtree):\n",
- " """Add a branch. If self.attr = val, go to the given subtree."""\n",
- " self.branches[val] = subtree\n",
- "\n",
- " def display(self, indent=0):\n",
- " name = self.attr_name\n",
- " print('Test', name)\n",
- " for (val, subtree) in self.branches.items():\n",
- " print(' ' * 4 * indent, name, '=', val, '==>', end=' ')\n",
- " subtree.display(indent + 1)\n",
- "\n",
- " def __repr__(self):\n",
- " return 'DecisionFork({0!r}, {1!r}, {2!r})'.format(self.attr, self.attr_name, self.branches)\n",
- "class DecisionLeaf:\n",
- " """A leaf of a decision tree holds just a result."""\n",
- "\n",
- " def __init__(self, result):\n",
- " self.result = result\n",
- "\n",
- " def __call__(self, example):\n",
- " return self.result\n",
- "\n",
- " def display(self):\n",
- " print('RESULT =', self.result)\n",
- "\n",
- " def __repr__(self):\n",
- " return repr(self.result)\n",
- "def DecisionTreeLearner(dataset):\n",
- " """[Figure 18.5]"""\n",
- "\n",
- " target, values = dataset.target, dataset.values\n",
- "\n",
- " def decision_tree_learning(examples, attrs, parent_examples=()):\n",
- " if len(examples) == 0:\n",
- " return plurality_value(parent_examples)\n",
- " if all_same_class(examples):\n",
- " return DecisionLeaf(examples[0][target])\n",
- " if len(attrs) == 0:\n",
- " return plurality_value(examples)\n",
- " A = choose_attribute(attrs, examples)\n",
- " tree = DecisionFork(A, dataset.attr_names[A], plurality_value(examples))\n",
- " for (v_k, exs) in split_by(A, examples):\n",
- " subtree = decision_tree_learning(exs, remove_all(A, attrs), examples)\n",
- " tree.add(v_k, subtree)\n",
- " return tree\n",
- "\n",
- " def plurality_value(examples):\n",
- " """\n",
- " Return the most popular target value for this set of examples.\n",
- " (If target is binary, this is the majority; otherwise plurality).\n",
- " """\n",
- " popular = argmax_random_tie(values[target], key=lambda v: count(target, v, examples))\n",
- " return DecisionLeaf(popular)\n",
- "\n",
- " def count(attr, val, examples):\n",
- " """Count the number of examples that have example[attr] = val."""\n",
- " return sum(e[attr] == val for e in examples)\n",
- "\n",
- " def all_same_class(examples):\n",
- " """Are all these examples in the same target class?"""\n",
- " class0 = examples[0][target]\n",
- " return all(e[target] == class0 for e in examples)\n",
- "\n",
- " def choose_attribute(attrs, examples):\n",
- " """Choose the attribute with the highest information gain."""\n",
- " return argmax_random_tie(attrs, key=lambda a: information_gain(a, examples))\n",
- "\n",
- " def information_gain(attr, examples):\n",
- " """Return the expected reduction in entropy from splitting by attr."""\n",
- "\n",
- " def I(examples):\n",
- " return information_content([count(target, v, examples) for v in values[target]])\n",
- "\n",
- " n = len(examples)\n",
- " remainder = sum((len(examples_i) / n) * I(examples_i) for (v, examples_i) in split_by(attr, examples))\n",
- " return I(examples) - remainder\n",
- "\n",
- " def split_by(attr, examples):\n",
- " """Return a list of (val, examples) pairs for each val of attr."""\n",
- " return [(v, [e for e in examples if e[attr] == v]) for v in values[attr]]\n",
- "\n",
- " return decision_tree_learning(dataset.examples, dataset.inputs)\n",
- "def RandomForest(dataset, n=5):\n",
- " """An ensemble of Decision Trees trained using bagging and feature bagging."""\n",
- "\n",
- " def data_bagging(dataset, m=0):\n",
- " """Sample m examples with replacement"""\n",
- " n = len(dataset.examples)\n",
- " return weighted_sample_with_replacement(m or n, dataset.examples, [1] * n)\n",
- "\n",
- " def feature_bagging(dataset, p=0.7):\n",
- " """Feature bagging with probability p to retain an attribute"""\n",
- " inputs = [i for i in dataset.inputs if probability(p)]\n",
- " return inputs or dataset.inputs\n",
- "\n",
- " def predict(example):\n",
- " print([predictor(example) for predictor in predictors])\n",
- " return mode(predictor(example) for predictor in predictors)\n",
- "\n",
- " predictors = [DecisionTreeLearner(DataSet(examples=data_bagging(dataset), attrs=dataset.attrs,\n",
- " attr_names=dataset.attr_names, target=dataset.target,\n",
- " inputs=feature_bagging(dataset))) for _ in range(n)]\n",
- "\n",
- " return predict\n",
- "def NaiveBayesDiscrete(dataset):\n",
- " """\n",
- " Just count how many times each value of each input attribute\n",
- " occurs, conditional on the target value. Count the different\n",
- " target values too.\n",
- " """\n",
- "\n",
- " target_vals = dataset.values[dataset.target]\n",
- " target_dist = CountingProbDist(target_vals)\n",
- " attr_dists = {(gv, attr): CountingProbDist(dataset.values[attr]) for gv in target_vals for attr in dataset.inputs}\n",
- " for example in dataset.examples:\n",
- " target_val = example[dataset.target]\n",
- " target_dist.add(target_val)\n",
- " for attr in dataset.inputs:\n",
- " attr_dists[target_val, attr].add(example[attr])\n",
- "\n",
- " def predict(example):\n",
- " """\n",
- " Predict the target value for example. Consider each possible value,\n",
- " and pick the most likely by looking at each attribute independently.\n",
- " """\n",
- "\n",
- " def class_probability(target_val):\n",
- " return (target_dist[target_val] * product(attr_dists[target_val, attr][example[attr]]\n",
- " for attr in dataset.inputs))\n",
- "\n",
- " return max(target_vals, key=class_probability)\n",
- "\n",
- " return predict\n",
- "def NaiveBayesContinuous(dataset):\n",
- " """\n",
- " Count how many times each target value occurs.\n",
- " Also, find the means and deviations of input attribute values for each target value.\n",
- " """\n",
- " means, deviations = dataset.find_means_and_deviations()\n",
- "\n",
- " target_vals = dataset.values[dataset.target]\n",
- " target_dist = CountingProbDist(target_vals)\n",
- "\n",
- " def predict(example):\n",
- " """Predict the target value for example. Consider each possible value,\n",
- " and pick the most likely by looking at each attribute independently."""\n",
- "\n",
- " def class_probability(target_val):\n",
- " prob = target_dist[target_val]\n",
- " for attr in dataset.inputs:\n",
- " prob *= gaussian(means[target_val][attr], deviations[target_val][attr], example[attr])\n",
- " return prob\n",
- "\n",
- " return max(target_vals, key=class_probability)\n",
- "\n",
- " return predict\n",
- "def NaiveBayesSimple(distribution):\n",
- " """\n",
- " A simple naive bayes classifier that takes as input a dictionary of\n",
- " CountingProbDist objects and classifies items according to these distributions.\n",
- " The input dictionary is in the following form:\n",
- " (ClassName, ClassProb): CountingProbDist\n",
- " """\n",
- " target_dist = {c_name: prob for c_name, prob in distribution.keys()}\n",
- " attr_dists = {c_name: count_prob for (c_name, _), count_prob in distribution.items()}\n",
- "\n",
- " def predict(example):\n",
- " """Predict the target value for example. Calculate probabilities for each\n",
- " class and pick the max."""\n",
- "\n",
- " def class_probability(target_val):\n",
- " attr_dist = attr_dists[target_val]\n",
- " return target_dist[target_val] * product(attr_dist[a] for a in example)\n",
- "\n",
- " return max(target_dist.keys(), key=class_probability)\n",
- "\n",
- " return predict\n",
- "def PerceptronLearner(dataset, learning_rate=0.01, epochs=100):\n",
- " """Logistic Regression, NO hidden layer"""\n",
- " i_units = len(dataset.inputs)\n",
- " o_units = len(dataset.values[dataset.target])\n",
- " hidden_layer_sizes = []\n",
- " raw_net = network(i_units, hidden_layer_sizes, o_units)\n",
- " learned_net = BackPropagationLearner(dataset, raw_net, learning_rate, epochs)\n",
- "\n",
- " def predict(example):\n",
- " o_nodes = learned_net[1]\n",
- "\n",
- " # forward pass\n",
- " for node in o_nodes:\n",
- " in_val = dot_product(example, node.weights)\n",
- " node.value = node.activation(in_val)\n",
- "\n",
- " # hypothesis\n",
- " return find_max_node(o_nodes)\n",
- "\n",
- " return predict\n",
- "def LinearLearner(dataset, learning_rate=0.01, epochs=100):\n",
- " """\n",
- " [Section 18.6.3]\n",
- " Linear classifier with hard threshold.\n",
- " """\n",
- " idx_i = dataset.inputs\n",
- " idx_t = dataset.target\n",
- " examples = dataset.examples\n",
- " num_examples = len(examples)\n",
- "\n",
- " # X transpose: the actual value of each input feature across the examples\n",
- " X_col = [[example[i] for example in examples] for i in idx_i] # vertical columns of X\n",
- "\n",
- " # add dummy\n",
- " ones = [1 for _ in range(len(examples))]\n",
- " X_col = [ones] + X_col\n",
- "\n",
- " # initialize random weights\n",
- " num_weights = len(idx_i) + 1\n",
- " w = random_weights(min_value=-0.5, max_value=0.5, num_weights=num_weights)\n",
- "\n",
- " for epoch in range(epochs):\n",
- " err = []\n",
- " # pass over all examples\n",
- " for example in examples:\n",
- " x = [1] + [example[i] for i in idx_i]\n",
- " y = np.dot(w, x)\n",
- " t = example[idx_t]\n",
- " err.append(t - y)\n",
- "\n",
- " # update weights\n",
- " for i in range(len(w)):\n",
- " w[i] = w[i] + learning_rate * (np.dot(err, X_col[i]) / num_examples)\n",
- "\n",
- " def predict(example):\n",
- " x = [1] + [example[i] for i in idx_i]\n",
- " return np.dot(w, x)\n",
- "\n",
- " return predict\n",
- "def EnsembleLearner(learners):\n",
- " """Given a list of learning algorithms, have them vote."""\n",
- "\n",
- " def train(dataset):\n",
- " predictors = [learner(dataset) for learner in learners]\n",
- "\n",
- " def predict(example):\n",
- " return mode(predictor(example) for predictor in predictors)\n",
- "\n",
- " return predict\n",
- "\n",
- " return train\n",
- "def ada_boost(dataset, L, K):\n",
- " """[Figure 18.34]"""\n",
- "\n",
- " examples, target = dataset.examples, dataset.target\n",
- " n = len(examples)\n",
- " eps = 1 / (2 * n)\n",
- " w = [1 / n] * n\n",
- " h, z = [], []\n",
- " for k in range(K):\n",
- " h_k = L(dataset, w)\n",
- " h.append(h_k)\n",
- " error = sum(weight for example, weight in zip(examples, w) if example[target] != h_k(example))\n",
- " # avoid divide-by-0 from either 0% or 100% error rates\n",
- " error = np.clip(error, eps, 1 - eps)\n",
- " for j, example in enumerate(examples):\n",
- " if example[target] == h_k(example):\n",
- " w[j] *= error / (1 - error)\n",
- " w = normalize(w)\n",
- " z.append(np.log((1 - error) / error))\n",
- " return weighted_majority(h, z)\n",
- "def WeightedLearner(unweighted_learner):\n",
- " """\n",
- " [Page 749 footnote 14]\n",
- " Given a learner that takes just an unweighted dataset, return\n",
- " one that takes also a weight for each example.\n",
- " """\n",
- "\n",
- " def train(dataset, weights):\n",
- " return unweighted_learner(replicated_dataset(dataset, weights))\n",
- "\n",
- " return train\n",
- "def cross_validation(learner, dataset, size=None, k=10, trials=1):\n",
- " """\n",
- " Do k-fold cross_validate and return their mean.\n",
- " That is, keep out 1/k of the examples for testing on each of k runs.\n",
- " Shuffle the examples first; if trials > 1, average over several shuffles.\n",
- " Returns Training error, Validation error\n",
- " """\n",
- " k = k or len(dataset.examples)\n",
- " if trials > 1:\n",
- " trial_errT = 0\n",
- " trial_errV = 0\n",
- " for t in range(trials):\n",
- " errT, errV = cross_validation(learner, dataset, size, k, trials)\n",
- " trial_errT += errT\n",
- " trial_errV += errV\n",
- " return trial_errT / trials, trial_errV / trials\n",
- " else:\n",
- " fold_errT = 0\n",
- " fold_errV = 0\n",
- " n = len(dataset.examples)\n",
- " examples = dataset.examples\n",
- " random.shuffle(dataset.examples)\n",
- " for fold in range(k):\n",
- " train_data, val_data = train_test_split(dataset, fold * (n // k), (fold + 1) * (n // k))\n",
- " dataset.examples = train_data\n",
- " h = learner(dataset, size)\n",
- " fold_errT += err_ratio(h, dataset, train_data)\n",
- " fold_errV += err_ratio(h, dataset, val_data)\n",
- " # reverting back to original once test is completed\n",
- " dataset.examples = examples\n",
- " return fold_errT / k, fold_errV / k\n",
- "class PartialOrderPlanner:\n",
- " """\n",
- " [Section 10.13] PARTIAL-ORDER-PLANNER\n",
- "\n",
- " Partially ordered plans are created by a search through the space of plans\n",
- " rather than a search through the state space. It views planning as a refinement of partially ordered plans.\n",
- " A partially ordered plan is defined by a set of actions and a set of constraints of the form A < B,\n",
- " which denotes that action A has to be performed before action B.\n",
- " To summarize the working of a partial order planner,\n",
- " 1. An open precondition is selected (a sub-goal that we want to achieve).\n",
- " 2. An action that fulfils the open precondition is chosen.\n",
- " 3. Temporal constraints are updated.\n",
- " 4. Existing causal links are protected. Protection is a method that checks if the causal links conflict\n",
- " and if they do, temporal constraints are added to fix the threats.\n",
- " 5. The set of open preconditions is updated.\n",
- " 6. Temporal constraints of the selected action and the next action are established.\n",
- " 7. A new causal link is added between the selected action and the owner of the open precondition.\n",
- " 8. The set of new causal links is checked for threats and if found, the threat is removed by either promotion or\n",
- " demotion. If promotion or demotion is unable to solve the problem, the planning problem cannot be solved with\n",
- " the current sequence of actions or it may not be solvable at all.\n",
- " 9. These steps are repeated until the set of open preconditions is empty.\n",
- " """\n",
- "\n",
- " def __init__(self, planning_problem):\n",
- " self.tries = 1\n",
- " # safety bounds for the backtracking search in execute(): the maximum\n",
- " # number of actions a plan may contain (iterative-deepening target) and\n",
- " # the maximum number of node expansions per deepening level\n",
- " self._max_plan_actions = 12\n",
- " self._max_expansions = 20000\n",
- " self.planning_problem = planning_problem\n",
- " self.causal_links = []\n",
- " self.start = Action('Start', [], self.planning_problem.initial)\n",
- " self.finish = Action('Finish', self.planning_problem.goals, [])\n",
- " self.actions = set()\n",
- " self.actions.add(self.start)\n",
- " self.actions.add(self.finish)\n",
- " self.constraints = set()\n",
- " self.constraints.add((self.start, self.finish))\n",
- " self.agenda = set()\n",
- " for precond in self.finish.precond:\n",
- " self.agenda.add((precond, self.finish))\n",
- " self.expanded_actions = planning_problem.expand_actions()\n",
- "\n",
- " def find_open_precondition(self):\n",
- " """\n",
- " Find the open precondition with the least number of achieving actions\n",
- " (a most-constrained-variable heuristic). Returns the triple\n",
- " (precondition, action_that_needs_it, [achieving_actions]). Iteration is\n",
- " ordered deterministically so the search does not depend on set/hash\n",
- " ordering. Returns (None, None, None) when some open precondition has no\n",
- " achiever at all, which is a dead end for the current partial plan.\n",
- " """\n",
- " possible_actions = list(self.actions) + self.expanded_actions\n",
- " number_of_ways = dict()\n",
- " actions_for_precondition = dict()\n",
- " for open_precondition, act in sorted(self.agenda, key=str):\n",
- " if open_precondition in number_of_ways:\n",
- " continue\n",
- " achievers = [action for action in possible_actions\n",
- " if any(effect == open_precondition for effect in action.effect)]\n",
- " if not achievers:\n",
- " return None, None, None\n",
- " number_of_ways[open_precondition] = len(achievers)\n",
- " actions_for_precondition[open_precondition] = achievers\n",
- "\n",
- " if not number_of_ways:\n",
- " return None, None, None\n",
- "\n",
- " chosen = min(number_of_ways, key=lambda p: (number_of_ways[p], str(p)))\n",
- " act1 = next(act for precond, act in sorted(self.agenda, key=str) if precond == chosen)\n",
- " return chosen, act1, actions_for_precondition[chosen]\n",
- "\n",
- " def find_action_for_precondition(self, oprec):\n",
- " """Find action for a given precondition"""\n",
- "\n",
- " # either\n",
- " # choose act0 E Actions such that act0 achieves G\n",
- " for action in self.actions:\n",
- " for effect in action.effect:\n",
- " if effect == oprec:\n",
- " return action, 0\n",
- "\n",
- " # or\n",
- " # choose act0 E Actions such that act0 achieves G\n",
- " for action in self.planning_problem.actions:\n",
- " for effect in action.effect:\n",
- " if effect.op == oprec.op:\n",
- " bindings = unify_mm(effect, oprec)\n",
- " if bindings is None:\n",
- " break\n",
- " return action, bindings\n",
- "\n",
- " def generate_expr(self, clause, bindings):\n",
- " """Generate atomic expression from generic expression given variable bindings"""\n",
- "\n",
- " new_args = []\n",
- " for arg in clause.args:\n",
- " if arg in bindings:\n",
- " new_args.append(bindings[arg])\n",
- " else:\n",
- " new_args.append(arg)\n",
- "\n",
- " try:\n",
- " return Expr(str(clause.name), *new_args)\n",
- " except:\n",
- " return Expr(str(clause.op), *new_args)\n",
- "\n",
- " def generate_action_object(self, action, bindings):\n",
- " """Generate action object given a generic action and variable bindings"""\n",
- "\n",
- " # if bindings is 0, it means the action already exists in self.actions\n",
- " if bindings == 0:\n",
- " return action\n",
- "\n",
- " # bindings cannot be None\n",
- " else:\n",
- " new_expr = self.generate_expr(action, bindings)\n",
- " new_preconds = []\n",
- " for precond in action.precond:\n",
- " new_precond = self.generate_expr(precond, bindings)\n",
- " new_preconds.append(new_precond)\n",
- " new_effects = []\n",
- " for effect in action.effect:\n",
- " new_effect = self.generate_expr(effect, bindings)\n",
- " new_effects.append(new_effect)\n",
- " return Action(new_expr, new_preconds, new_effects)\n",
- "\n",
- " def cyclic(self, graph):\n",
- " """Check cyclicity of a directed graph"""\n",
- "\n",
- " new_graph = dict()\n",
- " for element in graph:\n",
- " if element[0] in new_graph:\n",
- " new_graph[element[0]].append(element[1])\n",
- " else:\n",
- " new_graph[element[0]] = [element[1]]\n",
- "\n",
- " path = set()\n",
- "\n",
- " def visit(vertex):\n",
- " path.add(vertex)\n",
- " for neighbor in new_graph.get(vertex, ()):\n",
- " if neighbor in path or visit(neighbor):\n",
- " return True\n",
- " path.remove(vertex)\n",
- " return False\n",
- "\n",
- " value = any(visit(v) for v in new_graph)\n",
- " return value\n",
- "\n",
- " def add_const(self, constraint, constraints):\n",
- " """Add the constraint to constraints if the resulting graph is acyclic"""\n",
- "\n",
- " if constraint[0] == self.finish or constraint[1] == self.start:\n",
- " return constraints\n",
- "\n",
- " new_constraints = set(constraints)\n",
- " new_constraints.add(constraint)\n",
- "\n",
- " if self.cyclic(new_constraints):\n",
- " return constraints\n",
- " return new_constraints\n",
- "\n",
- " def is_a_threat(self, precondition, effect):\n",
- " """Check if effect is a threat to precondition"""\n",
- "\n",
- " if (str(effect.op) == 'Not' + str(precondition.op)) or ('Not' + str(effect.op) == str(precondition.op)):\n",
- " if effect.args == precondition.args:\n",
- " return True\n",
- " return False\n",
- "\n",
- " def protect(self, causal_link, action, constraints):\n",
- " """Check and resolve threats by promotion or demotion"""\n",
- "\n",
- " threat = False\n",
- " for effect in action.effect:\n",
- " if self.is_a_threat(causal_link[1], effect):\n",
- " threat = True\n",
- " break\n",
- "\n",
- " if action != causal_link[0] and action != causal_link[2] and threat:\n",
- " # try promotion\n",
- " new_constraints = set(constraints)\n",
- " new_constraints.add((action, causal_link[0]))\n",
- " if not self.cyclic(new_constraints):\n",
- " constraints = self.add_const((action, causal_link[0]), constraints)\n",
- " else:\n",
- " # try demotion\n",
- " new_constraints = set(constraints)\n",
- " new_constraints.add((causal_link[2], action))\n",
- " if not self.cyclic(new_constraints):\n",
- " constraints = self.add_const((causal_link[2], action), constraints)\n",
- " else:\n",
- " # both promotion and demotion fail\n",
- " print('Unable to resolve a threat caused by', action, 'onto', causal_link)\n",
- " return\n",
- " return constraints\n",
- "\n",
- " def convert(self, constraints):\n",
- " """Convert constraints into a dict of Action to set orderings"""\n",
- "\n",
- " graph = dict()\n",
- " for constraint in constraints:\n",
- " if constraint[0] in graph:\n",
- " graph[constraint[0]].add(constraint[1])\n",
- " else:\n",
- " graph[constraint[0]] = set()\n",
- " graph[constraint[0]].add(constraint[1])\n",
- " return graph\n",
- "\n",
- " def toposort(self, graph):\n",
- " """Generate topological ordering of constraints"""\n",
- "\n",
- " if len(graph) == 0:\n",
- " return\n",
- "\n",
- " graph = graph.copy()\n",
- "\n",
- " for k, v in graph.items():\n",
- " v.discard(k)\n",
- "\n",
- " extra_elements_in_dependencies = _reduce(set.union, graph.values()) - set(graph.keys())\n",
- "\n",
- " graph.update({element: set() for element in extra_elements_in_dependencies})\n",
- " while True:\n",
- " ordered = set(element for element, dependency in graph.items() if len(dependency) == 0)\n",
- " if not ordered:\n",
- " break\n",
- " yield ordered\n",
- " graph = {element: (dependency - ordered)\n",
- " for element, dependency in graph.items()\n",
- " if element not in ordered}\n",
- " if len(graph) != 0:\n",
- " raise ValueError('The graph is not acyclic and cannot be linearly ordered')\n",
- "\n",
- " def display_plan(self):\n",
- " """Display causal links, constraints and the plan"""\n",
- "\n",
- " print('Causal Links')\n",
- " for causal_link in self.causal_links:\n",
- " print(causal_link)\n",
- "\n",
- " print('\\n_constraints')\n",
- " for constraint in self.constraints:\n",
- " print(constraint[0], '<', constraint[1])\n",
- "\n",
- " print('\\n_partial Order Plan')\n",
- " print(list(reversed(list(self.toposort(self.convert(self.constraints))))))\n",
- "\n",
- " def execute(self, display=True):\n",
- " """\n",
- " Execute the algorithm with backtracking, using iterative deepening on the\n",
- " number of actions in the plan. The original greedy version committed to\n",
- " the first achiever it happened to iterate over and could not recover when\n",
- " that action's own preconditions turned out to be unsatisfiable, so it\n",
- " depended on hash ordering and often printed 'Probably Wrong' / "Couldn't\n",
- " find a solution". Backtracking over both action choices and threat\n",
- " resolution (promotion vs demotion), together with the deterministic\n",
- " selection in find_open_precondition and a smallest-plan-first deepening\n",
- " bound, makes the planner solve the standard problems reproducibly and\n",
- " return a short, valid plan.\n",
- " """\n",
- " pristine = self._snapshot()\n",
- " for limit in range(1, self._max_plan_actions + 1):\n",
- " self._restore(pristine)\n",
- " if self._search([self._max_expansions], limit):\n",
- " if display:\n",
- " self.display_plan()\n",
- " else:\n",
- " return self.constraints, self.causal_links\n",
- " return\n",
- " print("Couldn't find a solution")\n",
- " if not display:\n",
- " return None, None\n",
- "\n",
- " def _reachable(self, source, target):\n",
- " """True if target is forced to come after source by the ordering constraints"""\n",
- "\n",
- " stack, seen = [source], set()\n",
- " while stack:\n",
- " node = stack.pop()\n",
- " if node == target:\n",
- " return True\n",
- " if node in seen:\n",
- " continue\n",
- " seen.add(node)\n",
- " stack.extend(b for a, b in self.constraints if a == node)\n",
- " return False\n",
- "\n",
- " def _open_threat(self):\n",
- " """\n",
- " Return an (action, causal_link) threat that is not yet resolved by the\n",
- " ordering constraints, or None if every causal link is protected. A\n",
- " causal link (a0, p, a1) is threatened by an action whose effect negates p\n",
- " unless the action is already ordered before a0 (promotion) or after a1\n",
- " (demotion).\n",
- " """\n",
- " for a0, p, a1 in self.causal_links:\n",
- " for action in self.actions:\n",
- " if action == a0 or action == a1:\n",
- " continue\n",
- " if any(self.is_a_threat(p, effect) for effect in action.effect):\n",
- " if not (self._reachable(action, a0) or self._reachable(a1, action)):\n",
- " return action, (a0, p, a1)\n",
- " return None\n",
- "\n",
- " def _snapshot(self):\n",
- " return set(self.actions), set(self.constraints), list(self.causal_links), set(self.agenda)\n",
- "\n",
- " def _restore(self, snapshot):\n",
- " self.actions, self.constraints, self.causal_links, self.agenda = (\n",
- " set(snapshot[0]), set(snapshot[1]), list(snapshot[2]), set(snapshot[3]))\n",
- "\n",
- " def _search(self, budget, limit):\n",
- " """\n",
- " Recursively complete the partial plan, backtracking on failure. Three\n",
- " kinds of choice points are explored: which action satisfies an open\n",
- " precondition, how each threat is resolved (promotion vs demotion), and -\n",
- " bounded by 'limit' - whether to introduce a new action at all. Returns\n",
- " True and leaves the solution in self.* on success.\n",
- " """\n",
- " if budget[0] <= 0:\n",
- " return False\n",
- " budget[0] -= 1\n",
- "\n",
- " # first, resolve any outstanding threat to a causal link (choice point)\n",
- " threat = self._open_threat()\n",
- " if threat is not None:\n",
- " action, (a0, p, a1) = threat\n",
- " snapshot = self._snapshot()\n",
- " for ordering in ((action, a0), (a1, action)): # promotion, then demotion\n",
- " new_constraints = self.add_const(ordering, self.constraints)\n",
- " if ordering in new_constraints: # ordering was consistent (acyclic and allowed)\n",
- " self.constraints = new_constraints\n",
- " if self._search(budget, limit):\n",
- " return True\n",
- " self._restore(snapshot)\n",
- " return False\n",
- "\n",
- " # no open threats: a plan with an empty agenda is a complete solution\n",
- " if not self.agenda:\n",
- " return True\n",
- "\n",
- " # select <G, act1> from the agenda (most-constrained precondition first)\n",
- " G, act1, possible_actions = self.find_open_precondition()\n",
- " if G is None: # an open precondition has no achiever -> dead end\n",
- " return False\n",
- "\n",
- " # number of actions already introduced, excluding the dummy Start/Finish\n",
- " introduced = len(self.actions) - 2\n",
- " snapshot = self._snapshot()\n",
- " # try each achiever deterministically, reusing existing actions first\n",
- " for act0 in sorted(set(possible_actions), key=lambda a: (a not in self.actions, str(a))):\n",
- " is_new = act0 not in self.actions\n",
- " if is_new and introduced >= limit: # deepening bound on plan size\n",
- " continue\n",
- " self.agenda.discard((G, act1))\n",
- " self.actions.add(act0)\n",
- " self.constraints = self.add_const((self.start, act0), self.constraints)\n",
- " self.constraints = self.add_const((act0, act1), self.constraints)\n",
- " # the causal link act0 --G--> act1 requires act0 strictly before act1\n",
- " # (and after start); add_const drops an ordering that would create a\n",
- " # cycle, so reject the choice when the required ordering is not enforced\n",
- " if ((act0 == act1 or self._reachable(act0, act1)) and\n",
- " (act0 == self.start or self._reachable(self.start, act0))):\n",
- " if (act0, G, act1) not in self.causal_links:\n",
- " self.causal_links.append((act0, G, act1))\n",
- " if is_new: # a freshly introduced action contributes its own preconditions\n",
- " for precondition in act0.precond:\n",
- " self.agenda.add((precondition, act0))\n",
- " if self._search(budget, limit):\n",
- " return True\n",
- " # undo and try the next achiever\n",
- " self._restore(snapshot)\n",
- " return False\n",
- "class TrivialVacuumEnvironment(Environment):\n",
+ "class TrivialVacuumEnvironment(Environment):\n",
+ " \"\"\"This environment has two locations, A and B. Each can be Dirty\n",
+ " or Clean. The agent perceives its location and the location's\n",
+ " status. This serves as an example of how to implement a simple\n",
+ " Environment.\"\"\"\n",
"\n",
- " """This environment has two locations, A and B. Each can be Dirty\n",
- " or Clean. The agent perceives its location and the location's\n",
- " status. This serves as an example of how to implement a simple\n",
- " Environment."""\n",
+ " def __init__(self):\n",
+ " super().__init__()\n",
+ " self.status = {loc_AA: random.choice(['Clean', 'Dirty']),\n",
+ " loc_AB: random.choice(['Clean', 'Dirty']),\n",
+ " loc_BB: random.choice(['Clean', 'Dirty']),\n",
+ " loc_BA: random.choice(['Clean', 'Dirty'])\n",
+ " }\n",
"\n",
- " def __init__(self):\n",
- " super().__init__()\n",
- " self.status = {loc_A: random.choice(['Clean', 'Dirty']),\n",
- " loc_B: random.choice(['Clean', 'Dirty'])}\n",
+ " def thing_classes(self):\n",
+ " \"\"\"Return the Thing/Agent classes that may populate this vacuum world.\"\"\"\n",
+ " return [Wall, Dirt, ReflexVacuumAgent, RandomVacuumAgent, TableDrivenVacuumAgent, ModelBasedVacuumAgent]\n",
"\n",
- " def thing_classes(self):\n",
- " return [Wall, Dirt, ReflexVacuumAgent, RandomVacuumAgent,\n",
- " TableDrivenVacuumAgent, ModelBasedVacuumAgent]\n",
+ " def percept(self, agent):\n",
+ " \"\"\"Returns the agent's location, and the location status (Dirty/Clean).\"\"\"\n",
+ " return agent.location, self.status[agent.location]\n",
"\n",
- " def percept(self, agent):\n",
- " """Returns the agent's location, and the location status (Dirty/Clean)."""\n",
- " return (agent.location, self.status[agent.location])\n",
+ " def execute_action(self, agent, action):\n",
+ " \"\"\"Change agent's location and/or location's status; track performance.\n",
+ " Score 10 for each dirt cleaned; -1 for each move.\"\"\"\n",
+ " if action == 'Right':\n",
+ " agent.location = loc_B\n",
+ " agent.performance -= 1\n",
+ " elif action == 'Left':\n",
+ " agent.location = loc_A\n",
+ " agent.performance -= 1\n",
+ " elif action == 'Suck':\n",
+ " if self.status[agent.location] == 'Dirty':\n",
+ " agent.performance += 10\n",
+ " self.status[agent.location] = 'Clean'\n",
"\n",
- " def execute_action(self, agent, action):\n",
- " """Change agent's location and/or location's status; track performance.\n",
- " Score 10 for each dirt cleaned; -1 for each move."""\n",
- " if action == 'Right':\n",
- " agent.location = loc_B\n",
- " agent.performance -= 1\n",
- " elif action == 'Left':\n",
- " agent.location = loc_A\n",
- " agent.performance -= 1\n",
- " elif action == 'Suck':\n",
- " if self.status[agent.location] == 'Dirty':\n",
- " agent.performance += 10\n",
- " self.status[agent.location] = 'Clean'\n",
- "\n",
- " def default_location(self, thing):\n",
- " """Agents start in either location at random."""\n",
- " return random.choice([loc_A, loc_B])\n",
- "
\n",
- "\n",
- "\n"
+ " def default_location(self, thing):\n",
+ " \"\"\"Agents start in either location at random.\"\"\"\n",
+ " return random.choice([loc_A, loc_B])\n",
+ "