Criar pacotes de ML

Os cientistas de dados criam modelos pré-treinados usando Python ou usando uma plataforma AutoML. Esses modelos são consumidos pelos desenvolvedores do RPA dentro de um fluxo de trabalho.

Estruturação de Pacotes de ML

Um pacote deve cumprir um pequeno conjunto de requisitos. Esses requisitos são separados em componentes necessários para servir um modelo e componentes necessários para treinar um modelo.

Componente de serviço

Um pacote deve fornecer pelo menos o seguinte:

• Uma pasta contendo um arquivo main.py na raiz dessa pasta.
• Neste arquivo, uma classe chamada Main, que implementa pelo menos duas funções:
  • __init__(self): não recebe nenhum argumento e carrega seu modelo e/ou dados locais para o modelo (por exemplo, incorporações de palavras).
  • predict(self, input): uma função a ser chamada no tempo de serviço do modelo e que retorna uma String.
• Um arquivo chamado requirements.txt com dependências necessárias para executar o modelo.

Pense no componente de serviço de um pacote como o modelo no tempo de inferência. No tempo de serviço, uma imagem do contêiner é criada usando o arquivo requirements.txt fornecido, e a função predict é usada como o endpoint do modelo.

Componente de treinamento e avaliação

Além da inferência, um pacote pode ser usado opcionalmente para treinar um modelo de machine learning. Isso é feito fornecendo o seguinte:

• Na mesma pasta raiz com o arquivo main.py, forneça um arquivo chamado train.py.
• Neste arquivo, forneça uma classe chamada Main, que implementa pelo menos quatro funções. Todas as funções abaixo, exceto __init__, são opcionais, mas limitam o tipo de pipelines que podem ser executados com o pacote correspondente.
  • __init__(self): não recebe nenhum argumento e carrega seu modelo e/ou dados locais para o modelo (por exemplo,incorporações de palavras).
  • train(self, training_directory): recebe como entrada um diretório com dados estruturados arbitrariamente e executa todo o código necessário para treinar um modelo. Essa função é chamada sempre que um pipeline de treinamento é executado.
  • evaluate(self, evaluation_directory): recebe como entrada um diretório com dados estruturados arbitrariamente, executa todo o código necessário para avaliar um modo e retorna uma única pontuação para essa avaliação. Essa função é chamada sempre que um pipeline de avaliação é executado.
  • save(self): não aceita argumentos. Essa função é chamada após cada chamada da função train para persistir seu modelo.
  • process_data(self, input_directory): recebe uma entrada input_directory com dados estruturados arbitrariamente. Esta função só é chamada sempre que um pipeline completo é executado. Na execução de um pipeline completo, essa função pode executar transformações de dados arbitrários e pode dividir dados. Especificamente, quaisquer dados salvos no caminho apontado pela variável de ambiente training_data_directory são a entrada para a função train, e quaisquer dados salvos no caminho apontado pela variável de ambiente evaluation_data_directory são a entrada para a função evaluation acima.

Manuseio de tipos de dados

Para tornar o AI Center™ mais fácil de usar dentro de um fluxo de trabalho RPA, o pacote pode ser indicado para ter um dos três tipos de entrada: String, Arquivo e Arquivos (definidos durante o tempo de upload do pacote).

Dados de string

Essa é uma sequência de caracteres. Quaisquer dados que podem ser serializados podem ser usados com um pacote. Se forem usados dentro de um fluxo de trabalho do RPA, os dados podem ser serializados pelo Robô (por exemplo, usando uma atividade personalizada) e enviados como uma string. O carregador de pacotes deve ter selecionado json como o tipo de entrada do pacote.

A desserialização de dados é feita na função predict. Abaixo estão alguns exemplos para desserializar dados no Python:

Robot sends raw string to ML Skill Activity
# E.g. skill_input='a customer complaint'`
def predict(self, skill_input):
  example = skill_input  # No extra processing
    
# Robot sends json formatted string to ML Skill Activity
# E.g skill_input='{'email': a customer complaint', 'date': 'mm:dd:yy'}'
def predict(self, skill_input):
  import json
  example = json.loads(skill_input)
  
# Robot sends json formatted string with number array to ML Skill Activity
# E.g. skill_input='[10, 15, 20]'
def predict(self, skill_input):
  import json
  import numpy as np
  example = np.array(json.loads(skill_input))
  
# Robot sends json formmatted pandas dataframe
# E.g. skill_input='{"row 1":{"col 1":"a","col 2":"b"},
#                    "row 2":{"col 1":"c","col 2":"d"}}'
def predict(self, skill_input):
  import pandas as pd
  example = pd.read_json(skill_input)

Dados de arquivo

Isso informa à Atividade ML Skill que faz chamadas para esse modelo para esperar um caminho para um arquivo. Especificamente, a atividade lê o arquivo do sistema de arquivos e o envia para a função predict como uma string de bytes serializados. Assim, o desenvolvedor do RPA pode passar um caminho para um arquivo, em vez de ter que ler e serializar o arquivo no próprio fluxo de trabalho.

Dentro do fluxo de trabalho, a entrada para a atividade é apenas o caminho para o arquivo. A atividade lê o arquivo, o serializa e envia os bytes do arquivo para a função predict. A desserialização de dados também é feita na função predict; o caso geral é apenas ler os bytes diretamente em um objeto semelhante ao arquivo conforme abaixo:

ML Package has been uploaded with *file* as input type. The ML Skill Activity
# expects a file path. Any file type can be passed as input and it will be serialized.
def predict(self, skill_input):
  import io
  file_like = io.BytesIO(skill_input)

Ler os bytes serializados conforme acima é equivalente à abertura de um arquivo com o sinalizador binário de leitura ativado. Para testar o modelo localmente, leia um arquivo como um arquivo binário. A seguir encontra-se um exemplo de leitura de um arquivo de imagem e de seu teste localmente:

main.py where model input is an image
class Main(object):
   ...
    
   def predict(self, skill_input): 
      import io
      from PIL import Image
      image = Image.open(io.BytesIO(skill_input))
   ...
  
if__name__ == '_main_':
   # Test the ML Package locally
   with open('./image-to-test-locally.png', 'rb') as input_file:
      file_bytes = input_file.read()
     m = Main()
     print(m.predict(file bytes))

A seguir encontra-se um exemplo de leitura de um arqiuvo csv e do uso de um dataframe pandas na função predict:

main.py where model input is a csv file
class Main(object):
   ...
   def predict(self, skill_input): 
      import pandas as pd
      data frame = pd.read_csv(io.BytesIO(skill_input))
      ...
      
if name == '_main_':
   # Test the ML Package locally
   with open('./csv—to—test—locally.csv', 'rb') as input_file:
      bytes = input_file.read()
   m = Main()
   print(m.predict(bytes))

Dados de arquivos

Isso informa ao AI Center que a Atividade ML Skill que faz chamadas para esse modelo espera uma lista de caminhos de arquivos. Como no caso anterior, a atividade lê e serializa cada arquivo e envia uma lista de strings de bytes para a função predict.

Uma lista de arquivos pode ser enviada para uma habilidade. Dentro do fluxo de trabalho, a entrada para a atividade é uma string com caminhos para os arquivos, separados por uma vírgula.

Ao carregar um pacote, o cientista de dados seleciona a lista de arquivos como tipo de entrada. Em seguida, o cientista de dados deve desserializar cada um dos arquivos enviados (conforme explicado acima). A entrada para a função predict é uma lista de bytes na qual cada elemento na lista é a string de bytes do arquivo.

Dados arbitrários persistentes

Em train.py, qualquer pipeline executado podem persistir dados arbitrários, chamados de saída do pipeline. Quaisquer dados que são gravados no caminho do diretório a partir de artefatos de variáveis de ambiente persistem e podem surgir em qualquer ponto, acessando a página Detalhes do pipeline. Normalmente, quaisquer tipos de gráficos ou estatísticas dos trabalhos de treinamento/avaliação podem ser salvos no diretório artifacts e são acessíveis a partir da UI no final da execução do pipeline.

train.py where some historical plot are saved in ./artifacts directory during Full Pipeline execution
# Full pipeline (using process_data) will automatically split data.csv in 2/3 train.csv (which will be in the directory passed to the train function) and 1/3 test.csv
import pandas as pd
from sklearn.model_selection import train_test_split
class Main(object):
   ...
   def process_data(self, data_directory):
     d = pd.read_csv(os.path.join(data_directory, 'data.csv')) 
     d = self.clean_data(d)
     d_train, d_test = train_test_split(d, test_size=0.33, random_state=42)
     d_train.to_csv(os.path.join(data_directory , 'training', 'train.csv'), index=False)
     d_test.to_csv (os.path.join(data__directory , 'test' , 'test.csv'), index=False)
     self.save_artifacts(d_train, 'train_hist.png', os.environ["artifacts"])
     self.save_artifacts(d_test, 'test_hist.png', os.environ["artifacts"])
  ...
  
   def save_artifacts(self, data, file_name, artifact_directory):
      plot = data.hist() 
      fig = plot[0][0].get_figure()
      fig.savefig(os.path.join(artifact_directory, file_name))
...

Usando o TensorFlow

Durante o desenvolvimento do modelo, o gráfico do TensorFlow deve ser carregado no mesmo thread usado para servir. Para fazê-lo, deve-se usar o gráfico padrão.

Abaixo encontra-se um exemplo com as modificações necessárias:

import tensorflow as tf
class Main(object):
  def __init__(self):
    self.graph = tf.get_default_graph() # Add this line
    ...
    
  def predict(self, skill_input):
    with self.graph.as_default():
      ...

Informações sobre o uso da GPU

Quando a GPU está habilitada no momento de criação de habilidades, ela é implantada em uma imagem com o driver da GPU NVIDIA 418, CUDA Toolkit 10.0 e uma biblioteca de runtime CUDA Deep Neural Network Library (cuDNN) 7.6.5.

Exemplos

Modelo de ML simples pronto para servir sem nenhum treinamento

Neste exemplo, o problema de negócios não requer o retreinamento do modelo e, portanto, o pacote deve conter o modelo serializado IrisClassifier.sav que será servido.

1. Árvore de projeto inicial (sem main.py e requirements.txt):

   IrisClassifier/
     - IrisClassifier.sav

2. Amostra de main.py a ser adicionada à pasta raiz:

   from sklearn.externals import joblib 
   import json
   class Main(object):
      def __init__(self):
         self.model = joblib.load('IrisClassifier.sav')
      def predict(self, X):
         X = json.loads(X)
         result = self.model.predict_proba(X)
         return json.dumps(result.tolist())

3. Adicione requirements.txt:

   scikit-learn==0.19.0

   Observação: existem algumas restrições que precisam ser respeitadas nas bibliotecas pip. Certifique-se de que você pode instalar bibliotecas sob os seguintes arquivos de restrição:

   itsdangerous<2.1.0
   Jinja2<3.0.5
   Werkzeug<2.1.0
   click<8.0.0

   Para testar isso, você pode usar o seguinte comando em um ambiente novo e certificar-se de que todas as bibliotecas estejam instaladas corretamente:

   pip install -r requirements.txt -c constraints.txt

4. Estrutura final da pasta:

   IrisClassifier/
     - IrisClassifier.sav
     - main.py
     - requirements.txt

Modelo simples pronto para implementação com treinamento habilitado

Neste exemplo, o problema de negócios requer que o modelo seja retreinado. Com base no pacote descrito acima, é possível ter o seguinte:

1. Árvore de projeto inicial (pacote apenas de serviço):

   IrisClassifier/
     - IrisClassifier.sav
     - main.py
     - requirements.txt

2. Amostra de train.py a ser adicionada à pasta raiz:

   import pandas as pd 
   import joblib
   class Main(object): 
      def __init__(self):
          self.model_path = './IrisClassifier.sav' 
          self.model = joblib.load(self.model_path)
         
      def train(self, training_directory):
          (X,y) = self.load_data(os.path.join(training_directory, 'train.csv'))
          self.model.fit(X,y)
      def evaluate(self, evaluation_directory):
          (X,y) = self.load_data(os.path.join(evaluation_directory, 'evaluate.csv'))
          return self.model.score(X,y)
      def save(self):
          joblib.dump(self.model, self.model_path)
      def load_data(self, path):
          # The last column in csv file is the target column for prediction.
          df = pd.read_csv(path)
          X = df.iloc[:, :-1].get_values()
          y = df.iloc[:, 'y'].get_values()
          return X,y

3. Edite requirements.txt se necessário:

   pandas==1.0.1
   scikit-learn==0.19.0

4. Estrutura final da pasta (pacote):

   IrisClassifier/
     - IrisClassifier.sav
     - main.py
     - requirements.txt
     - train.py

   Observação: esse modelo agora pode ser primeiro servido e, conforme novos pontos de dados entrem no sistema por meio do robô ou da participação de seres humanos, os pipelines de treinamento e avaliação podem ser criados aproveitando o train.py.