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

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

以下の条件を満たす Python 関数は、foundry_ml.function_stages.pandas_function_stage デコレータで注釈を付けることができます。これにより、stageが生成され、必要に応じて他の stage と組み合わせて Python の modelを形成することができます。これらの stage を含むモデルは、Batch と Live のデプロイメントと完全に互換性があり、pandas または pyspark の DataFrames で呼び出すことができます（ただし、後者の場合でも、処理はローカルの 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 のドキュメントに提供されている simple SIR model を pandas_function_stage デコレータを使用して FoundryML エコシステムで再実装することに焦点を当てます。

次のコードは、モデルのコア機能を実装します：

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

# 全人口、N。
N = 1000
# 初期の感染者と回復者の数、I0とR0。
I0, R0 = 1, 0
# それ以外の全員、S0は最初は感染に対して感受性があります。
S0 = N - I0 - R0
# 接触率、ベータ、平均回復率、ガンマ（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 を作成し、データセットを環境にインポートするための導入ダイアログをスキップしてください。次に、新しい Transform を作成し、言語として Python を選択します。また、id: string、population: int というフィールドを持つ、州と対応する人口のデータセットがあることを前提としています。データセットがない場合は、次のコードで Pandas DataFrame を定義し、テストに使用できます。

Copied!
1
2
3
4
5
6
7
8
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モデルのパラメータを定義するクラス
class SIRParameters(NamedTuple):
    n_days: int = 2  # シミュレーションを行う日数
    I0: int = 1  # 初期感染者数
    R0: int = 0  # 初期回復者数
    beta: float = 0.2  # 感染率（一人が一日に他の人に感染させる確率）
    gamma: float = 0.1  # 回復率（一人が一日で回復する確率）

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

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
from scipy.integrate import odeint
import pandas as pd

# SIRモデルを計算する関数
def sir_model(data: pd.DataFrame, params: SIRParameters) -> pd.DataFrame:
    rows = data.to_dict('rows')
    results = []
    for state in rows:
        id = state['id']
        population = state['population']
        state_df = forecast(population, params)
        state_df['state'] = 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_days, beta, gamma))
    S, I, R = ret.T
    return pd.DataFrame({
        'susceptible': S,
        'infected': I,
        'recovered': R,
        'n_days': range(n_days),
    })

# 微分方程式を表現する関数
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
from foundry_ml.function_stages import pandas_function_stage

# SIRモデルを実装する関数
@pandas_function_stage()
def sir_model(data: pd.DataFrame, params: SIRParameters) -> pd.DataFrame:
    # ここに実装を記述...
    ...

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

Copied!
1
2
3
4
5
6
7
# foundry_mlというライブラリからModelとStageをインポートします
from foundry_ml import Model, Stage

# モデル関数を定義します
def model():
    # Modelクラスのインスタンスを生成し、その中にsir_modelをステージとして設定します
    return Model(Stage(sir_model))

モデルは他のモデルと同じように、transform 関数で実行できます。

Copied!
1
2
3
4
5
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

# このコマンドは、HTTP/2で指定されたURLにJSONデータをPOSTリクエストで送信します。
# ヘッダーには、コンテントタイプ（Content-Type）と認証トークン（Authorization）が含まれています。
# 送信するデータには、2つの州の情報（IDと人口）と、リクエストパラメータ（日数）が含まれています。