Python 関数をモデルとして保存する

FoundryML は、単一の pandas DataFrame 上で動作する Python 関数を組み込むためのネイティブラッパーを提供します。これにより、任意の処理や迅速なプロトタイピングが可能となり、予測モデル、シミュレーション、最適化など、全体のデータセットに対して一度に操作する必要がある非行単位のモデルの実装が可能になります。

以下の条件を満たす Python 関数には foundry_ml.function_stages.pandas_function_stage デコレータを注釈として付けることができます。これにより ステージ が生成され、必要に応じて他のステージと組み合わせて Python モデル を形成することができます。これらのステージを含むモデルは、バッチおよびライブデプロイメントと完全に互換性があり、pandas または pyspark DataFrame 上で呼び出すことができます（ただし後者の場合、処理はローカルで pandas で実行されます）。

デコレートされる関数の主な要件は以下の通りです:

1. デコレートされた関数のシグネチャは以下のようにする必要があります。

   (data: pandas.DataFrame, params: NamedTuple) -> pandas.DataFrame
   

2. data は必須の kwarg（キーワード引数）であり、pandas DataFrame でなければなりません。
3. params は必須の kwarg であり、NamedTuple でなければなりません。

例: SIR モデルの実装

このチュートリアルでは、Foundry で SIR（Susceptible, Infected, Recovered）モデルを実装する方法を示します。これはパンデミックの拡散を予測するために使用される「区画化」モデルの一部であるシンプルなバージョンです。この例では、SciPy ドキュメントに記載されている シンプルな SIR モデル ↗ を FoundryML エコシステム内で pandas_function_stage デコレータを使用して再実装することに焦点を当てます。

以下のコードは、モデルのコア機能を実装しています:

Copied!
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
import numpy as np
from scipy.integrate import odeint

# 総人口, N.
N = 1000
# 初期の感染者数と回復者数, I0 と R0.
I0, R0 = 1, 0
# 残りの人々, S0, は初めは全員が感染に対して感受性がある。
S0 = N - I0 - R0
# 接触率, beta, と平均回復率, gamma, (単位: 日^-1).
beta, gamma = 0.2, 1./10
# 時間のグリッド (単位: 日)
t = np.linspace(0, 160, 160)

# SIRモデルの微分方程式.
def deriv(y, t, N, beta, gamma):
    S, I, R = y
    dSdt = -beta * S * I / N # 感受性がある人々が減少する速度
    dIdt = beta * S * I / N - gamma * I # 感染者が増減する速度
    dRdt = gamma * I # 回復者が増加する速度
    return dSdt, dIdt, dRdt

# 初期条件ベクトル
y0 = S0, I0, R0
# 時間グリッドtにわたってSIR方程式を積分する.
ret = odeint(deriv, y0, t, args=(N, beta, gamma))
S, I, R = ret.T

はじめに

新しいCode Workbookを作成し、導入ダイアログをスキップしてデータセットを環境にインポートします。次に、新しいトランスフォームを作成し、言語として Python を選択します。また、州と対応する人口のデータセットがあり、フィールドとして id: string、population: int が含まれていることを前提とします。データセットを持っていない場合は、以下のコードを使用してテスト用のPandas DataFrameを定義することもできます:

Copied!
1
2
3
4
5
6
7
import pandas as pd

input_df: pd.DataFrame = pd.DataFrame([
                                ['TX', 29000000],  # テキサス州の人口
                                ['CO', 5800000],   # コロラド州の人口
                                ['NY', 8400000]],  # ニューヨーク州の人口
                                columns=['id', 'population'])  # 'id'列と'population'列を指定

pandas_function_stageデコレータを使用する

このモデルを FoundryML と統合するために、pandas_function_stageデコレータを使用します。このデコレータは、関数を Foundry ML Stage にラップできるように準備し、その後 Foundry ML Model を作成するために使用できます。

これを行うために、最初にNamedTupleを定義し、入力パラメーターに型を適用します。

Copied!
1
2
3
4
5
6
7
8
9
from typing import NamedTuple

# SIRモデルのパラメータを格納するためのNamedTupleクラス
class SIRParameters(NamedTuple):
    n_days: int = 2  # シミュレーションする日数
    I0: int = 1      # 初期感染者数
    R0: int = 0      # 初期回復者数
    beta: float = 0.2  # 感染率
    gamma: float = 0.1  # 回復率

次に、単一の状態に対して予測を実行するための forecast メソッドを作成し、この関数を各状態の DataFrame の行に適用するために sir_model を使用します。

Copied!
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
from scipy.integrate import odeint
import pandas as pd
import numpy as np  # npを利用するためにインポート

# SIRモデルを実行する関数
def sir_model(data: pd.DataFrame, params: SIRParameters) -> pd.DataFrame:
    rows = data.to_dict('rows')  # データフレームを辞書形式に変換
    results = []
    for state in rows:
        id = state['id']  # 各状態のIDを取得
        population = state['population']  # 各状態の人口を取得
        state_df = forecast(population, params)  # 予測を実行
        state_df['state'] = id  # 予測結果に状態IDを追加
        results.append(state_df)  # 結果をリストに追加
    return pd.concat(results, ignore_index=True)  # 結果をデータフレームとして結合して返す

# 予測を行う関数
def forecast(n_population, parameters):
    n_days = parameters.n_days  # 予測日数を取得
    I0 = parameters.I0  # 初期感染者数
    R0 = parameters.R0  # 初期回復者数
    S0 = n_population - I0 - R0  # 初期感受性者数

    beta = parameters.beta  # 感染率
    gamma = parameters.gamma  # 回復率
    t = np.linspace(0, n_days, n_days)  # 時間の配列を生成

    y0 = S0, I0, R0  # 初期状態
    ret = odeint(deriv, y0, t, args=(n_population, beta, gamma))  # 微分方程式を解く
    S, I, R = ret.T  # 結果を転置してそれぞれの状態を取得
    return pd.DataFrame({
        'susceptible': S,
        'infected': I,
        'recovered': R,
        'n_days': range(n_days),
    })  # 予測結果をデータフレームとして返す

# SIRモデルの微分方程式
def deriv(y, t, N, beta, gamma):
    S, I, R = y
    dSdt = -beta * S * I / N  # 感受性者の変化率
    dIdt = beta * S * I / N - gamma * I  # 感染者の変化率
    dRdt = gamma * I  # 回復者の変化率
    return dSdt, dIdt, dRdt  # 各状態の変化率を返す

最後に、sir_model 関数に pandas_function_stage デコレーターを付けてください。

Copied!
1
2
3
4
5
6
7
8
from foundry_ml.function_stages import pandas_function_stage

@pandas_function_stage()
def sir_model(data: pd.DataFrame, params: SIRParameters) -> pd.DataFrame:
    # SIRモデルのデータ処理を行う関数
    # data: SIRモデルの入力データを含むDataFrame
    # params: SIRモデルのパラメータを含むオブジェクト
    ...

関数プロパティをデコレートしたので、Foundry ML モデルとしてモデルを保存できます。

Copied!
1
2
3
4
5
6
7
from foundry_ml import Model, Stage

def model():
    return Model(Stage(sir_model))

# このコードは、foundry_mlライブラリからModelおよびStageをインポートし、
# sir_modelを使用してModelオブジェクトを生成する関数を定義しています。

transform 関数を使用して、他の Model と同じようにモデルを実行できます。

Copied!
1
2
3
4
5
6
def execute_model(model, states_df: pd.DataFrame):
    # モデルを実行してオーバーライドなしの結果を取得
    result_no_overrides = model.transform(states_df)
    
    # モデルを実行してオーバーライドありの結果を取得（n_daysを10に設定）
    result_with_override = model.transform(states_df, params={'n_days': 10})

ライブデプロイメントでモデルを実行する

モデルをライブデプロイメントを通じて提供している場合、ライブデプロイメント推論 API を使用してデプロイメントを実行できます:

curl --http2 -H "Content-Type: application/json" -H "Authorization: <$BEARER_TOKEN>" -d '{"requestData":[{"id":"TX","population":29000000}, {"id":"CO","population":5800000}], "requestParams":{"params":{"num_days":10}}}' --request POST $STACK_URL/foundry-ml-live/api/inference/transform/$DEPLOYMENT_RID

# このコマンドは、curlを使用してHTTP POSTリクエストを送信します。
# --http2: HTTP/2プロトコルを使用します。
# -H "Content-Type: application/json": リクエストヘッダーにコンテンツタイプとしてJSONを指定します。
# -H "Authorization: <$BEARER_TOKEN>": リクエストヘッダーに認証トークンを含めます。
# -d '{"requestData":[{"id":"TX","population":29000000}, {"id":"CO","population":5800000}], "requestParams":{"params":{"num_days":10}}}': リクエストのデータ部分にJSON形式のデータを含めます。
# --request POST: HTTP POSTリクエストを指定します。
# $STACK_URL/foundry-ml-live/api/inference/transform/$DEPLOYMENT_RID: リクエストを送信するエンドポイントURLです。