Laboratório 6 - Functional API TensorFlow

Nesse laboratório, vamos aprender como utilizar a API Funcional do TensorFlow para permitir personalizações no treinamento de modelos, com treinamentos, funções e arquiteturas customizadas.

Sequential API

Anteriormente vimos a definição de modelos utilizando a estrutura Sequential, a qual adiciona as camadas sequencialmente para criação do modelo. Vamos utilizar o seguinte modelo como base:

def cria_modelo_com_sequencial_api():
    seq_model = Sequential([Flatten(input_shape=(28, 28)),
                            Dense(128, activation=tf.nn.relu),
                            Dense(10, activation='softmax')
                          ])

    return seq_model

Functional API

Por outro lado, para criar modelos utilizando a Functional API devemos tratar cada camada do modelo como sendo uma função, que recebe como parâmetro a camada anterior. Assim, para criar o mesmo modelo criado anteriormente, utilizamos a seguinte estrutura:

def cria_modelo_com_functional_api():
    input_layer   = Input(shape=(28, 28))

    # empilha as camadas usando: new_layer()(previous_layer)
    flatten_layer = Flatten()(input_layer)
    first_dense   = Dense(128, activation='relu')(flatten_layer)
    output_layer  = Dense(10, activation='softmax')(first_dense)

    # declara inputs e outputs
    func_model = Model(inputs=input_layer, outputs=output_layer)

    return func_model

Com a estrutura do modelo definida, retornamos um objeto do tipo Model que representa o modelo criado. Dessa forma, esse modelo agora pode utilizar os métodos de treinamento padrão utilizados pelo keras.

Note

Ao definir as camadas, podemos definir o fluxo que desejamos seguir, onde uma memsa camada pode ser ligada com outras gerando ramificações e fluxos diferentes de dados.

Visualizando o Modelo

Para verifica que ambos os modelos são iguais, podemos plotar sua estrutura. Para isso, vamos precisar dos pacotes pydot e graphviz que podem ser instalados utilizando o seguinte comando:

!pip install pydot graphviz

Em seguida, com os pacotes instalados podemos utilizar o plot_model() para gerar a visualização do modelo, como mostrado no exemplo a seguir:

import pydot

m1 = cria_modelo_com_sequencial_api()
m2 = cria_modelo_com_functional_api()

tf.keras.utils.plot_model(m1, show_shapes=True, show_layer_names=True)
tf.keras.utils.plot_model(m2, show_shapes=True, show_layer_names=True)

Treinando o Modelo

Vamos fazer um treinamento simples com o modelo definido utilizando o fashion mnist. O dataset Fashion MNIST é um conjunto de dados amplamente utilizado em tarefas de classificação de imagens, desenvolvido como uma alternativa ao popular MNIST, que contém imagens de dígitos manuscritos. Ele consiste em 70.000 imagens em escala de cinza, de 28x28 pixels, organizadas em 10 categorias diferentes de roupas e acessórios, como camisetas, calçados, bolsas, e casacos. O dataset está dividido em 60.000 imagens de treino e 10.000 imagens de teste, e cada imagem está associada a um rótulo correspondente à sua categoria.

Podemos treinar o modelo da forma tradicional, utilizando o seguinte código:

#carrega dataset
mnist = tf.keras.datasets.fashion_mnist
(training_images, training_labels), (test_images, test_labels) = mnist.load_data()

#normaliza dados
training_images = training_images / 255.0
test_images     = test_images / 255.0

#configura modelo
modelo = cria_modelo_com_functional_api()
modelo.compile(optimizer=tf.optimizers.Adam(),
              loss='sparse_categorical_crossentropy',
              metrics=['accuracy'])

#treina modelo
history = modelo.fit(training_images, training_labels, epochs=20)

Podemos utilizar o código do Laboratório 5 para plotar os resultados do modelo

Modelos com Múltiplos Entradas e Saídas

Vamos explorar arquiteturas diferentes agora com a Functional API, para isso vamos utilizar um exemplo para treinar um modelo com duas saídas.

Preparando os Dados

Vamos utilizar o seguinte dataset para realizar o treinamento disponível no UCI. O objetivo principal é criar um modelo capaz de identificar a qualidade do vinho e também o tipo desse vinho (i.e., red, white). Podemos importar os dados utilizado o seguinte código:

data_url_red   = 'https://archive.ics.uci.edu/ml/machine-learning-databases/wine-quality/winequality-red.csv'
data_url_white = 'https://archive.ics.uci.edu/ml/machine-learning-databases/wine-quality/winequality-white.csv'

# Carregar o dataset
data_red             = pd.read_csv(data_url_red, delimiter=';')
data_red['classe']   = 1 
data_white           = pd.read_csv(data_url_white, delimiter=';')
data_white['classe'] = 0
data                 = pd.concat([data_red, data_white])
data

Note

É importante destacar que esse modelo trabalha com duas saídas com diferentes problemas de aprendizado supervisionado. Por exemplo, a qualidade do modelo é um problema regressão, enquanto o tipo é um problema de classificação

Normalizando e Separando os Dados

from sklearn.model_selection import train_test_split
from sklearn.preprocessing import StandardScaler

# Separar as entradas e saídas
y     = data['quality']
y_bin = data['classe']

X = data.drop(columns=['quality', 'classe'])


# Dividir em conjunto de treino e teste
X_train, X_test, y_train, y_test, y_bin_train, y_bin_test = train_test_split(X, y, y_bin, test_size=0.2, random_state=42)

# Normalizar os dados
scaler  = StandardScaler()
X_train = scaler.fit_transform(X_train)
X_test  = scaler.transform(X_test)

Criando o Modelo

Ao instanciar o Model(inputs=[], outputs=[]) devemos passar como parâmetros as entradas e saídas do modelo. Caso desejamos utilizar múltiplas entradas passamos essas entradas via lista Model(inputs=[input_1, input_2]). O mesmo se repete para as saídas

def cria_modelo_multiplas_saidas():
    entrada = Input(shape=(11, ))

    dense_1 = Dense(128, activation='relu')(entrada)
    dense_2 = Dense(64, activation='relu')(dense_1)

    saida_1 = Dense(1, name='qualidade')(dense_2)

    dense_3 = Dense(64, activation='relu')(dense_2)
    saida_2 = Dense(1, activation='sigmoid', name='classe')(dense_3)

    modelo_2_saidas = Model(inputs=entrada, outputs=[saida_1, saida_2])

    return modelo_2_saidas

Visualizando o Modelo

Podemos tanto visualizar o modelo utilizando uma visualização de texto com o modelo.summary() ou com o plot_model(), os quais são apresentados a seguir:

modelo.summary()

tf.keras.utils.plot_model(modelo, show_shapes=True, show_layer_names=True)

Compilando o Modelo

O código configura a compilação de um modelo de aprendizado de máquina em TensorFlow/Keras com múltiplas saídas, onde cada saída tem uma função de loss e métricas diferentes. Ele utiliza o otimizador Adam e define duas saídas: "qualidade" e "classe". Para a saída "qualidade", que provavelmente é uma tarefa de regressão, a função de loss é o erro quadrático médio (MSE), e a métrica de desempenho é o erro absoluto médio (MAE). Já para a saída "classe", que provavelmente envolve uma tarefa de classificação binária, a loss utilizada é a binary_crossentropy (entropia cruzada binária), e a métrica de avaliação é a acurácia. Esse tipo de configuração é útil em modelos que têm múltiplas tarefas a serem aprendidas simultaneamente.

modelo.compile(
    optimizer='adam',
    loss={
        'qualidade': 'mse',
        'classe': 'binary_crossentropy'
    },
    metrics={
        'qualidade': 'mae',
        'classe': 'accuracy'
    }
)

Note

Podemos utilizar losses e métricas diferentes para cada saída!

Treinando o Modelo

Assim, podemos realizar o treinamento de um modelo de aprendizado de máquina utilizando o método fit() do TensorFlow/Keras. Ele utiliza o conjunto de dados de entrada X_train e dois conjuntos de rótulos, y_train para a saída qualidade e y_bin_train para a saída classe, refletindo que o modelo tem múltiplas saídas. O treinamento é executado por 10 épocas com um tamanho de batch de 32, ou seja, o modelo atualizará seus pesos a cada grupo de 32 amostras. Além disso, 10% dos dados de treinamento (validation_split=0.1) são separados automaticamente para validação durante o treinamento, o que permite monitorar o desempenho do modelo em dados que ele não está diretamente treinando a cada época. Este processo auxilia no ajuste do modelo e no monitoramento de métricas de desempenho em tempo real.

history = modelo.fit(
    X_train, {'qualidade': y_train, 'classe': y_bin_train},
    epochs=10,
    batch_size=32,
    validation_split=0.1
) 

Visualizando Resultados

import matplotlib.pyplot as plt
import seaborn as sns

fig, axs = plt.subplots(1, 3, figsize=(10, 3))
axs      = axs.flatten()

for idx, chave in enumerate(history.history.keys()):
    axs[idx%3].plot(history.history[chave], label=chave, marker='o')
    axs[idx%3].set_title(chave)
    axs[idx%3].grid(True, linestyle=':')
    axs[idx%3].legend()

plt.tight_layout()
plt.show()

Função de Loss Personalizada

Podemos utilizar funções de perda pernsonalizadas para mensurar o erro do modelo para cada saída. O código a seguir, mostra um exemplo de uma função que será utilizada como loss no próximo treinamento.

def calcula_loss(model, X, y_quality, y_class):
    mse_loss = tf.keras.losses.MeanSquaredError()
    binary_crossentropy_loss = tf.keras.losses.BinaryCrossentropy()

    predictions    = modelo(X)

    qualidade_loss = mse_loss(y_quality, predictions[0])
    classe_loss    = binary_crossentropy_loss(y_class, predictions[1])
    total_loss     = qualidade_loss + classe_loss

    return total_loss

Note

É importante destacar que a função apresentada tem o mesmo comportamento do modelo anterior, apenas codificamos manualmente. Entretanto, métodos e funções da sua escolha podem ser implementadas.

Definindo Função de Treinamento

para alguns problemas, precisamos ter controle do fluxo e treinamento dos modelos, para salvar estados, tomar decisões, etc. Dessa forma, vamos construir uma função de treinamento manual para o modelo de exemplo, utilizando a função de perda anteriror.

O código define a função treina_modelo, que realiza uma etapa de treinamento manual para um modelo com múltiplas saídas, utilizando TensorFlow e Keras. Dentro de um contexto de tf.GradientTape(), a função calcula a função de perda (loss) chamando a função calcula_loss, que leva em consideração as previsões do modelo e as saídas reais para as variáveis y_qualidade (uma tarefa de regressão) e y_classe (tarefa de classificação). Em seguida, os gradientes da função de perda em relação aos pesos do modelo são calculados e aplicados ao modelo por meio do otimizador fornecido. Após atualizar os pesos, as previsões do modelo são geradas para os dados de entrada X, e as métricas de desempenho (train_mae para a saída de qualidade e train_accuracy para a saída de classe) são atualizadas com base nas previsões e nos valores reais. Finalmente, a função retorna o valor da loss, completando uma iteração de treinamento.

def treina_modelo(modelo, X, y_qualidade, y_classe, optimizer, train_mae, train_accuracy):
    with tf.GradientTape() as tape:
        loss = calcula_loss(modelo, X, y_qualidade, y_classe)

    # Calcular os gradientes
    gradients = tape.gradient(loss, modelo.trainable_variables)

    # Atualizar os pesos
    optimizer.apply_gradients(zip(gradients, modelo.trainable_variables))

    #predict
    predictions = modelo.predict(X, verbose=0)

    train_mae.update_state(y_qualidade, predictions[0])
    train_accuracy.update_state(y_classe, predictions[1])

    return loss

Treinando o Modelo

Dessa forma, podemos treinar o modelo com o seguinte código que treina um modelo com múltiplas saídas por 100 épocas, utilizando o otimizador Adam e monitorando duas métricas: Mean Absolute Error (MAE) para a saída de regressão e Binary Accuracy para a saída de classificação. O treinamento é conduzido dentro de um loop de épocas, onde a função treina_modelo é chamada em cada iteração para atualizar os pesos do modelo com base nos dados de treinamento (X_train, y_train, y_bin_train). Após cada época, a função armazena os valores da loss, MAE e acurácia (accuracy) em listas (losses, accs, maes). As métricas são resetadas a cada nova época e, a cada 100 épocas, os resultados de perda, MAE e acurácia são exibidos no console para monitoramento do progresso do treinamento.

modelo         = cria_modelo_multiplas_saidas()
optimizer      = tf.keras.optimizers.Adam()
train_mae      = tf.keras.metrics.MeanAbsoluteError()
train_accuracy = tf.keras.metrics.BinaryAccuracy()

losses, accs, maes = [], [], []

epochs = 100
for epoch in range(epochs):
    train_mae.reset_state()
    train_accuracy.reset_state()

    loss = treina_modelo(modelo, X_train, y_train, y_bin_train, optimizer, train_mae, train_accuracy)
    losses.append(loss.numpy())
    accs.append(train_accuracy.result().numpy())
    maes.append(train_mae.result().numpy())
    if epoch % 100 == 0:
        print(f"Epoch {epoch}, Loss: {loss.numpy()}, MAE {train_mae.result().numpy()}, Accuracy {train_accuracy.result().numpy()}")

Visualizando Resultados

fig, axs = plt.subplots(1, 3, figsize=(10, 3))
axs      = axs.flatten()

axs[0].plot(losses, label='Loss', color='b')
axs[0].set_title('Loss')

axs[1].plot(accs, label='Acurácia', color='r')
axs[1].set_title('Acurácia')

axs[2].plot(maes, label='MAE', color='k')
axs[2].set_title('Loss')


for i in range(3):
  axs[i].grid(True, linestyle=':')
  axs[i].legend()

Exercício - Redes Siamesas (Entrega)

O objetivo desse laboratório e implementar uma rede siamesa para identificar similaridade entre imagens. Redes siamesas são um tipo de arquitetura de rede neural que consiste em duas (ou mais) sub-redes idênticas, que compartilham os mesmos pesos e parâmetros. Elas são usadas para aprender representações comparáveis entre duas entradas, processando ambas de forma independente e comparando suas saídas. A arquitetura é frequentemente usada em tarefas de comparação ou verificação de similaridade, como reconhecimento facial, verificação de assinaturas, ou para encontrar pares de imagens ou sequências de texto que sejam semanticamente similares. O objetivo é aprender uma função de distância que identifique se dois inputs são semelhantes ou dissimilares, com a ajuda de funções de perda como a Contrastive Loss ou Triplet Loss.

Para isso, vamos utilizar o dataset fashion mnist, onde a ideia é fornecer duas imagens para um modelo base, em seguida comparar a distância euclidiana dos vetores de saída de cada modelo.

As seguintes etapas devem ser concluídas:

• Preparar o daset Fashion MNIST em pares (i.e., com amostras aleatórias), esses pares agora serão a entrada da rede. Assim, devemos criar um novo rótulo para esses pares atribuindo 1 caso eles sejam similares e 0 caso contrário

• Definir o modelo base que será utilizados para as duas imagens. Esse modelo deve ter como saída um vertor.

• Implementar uma camada que calcule a distância euclidia dos dois vetores das imagens de entrada e retorne um valor de saída a distância desses vetores.

Dica

Você deve utilizar a camada Lambda que permite transformar uma função em uma camada para o modelo.

• Seu modelo deve ser treinado utilizado uma Contrastive Loss que é usada para treinar redes neurais em tarefas de aprendizado de representação, como o aprendizado de similaridade entre pares de amostras. Ela funciona penalizando a distância entre pares de exemplos com base em se eles são similares ou dissimilares. A fórmula é definida como: $$L = (1 - y) \cdot D^2 + y \cdot max(0, m - D)^2$$ onde, y é o rótulo (1 para pares similares e 0 para dissimilares). Dé a distância Euclidiana entre as representações aprendidas dos dois exemplos, m é uma margem, que define uma distância mínima que os exemplos dissimilares devem manter.

• Apresente uma visualização para mostrar o comportamento da função de perda ao longo das rodadas. Em seguida, utilize a seguinte visualização para mostrar o resultado do modelo treinado:

Atenção

Agora você deve implementar cada rede em processos (i.e., clientes) diferentes e reportar o vetor de saída via gRPC para um servidor para calcular a distância e atualizar os gradientes da camada Lambda que define a similaridade das camadas.