AI: Logic Programming in Scala

I decided to write some weakly related posts about the application of AI algorithms. One of my all time favorites is unification + backward chaining. These two algorithms together form the basis of logic programming or declarative programming in Prolog. In this post I will talk about Prolog and actually implement my own version in Scala.

Prolog, Logic Programming and First Order Logic

Atoms, Variables, Rules and Clauses

When writing procedural code, we write down statements and assignments in order to compute a certain value. In contrast, Prolog is a declarative language. We only write down facts and certain rules and the variable assignments are computed using "logical" inference. Then given a certain goal or query , Prolog tries to very that the goal is true. For example, the following definition defines an ancestor relationship:

// Some facts about family 
parent(sam, peter).
parent(sam, joana).
parent(peter, maria).
parent(maria, john).

In Prolog facts are written down as predicates of the form: name(value1, ..., valueN).
Here the parent(sam, peter) predicate states that sam has a parent called peter. The values
"sam" and "peter" are called atoms. Atoms can be strings or integer numbers. Now given
that definition we can start asking Prolog questions. For example:

? parent(sam, peter).

will return true.

? parent(sam, john).

will return false.

Prolog tries to find a matching predicate in the database and returns true if it does and false otherwise.
However, this is not really programming yet. One thing we are missing is variables.
Variables in Prolog start with an uppercase letter (atoms and predicates do not).
With variables we could ask Prolog to return all parents of same:

? parent(sam, Parent).

Will return:

Parent = peter
Parent = joana
true

Since both are valid options for Parent. Now maybe we want to find all ancestors of sam.
Besides facts we can also write rules in Prolog:

ancestor(Person,Ancestor):-
   parent(Person, Ancestor).

ancestor(Person, Ancestor):-
   parent(Person,Parent),
   ancestor(Parent, Ancestor).

Rules in Prolog equate Horn- Clauses in First Order Logic: y <- x AND y AND z.
In other words, y is true if x and y and z are true. In Prolog y is called the head.
The head predicate is seperated by a  ":-" from the body of the clause.
The clause and the body are other predicates separated by commas. If all
statements in the body evaluate true, the clause is true, too.
One detail that will make implementation easier in a second is that facts such as parent are a clause with an empty body. Anyways, in the example, we define the clause: ancestor(P, A) <- parent(P,A).
In other terms A is an ancestor of P if A is a parent of P. This is simply the search we did earlier when searching for parents. The other ancestor clause defines an ancestor recursively:
ancestor(P, A) <- parent(P,Z), ancestor(Z,A). Meaning A is an ancestor of P if there is a person
Z that is a parent of P and has an ancestor(A). Using the facts and the rules, we can ask Prolog again:

? ancestor(sam, john).

Will evaluate true and:

? ancestor(sam, X).

Will return

 X = peter
 X = joana
 X = maria
 X = john

Data Structures in Prolog.

Data structures such as lists, trees can be implemented in Prolog as recursive predicates.
For example a simply binary tree can be defined as:

node(
    leaf(a), 
    node(
       leaf(b),
       leaf(c)
    ) 
 ).

Lists are also represented as binary trees. Every left node is a leaf with the "head"
of the list and the rest (or "tail") is saved recursively in the right child:

node(
  leaf(1),
  node(
    leaf(2),
    node(
       leaf(3),
       nil
    )
).

defines the list: [1,2,3]. Once I introduced how Prolog actually works, I will give some examples. On how to program in Prolog. For now lets start implementing a very basic Prolog.

Coding the logic model

    
sealed trait Term

case class Atom[T](val x: T) extends Term

case class Variable(name: String) extends Term

case class Predicate(functor: String, terms: List[Term]) extends Term {

  def arity: Int = terms.size

  def matches(other: Predicate): Boolean = other.functor == functor && other.arity == arity

}

case class Clause(head: Predicate, body: List[Predicate])

In this scala implementation predicates, atoms and variables are all terms.
Atoms can be of any type (since we need to compare atoms T should implent a proper equals method). Variables are simply names. Predicates have a name and are an ordered collection
of Terms. These can be atoms (such as in facts), variables (as in rules) and predicates again.
The number of terms in a predicate is also called the arity of the predicate. The name of the predicate (or functor) and the arity define a predicate. If both match, the predicate is the same.
As stated earlier a clause has a head and a body. The head is a predicate that we try to prove true
and the body is a list of predicates that have to be true.

Unification, Depth First Search and Backtracking

Unification

In order to implement a Prolog version the first step is to implement how to match terms.
This process is called unification in Prolog.

  
  def unify(x: Term, y: Term, subs: mutable.Map[Variable, Term]): Boolean = (x, y) match {
    case (Atom(x), Atom(y)) => x == y
    case (x: Variable, y) => unifyVar(x, y, subs)
    case (x, y: Variable) => unifyVar(y, x, subs)
    case (x: Predicate, y: Predicate) if x matches y => {
      x.terms.zip(y.terms).map{ case (t1, t2) => unify(t1, t2, subs)}.reduce(_ && _)
    }
    case _ => false
  }

If we unify two atoms x and y, unification succeeds (returns true) if x and y are equal.
For matching predicates unification succeeds if their functor and arity are the same and if
the recursive unification succeeds.

In order to unify variables, we keep a substitution map that assigns variables to terms.
When unifying a term with an already assigned variable, and that term is not equal
to the assigned value unification fails. Otherwise, unification succeeds and the term
is assigned to the variable. This is implemented in the unifyVar function.

 
 def unifyVar(x: Variable, y: Term, subs: mutable.Map[Variable, Term]): Boolean = {
    if(subs.contains(x)) unify(subs(x), y, subs)
    else if(y.isInstanceOf[Variable] && subs.contains(y.asInstanceOf[Variable])) unify(x, subs(y.asInstanceOf[Variable]), subs)
    else {
      subs += (x -> y)
      true
    }
  }

In order to understand unification better consider the following examples:

unification(Atom("sam"), Atom("sam"), Map) == true, the map is empty
unification(Atom("sam"), Atom("peter")) == false, the map is empty

in other words two atoms match if there value is the same. Furthermore,
no substitutions are given since no variables are involved.

unification(
  Predicate(parent, List(Atom(sam), Atom(peter))),
  Predicate(parent, List(Atom(sam), Atom(peter)))
) == true, the map is empty

The above expression shows that two predicates match if their parts matches, too.
Last but not least the expression:

unification(
  Predicate(parent, List(Variable(X), Atom(peter))),
  Predicate(parent, List(Atom(sam), Variable(Y)))
) == true, the map is (X -> sam, Y -> peter)

So the variables are assigned with the matching atoms. Variables can also be bound to other variables.
In this case their values will be the same as in the following map:

X -> Y
Y -> sam

Searching in a knowledge database

In the section above we just saw how we can match terms such as atoms and predicates and
also how to bind variables. In prolog all statements are given in the form of horn clauses.
A prolog program can be seen as a list of clauses. A query to that database is a goal list of predicates.
All goals have to be found true in the database in order to full fill the query.

 
  protected def inferAll(goals: List[Predicate], solution: Substitution): Solutions =
    if(goals.isEmpty) List(solution)
    else {
      val goal   = goals.head
      val other  = goals.tail
      val answer = for (clause <- program(goal.functor)) yield {
        val Clause(standarizedHead, standarizedBody) = standarizer.standarizeAppart(clause)
        val substitution = mutable.Map.empty[Variable, Term] ++ solution
        val unifyable    = unify(goal, standarizedHead, substitution)
        if(unifyable) {
          val newGoals = substitute(standarizedBody ::: other, substitution.toMap)
          val x = inferAll(newGoals.collect{case x: Predicate => x}, substitution.toMap)
          x
        } else Nil
      }
      answer.flatten
    }

The search algorithm that finds all solutions is a depth first search algorithm with backtracking.
If the goal list is empty the program returns the current substitution. In the implementation substitution is a map from variables to assigned terms.

The search algorithm checks the complete database for a match of a clause head with the first goal.
If found, all the variables in the clause head and body are renamed so that there is no collisions
and the clause head is then unified with the current goal. If the unification succeeds, the body is added to the rest of the goal list (all elements except the first) and then all the substitutions that can be applied will be performed.  The search continues recursively with the new goal list. The algorithm
returns a list of assignments to the variables in the query.

The next listing shows the helper functions

 

def substitute(x: List[Term], subs: Substitution): List[Term] = x.map {
  case term: Variable           => subs.get(term).getOrElse(term)
  case Predicate(functor, body) => Predicate(functor, substitute(body, subs))
  case term                     => term
}

  
class VariableStandarization(var idx: Int = 0) {

  def standarizeAppart(clause: Clause): Clause = {
    next
    clause match {
      case Clause(Predicate(functor, body), terms) ==> Clause(
        Predicate(functor, standarize(body)),
        standarize(terms).map(_.asInstanceOf[Predicate])
      )
    }
  }

  def standarize(terms: List[Term]): List[Term] =
    terms.map {
      case term: Atom[_] => term
      case Predicate(functor, body) => Predicate(functor, standarize(body))
      case x: Variable ==> Variable(x.name + s"_${idx}")
    }

  def next: Unit = idx += 1

}

Standarization basically appends an id to the variable name. The id is increased whenever we take take a clause from the database.

The last thing we need is to output all the assignments for the variables in the query from the result list. Since we renamed them multiple times, we have to find a path from the original variable name
to a term that is not a variable.

 
  def assign(predicate: List[Predicate], substitution: Substitution): Substitution = {
    val variables: List[Variable] = predicate.flatMap {
      case Predicate(_, body) => body.flatMap(getAllVariables)
    }.distinct
    variables.map(
      x => x -> findAssignment(x, substitution)
    ).toMap
  }

  def getAllVariables(term: Term): List[Variable] = term match {
    case x: Atom[_]         => List.empty
    case x: Variable        => List(x)
    case Predicate(_, body) => body.flatMap(getAllVariables)
  }

  def findAssignment(x: Term, substitution: Substitution): Term = x match {
    case x: Variable => findAssignment(substitution(x), substitution)
    case Predicate(functor, body) => Predicate(functor, body.map(x => findAssignment(x, substitution)))
    case x => x
  }

Syntactic sugar and an example

First let us add an easy way to handle lists. Especially to create lists in the binary tree form. First we define a special predicate name for all list elements '.'. The same name is used in prolog.
Then we define an atom indicating the end of a list called nil. Given a scala list we can create a prolog list by recursively building the binary tree. We save the next element in the list as the left child
and the right child is the rest of the list. Furthermore, we can create a split of a list. A 'split' splits the list into its head and the rest of the list. In prolog this is written as '[A|B]'. A is the head and B is the tail. The split can actually be matched directly to a list predicate.

 
object ListBuilder {

  final val ListFunctor = "."

  final val End = Atom("nil")

  def split(head: String, tail: String) = Predicate(ListFunctor, List(Variable(head), Variable(tail)))

  def split(head: Term, tail: Term) = Predicate(ListFunctor, List(head, tail))

  def apply(x: List[Term]):Term = binary(x)

  def binary(x: List[Term]): Term =
    if(x.isEmpty) End
    else Predicate(ListFunctor, List(x.head, binary(x.tail)))

  def flatten(x: Predicate): List[Term] = x match {
    case Predicate(_, List(h, End)) => List(h)
    case Predicate(_, List(h, tail)) => h :: flatten(tail.asInstanceOf[Predicate])
   }

}

Now we can write a small prolog program to find out if an element is in the list.

Predicate("member", List(Variable("X"), Predicate(ListFunctor, List(Variable("X"), Variable("T"))))) 
Predicate("member", List(Variable("X"), Predicate(ListFunctor, List(Variable("H"), Variable("T"))))), List(
      Predicate("member", List(Variable("X"), Variable("T")))
)

So we found a member if we can unify the value of variable 'X' with the head of the list. If we can not we search in the tail recursively.

Now lets turn to our example. It is a quicksort in prolog. Since our version of prolog does not support numeric operations we introduce a predicate for greater(X, Y) and smaller(X,Y) meaning X is greater than Y or smaller or equal. The partition function sorts the values into a sublist with all elements smaller than a pivot and another lists with all elements bigger. Append will merge two lists. Quicksort then sorts the values into sublists around a pivot element and then recombines the resulting lists. A quicksort in prolog.


greater(1, 0).
greater(2, 0).
greater(2, 1).
greater(3, 0).
greater(3, 1).
greater(3, 2).
greater(4, 0).
greater(4, 1).
greater(4, 2).
greater(4, 3).

smaller(0, 0).
smaller(0, 1).
smaller(0, 2).
smaller(0, 3).
smaller(0, 4).
smaller(1, 1).
smaller(1, 2).
smaller(1, 3).
smaller(1, 4).
smaller(2, 2).
smaller(2, 3).
smaller(2, 4).
smaller(3, 3).
smaller(3, 4).
smaller(4, 4).

quicksort([X|Xs],Ys):-
    partition(Xs,X,Left,Right),
    quicksort(Left,Ls),
    quicksort(Right,Rs),
    append(Ls,[X|Rs],Ys).
quicksort([],[]).

partition([X|Xs],Y,[X|Ls],Rs):-
    smaller(X,Y),
    partition(Xs,Y,Ls,Rs).
partition([X|Xs],Y,Ls,[X|Rs]):-
    greater(X,Y),
    partition(Xs,Y,Ls,Rs).
partition([],Y,[],[]).

append([],Ys,Ys).
append([X|Xs],Ys,[X|Zs]):-
    append(Xs,Ys,Zs).


I also put a "full" implementation into github. Enjoy!
So if you like the full code click there [github].
As always I also attached some books to read :)

References

[1] Ivan Bratko: "Prolog Programming for Artificial Intelligence", 4th ed., Chapter 2, Adison Wesley, 2011

[2]  Stuart Russel, Peter Norvig: "Artificial Intelligence a Modern Approach", Pearson, 2011