Python 함수를 모델로 저장하기

FoundryML은 단일 pandas DataFrame에서 작동하는 Python 함수를 통합하기 위한 기본 래퍼를 제공합니다. 이를 통해 임의의 처리와 빠른 프로토타이핑을 가능하게 하며, 일부 예측 모델, 시뮬레이션, 최적화 등과 같이 전체 데이터셋에 대해 한 번에 작동해야 하는 비행단위 모델의 구현을 가능하게 합니다.

아래의 기준을 충족하는 Python 함수는 foundry_ml.function_stages.pandas_function_stage 데코레이터로 주석 처리될 수 있습니다. 이는 단계를 생성하며, 필요한 경우 다른 단계와 결합하여 Python 모델을 형성할 수 있습니다. 이러한 단계를 포함하는 모델은 배치 및 실시간 배포와 완전히 호환되며, pandas 또는 pyspark DataFrames에서 호출될 수 있습니다(그러나 후자의 경우, 처리는 여전히 로컬에서 pandas에서 발생합니다).

함수에 데코레이터를 추가하기 위한 하이레벨 요구 사항은 다음과 같습니다:

1. 데코레이터가 추가된 함수의 서명은 일치해야 합니다.

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

2. data는 필수 kwarg(키워드 인수)이며 pandas DataFrame이어야 합니다.
3. params는 필수 kwarg이며 NamedTuple이어야 합니다.

예시: SIR 모델 구현하기

이 튜토리얼은 Foundry에서 SIR(감염 가능, 감염, 회복) 모델을 구현하는 방법을 보여줍니다. 이는 팬데믹의 확산을 예측하는 데 사용되는 "구획화"된 모델의 더 큰 클래스의 간단한 버전입니다. 이 예시에서는 우리는 pandas_function_stage 데코레이터를 사용하여 FoundryML 생태계에서 SciPy 문서에 제공된 단순 SIR 모델을 다시 구현하는 데 중점을 둘 것입니다.

다음 코드는 모델의 핵심 기능을 구현합니다:

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 = 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을 만들고 데이터셋을 환경으로 가져오기 위해 소개 대화 상자를 건너뛰십시오. 그런 다음 새로운 변환을 만들고 언어로 파이썬을 선택합니다. 또한, 주와 해당 인구의 데이터셋이 있으며, 필드로 id: string, population: int를 사용한다고 가정합니다. 주의: 데이터셋이 없는 경우 아래 코드를 사용하여 테스트용 Pandas DataFrame을 정의할 수도 있습니다:

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

# 입력 데이터를 판다스 DataFrame으로 변환합니다.
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

# 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
44
45
46
47
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']
        # 각각의 상태에 대한 예측값을 DataFrame 형태로
        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
    # 감염, 회복률에 따른 새로운 S, I, R 값을 구함
    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),  # 날짜 범위
    })

# 감염, 회복률에 따른 새로운 S, I, R 값을 구하는 함수
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
9
10
from foundry_ml.function_stages import pandas_function_stage

# pandas_function_stage 데코레이터를 사용하여 함수를 정의합니다.
@pandas_function_stage()
def sir_model(data: pd.DataFrame, params: SIRParameters) -> pd.DataFrame:
    # 이 함수는 SIR 모델을 실행하는 함수입니다.
    # data 파라미터는 pd.DataFrame 형태의 데이터를 받아옵니다.
    # params 파라미터는 SIRParameters 형태의 파라미터를 받아옵니다.
    # 이 함수는 pd.DataFrame 형태의 결과를 반환합니다.
    ...

함수 속성이 장식된 상태에서 이제 Foundry ML 모델로 모델을 저장할 수 있습니다.

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

# 모델 함수 정의
def model():
    # SIR 모델을 포함하는 단계를 가진 Model 객체를 반환합니다.
    return Model(Stage(sir_model))

모델은 다른 모델처럼 transform 함수를 사용하여 실행할 수 있습니다.

Copied!
1
2
3
4
5
def execute_model(model, states_df: pd.DataFrame):
    # 모델에 states_df를 입력하여 결과를 얻습니다. (오버라이드 없음)
    result_no_overrides = model.transform(states_df)
    # 모델에 states_df와 n_days 매개변수를 입력하여 결과를 얻습니다. (오버라이드 있음)
    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를 사용하여 POST 요청을 보내는 curl 명령입니다. 명령의 각 부분은 다음과 같습니다:

• --http2 : HTTP/2 프로토콜을 사용하도록 curl에 지시합니다.
• -H "Content-Type: application/json" : 요청 헤더로 'Content-Type'을 'application/json'으로 설정합니다. 이는 보내는 데이터가 JSON 형식임을 알립니다.
• -H "Authorization: <$BEARER_TOKEN>" : 'Authorization' 헤더를 설정합니다. <$BEARER_TOKEN>는 실제 베어러 토큰으로 대체되어야 합니다.
• -d '{"requestData":[{"id":"TX","population":29000000}, {"id":"CO","population":5800000}], "requestParams":{"params":{"num_days":10}}}' : POST 요청과 함께 전송되는 데이터입니다. 이 데이터는 JSON 형식이며, 두 개의 객체를 포함하는 'requestData' 배열과 하나의 객체를 포함하는 'requestParams' 객체를 포함하고 있습니다.
• --request POST $STACK_URL/foundry-ml-live/api/inference/transform/$DEPLOYMENT_RID : POST 메소드를 사용하여 특정 URL로 요청을 보냅니다. $STACK_URL은 요청을 보낼 서버의 URL을 나타내고, $DEPLOYMENT_RID는 배포 리소스의 식별자입니다. 이 두 변수는 실제 값으로 대체되어야 합니다.