将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（易感者、感染者、康复者）模型。这是用于预测疫情传播的更大类“分区”模型的简单版本。在本示例中，我们将专注于在FoundryML生态系统中使用pandas_function_stage装饰器重新实现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 (以每天为单位)。
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  # 转置结果矩阵，以便分别获取 S, I, R 的时间序列

起始

创建一个新的代码工作簿，跳过介绍对话框，将数据集导入环境。然后，创建一个新的变换并选择 Python 作为语言。我们也假设有一个包含州和相应人口的数据集，字段为 id: string，population: int。注意，如果您没有数据集，也可以定义一个 Pandas DataFrame，您可以使用以下代码进行测试：

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

# 创建一个包含州名（id）和人口（population）的数据框
input_df: pd.DataFrame = pd.DataFrame([
                                ['TX', 29000000],  # 德克萨斯州，人口2900万
                                ['CO', 5800000],   # 科罗拉多州，人口580万
                                ['NY', 8400000]],  # 纽约州，人口840万
                                columns=['id', 'population'])  # 列名分别为'id'和'population'

使用 pandas_function_stage 装饰器

为了将此模型与 FoundryML 集成，我们将使用 pandas_function_stage 装饰器。这个装饰器会准备我们的函数，以便它可以包装在 Foundry ML 阶段中，然后我们可以用它来创建一个 Foundry ML 模型。

为此，我们首先定义一个 NamedTuple，以对我们的输入参数应用一些类型注释。

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

class SIRParameters(NamedTuple):
    # 定义用于存储SIR模型参数的命名元组
    n_days: int = 2  # 模拟的天数
    I0: int = 1      # 初始感染者数量
    R0: int = 0      # 初始恢复者数量
    beta: float = 0.2  # 传染率系数
    gamma: float = 0.1 # 恢复率系数

然后我们将创建一个forecast方法以运行单个状态的预测，并使用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
45
46
from scipy.integrate import odeint
import pandas as pd

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  # 初始状态
    # 使用odeint求解微分方程
    ret = odeint(deriv, y0, t, args=(n_population, beta, gamma))
    S, I, R = ret.T  # 转置结果以获取S, I, R
    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模型，用于流行病学研究中的传染病传播模拟。通过输入初始条件和参数，可以预测在特定天数内的易感者、感染者和康复者的数量变化。 最后，使用 pandas_function_stage 装饰器装饰您的 sir_model 函数。

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

@pandas_function_stage()
def sir_model(data: pd.DataFrame, params: SIRParameters) -> pd.DataFrame:
    # 这里定义了一个 SIR 模型的函数，该函数使用 pandas DataFrame 作为输入数据。
    # SIRParameters 是模型的参数类型。
    ...

在装饰了函数属性后，您现在可以将模型保存为 Foundry 机器学习模型。

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

def model():
    # 创建一个 Model 实例，使用 Stage 包装 sir_model
    return Model(Stage(sir_model))

您可以像其他模型一样使用变换函数执行该模型。

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执行部署：

Copied!
1
2
3
4
5
curl --http2 \
-H "Content-Type: application/json" \
-H "Authorization: <$BEARER_TOKEN>" \  # 使用 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/2 POST 请求。请求头中指定了内容类型为 JSON，并使用 Bearer Token 进行授权。请求体中包含了一组请求数据和请求参数，将其发送到指定的 API 端点，该端点用于机器学习模型的推理转换。