Writing communicating agents with no effort, step I

Introduction

If there's a pattern that arises over and over again in distributed computing, it is that of Communicating Agents. Communicating Agents are SW components (e.g. classes) that get instantiated at one or more computers interconnected by some kind of network, each instance able to live on its own but exchanging messages with other instances with the aim of reaching a common goal. For example, when the SW components being instantiated are processes at a single computer, we have Hoare's well-known CSP programming model.

Along the history of distributed computing we've seen multiple versions of the Communicating Agents pattern. From the early days of the Internet, where very simple deployments (typically client-server with a couple of servers and a handful of clients) were the norm, to these days where huge enterprise applications composed of hundreds of agents distributed across dozens of machines talk over a number of disparate networks.

Every seasoned programmer has met this pattern at least once in a life. Depending on the needs, context and tools at hand, one might decide to leverage existing distributed computing infrastructure like e.g. CORBA or EJB or simpler tools like e.g. Google's ProtocolBuffers; on the other hand, one might opt for developing own tools. This post deals with the latter case.

Our goal

We'll start by defining what we want to achieve. Say we are in the need of developing a distributed application built from multiple agents that cooperate by exchanging messages over some network linking the machines these agents run on. For the time being we're not concerned with run-time infrastructure, i.e. the middleware managing the agents' execution and life-cycle, so we will focus on achieving message definition and their exchange between our agents.

We have very little time and resources to do this so we'll be using Python as our rapid prototyping language, but we don't want to constrain ourselves to Python so we'd like our messages to be easily encoded/decoded using other languages as well. Thus we'll be using JSON as our encoding/decoding machinery of choice.

We want our message passing mechanism to be independent of the underlying inter-networking technology. We would also like to be able to change the inter-networking technology our agents use without impacting our agents' implementations.

We want our agents to be easily extensible so they can send and receive new messages. New -and changed- messages must have minimum to no impact on existing code.

Finally, we want our messages and message handling specifications to be defined in the code, so we don't have to resource to additional/external tools like IDL compilers, .proto files or the like.

We'll see that using advanced Python facilities like decorators and meta-classes achieving the above goals is quick and simple.  By writing a few lines of code we'll be able to quickly create classes implementing arbitrarily complex message protocols, using whatever communication means, and all this as easily as we'd write classes local to a single module or program.

Let me follow the TDD paradigm to reach our goal. In TDD, you write your tests before anything else, then along multiple iterations you write the code that eventually shall successfully pass those tests. Using PyUnit, we might write a test like the following:

import unittest ... class TestAgent ( object ):     def TestMsg ( *args, **kwargs ):         pass                def address ( self ):         pass                   def send ( self, msg, target ):         pass     def testMsgHandler ( self, msg, src ):         pass                     def __iter__ ( self ):         return iter([]) ... class CommAgentTest(unittest.TestCase):     def testCommAgent ( self ):         testmsgs = [             TestAgent.TestMsg(a=1, b='Hi '),             TestAgent.TestMsg(a=2, b='there!'),         ]         agent1, agent2 = TestAgent(), TestAgent()         for msg in testmsgs:             agent1.send(msg, agent2.address())                  self.assertListEqual(testmsgs, list(agent2), "Lists not equal") ... if __name__ == "__main__":     unittest.main()

Snippet 0 - PyUnit test of a dummy agent class

The test above runs but fails, as expected from the first iteration of a TDD. Let's complete the TestAgent class so the test passes:

import unittest from collections import namedtuple ... class TestAgent ( object ):     TestMsg = namedtuple('TestMsg', 'a,b')                def __init__ ( self ):         self.__rcvdmsgs = []     def address ( self ):         return self                   def send ( self, msg, target ):         return target.testMsgHandler(msg, self)     def testMsgHandler ( self, msg, src ):         self.__rcvdmsgs.append(msg)                     def __iter__ ( self ):         return iter(self.__rcvdmsgs) ... class CommAgentTest(unittest.TestCase):     def testCommAgent ( self ):         testmsgs = [             TestAgent.TestMsg(a=1, b='Hi '),             TestAgent.TestMsg(a=2, b='there!'),         ]         agent1, agent2 = TestAgent(), TestAgent()         for msg in testmsgs:             agent1.send(msg, agent2.address())                  self.assertListEqual(testmsgs, list(agent2), "Lists not equal") ... if __name__ == "__main__":     unittest.main()

Snippet 1 - PyUnit test of a prototypical Communicating Agent

In the test above, we define an agent class (TestAgent) whose protocol is made of just one message (TestMsg). We've decided to implement messages as instances of class namedtuple, which is as close as you can get to a C struct in Python. Agent functionality comes down to storing received messages in a private list (self.__rcvdmsgs), and is implemented in method TestAgent.testMsgHandler(). The class supports the iterator protocol (the __iter__() method) so we can easily obtain the messages received by an instance of the class.

Then we instantiate two of those agents and send a pre-defined set of messages from the first agent to the second, after which we test the second agent's received messages list against the pre-defined message set.

The unit test above runs OK. However, it falls short of reaching our goal, for the following -otherwise obvious- reasons:

• our agents are only able to talk to each other when they run on the same processor and within the same memory space, since our "network" is the function call stack
• we can't change our "network" without modifying our agents' implementations
• we can't exchange messages with agents written in Java (unless we build a Python-C-Java bridge or we run on Jython), and exchanging messages with agents written in C/C++ forces us to use the Python-C interface which is cumbersome when you're short of time
• implementing complex protocols would be difficult to maintain, since each agent class needs to know the method names of other agents' classes that handle each message of the protocol

In the following sections we'll fix some of the problems of this prototypical implementation.

Enabling network communication between agents

In order to enable our agents to talk to remote agents over a network, we need some networking code. We might write it ourselves as part of a base CommAgent class, but why bother when we have the fancy socketserver module?.

To retro-fit our agent class with networking capabilities, all we need is inheriting from a class in the socketserver module, and providing a handler class that manages the messages received from remote agents.

Let's take a first stake at networking our agents following the approach above:

import unittest from collections import namedtuple from socketserver import UDPServer, BaseRequestHandler import socket ... class TestAgentHandler ( BaseRequestHandler ):     def handle ( self ):         msg = self.request[0].strip()         src = self.client_address         self.server.testMsgHandler(msg, src) class TestAgent ( UDPServer ):     TestMsg = namedtuple('TestMsg', 'a,b')                  def __init__ ( self, local_address ):         super(TestAgent, self).__init__(local_address, TestAgentHandler)         self.__rcvdmsgs = []             def address ( self ):         return self.server_address     def send ( self, msg, dst ):         return self.socket.sendto(msg, dst)     def testMsgHandler ( self, msg, src ):         self.__rcvdmsgs.append(msg)                      def __iter__ ( self ):         return iter(self.__rcvdmsgs) ... class CommAgentTest(unittest.TestCase):     ...     def testCommAgent ( self ):         testmsgs = [             TestAgent.TestMsg(a=1, b='Hi '),             TestAgent.TestMsg(a=2, b='there!'),         ]         host = socket.gethostbyname(socket.gethostname())         ports = (2013, 2014)         agent1, agent2 = \             TestAgent((host, ports[0])), TestAgent((host, ports[1]))         [agent1.send(msg, agent2.address()) for msg in testmsgs]         try:             Timer(1, lambda: sleep(3) or agent2.shutdown()).start()             agent2.serve_forever()                         self.assertListEqual(testmsgs, list(agent2), "Lists not equal")         finally:             agent1.socket.close()             agent2.socket.close() ... if __name__ == "__main__":     unittest.main()

Snippet 2 - First stake at a networked Communicating Agent

It didn't take much pain to network-enable our agent, dit it?. I chose to use UDP due to its ease of use, but using a TCPServer instead of UDPServer shouldn't be much harder (I'll leave this to you fellow readers as an exercise).

I had to enhance our unit test a bit. When real networking comes up on stage we need to consider threading issues. We can't run agent2's server loop with agent2.serve_forever() and later on cause the loop to end with agent2.shutdown() from within the same thread. Since our interpreter's main thread blocks on the server loop we need an additional thread that calls agent2.shutdown(), and that's what we get with the Timer class. We're scheduling execution of agent2.shutdown() after 4 seconds (1 second until timer thread start, to provide some time for preparations, and 3 seconds before the shutdown() method is actually called).

Language-independence (of messages)

Unfortunately, if you run the test case above you'll get an annoying exception when calling agent1's send() method: "'TestMsg' does not support the buffer interface".

What does that mean? If you check the doc for the socket class (just type "import socket; help(socket.socket)" at the Python interpreter's prompt), you'll see its sendto() method refers you to its send() method, which reads "sends a data string to the socket". This is ambiguous at the very least, but the key lies in the string word: what it actually means is whatever data you pass to the sendto() method must either be an instance of class bytes, or of some other class that can be converted to an instance of bytes somehow. Class string is one example of the latter, which can be converted as bytes('<any string of characters here>', 'utf8').

Hence we need to take an additional step in order to get to a working networked agent implementation: we need to write a function that converts our protocol messages to instances of class bytes. Since we decided to use JSON for message exchange between our agents, let's add JSON encoding to our design:

import unittest from collections import namedtuple from socketserver import UDPServer, BaseRequestHandler import socket import json ... class TestAgent ( UDPServer ):     ...     def send ( self, msg, dst ):         jsonencodedmsg = \             type(msg).__name__ + ':' + json.dumps(msg.__dict__)         return self.socket.sendto(bytes(jsonencodedmsg, 'utf8'), dst)     ...

Snippet 3 - Networked Communicating Agent using JSON encoding

OK, it took just one line in the agent's send() method to enhance our agent (plus the corresponding import at the beginning). Why are we packing "type(msg).__name__ + ':'" ahead of the JSON encoding of our message instance? The answer is simple: when receiving the message, the receiving agent needs a way to tell the message apart the other messages in its protocol. It won't be able to do that just from the JSON encoding, hence we're pre-pending the encoded message with the message class name so the receiver knows which class it needs to instantiate to rebuild the message.

Now the test case runs, but fails. This indicates that we're still missing one last piece in the puzzle: we're sending JSON encoded messages over the network, but the receiving agent is not JSON-decoding the received message into a message instance that satisfies the assertEquals() check. Let's enhance our agent to do so:

import unittest from collections import namedtuple from socketserver import UDPServer, BaseRequestHandler import socket import json ... class TestAgentHandler ( BaseRequestHandler ):     def handle ( self ):         jsonencodedmsg = self.request[0].strip()         src = self.client_address         msgname, msgbody = jsonencodedmsg.decode().split(':', 1)         if msgname == "TestMsg":             msg = TestAgent.TestMsg(**json.loads(msgbody))             self.server.testMsgHandler(msg, src)         else:             pass    # ignore the message ...

Snippet 4 - Networked Communicating Agent using JSON encoding/decoding

With just a handful of new lines, in addition to JSON-decoding the received message I've added an 'if ...' clause to the TestAgentHandler class as a placeholder for future extensions of the protocol spoken by our agent.

This completes our first fully functional communicating agent. It's fairly simple to extend the protocol our agent handles - just add a new namedtuple sub-class per new message to the agent class, new handler method for each new namedtuple, and extend the 'if ...' clause inside the TestMsgHandler.handle() method to call the right handler method. I'll let the fun of doing so all for you.

This is how we'd go if we'd use a primitive language like C++ or Java. However, we're still far from reaching the goal we set for our solution above. Even if our agents do talk to each other over a network using JSON, changing the networking used implies changes in the agents' code (you already know that if you did the exercise of changing to TCPServer as networking support). Additionally, modifying or extending the protocol our agents talk requires code changes as well (as you know if you did the second exercise of extending the single-message protocol used in our example).

In the next post, we'll see how we can use advanced features of the Python language to solve those issues and reach an elegant, non-intrusive solution for writing communicating agents without the pain. Catch you all there!

Tidying up the house

Introduction

In the series of posts on problem solving by means of search, either blind of informed, we ended up with a single Problem class containing all the data and functions/methods necessary for declaring, solving and obtaining solutions for complex problems. We applied the Problem class to a number of real-world problems like the well-known missionaries&cannibals and N-queens problems, and rounded it all up with a road  navigation problem across the beautiful land of Romania.

I've decided I'm going to write a new series on solving problems in continuous spaces by means of local search and optimization. This is the kind of problems that a rover robot meets when travelling the world, for  example.

But, as my Mom uses to say, "before hosting new guests tidy up the house". So we're going to revise the implementation used across the previous series and clean it up according to OO principles. Thus this post serves as a quick introduction of OO design as well. Are you ready? Let's go then, I hope you enjoy the trip!

Problems with the Problem class (no pun intended)

The main issue with the Problem class we developed is that it mixes two different concepts in one single class. The class does not only model a problem but also its solution and the algorithm used to reach it.

This fact shows up notoriously in the need for two methods, treeSearch() and treeSearchWithCost(), which perform the same function using different algorithm classes. Moreover, there is no clue whatsoever in the class definition about which algorithms can be used with which algorithm class (i.e. which generators can be passed to each method).

Another indication of this mix up is that we cannot -at least not simultaneously- get multiple solutions to the same problem, not even using different algorithms.

Here's the definition of the Problem class we ended up with after the previous series (I've omitted class' private members for clarity):

class Problem(object):     class Node(tuple): ...                  def __init__ ( self, initial, goal, successor, cost ): ...              performance = property(getPerformance)          resources = property(getResources)          solution = property(getSolution)          def solved ( self ): ...          def solutionCost ( self ): ...              def graphSearch ( self, gen ): ...          def treeSearch ( self, gen, chk_dups = False ): ...     def treeSearchWithCost ( self, gen ): ...                              @staticmethod     def depthFirstGenerator ( successor ): ...              @staticmethod     def breadthFirstGenerator ( successor ): ...                  @staticmethod     def depthLimitedGenerator ( successor, max_level ): ...              @staticmethod     def uniformCostGenerator ( successor, cost ): ...                          @staticmethod     def iterativeDeepeningGenerator ( successor, anchor ): ...     @staticmethod     def iterativeLengtheningGenerator ( successor, cost, anchor ): ...                          @staticmethod     def bidirectionalSearchGenerator ( goal ): ...     @staticmethod ...     def greedyBestFirstGenerator ( successor, h ):          @staticmethod     def aStarGenerator ( successor, cost, h ): ...

Snippet 1 - Class structure of class Problem from the previous series

Our design goals with this exercise are as follows:

1. Having a Problem class that models exclusively a single problem;
2. Having a ProblemSolver class that solves problems expressed as instances of Problem;
3. Separating the blind and cost-based algorithm classes so it can be chosen which algorithm class to use, if possible using the same interface

So we're going to split the Problem class into a -shorter- Problem class and a new ProblemSolver class. As criteria for what to put where, we'll use a simple but effective rule-of-thumb: if something can't live without a class, then it's part of that class. This criterion is not only used in OO design but in Entity-Relationship modelling as well.

Let's start with the Node sub-class. A Node is an artifact used to hold book-keeping information in a search tree while a problem is being solved by means of a given algorithm. Thus, from its definition and the fact it does not need a Problem instance to live, we'll remove it from Problem. We'll include it in ProblemSolver though, not because that's the only other place it could go but because it makes sense. After all, even if a Node can live without a ProblemSolver, what's the point anyway?.

Then with Problem's properties. Neither of those are properties of Problem (they can happily live without it), but seem to be part of ProblemSolver (apart from solution, which may be debatable, the others make no sense without being attached to a ProblemSolver).

And finally with the class' methods; none of the class nor instance methods are part of Problem according to our rule-of-thumb criterion, but seem to be part of ProblemSolver instead.

Thus we're moving the Node sub-class, properties and methods over to the new ProblemSolver class. We will end up with something like this:

class Problem(object):     def __init__ ( self, initial, goal, successor, cost ): ... class ProblemSolver(object):              class Node(tuple): ...                  performance = property(getPerformance)          resources = property(getResources)          solution = property(getSolution)          def solve ( self, problem, gen, chk_dups = False ): ...     def solved ( self ): ...          def solutionCost ( self ): ...              def graphSearch ( self, gen ): ...          def treeSearch ( self, gen, chk_dups = False ): ...     def treeSearchWithCost ( self, gen ): ...                              @staticmethod     def depthFirstGenerator ( successor ): ...              @staticmethod     def breadthFirstGenerator ( successor ): ...                  @staticmethod     def depthLimitedGenerator ( successor, max_level ): ...              @staticmethod     def uniformCostGenerator ( successor, cost ): ...                          @staticmethod     def iterativeDeepeningGenerator ( successor, anchor ): ...     @staticmethod     def iterativeLengtheningGenerator ( successor, cost, anchor ): ...                          @staticmethod     def bidirectionalSearchGenerator ( goal ): ...     @staticmethod ...     def greedyBestFirstGenerator ( successor, h ):          @staticmethod     def aStarGenerator ( successor, cost, h ): ...

Snippet 2 - Class split between Problem and ProblemSolver

We've taken a first stab at the design but we're not done yet. Still our ProblemSolver class mixes up the two algorithm classes and provides no clue at which generators can be applied to which search method. Since we're talking about algorithm classes here, the design seems to naturally lend towards separating ProblemSolver into two different classes: one using blind search and another one using cost-based search. In order to offer the same interface for both algorithm classes, we might devise a inheritance scheme where the basic ProblemSolver provides the foundation for problem solving like properties, initialization method etc. plus an implementation of the interface using blind search, whereas a derived CostBasedProblemSolver overloads the interface to use cost-based search. We'd end up with something like in the following snippet:

class Problem(object):     def __init__ ( self, initial, goal, successor, cost ): ... class ProblemSolver(object):              class Node(tuple): ...                  performance = property(getPerformance)          resources = property(getResources)          solution = property(getSolution)          def solve ( self, problem, gen, chk_dups = False ): ...     def solved ( self ): ...          def solutionCost ( self ): ...              def graphSearch ( self, gen ): ...          def treeSearch ( self, gen, chk_dups = False ): ...                              @staticmethod     def depthFirstGenerator ( successor ): ...              @staticmethod     def breadthFirstGenerator ( successor ): ...                  @staticmethod     def depthLimitedGenerator ( successor, max_level ): ...              @staticmethod     def iterativeDeepeningGenerator ( successor, anchor ): ...                          @staticmethod     def bidirectionalSearchGenerator ( successor, goal ): ...     @staticmethod ...     def greedyBestFirstGenerator ( successor, h ): class CostBasedProblemSolver(ProblemSolver):     def treeSearch ( self, gen, chk_dups = False ): ...     @staticmethod     def uniformCostGenerator ( successor, cost ): ...                          @staticmethod     def iterativeLengtheningGenerator ( successor, cost, anchor ): ...          @staticmethod     def aStarGenerator ( successor, cost, h ): ...

Snippet 3 - Final class split fulfilling all our design goals

This re-factoring forces us to update all our unit tests to adapt to the new classes. The update is no big deal, just slight modifications to the test code. For example:

initial = ...     goal = ...     cost = lambda s, n: ...     successor = lambda n: ...     h = lambda n: ...                 problem = Problem(initial, goal, successor, cost)              solver = ProblemSolver()     solver.solve(problem, ...)     self.assertTrue(solver.solved(problem))

Snippet 4 - Modified unit test code using the new class split

I specially love how much sense the solver.solved(problem) call seems to have. When calls to your interface seem to mean what they actually do, it's good sign that you're on the right track with your design.

The complete code for the re-factored search framework can be obtained here. As always, if you're actually leveraging this code in your designs please drop me a line so I know this work is being useful to someone else.

In the next post we'll implement Local Search and Optimization based problem solvers that enable us to solve problems in continuous spaces. I hope to see you there.

TCP Dynamic Behavior

This post has been moved here.

Finally: informed search algorithms

Introduction

In this post we'll look at how to implement informed search, a.k.a. heuristic search. Heuristic search leverages knowledge of the problem space to speed up search, which is a very nice feature when search trees are very wide and/or deep. Previous knowledge of the problem space is modeled as a heuristic function, which is a function receiving a state as argument and returning a value representing how far the heuristic function thinks that state is from a goal state.

The effectiveness of a heuristic search algorithm when compared to an uninformed search counterpart depends heavily on the heuristic function's ability to provide accurate hints about how far from a goal the current node being examined is. Devising accurate heuristics is an art in itself, though there are automated methods that may help. We won't be discussing heuristic devising in this post, but you can find additional information in the AI book I've used as reference across the series.

Heuristic function characterization

So can we use any function taking a state as input and providing a numerical value as output as an effective heuristic? The obvious answer is no. In order to be effective with a search algorithm, a heuristic function h must fulfill a number of properties:

1. Returned value must be positive and not zero for all but the solution state, and must be 0 for the solution state;
2. The function must be admissible, i.e. it must always underestimate the actual cost of reaching a solution from any given state;
3. In addition to being admissible the function should be consistent, i.e. when evaluated for a state actually closer to a solution than a previous state it must return a value lower than that returned for said previous state (this condition is required only when checking for duplicates, which concerns us since if you check the treeSearchWithCost() method from the previous post you'll see that's always the case)

The above conditions are necessary for a heuristic search algorithm to be complete and optimal, though they are not sufficient, as we we'll see shortly.

It turns out that it is not very common to meet not consistent heuristics, specially with real-world problems like the one we used in the previous post. Let's continue using the same problem. A consistent heuristic for that problem is the straight line distance SLD between two towns. Since the straight line distance is the shortest possible distance between two points (in 3 dimensions, for the astro-physicists amongst you), it must be admissible. And applying the triangle inequality it can be seen that it is also consistent. We can model this heuristic function in our problem definition as follows:

sld_bucharest = {     "Arad": 366,     "Bucharest": 0,     "Craiova": 160,     "Drobeta": 242,     "Eforie": 161,     "Fagaras": 176,     "Giurgiu": 77,     "Hirsova": 151,     "Iasi": 226,     "Lugoj": 244,     "Mehadia": 241,     "Neamt": 234,     "Oradea": 380,     "Pitesti": 100,     "Rimnicu Vilcea": 193,     "Sibiu": 253,     "Timisoara": 329,     "Urcizeni": 80,     "Vaslui": 199,     "Zerind": 374, } h = lambda n: sld_bucharest[n]

Snippet 1 - Modeling the consistent SLD heuristic

We will be using the SLD heuristic with the algorithms that follow.

Greedy best-first search

The simplest form of heuristic search algorithm is called greedy best-first search. In this algorithm search always follows blindly the path with the lowest value of h.

Implementing greedy best-first search in our generic search framework is stupidly simple; we just have to apply breadth-first search sorting the fringe in order of increasing h. See the following code.

@staticmethod def greedyBestFirstGenerator ( successor, h ):     heuristicSuccessor = lambda n: sorted(successor(n), key=h)     return Problem.breadthFirstGenerator(heuristicSuccessor)

Snippet 2 - Greedy best-first search algorithm

We wrap the passed successor() function into a heuristicSuccessor() function that merely sorts the children returned according to their h values. At every level down the tree, the search algorithm expands states in increasing order of estimated cost to the goal state. The first solution found at a given level is therefore the shortest solution down to that level. I underline "down to that level" because, as you'll see if you examine carefully the Romania road map from the previous post, the first solution popping up on the fringe is not actually the shortest solution (we know from the previous post that the first solution pops up at level 4 while the best solution pops up at level 6).

Therefore, despite using a consistent heuristic function greedy best-first search is complete but not optimal.

Notice that greedy-best first search is not a cost-based search algorithm, thus it must be used with the initial, unmodified search algorithm from the first post:

problem.graphSearch(Problem.greedyBestFirstGenerator(successor, h)) self.assertTrue(problem.solved()) print ("Solution found using greedy-best-first search: ", str(problem.solution)) print("Algorithm performance: ", problem.performance)

Snippet 3 - Testing greedy best-first search

A* search

Let's move on to a heuristic search algorithm that is both complete and optimal, as long as the heuristic function is consistent. It is the "a star" algorithm, which uses as sorting function a combination of what we know and what we believe, i.e.: f(s) = c(s) + h(s), being c(s) the actual cost of reaching state s from the initial state and h(s) the estimated cost of reaching the goal state from state s.

The mathematical proof that using a consistent heuristic function h with A* makes it optimal and complete can be found in the book. It is a beautiful while easy to follow demonstration so I encourage you to check it out.

The generator that enables A* in our common search framework is as follows:

@staticmethod def aStarGenerator ( successor, cost, h ):     def nestedGenerator ( fringe ):         best_node = heapq.heappop(fringe)         descendants = successor(best_node.item()) or []         for desc in descendants:             if desc != best_node.parent():  # going back where we come from                 f = best_node.cost() + cost(desc, best_node.item()) + h(desc)                 heapq.heappush(fringe, Problem.Node((f, desc, best_node)))         yield fringe     return nestedGenerator

Snippet 4 - Generator implementing A* search

Once more we're using the Python heap functions to deal with ordering the states according to their respective f values. If you compare the A* generator with the uniform cost generator from the previous post, you'll see the only difference is in the line that calculates the value of f. I could actually have used the same generator code for both but I think it is not a good idea to mix both algorithms under the same signature.

To test the A* algorithm with our example problem:

problem.treeSearchWithCost(Problem.aStarGenerator(successor, cost, h)) self.assertTrue(problem.solved()) print ("Solution found using A*: " + str(problem.solution)) print("Algorithm performance: ", problem.performance)

Snippet 4 - Generator implementing A* search

Voilá! If you check the output of this test, you'll see this algorithm is the only one that has been able to find the optimum solution for driving from Arad to Bucharest. This confirms the A* algorithm is both complete and optimal. However, this comes at a cost. The algorithm performance, measured as the number of states expanded, is by large the worst one we've seen so far.

There are several variants of the A* algorithm that tackle its weaknesses in order to make better usage of the  computing resources made available to the algorithm execution. I recommend that you check out the book to  round up your knowledge of heuristic search.

And this is it, we're done! I hope you've enjoyed this series of posts as much as I enjoyed writing it. One last thing: if you decide to use the common search framework in your own project, please leave a comment or something so I know my work is being useful to someone else.

Thanks for reading and catch you all around!

One further notch: cost-based search

Introduction

If you've followed closely the previous posts, you might be wondering what the cost formal argument in the generators' signatures is for. We've been consistently passing a constant lambda function as actual argument, and the code doesn't seem to be using it at all. Well, the time has come for providing an explanation: there's a kind of uninformed search algorithms that try to improve search time and/or space by filtering out nodes based on the actual cost of reaching a node. These algorithms are useful when the problem space allows allocating a numeric cost to the action of moving from a parent node to a child node. So far we haven't met such a problem space, hence we've been using a constant function (known as unitary cost function).

A problem space that lends easily to cost-based search is that of finding the shortest path by car between two towns. Here the cost function is the driving distance between two cities connected by a road. If we map nodes to cities and paths to roads, we might use a cost-based search algorithm to try to find the optimum way of driving from city A to city B.

Let's define such a problem space to use it with our search algorithms. We'll use Romanian towns and roads to honor the AI book all these posts are based on. Graphically the map looks as in the following figure (ignore the figure within braces inside each state, we'll see what's that about in the next post):

Figure 1 - Romania road map with driving distance between cities

We can model that map easily using a Python dictionary:

distance = {     ("Oradea","Zerind"): 71,     ("Oradea","Sibiu"): 151,     ("Neamt","Iasi"): 87,     ("Zerind","Arad"): 75,     ("Iasi","Vaslui"): 92,     ("Arad","Sibiu"): 140,     ("Arad","Timisoara"): 118,     ("Sibiu","Fagaras"): 99,     ("Sibiu","Rimnicu Vilcea"): 80,     ("Fagaras","Bucharest"): 211,     ("Vaslui","Urziceni"): 142,     ("Rimnicu Vilcea","Pitesti"): 97,     ("Rimnicu Vilcea","Craiova"): 146,     ("Timisoara","Lugoj"): 111,     ("Lugoj","Mehadia"): 70,     ("Pitesti","Bucharest"): 101,     ("Pitesti","Craiova"): 138,     ("Hirsova","Urziceni"): 98,     ("Hirsova","Eforie"): 86,     ("Urcizeni","Bucharest"): 85,     ("Mehadia","Drobeta"): 75,     ("Bucharest","Giurgiu"): 90,     ("Drobeta","Craiova"): 120, }

Instead of defining classes to model a graph or using a data structures library, we're defining the problem space as a simple dictionary mapping pairs of towns to the driving distance between them.  Pairs not present in the dictionary do not have a road connecting them.

Now let's define a cost function that uses the above problem space knowledge to measure the cost of driving  from town A to town B, as long as A and B are connected by a single road segment (i.e. no intermediate towns need to be traversed) :

cost = lambda s, n: distance.get((s,n), distance.get((n,s))) or (s==n and 0)

We're using the get() method of the dict class instead of direct subscripting, like in 'distance[("A","B")]'. The reason is that subscripting raises an exception if the index is not found in the dictionary, whereas get() gives us the chance of providing an additional argument that shall be used if the index is not found. In addition get() doesn't raise an exception if the search fails, but just returns None. That semantic allows us to condense the dictionary access code in a convenient single lambda function. Notice the driving distance between town A and itself is 0, so we add a final consideration to our cost function instead of adding all those stupid { ("A","A"): 0 } items to the dictionary.

The successor() function for this problem space can also be written as a simple lambda function:

successor = lambda n: [p[1-i] for i in range(1) for p in distance if p[i] == n]

I'll let you try to find out how the successor() function does its job. From the second for loop in the comprehension it can be deduced this is not the best way of obtaining the successors of a node. A tree-like data structure would allow us to be much more efficient, but it would make the code more complex too. This is a trade-off that you'll find over and over when using Python: you'll be able to express complex concepts easy and fast, sacrificing run-time performance. But you'll have your system ready to show very quickly and nothing prevents you from optimizing the heavy parts as soon there's time for it.

Cost-based search: the basics

The main difference between the brute-force approaches we've seen so far and cost-based search is the order in which states are expanded. Cost-based search algorithms pick the next state to expand from the current fringe using a function of the cost of reaching that state from the initial state. An important implication of that behavior is that it is not possible to pick a set of states to expand at one time, as e.g. the breadth-first search algorithm does; the reason being that with each state expanded, new child states cheaper than the ones already in the fringe might be generated, thus altering any order established before the expansion.

One advantage of the recursive algorithm proposed in the first post is that we don't need to perform housekeeping of the data structure holding the current fringe; all the states expanded down to a given recursion level are kept in the successive stacks of the enclosing recursion levels. This is possible because within a recursion level we don't need to access the states generated in an enclosing level. However, that status quo changes with cost-based search. After expanding a number of states at a given recursion level, the candidate to be expanded next might have been generated at any enclosing recursion level, in addition to the current level. This breaks our previous scheme and forces us to make slight modifications to the algorithm; unfortunately, these modifications turn the new algorithm incompatible with the previous one. The good news are that the new algorithm can efficiently behave as the previous one if the appropriate cost functions are used.

Another advantage of the previous algorithm is that it is able to model the search tree very efficiently by means of a dictionary relating children (keys) with parent (value). That has to do as well with the recursive nature of the algorithm and the order in which states are expanded; when returning from en enclosed recursion level because a solution has been found, the enclosing level can safely assume the parent of the state added to the solution by the enclosed level is held at its local stack and can be obtained as desc[sol(0)]. From the paragraph above you can conclude this situation changes as well, since the parent of a state included in the solution by a recursion level might be in the local stack of any enclosing level. Thus we need another way to model the parent-child relationship, while adding the step cost of going from parent to child to the equation.

Let's see how the slightly modified algorithm looks like:

class Problem(object):     ...     class Node(tuple):         '''         A simple tuple wrapper to make the algorithm implementations         easier to understand         '''         cost = lambda n: n[0]         item = lambda n: n[1]         parent = lambda n: n[2]     def getPerformance ( self ):         return self.__performance     performance = property(getPerformance)         ... def treeSearchWithCost ( self, gen ):     self.__performance = 0     fringe = [Problem.Node((0, self.__initial, None))]     visited = set()     h = lambda n: fringe[0] == n and n not in visited     sol = self.__recurseWithCost(fringe, gen, visited, h)     if sol:         self.__solution = [n.item() for n in sol if n]                          def __recurseWithCost ( self, fringe, gen, visited, h ):     for n in filter(h, fringe):         self.__performance += 1         if self.__goal(n.item()):             return [n]                    # the base case                      for desc in gen(fringe):         try:             visited.update(\                 map(Problem.Node.item, map(Problem.Node.parent, desc)))         except TypeError:                # 'NoneType' is not subcsriptable             visited and visited.clear()  # desc is the initial state, IDDF         except AttributeError:           # 'NoneType' has no attribute 'visited'             pass                         # user doesn't want us to check for dups                               sol = self.__recurseWithCost(desc, gen, visited, h) # the recursive case         if sol:             try:                 sol.insert(0, sol[0].parent())             except:                 pass             return sol

Snippet 1 - Modified code for cost-based search

We've added a Node class to our definition of Problem. This is just a façade class for a tuple; we'll be using a tuple to hold the cost and the parent of a state, and using the façade class makes the code much more inteligible than using indexing, like in cost = node[0], parent = node[2] etc. In exchange for this syntactic sugar we lose some performance since a virtual function call is more expensive than indexing a small tuple, but in my oppinion that's a small loss compared to the understanding of the code we gain.

Another addition to the class is the performance property. We're going to use a performance figure to evaluate the relative quality of the generators we explore in the coming sections, and the performance property is the realization of that figure. As you can see in the code, the value of the property is reset each time we call the treeSearchWithCost() method, and is increased with every node selected for expansion. thus, at the end of each algorithm run the performance property holds the number of expanded states. The smaller this value, the more efficient the algorithm is in finding a solution to the problem.

The single remarkable change that remains to explain is within the treeSearchWithCost() method. As you can see, the filter function h is set to fringe[0] == n and n not in visited. If you remember the previous post, this function was used to prevent expanding already expanded states, returning True when the state is not in the list of visited nodes. Here we're enhancing it so in addition to that, it returns only the first element in the fringe. But why's that?

Remember brute-force algorithms just find a solution, they don't care whether it's the best one or if there are others (well, in fact breadth-first search shall find the solution with the smallest number of steps). However a cost-based optimal algorithm cannot go by that simple rule. Imagine a solution pops up in the fringe at absolute cost 200, while there are some other, not yet expanded states in that fringe with absolute costs 100, 150 and 185. Any of those three states has the potential to reach a solution with a cost lower than 200, so we cannot just return the solution that popped up and go to sleep.  The algorithm must explore all the states, in cost order, until it makes sure the first solution in the fringe is the best one. And that happens when the heap's head is a solution, because then there's no other unexpanded state whose absolute cost is lower than that of the head. Thus we'll check only the heap's head for deciding if we've found the solution.

Uniform cost-based search

Let's move on to the search algorithms themselves. We'll start with the simplest form of cost-based search, called uniform cost-based search. The generator that runs uniform cost-based search on our problem space is as follows:

@staticmethod def uniformCostGenerator ( successor, cost ):     def nestedGenerator ( fringe ):         best_node = heapq.heappop(fringe)         descendants = successor(best_node.item()) or []         for desc in descendants:             if desc != best_node.parent():  # going back to where we come from                 f = best_node.cost() + cost(desc, best_node.item())                 heapq.heappush(fringe, Problem.Node((f, desc, best_node)))         yield fringe     return nestedGenerator

Snippet 2 - Uniform cost search generator function

Surprisingly simple, isn't it?. But hidden in the apparent simplicity lies a major change with respect to the previous generators: as opposed to those, this generator modifies the fringe it receives from its caller in-place, and yields that same fringe instead of a newly generated one. This is because after yielding, the next state to expand might indeed be in the fringe received (if e.g. all the states just generated are more expensive than some state in that fringe). We could have created a copy of the fringe received and returned the modified copy instead, but when the fringe is large the performance penalty would be unacceptable.

Other than that, the generator uses a heap to sort the descendants of the state it expands in the current iteration by cost order, with the node having the lowest cost sitting at the beginning of the heap. Actually, the only requirement the algorithm has on the heap is to keep the lowest cost node at the head (remember that treeSearchWithCost() always expands the node at the head of the fringe), so using the Python heap implementation is very convenient due to the space and performance gains attained by partial ordering with respect to a fully sorted heap. For further details about the Python heap container see the Python Library Reference.

The following image shows the patch followed by uniform-cost search down to the fringe leading to Bucharest:

Figure 2 - Path followed by Uniform Cost-based search from Arad down to Bucharest

Like breadth-first search, uniform-cost search will find a solution if there's one, i.e. it is complete. If we consider path costs uniform-cost search will not only find a solution, but the best solution in terms of path cost, i.e. it is optimal. Notice breadth-first search might not be optimal in cost terms, since it always finds the shortest path in terms of number of steps but the cost of taking those steps might actually be higher than that of the steps in another solution with a higher number of steps.

We can test uniform-cost search with our brand new problem definition, as follows (I suggest you put the problem space together with the code that follows in a testNavigation() method)  :

initial = "Arad" goal = "Bucharest" problem = Problem(initial, goal, successor, cost) problem.treeSearch(Problem.uniformCostGenerator(successor, cost)) self.assertTrue(problem.solved()) print ("Solution found using uniform-cost search: " + str(problem.solution)) print ("Algorithm performance: ", problem.performance)

Snippet 3 - Testing uniform cost search

If you check the solution the algorithm found, you'll see that it is not the apparently shortest Arad->Sibiu->Fagaras->Bucharest (4 steps). This is because the cost of that path is indeed higher than that of the Arad->Sibiu->R.Vilcea->Pitesti->Bucharest path (you can check that yourself by carefully looking at each step's cost in our problem state definition above).

Iterative Lengthening search

The same as we derived an iterative algorithm from breadth-first search (see the previous post), a cost-based iterative algorithm can be derived from uniform-cost search. The idea would be to keep the completeness and optimality guarantees of uniform-cost search while avoiding its excessive memory requirements (recall that uniform-cost search, as breadth-first search, keeps all the expanded nodes in local memory). This algorithm is called iterative lengthening search, and is coded as follows:

@staticmethod def iterativeLengtheningGenerator ( successor, cost, anchor ):     var = { "max_cost": 0, "new_max_cost": 0 }     def nestedGenerator ( fringe ):         max_cost = var["max_cost"]         new_max_cost = var["new_max_cost"]         best_node = heapq.heappop(fringe)         descendants = successor(best_node.item()) or []                     for desc in descendants:             if desc != best_node.parent():  # going back to where we come from                 f = best_node.cost() + cost(desc, best_node.item())                 if f <= max_cost:                     heapq.heappush(fringe, Problem.Node((f, desc, best_node)))                 else:                     new_max_cost = min(f, new_max_cost or f)         var["new_max_cost"] = new_max_cost                      if len(fringe) == 0:             var["new_max_cost"] = 0             var["max_cost"] = new_max_cost             heapq.heappush(fringe, Problem.Node((0, anchor, None)))                          yield fringe     return nestedGenerator

Snippet 4 - Iterative Lengthening search generator

If you check the iterative deepening search generator from previous post you'll find some resemblances here. The differences between that one and iterative lengthening search are that where iterative deepening search expands states in the current level in sequential order, iterative lengthening search expands states by cost order, starting with the one with the lowest cost; and instead of using a number of steps as limit, it uses the smallest cost from the previous iteration. The latter allows the algorithm to discard paths further down the tree which cost is higher than that of other state previously expanded. The search shall trace back and forth until it finds the optimum solution in terms of cost.

Unfortunately the addition of path costs to the equation considerably widens the alternative paths the algorithm has to explore, so it suffers from excessive trace back and forth overhead. Moreover, as I mentioned in the previous post this way of implementing an iterative search algorithm is far from optimal, since it keeps all the states expanded in a previous iteration in main memory. Actually, if you test this algorithm you'll probably get a recursion limit reached error from the Python interpreter.

Once again, we can test this algorithm with our problem space as follows:

initial = "Arad" goal = "Bucharest" problem = Problem(initial, goal, successor, cost) problem.treeSearchWithCost(     Problem.iterativeLengtheningGenerator(successor, cost, initial)) self.assertTrue(problem.solved()) print ("Solution found using ILS: " + str(problem.solution)) print ("Algorithm performance: ", problem.performance)

Snippet 5 - Testing iterative lengthening search

Conclusion

In the next -and last- post, we shall look at a different, smarter breed of algorithms that leverage additional knowledge about the problem space to improve its search performance. Nonetheless, the new algorithms fit seamlessly in the common treeSearchWithCost() search framework. I hope you follow me through to that point. While you wait eagerly :), there are two interesting exercises you can do on your own:

1. Modify the treeSearch() algorithm from the previous post so it uses the Node class, instead of a simple dictionary-like structure. Also disable the optionality of duplicate checking, making it mandatory as in treeSearchWithCost(). diff what you get with the treeSearchWithCost() code from this post: there's a remarkable difference between both we haven't talked about, can you reason where it comes from?
2. Enhance the treeSearch() algorithm from the previous post with performance measuring. Run the different brute-force algorithms from the previous post on the Romania road map and check the differences with respect to uniform-cost based search.

Solving well-known algorithmic problems

Introduction

Before looking at informed search, let me show you how to apply the search framework we've created to two well-known algorithmic problems. I hope this part turns out to be refreshing after the hard reading of the previous post.

The missionaries and cannibals problem

In this problem the world is made of a river, a boat, a boat operator, three missionaries and three cannibals. The boat operator can use the boat to cross the river, carrying at most 2 passengers in every trip. When the boat operator leaves a group alone on one side of the river, the number of missionaries and cannibals in that group must be balanced to prevent one group from overwhelming and -probably- killing the other while the boat operator is busy crossing the river.

The initial state of this world is three missionaries and three cannibals sitting on one side of the river. The goal of the problem is to cross everyone to the opposite side of the river, while keeping everyone alive (consider the boat operator is once a hardened marine and has no fear of cannibals, not even of missionaries!).

Before delving into the algorithmic solution to this problem, have a try at solving it using just your grey matter. First tackle it using a casual approach, and if that doesn't cut it use a systematic approach along the lines of problem formulation as described in the previous post.

A possible formulation of the problem that allows us to solve it using our search framework could be as follows:

initial = ((3,3), (0,0)) goal = ((0,0), (3,3)) cost = lambda x : 1 def successor ( s ):     mis = lambda b: b[0]     can = lambda b: b[1]     ouch = lambda a: mis(a) > 0 and can(a) > 0 and mis(a) != can(a)     def trip(a, b, a_to_b):         if mis(a_to_b) > mis(a) or can(a_to_b) > can(a): # check physical limits             raise Exception("Not enough people!")         nxt_a = (mis(a) - mis(a_to_b), can(a) - can(a_to_b))         if ouch(nxt_a):                        # discard fail a->b trips             raise Exception("Holocaust!")         nxt_b = (mis(b) + mis(a_to_b), can(b) + can(a_to_b))         return nxt_a, nxt_b                          result = []     a, b = s[0], s[1]     boat_floats = ((0,0), (1,0), (2,0), (1,1), (0,1), (0,2))     for a_to_b in boat_floats:         try:             tmp_a, tmp_b = trip(a, b, a_to_b)         except Exception as e:             continue         for b_to_a in filter(lambda b_to_a: b_to_a != a_to_b, boat_floats):             try:                 nxt_b, nxt_a = trip(tmp_b, tmp_a, b_to_a)             except Exception as e:                 continue             if not ouch(nxt_a):                  # discard wrong b->a trips                 result.append((nxt_a, nxt_b))     return result

Snippet 1 - Formulation of the missionaries and cannibals problem

From the snippet above, the toughest part is the function obtaining all the successors of a given state. We model each state as a 2-elements tuple, each element representing one side of the river; each river side in turn is modelled as a 2-elements tuple where the first element represents a number of missionaries and the second element represents a number of cannibals. Using this approach, the initial state is given by the expression ((3,3), (0,0)) whereas the goal state is given by the expression ((0,0), (3,3)). To model the boat contents we use a 2-elements tuple with the same meaning as a river side tuple.

Within the successor() function, we define three utility lambdas returning the number of missionaries and cannibals respectively in a side tuple (mis(side) and can(side)) and the validity of a side tuple (ouch(side)). Then we define a function simulating a boat trip between two sides (trip(side_a, side_b, boat_contents)) that returns what would be the new world state after performing that trip.

The main task of the successor() function is performed in the loop that follows. Given a world state, the loop tries all the possible combinations of trips side a -> side b and backwards, recording each valid combination in a list that is returned as output of the function.

Now we can apply the uninformed search algorithm from the previous post to the problem. Let's write it as a test method that we can integrate into our unit test suite:

def testMissionariesAndCannibals(self):         <paste code from snippet 1 right here>         problem = Problem(initial, goal, successor, cost)         problem.treeSearch(Problem.depthFirstGenerator(successor), True)         self.assertTrue(problem.solved())

Snippet 2 - Solving the missionaries and cannibals problem

Observe how we've used the duplicate-checking version of the algorithm. If you've analyzed the problem a little bit as I suggest above, you've seen the state space is actually a graph. Therefore we try to avoid the search to get stuck in a loop within the graph. This problem is simple enough to be solved very quickly using depth-first search, but I suggest you do your own experiments using other search methods, with and without duplicate checking, and see the results.

The 8-queens problem

This well-known game theory problem consists of placing 8 queens on a chess board so no queen attacks any other. The problem can be extended to any number of queens since its solution does not depend on board size. However increasing board size has a big impact on the world's state space thus the resources used by the search, since the number of states is at least the cubic root of n!, being n the board size.

Let's formulate this problem in a way suitable for our search algorithm:

initial = () cost = lambda x : 1 goal = lambda b : b == n def successor ( board ):     def attacks ( q1, q2 ):         return q1[0]==q2[0] or q1[1]==q2[1] or abs(q1[0]-q2[0])==abs(q1[1]-q2[1])                  result = []     for col in range(len(board), n):         for row in range(n):             new_queen = (row, col)             hostiles = [q for q in board if attacks(new_queen, q)]             if len(hostiles) == 0:                 nxt_board = board + (new_queen,)                 result.append(nxt_board)                 return tuple(result)

Snippet 3 - Formulation of the n-queens problem

Examine snippet 3 carefully. Can you spot anything unusual? In previous problems the goal was a definite state of the world, modelled as a constant (either some number or a tuple as in the missionaries&cannibals problem above).  That's because the solutions to those problems were exactly those sought states. In this case however, there are many valid solutions and the context of the problem does not tell us that one is preferred over the other. Moreover, specifying a single solution requires us to solve the problem in the first place! To find out whether a state is a solution to our problem we need to define a function taking the state as input and validating it against the goal constraints. Instead of checking the relative positions of all the queens currently on the board each time we call the goal() function, we chose to generate only valid states in the successor() function and then cheking if we have placed n queens on the board. By generating only valid states we severely decrease the number of states that have to be generated, which helps quite a bit memory-hungry search methods like breadth-first.

But for the goal() function to be used with our search algorithm, we need to make one (and just one) little  change. Goal checking must change from using the == operator to evaluating the goal() function. The following code snippet shows the updated search algorithm using the goal() function:

def treeSearch ( self, gen ):     fringe = [self.__initial]     self.__solution = self.__recurse(fringe, gen) def __recurse ( self, fringe, gen ):     for n in fringe:         if self.__goal(n):             return [n] # the base case     for desc in gen(fringe):         sol = self.__recurse(desc, gen) # the recursive case         if sol:             try:                 sol.insert(0, desc[sol[0]])             except:                 pass             return sol

Snippet 4 - Updated common search algorithm

I'll let as an exercise to you the modification of the unit tests we've coded so far so they work with the updated algorithm.

Now we can solve the problem surprisingly fast using duplicate-checking depth-first search, as follows:

def testQueens ( self, n = 8 ):         <paste code from snippet 3 right here>         problem = Problem(initial, goal, successor, cost)         problem.treeSearch(Problem.depthFirstGenerator(successor), True)         self.assertTrue(problem.solved())               print ("Solution found: " + str(problem.solution[-1]))

Snippet 5 - Solving the 8-queens problem

As before, I urge you to experiment with other search methods and obtain your own conclusions.

In the next post I'll show you how to use cost-based search with this algorithm framework. Till then, spend some time sharpening your python coding skills.