domingo, 15 de noviembre de 2015

Trasteando con el motor de Inteligencia Artificial de Google (TensorFlow)

"Cuando entendamos el cerebro, la humanidad se entenderá a sí misma" (Rafael Yuste)

Hace unos días, Google ha decidido liberar su motor de inteligencia artificial más actual: el conocido como TensorFlow.

Esta librería ha sido, de hecho, la utilizada por los chicos de Mountain View para desarrollar el software más avanzado hasta estos momentos relacionados con la IA: por mencionar unos pocos ejemplos, valga decir que Google ha usado TensorFlow para desarrollar su Google Translator, GoogleFotos, su novedoso Smart Reply, el reconocimiento de voz para Android, el proyecto que le permite poseer el mejor software hasta el momento en cuanto a localización y reconocimiento de objetos dentro de una imagen (Inception), etc., etc.

Pues bien, ni corto ni perezoso, no pude evitar abalanzarme sobre esta maravillosa herramienta, y aprovechar la abundante documentación ofrecida en la web oficial del proyecto para aprender a manejar semejante maravilla tecnológica en el terreno del software.

Y qué mejor modo de hincar el diente a esta herramienta que reescribiendo un ejemplo que realicé desde cero (y sin usar ningún framework) hace unos meses. Me refiero en concreto al ejemplo mediante el cual conseguí entrenar una red neuronal para que fuese capaz de aprender de manera autónoma a sumar dos unidades de enteros. Podéis ver todo lo relacionado con este ejemplo en esta entrada del blog: http://quevidaesta2010.blogspot.com.es/2015/04/aprendizaje-funcional-automatico.html

Así pues, el objetivo era conseguir realizar la misma tarea, pero usando esta vez la API desarrollada por Google, y comprobar de primera mano qué curva de aprendizaje requiere, y como de accesible es la susodicha y famosa herramienta...

Resultado: ¡es una verdadera maravilla!

La versatilidad, el enorme número de utilidades disponibles, el modo en que lo han enfocado todo alrededor de modelos basados en operaciones dentro de nodos en un grafo, la documentación que ofrecen, la potencia de poder utilizar para los cálculos varias GPU además de procesadores, e incluso poder distribuir dichos grafos en un cluster de máquinas funcionando en paralelo (aunque esta funcionalidad distributiva no la han liberado aún). En resumen:  esta herramienta es una verdadera revolución, y una muestra más de que Google siempre intenta hacer las cosas bien.

Sin entrar mucho en detalles, os dejo a continuación el código fuente capaz de hacer la misma tarea programada aquí (es decir, entrenar una red neuronal para que aprenda de manera autónoma a sumar dos operandos), pero mediante los modelos basados en grafos de TensorFlow:

Grafo dirigido utilizado en un ejemplo de TensorFlow
Se trata de una red neuronal con 20 nodos de entrada, una capa intermedia de otros 20 nodos, y un nodo más para el resultado de salida. En total 420 pesos entre nodos (el equivalente a 420 "sinapsis" entre neuronas).

He separado el código en tres ficheros: uno para el entrenamiento de la red y el almacenamiento de la misma, otro para recuperar una red previamente entrenada en algún momento y usarla para probar su eficacia, y un último fichero de utilidades donde agrupo algunas funciones auxiliares.

Para probar este código, lo recomendable es seguir primero las instrucciones de instalación del paquete de TensorFlow que los chicos de Google especifican aquí.

Una vez instalada la herramienta, simplemente debes abrir tu IDE para Python favorito (yo uso Ninja IDE), y copiar el código que os dejo a continuación:

1) entrenamiento_modelo.py:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
import utilidades as util
import tensorflow as tf

# Definimos las constantes del modelo
flags = tf.app.flags
flags.DEFINE_integer("operando_maximo", 9, "Mayor valor que puede tomar un operando de la suma.")
flags.DEFINE_integer("nodos_red", 20, "Numero de nodos de la red neuronal.")
FLAGS = flags.FLAGS

# Inicializamos la sesion
sess = tf.InteractiveSession()

# Definimos el modelo de la red neuronal
x = tf.placeholder("float", shape=[FLAGS.nodos_red, None])
y_ = tf.placeholder("float")
y = tf.placeholder("float")

# Primera capa (layer) de FLAGS.nodos_red nodos que reciben entradas de FLAGS.nodos_red inputs de x, y
# dan salida (output) a FLAGS.nodos_red conexiones h1i
W = tf.Variable(tf.constant(0.2, shape=[1, FLAGS.nodos_red]))
b = tf.Variable(tf.constant(0.1, shape=[FLAGS.nodos_red]))
h1i = tf.nn.l2_normalize(tf.matmul(x, W) + b, 0, epsilon=1e-12, name=None)

# Segunda capa de 1 nodo que recibe como entrada la salida de los FLAGS.nodos_red nodos h1i
# de la primera capa, y conecta su salida al nodo respuesta y de la red neuronal
Wf = tf.Variable(tf.constant(0.2, shape=[FLAGS.nodos_red, 1]))
bf = tf.Variable(tf.constant(0.1, shape=[1]))
hf = tf.matmul(h1i, Wf) + bf

# Nodo de respuesta de la red
y = tf.reduce_sum(hf)

# Inicializamos las variables del modelo (Inicialmente los pesos se inicializan
# con valores constantes de tipo float = 0.2 y 0.1)
sess.run(tf.initialize_all_variables())

# Establecemos el nodo del grafo encargado del entrenamiento de la red neuronal
train_step = tf.train.GradientDescentOptimizer(0.01).minimize(tf.abs(tf.sub(y_, tf.floor(y))))

# Establecemos los nodos del grafo encargados de la verificacion del aprendizaje logrado
correct_prediction = tf.equal(y_, tf.floor(y))
accuracy = tf.reduce_sum(tf.cast(correct_prediction, "float"))
diffe = tf.cast(tf.add(tf.floor(y), 0), "float")

# Realziamos el entrenamiento ejecutando (run) el nodo de entrenamiento train_step repetidas veces
for i in range(5000):
    x_train, y_train = util.random_simple_operacion_size(FLAGS.nodos_red, 0, FLAGS.operando_maximo)
    train_step.run(feed_dict={x: x_train, y_: y_train})
    if i % 500 == 0:
        train_accuracy = accuracy.eval(feed_dict={x: x_train, y_: y_train})
        print "Paso %d(de 5000) del entrenamiento..." % (i)


# Probamos la eficiencia conseguida con el entrenamiento
print "\nEficiencia conseguida: "
k = 0
for j in range(100):
    x_prueba, y_prueba = util.random_simple_operacion_size(FLAGS.nodos_red, 0, FLAGS.operando_maximo)
    if accuracy.eval(feed_dict={x: x_prueba, y_: y_prueba}) == 1:
        k = k + 1
print "Numero de aciertos igual a %i (de 100 intentos)" % (k)

# Guardamos las variables de la red neuronal entrenada para su futuro uso
saver = tf.train.Saver()
saver.save(sess, "suma-model.ckpt")

2) uso_modelo_entrenado.py:


 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
import utilidades as util
import tensorflow as tf

# Definimos las constantes del modelo
flags = tf.app.flags
flags.DEFINE_integer("operando_maximo", 9, "Mayor valor que puede tomar un operando de la suma.")
flags.DEFINE_integer("nodos_red", 20, "Numero de nodos de la red neuronal.")
FLAGS = flags.FLAGS

# Inicializamos la sesion
sess = tf.InteractiveSession()

# Definimos el modelo de la red neuronal
x = tf.placeholder("float", shape=[FLAGS.nodos_red, None])
y_ = tf.placeholder("float")
y = tf.placeholder("float")

# Primera capa (layer) de FLAGS.nodos_red nodos que reciben entradas de FLAGS.nodos_red inputs de x, y
# dan salida (output) a FLAGS.nodos_red conexiones h1i
W = tf.Variable(tf.constant(0.2, shape=[1, FLAGS.nodos_red]))
b = tf.Variable(tf.constant(0.1, shape=[FLAGS.nodos_red]))
h1i = tf.nn.l2_normalize(tf.matmul(x, W) + b, 0, epsilon=1e-12, name=None)

# Segunda capa de 1 nodo que recibe como entrada la salida de los FLAGS.nodos_red nodos h1i
# de la primera capa, y conecta su salida al nodo respuesta y de la red neuronal
Wf = tf.Variable(tf.constant(0.2, shape=[FLAGS.nodos_red, 1]))
bf = tf.Variable(tf.constant(0.1, shape=[1]))
hf = tf.matmul(h1i, Wf) + bf

# Nodo de respuesta de la red
y = tf.reduce_sum(hf)

# Nodo de evaluacion del resultado
diffe = tf.cast(tf.add(tf.floor(y), 0), "float")

# Vamos a recuperar una red neuronal previamente entrenada.
saver = tf.train.Saver()
saver.restore(sess, "suma-model.ckpt")

# Iremos pidiendo los operandos por teclado y observando el resultado
print "\nIntroduce ahora los operandos para probar la red neuronal previamente entrenada:"
while True:
    pedir_operando = "Introduce un entero entre 0 y " + str(FLAGS.operando_maximo) + " (-1 para salir): "
    op1 = input(pedir_operando)
    if op1 == -1:
        break
    op2 = input(pedir_operando)
    if op2 == -1:
        break
    resultado = op1 + op2
    print "\nLa suma real es: %i + %i = %i" % (op1, op2, resultado)
    x_prueba, y_prueba = util.simple_operacion_size(FLAGS.nodos_red, op1, op2)
    print "Las suma prevista por la red neuronal es: %i + %i = %i" % (op1, op2, diffe.eval(feed_dict={x: x_prueba, y_: y_prueba}))

3) utilidades.py:


  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
import numpy as np


def random_operaciones(size):
    x_res = np.zeros((0, 20))
    y_res = np.zeros((0, 1))
    for i in range(size):
        operando1 = np.random.randint(0, 9)
        operando2 = np.random.randint(0, 9)
        aux_x = np.array([])
        for j in range(10):
            if operando1 > j:
                aux_x = np.append(aux_x, np.array([1]))
            else:
                aux_x = np.append(aux_x, np.array([0]))
        for j in range(10):
            if operando2 > j:
                aux_x = np.append(aux_x, np.array([1]))
            else:
                aux_x = np.append(aux_x, np.array([0]))
        resultado = operando1 + operando2
        aux_y = np.array([resultado])
        x_res = np.insert(x_res, 0, np.array(aux_x), axis=0)
        y_res = np.insert(y_res, 0, np.array(aux_y), axis=0)
    return x_res, y_res


def random_simple_operacion():
    x_res = np.zeros((20, 0))
    y_res = np.zeros((0, 1))

    operando1 = np.random.randint(0, 9)
    operando2 = np.random.randint(0, 9)
    aux_x = np.array([])
    for j in range(10):
        if operando1 > j:
            aux_x = np.append(aux_x, np.array([1]))
        else:
            aux_x = np.append(aux_x, np.array([0]))
    for j in range(10):
        if operando2 > j:
            aux_x = np.append(aux_x, np.array([1]))
        else:
            aux_x = np.append(aux_x, np.array([0]))
    resultado = operando1 + operando2
    aux_y = np.array([resultado])
    x_res = np.insert(x_res, 0, np.array(aux_x), axis=1)
    y_res = np.insert(y_res, 0, np.array(aux_y), axis=0)
    return x_res, y_res


def random_simple_operacion_size(size, min, max):
    x_res = np.zeros((size, 0))
    y_res = np.zeros((0, 1))

    operando1 = np.random.randint(min, max)
    operando2 = np.random.randint(min, max)
    aux_x = np.array([])
    for j in range(size / 2):
        if operando1 > j:
            aux_x = np.append(aux_x, np.array([1]))
        else:
            aux_x = np.append(aux_x, np.array([0]))
    for j in range(size / 2):
        if operando2 > j:
            aux_x = np.append(aux_x, np.array([1]))
        else:
            aux_x = np.append(aux_x, np.array([0]))
    resultado = operando1 + operando2
    aux_y = np.array([resultado])
    x_res = np.insert(x_res, 0, np.array(aux_x), axis=1)
    y_res = np.insert(y_res, 0, np.array(aux_y), axis=0)
    return x_res, y_res


def simple_operacion(op1, op2):
    x_res = np.zeros((20, 0))
    y_res = np.zeros((0, 1))

    operando1 = op1
    operando2 = op2
    aux_x = np.array([])
    for j in range(10):
        if operando1 > j:
            aux_x = np.append(aux_x, np.array([1]))
        else:
            aux_x = np.append(aux_x, np.array([0]))
    for j in range(10):
        if operando2 > j:
            aux_x = np.append(aux_x, np.array([1]))
        else:
            aux_x = np.append(aux_x, np.array([0]))
    resultado = operando1 + operando2
    aux_y = np.array([resultado])
    x_res = np.insert(x_res, 0, np.array(aux_x), axis=1)
    y_res = np.insert(y_res, 0, np.array(aux_y), axis=0)
    return x_res, y_res


def simple_operacion_size(size, op1, op2):
    x_res = np.zeros((size, 0))
    y_res = np.zeros((0, 1))

    operando1 = op1
    operando2 = op2
    aux_x = np.array([])
    for j in range(size / 2):
        if operando1 > j:
            aux_x = np.append(aux_x, np.array([1]))
        else:
            aux_x = np.append(aux_x, np.array([0]))
    for j in range(size / 2):
        if operando2 > j:
            aux_x = np.append(aux_x, np.array([1]))
        else:
            aux_x = np.append(aux_x, np.array([0]))
    resultado = operando1 + operando2
    aux_y = np.array([resultado])
    x_res = np.insert(x_res, 0, np.array(aux_x), axis=1)
    y_res = np.insert(y_res, 0, np.array(aux_y), axis=0)
    return x_res, y_res

Hay que utilizar primero entrenamiento_modelo.py hasta que obtengas los resultados de entrenamiento deseado. Posteriormente, ya pueder ejecutar uso_modelo_entrenado.py para comprobar qué responde la red neuronal entrenada a las operaciones que tú le vayas indicando.

Evidentemente, este ejemplo no es nada representativo de las enormes posibilades de TensorFlow (posibiliades que podéis comprobar mejor simplemente observando la capacidad que tiene un simple móvil Android para reconocer nuestra voz y transcribir lo que decimos en palabras), pero sí me ha servidor para hacerme con el funcionamiento de la herramienta. Prometo desarrollar y publicar algo más potente e interesante próximamente :).

Y nada más, por ahora. Sólo comentar que cualquiera que se quiera iniciar en el uso TensorFlow puede contactar conmigo y le ayudaré en todo lo que pueda.

Un saludo a todos.

4 comentarios: