ML パッケージを構築する

データ サイエンティストは、Python か AutoML プラットフォームを使用して、事前トレーニング済みのモデルを構築します。これらのモデルは、RPA 開発者によってワークフロー内で使用されます。

ML パッケージを構造化する

パッケージは一連のわずかな要件に従う必要があります。これらの要件は、モデルのサービングに必要な要素とモデルのトレーニングに必要な要素に分割されます。

サービング コンポーネント

パッケージでは、少なくとも以下のものを提供する必要があります。

• ルートに main.py ファイルがあるフォルダー
• このファイルでは、少なくとも以下の 2 つの関数を実装する Main クラス
  • __init__(self): 引数は取らず、モデルとモデルのローカル データのいずれか、あるいはその両方をロードします (例:単語埋め込み)。
  • predict(self, input): モデル サービング時に呼び出され、文字列を返す関数
• モデルの実行に必要な依存関係を含む requirements.txt という名前のファイル

パッケージのサービング コンポーネントを推論時のモデルとして考えます。サービング時には、提供された requirements.txt ファイルを使用してコンテナー イメージが作成されます。また、モデルへのエンドポイントとして predict 関数が使用されます。

トレーニングと評価コンポーネント

パッケージは推論に使用できるほか、必要に応じてマシン ラーニング モデルのトレーニングにも使用できます。これは、次の手順で行います。

• main.py ファイルと同じルート フォルダーにファイル train.py を用意します。
• このファイルでは、少なくとも以下の 4 つの関数を実装する Main という名前のクラスを提供します。__init__ を除き、以下の関数はすべて任意ですが、対応するパッケージで実行できるパイプラインの種類が制限されます。
  • __init__(self): 引数は取らず、モデルとモデルのデータのいずれか、あるいはその両方をロードします (例:単語埋め込み)。
  • train(self, training_directory): 任意に構築されたデータを含むディレクトリを入力として取ります。モデルのトレーニングに必要なコードをすべて実行します。この関数は、トレーニング パイプラインの実行時に常に呼び出されます。
  • c. evaluate(self, evaluation_directory): 任意に構築されたデータを含むディレクトリを入力として取ります。モードの評価に必要なコードをすべて実行し、その評価に対する単一のスコアを返します。この関数は、評価パイプラインの実行時に常に呼び出されます。
    
  • save(self): 引数を取りません。この関数は、モデルを保持するために train 関数のそれぞれの呼び出しの後に呼び出されます。
  • process_data(self, input_directory): 任意に構築されたデータを含む input_directory 入力を取ります。この関数はフル パイプラインの実行時にのみ呼び出されます。フル パイプラインの実行時に、この関数は任意のデータ変換を行い、データを分割できます。具体的には、環境変数 training_data_directory で指定したパスに保存されたデータが train 関数の入力となり、環境変数 evaluation_data_directory で指定したパスに保存されたデータが上記の evaluation 関数の入力となります。

データ型を処理する

RPA ワークフローで使いやすい AI Center™ にするために、String、File、Files の 3 種類の入力のいずれかを備えることをパッケージに指定できます (パッケージのアップロード時に設定します)。

String データ

これは一連の文字列です。シリアル化できるデータは、すべてパッケージで使用できます。RPA ワークフローで使用する場合、カスタム アクティビティの使用などによってデータを Robot でシリアル化し、文字列として送信できます。パッケージ アップローダーでは、パッケージの入力の種類として json を選択している必要があります。

データの逆シリアル化は、predict 関数の中で実行されます。以下に、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)

File データ

これにより、このモデルを呼び出す ML スキル アクティビティにファイル パスを要求することを通知します。具体的には、このアクティビティがファイル システムからファイルを読み取り、シリアル化したバイト文字列として predict 関数に送信します。したがって、RPA 開発者はワークフロー自体でファイルを読み取ってシリアル化することなく、ファイルにパスを渡すことができます。

ワークフローでは、アクティビティへの入力はファイルへのパスだけです。アクティビティがファイルを読み取ってシリアル化し、ファイル バイトを predict 関数に送信します。データの逆シリアル化は predict 関数でも実行されます。一般的には、以下のようにファイル形式のオブジェクトに直接バイトを読み込みます。

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)

上記のようにシリアル化されたバイトを読み取ることは、読み取りバイナリ フラグをオンにしてファイルを開くことと同じです。モデルをローカルでテストするには、ファイルをバイナリ ファイルとして読み取ります。以下に、イメージ ファイルを読み取ってローカルでテストする例を示します。

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))

以下に、csv ファイルを読み取って、predict 関数の pandas データフレームを使用する例を示します。

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))

Files データ

これにより、このモデルを呼び出す ML スキル アクティビティがファイル パスのリストを要求することを AI Center に通知します。前のケースと同様に、このアクティビティによって各ファイルが読み込まれ、シリアル化されたうえで、バイト文字列のリストが predict 関数に送信されます。

ファイルのリストをスキルに送信できます。ワークフローでは、アクティビティへの入力は、ファイルへのパスをコンマで区切った文字列です。

パッケージのアップロード時にデータ サイエンティストが入力の種類としてファイルのリストを選択します。その場合、データ サイエンティストは、送信されたファイルをそれぞれ逆シリアル化しなければなりません (前述のとおり)。predict 関数への入力は、バイトのリストであり、そのリストの各要素はファイルのバイト文字列です。

任意のデータを保持する

train.py では、実行されたすべてのパイプラインが、パイプライン出力と呼ばれる任意のデータを保持できます。環境変数のアーティファクトからディレクトリ パスに書き込まれるデータはすべて保持されます。パイプラインの詳細ページに移動すれば、このデータをいつでも確認できます。通常、トレーニング/評価ジョブのあらゆる種類のグラフと統計情報は artifacts ディレクトリに保存され、パイプラインの実行後に UI からアクセスできます。

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))
...

TensorFlow を使用する

モデルの開発時には、サービスの提供に使用するのと同じスレッドに TensorFlow グラフを読み込む必要があります。それには、既定のグラフを使用する必要があります。

以下に、必要な変更の例を示します。

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():
      ...

GPU の使用状況に関する情報

スキルの作成時に GPU を有効化した場合、スキルは NVIDIA GPU ドライバー 418、CUDA Toolkit 10.0、CUDA Deep Neural Network Library (cuDNN) 7.6.5 ランタイム ライブラリを有するイメージ上にデプロイされます。

例

トレーニングがないシンプルな Ready-to-Serve ML モデル

この例では、業務上の問題にモデルの再トレーニングは必要ありません。したがって、サービングするシリアル化したモデル IrisClassifier.sav をパッケージに置く必要があります。

1. 初期プロジェクト ツリー (main.py と requirements.txt がないもの)

   IrisClassifier/
     - IrisClassifier.sav

2. ルート フォルダーに追加する main.py の例:

   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. requirements.txt を追加します。

   scikit-learn==0.19.0

   注: pip ライブラリには考慮すべき制約がいくつかあります。以下の制約ファイル下にライブラリをインストールできることを確認してください。

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

   これをテストするには、新しい環境で次のコマンドを使用し、すべてのライブラリが適切にインストールされることを確認します。

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

4. 最終的なフォルダー構造は以下のとおりです。

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

トレーニングが有効化されている、すぐにサービング可能なシンプルなモデル

この例では、業務上の問題はモデルの再トレーニングが必要です。上記のパッケージに構築することで、以下が得られます。

1. 初期プロジェクト ツリー (サービング専用パッケージ):

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

2. ルート フォルダーに追加する train.py の例:

   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. 必要に応じて requirements.txt を編集します。

   pandas==1.0.1
   scikit-learn==0.19.0

4. 最終的なフォルダー (パッケージ) 構造は以下のとおりです。

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

   注: これで、このモデルはまずサービングできるようになりました。この後、Robot または人間参加型プロセスを介して新しいデータ ポイントがシステムに追加されると、train.py を活用してトレーニング パイプラインと評価パイプラインを作成できます。