Artificial Intelligence algorithms and Kubernetes II
How to create an artificial intelligence algorithm and deploy it to Kubernetes
In the previous post of this series, I wrote about planning problems and an algorithm, the A-star. Using some notation and a bit of Set Theory, we can create a representation of the world. Then, an agent can use algorithms to reason about the world and how different actions affect it. In this post, we are going to go a bit further and discuss a possible implementation of the algorithm in Python.
First, I will explain how to represent concepts such as actions and states. Second, we will look at some data structures needed to run the algorithm. And then, I will show you how to deploy it on a local cluster using Kubernetes. Finally, we will run some experiments and discuss how to improve the planning agent.
When coding an algorithm from scratch, we must aim for efficiency. But in this case, since my goal is to show how it works, I want to simplify the algorithm. I also want to make it more readable so the concepts are clear and concise. And then, we can improve from there. There are lots of ways to code an efficient algorithm. But most of them would not be very easy to understand at first glance. Using arrays of bits to describe states and arithmetic operations with them is one of them. We will talk a bit about it, without much detail, leaving it as an exercise for the reader.
Classes and Data Structure
If we look at the previous post, we will notice we rely on the concepts of state and actions a lot. Before continuing, I will clarify some words that I will be using throughout the text. For instance, a predicate represents the smallest element of information that can be true or false. Remember, we only consider predicates that are true, so our states will not only contain the whole lot of predicates of the problem. For example, a predicate can represent if the agent is at the position 1_1, or 2_1. Both of those can be true or false. We will only need to store the information that is true and discard the others. An example of a state can be as simple as:
{agent_at_pos_1_1}After a movement, we discard this predicate and include the new coordinates predicate.
Now, we will begin with the coding of the actions because it is the simplest thing to implement (in our version). An action consists of three different sets: preconditions, positive effects and negative effects. Preconditions are predicates that must be true before applying an action. Positive effects are predicates that become true after applying an action. And negative effects are predicates that are deleted after applying the action:
class Action():
def __init__(self,
name: str,
preconditions: list,
posEffects: list,
negEffects: list,
cost: int=None) -> None:
self.name=name
self.precond=frozenset(preconditions)
self.posEffects=posEffects
self.negEffects=negEffects
self.cost=costNote: I use set as a conceptual and abstract element. But notice that in the implementation I use frozenset. The main difference is that frozenset is immutable resulting in more efficient operations.
Now, for the state itself, we defined it as a configuration of the world, but how can we write that in Python? We will represent a state as a set of predicates. Now, we should think about the operations we can do with the state. On the line 13 of the algorithm (see previous post here), we can see two operations:
# Generate the successor states, by applying actions when possible
for each action:
if action.preconditions in current_state:
successor_state = apply_action(action, current_state)Notice that in each loop, the algorithm tests each action precondition to see if they are in the state set. We can reduce the operations on states to simple ‘add’, ‘remove’ and ‘contains’. Using sets, those operations are simple and easy to code in Python. Testing if we can apply an action is the same as testing if the preconditions are contained in the state set. Creating a new set is also the same, we will add or subtract predicates:
class State():
def __init__(self,
predicates: frozenset=None,
cost: int=None,
heuristic: int=0,
parent_action: str = "",) -> None:
self.heuristic = heuristic
self.parent_action = parent_action
self.predicates = predicates
def conditionHolds(self,
precondition: frozenset)->bool:
return precondition.issubset(self.predicates)
def applyEffects(self,
posEffect: frozenset,
negEffect: frozenset)->None:
self.predicates = self.predicates.difference(negEffect)
self.predicates = self.predicates.union(posEffect)
def getSuccessorState(self,
action: Action):
newPredicates = self.predicates.copy()
newPredicates = newPredicates.difference(action.negEffects)
newPredicates = newPredicates.union(action.posEffects)
return State(newPredicates, parent_action=action.name)
def isApplicable(self, action: Action)->bool:
return self.conditionHolds(action.precond)Notice that we do not add predicates to the state. Instead, the operations return a new copy of the modified state.
Finally, we will talk about the most important data structure, the priority queue. In the algorithm, the queue is used to store the open states and also to order them according to the cost of reaching them. Remember, we add the action cost to the state each time we apply an action. The value (cost) of a state is the sum of the actions to reach it from the initial state. The optimal solution is the lowest-cost solution. So, we need to expand states according to their cost (and also their heuristic cost). Priority queues on Python will allow us to retrieve always the state with a lesser cost:
from libs.classes.state import State
from queue import PriorityQueue
import time
class CostQueue():
def __init__(self) -> None:
self.queue = PriorityQueue()
# We use time as second parameter
def addElement(self, element: State)->None:
self.queue.put((element.cost+element.heuristic, time.time(), element))
def getNextState(self)->State:
return self.queue.get()
def isEmpty(self)->bool:
return self.queue.empty()When we add a state to the queue, notice that we use a tuple with three elements: the cost of the state, the time of insertion and the state.
There are only two lists left: open and closed. We need to test if a state is on the open list, meaning it is waiting to be expanded, or it has been already treated. For these, we can use dicts. The key of the dict is the predicate set of each state. And we should be able to test if a state is open or closed.
Algorithm in Python
Now that we have shown the main elements of the algorithm, let’s show the implementation:
while not self.queue.isEmpty():
# 1 First get lowest cost state from queue
currentStateFCost, _, current_state = self.queue.getNextState()
# 2 if it is a goal state, break
if self.isGoalState(current_state, self.goals):
break
...First of all, we feed the initial state to the queue, and then we start the loop. At each step of the loop, we retrieve the state with the lowest cost. This state is checked to verify if it is a goal state (by testing the goal conditions on the current state).
...
# 3 Generate its succesors
# Each successor has cost = parent_cost + action_cost
for action in self.actions:
# Verify if action is applicable, else continue:
if not current_state.isApplicable(action):
continue
next_state = self.expandState(current_state,
action)In case the state is not a goal, we create a new state by applying each action we can apply. We call this procedure to expand the state. We verify if an action is applicable by checking if the state predicates contain the preconditions.
next_state.cost = current_state.cost + action.cost
if self.openListContains(next_state):
# Verify if we have the same state with a lesser cost
if self.getStateFromOpenList(next_state).cost < next_state.cost:
continue
elif self.closedListContains(next_state):
# Verify if we have to reopen same state
if self.getStateFromClosedList(next_state).cost < next_state.cost:
continue
else:
# Obtain heuristic value for nextState
# for testing we will use 0 heuristic
next_state.heuristic = self.getHeuristicValue(next_state)
self.queue.addElement(next_state)
self.addStatetoOpenList(next_state)We initialize each state with the cost of the parent state (the state from which it comes) plus the action cost. Next, we check if the state: a) is in the open list, awaiting the expanded operation. Or b) is in the closed state, meaning it has already been expanded. Both checks serve to verify if we visited the state before, and we can discard it because we found a better way. In case neither of those cases is true, we continue and get the heuristic value to the goal for the next state. At the end of the iteration, we add the new states to the open list, and to the priority queue for future treatment.
An heuristic is a function that should return us an estimate of the cost to the goal. For now, we can create a function to return 0, as if the goal was one step ahead. This means the algorithm is extra-optimistic every time we expand a new state. But it also means all states are equally good. There are several types of heuristics we can use, and we talk about one of them later.
As an example of a problem for this algorithm, we can use a labyrinth (or maze) problem. In this problem, the agent located in a grid has to find the path from one corner to the opposite corner. We run the simpler version of this problem (4x4 grid), and we get the following results:
Adding predicate: at_agent_1_pos_1_1
Solution found: frozenset({'at_agent_1_pos_4_4'})
**********************
Expanded states: 14
Solution: ['', 'move_who_agent_1_from_pos_1_1_to_pos_2_1',
'move_who_agent_1_from_pos_2_1_to_pos_3_1',
'move_who_agent_1_from_pos_3_1_to_pos_3_2',
'move_who_agent_1_from_pos_2_3_to_pos_3_3',
'move_who_agent_1_from_pos_3_3_to_pos_4_3',
'move_who_agent_1_from_pos_4_3_to_pos_4_4']
Solution size: 7
python -m main 0,03s user 0,01s system 54% cpu 0,077 totalImproving the algorithm
There are several ways to improve the previous algorithm. The most important thing when considering a planner is also the heuristic. We designed the algorithm to consider a value of 0 for all new states. But this will indeed transform our A star algorithm into a Breadth-First Search — BFS. That is, an algorithm that will expand all states, and while it will find the best solution, it will visit all states. A heuristic narrows the search states, sending some states that have a higher cost to the back of the queue. One good heuristic used in planning is the Fast Forward heuristic. This heuristic solves a reduced version of the problem starting from each new state. Yes, to solve a problem, every time the algorithm expands a state, it solves an easier version of it. This is like when we are driving, and we mentally draw a straight line to our destination. We can use that in planning. But we will only consider the positive effects of the actions instead of all of them. In the maze problem, the heuristic will be almost as accurate as the original algorithm. This is due to the low number of relevant predicates, in this case, only the position of the agent.
We coded the heuristic and compared the results with the previous version. Notice the improvement in the number of expanded states. The search is more focused as the heuristic will expand only relevant states:
Problem size Time Time_h Expanded states Expanded states_h
4x4 0.04 0.04 14 14
5x5 0.04 0.04 23 20
6x6 0.04 0.05 33 25We noticed that the heuristic can take a bit more time because of the overhead of computing the value. But in higher instances this may payoff as we expand less states.
As for another improvement we can do, we can move from set logic to binary logic. We can imagine the state to be a large array S of N bits, where N is the total number of predicates in the problem. In this case, every bit n of S will have a value of 1 if it is true in the state, and 0 otherwise. We will move from set logic to binary arithmetics. Add and remove operations will be translated as binary addition or subtraction. A contains operation will be an AND and a comparison operation (in binary, a is subset of b if and only if a & b == a). Using pure bits will speed up the processing and reduce the space used in the computation of a solution. If someone wants to implement this solution, you can post your findings in the comments.
Packaging in Docker
At this step, we have the algorithm coded in Python. Now, we can package it so we can use the image everywhere. To do this part, I recommend to install Docker and Minikube. Docker will allow us to package our code and run it as if it were a virtual machine. Minikube is a testing tool that allows us to use most of the Kubernetes commands. And it runs on docker with its own docker daemon. I recommend using the same daemon as the Minikube engine. With it, you can create the docker images in Minikube and deploy them without using an external repo. Just run the following command, and Minikube will show you the configuration:
minikube docker-env
export DOCKER_TLS_VERIFY="1"
export DOCKER_HOST="tcp://127.0.0.1:63766"
export DOCKER_CERT_PATH="/Users/ignasi/.minikube/certs"
export MINIKUBE_ACTIVE_DOCKERD="minikube"
# To point your shell to minikube's docker-daemon, run:
# eval $(minikube -p minikube docker-env)I also recommend using K9s as UI for the Minikube cluster. Now, we need to create a container that will run the algorithm we created.
FROM python:3.12.2-slim-bookworm
ENV PYTHONDONTWRITEBYTECODE 1
WORKDIR /home/
RUN sh -c 'apt-get update -y && apt-get install curl python3-dev musl-dev -y && apt-get clean -y'
COPY Pipfile* /home/
RUN pip install --upgrade pip
RUN pip install pipenv
RUN pipenv install --system
ARG EXECUTABLE
ENV my_exec=$EXECUTABLE
COPY . /home/
USER root
ENTRYPOINT python -m ${my_exec}The most important steps of this Dockerfile are the following:
- Use a python base image: I used python:3.12.2-slim-bookworm;
- Set the working directory and install all needed libraries;
- Copy just the pipfile files;
- Install all libraries with pipfile;
- Copy all Python files and define an entry point.
Notice we need to copy the files in two steps. If you are not familiar with how docker works, it is enough to know that every line in the Dockerfile is saved as intermediary image. So when updating the image, it will not repeat the lines if the code has not changed. In this case, if we update the python code, when we recreate the image, it will start from step 5 (saving a lot of time).
With the Dockerfile ready, we build the image, and use it on the Yaml declaration files for Kubernetes:
Local tests in Kubernetes
Now the code is packaged, we can create the Kubernetes manifest files. The first question we have to ask is: what do we pretend with this code? Is it a long-running service? Is a job that executes on a given schedule? Or is it a punctual job we can run whenever we want? Since this algorithm is created to run once and return the solution, we can create a Kubernetes job with it.
A Kubernetes job is a simple code that will run once and can be customised to run until X completions. It can also be executed in parallel, but in this case, that is not yet important. We also need a config map to store some environment variables. In our example, we can use it to store the name of the problem we want to run:
apiVersion: v1
kind: ConfigMap
metadata:
name: astar-cm
data:
problem: libs/problems/problem1.yamlEvery manifest in Kubernetes has 4 parts that are common for all elements and objects. First, the apiVersion: name of the version used. If you are in doubt, please refer to the examples in the documentation. Second, the kind that is the name of the element of the manifest. Third, the metadata where we can assign the name and labels for easier identification. And for last, the spec/data. This part is where every different kind of object varies. Notice how in the configmap above, the variables mapping is under the data itself. In this case, there is a variable, called problem, containing the path for the problem. And for the job:
apiVersion: batch/v1
kind: Job
metadata:
name: astar-search-job
spec:
ttlSecondsAfterFinished: 100
template:
spec:
containers:
- name: astar-search
image: astar-search:latest
imagePullPolicy: IfNotPresent
envFrom:
- configMapRef:
name: astar-cm
restartPolicy: NeverNotice that in this case, we do not have a data section, and instead we have a spec section. On a job specification, we define the container using the name of the image created on docker before. Notice how the job pulls the environment variables from the config map defined above (astar-cm).
Now, to run it on Kubernetes, first of all, we have to create the config map using the command kubectl:
kubectl apply -f <PATH_TO_CONFIGMAP>
We have created the configmap, so now we are going to create the job using the same kubectl apply command:
Job appears with the status Completed, meaning it has finished its execution. Once a job runs and finishes successfully, the job is marked as completed and ends. We can examine its logs to verify if the algorithm found the solution:
Logs show how the job has reached the solution, expanding 14 states to find a solution that is a sequence of 7 actions.
Conclusions
In this post, we showed how to code the A-star algorithm using Python. We only showed the main classes and structures needed to run. The full code I used is in Github here (be warned, it is a different version, I am testing with other features). Use it as a help, but do not rely on it a lot. I’m sure the reader will be able to put the pieces together.
Please let me know how it works in the comments, and share your results here. I left some other improvements for the reader to test, such as using binary arrays. However, we can improve it by using parallelism and trying to solve more complex problems.
In the next post we will be improving this algorithm using parallelism in Kubernetes.
Stay tuned for the next article of this series!
