Palantir Foundry模型评估配置遇TypeError问题求助

模型运行符合预期，model_output能输出包含原列DIFF_BY_DAY和预测列prediction的DataFrame。但按照Palantir的Model Evaluation流程配置时，出现以下错误：

TypeError: Unable to represent an input/output object of type TransformOutput as a parameter. Please verify that your inputs/outputs conform to this model's API definition.

目标是利用Model Objective接口计算模型评估指标，以下是模型仓库中的3个脚本：

model_adapter.py 脚本 #

import palantir_models as pm
from palantir_models_serializers import DillSerializer, JsonSerializer
import pandas as pd
import numpy as np
import datetime
 
class ExampleModelAdapter(pm.ModelAdapter):
 
    @pm.auto_serialize(
        model=DillSerializer(),
        #config=JsonSerializer()
    )
    def __init__(self, model):
        self.model = model
        #self.config = config
 
    @classmethod
    def api(cls):
        columns = [
            ('EVENT_DATE', datetime.datetime),
            ('SAS_DIFF_BY_DAY', np.float64)
        ]
        # TODO: 编辑此方法定义模型API
        inputs = {
            "df_in": pm.Pandas(columns = columns),
            "param_in": pm.Parameter(type=int, default=12)
        }
        outputs = {
            "df_out": pm.Pandas(columns=columns+[("prediction", np.float64)])
        }
        return inputs, outputs
 
    def predict(self, df_in, param_in):
        #forecast_period = 12
        df_in = df_in[['EVENT_DATE', 'DIFF_BY_DAY']]
        df_in['EVENT_DATE'] = pd.to_datetime(df_in['EVENT_DATE'])
        df_in = df_in.set_index('EVENT_DATE')
        df_in = df_in.resample('M').mean()
 
        forecast = self.model.get_forecast(steps = int(param_in))#forecast_period
        forecast_df = forecast.predicted_mean.to_frame(name = 'prediction')
        #print(forecast.conf_int().__dict__)
        forecast_df = pd.DataFrame({
            'DATE': forecast_df.index,
            'prediction': forecast.predicted_mean,
            'lower_bound' : forecast.conf_int()['lower SAS_DIFF_BY_DAY'],
            'upper_bound' : forecast.conf_int()['upper SAS_DIFF_BY_DAY']
        })
 
        forecast_df['DATE'] = pd.to_datetime(forecast_df['DATE'])
        df_in = pd.merge(df_in, forecast_df, left_index=True, right_index=True, how = 'inner')
        #df_in['DIFF_BY_DAY'] = df_in['DIFF_BY_DAY'].astype(np.float32)
        #df_in['prediction'] = df_in['prediction'].astype(np.float32)
        print(df_in.dtypes)
        #return {"df_out": df_in.reset_index(drop = True)}
        return df_in.reset_index(drop = True)

model_training.py 脚本 #

from transforms.api import transform, Input
from palantir_models.transforms import ModelOutput
from main.model_adapters.adapter import ExampleModelAdapter
 
@transform(
    training_data_input=Input("/filepath/.../training_dataset"),
    model_output=ModelOutput("/filepath/.../Model Name"),
)
def compute(training_data_input, model_output):
    training_df = training_data_input.pandas()     # 从TransformsInput加载pandas dataframe
 
    model = train_model(training_df)               # 训练模型
 
    # 用ModelAdapter封装训练好的模型
    foundry_model = ExampleModelAdapter(model)     # 根据你的模型修改ExampleModelAdapter
 
    # 发布并将训练好的模型写入Foundry
    model_output.publish(
        model_adapter=foundry_model
    )
 
 
def train_model(training_df):
    import pandas as pd
    import numpy as np
    import statsmodels.api as sm
    from statsmodels.tsa.statespace.sarimax import SARIMAX
 
 
    training_df = training_df[['DATE', 'DIFF_BY_DAY']]
    training_df['DATE'] = pd.to_datetime(training_df['DATE'])
    training_df = training_df.set_index('DATE')
    training_df = training_df.resample('M').mean()
 
    # 定义SARIMA参数
    p, d, q = 1, 1, 1
    P, D, Q, s = 1, 1, 1, 12 # 假设月度季节性
 
    # 拟合SARIMA模型
    model = sm.tsa.statespace.SARIMAX(training_df['DIFF_BY_DAY'], order = (p, d, q), seasonal_order = (P, D, Q, s))
    results = model.fit()
 
    return results

run_inference.py 脚本 #

from transforms.api import transform, Input, Output
from palantir_models.transforms import ModelInput, ModelOutput
import pandas as pd
 
@transform(
    testing_data_input=Input("/filepath/.../testing_dataset"),
    model_input=ModelInput("/filepath/.../Model Name"),
    predictions_output=ModelOutput("/filepath/.../model_output"),
)
def compute(testing_data_input, model_input, predictions_output):
    inference_outputs = model_input.transform(testing_data_input)
    print(inference_outputs)
    predictions_output.write_pandas(inference_outputs.df_out)

核心问题 #

1. Model API定义与实际输出不匹配：api()方法声明输出是{"df_out": pm.Pandas(...)}，但predict()方法直接返回了DataFrame，而非包含df_out键的字典，导致评估流程校验失败。
2. 列名不一致：api()方法中声明的输入列是SAS_DIFF_BY_DAY，但实际代码中使用的是DIFF_BY_DAY，存在隐性不匹配。

修复步骤 #

1. 修正predict()方法的返回格式

修改model_adapter.py中的predict()方法，使其返回符合API定义的字典结构：

def predict(self, df_in, param_in):
    # 保留原有逻辑...
    print(df_in.dtypes)
    # 改为返回包含df_out键的字典
    return {"df_out": df_in.reset_index(drop = True)}

2. 对齐API定义与实际数据列

调整api()方法中的列名，与代码实际使用的列保持一致：

@classmethod
def api(cls):
    columns = [
        ('EVENT_DATE', datetime.datetime),
        ('DIFF_BY_DAY', np.float64)  # 改为实际使用的列名
    ]
    inputs = {
        "df_in": pm.Pandas(columns = columns),
        "param_in": pm.Parameter(type=int, default=12)
    }
    outputs = {
        "df_out": pm.Pandas(columns=columns+[("prediction", np.float64)])
    }
    return inputs, outputs

3. 验证流程一致性

重新发布模型后，检查训练、推理流程中的数据列是否与API定义完全匹配，避免因列名或格式差异导致的错误。

修复完成后，配置Model Evaluation即可正常读取输入输出，顺利计算评估指标。

内容的提问来源于stack exchange，提问作者Camille