如何持久化FunctionTransformer并避免函数定义变更影响其行为？

要解决pickle仅保存函数引用导致的问题，核心是让序列化过程保存函数的完整定义而非引用。以下是几种可行方案：

方法1：用cloudpickle替代标准pickle #

cloudpickle专门针对Python对象（包括自定义函数）做了序列化优化，能直接保存函数的完整实现，不受后续原函数修改的影响。

代码示例：

import cloudpickle
import pandas as pd
from sklearn.compose import ColumnTransformer, make_column_selector
from sklearn.preprocessing import FunctionTransformer

# 定义初始预处理函数
def preprocess_A(x):
    return x ** 2

# 构建管道
pipeline = ColumnTransformer([
    ('test', FunctionTransformer(preprocess_A), make_column_selector())
])

# 拟合并保存
df = pd.DataFrame({'col1': [1, 2, 4], 'col2': [3, 6, 9]})
pipeline.fit(df)

with open('test_cloudpickle.pkl', 'wb') as f:
    cloudpickle.dump(pipeline, f)

# 修改原函数用于新实验
def preprocess_A(x):
    return x ** 4

# 加载管道验证行为
with open('test_cloudpickle.pkl', 'rb') as f:
    loaded_pipeline = cloudpickle.load(f)

print(loaded_pipeline.transform(df))
# 输出：[[ 1  9]
#        [ 4 36]
#        [16 81]] （依然保留旧的平方逻辑）

方法2：将预处理逻辑封装为类 #

把函数逻辑封装成实现__call__方法的类，让FunctionTransformer调用类的实例。pickle会保存实例的状态，后续修改类定义不会影响已序列化的实例。

代码示例：

import pickle
import pandas as pd
from sklearn.compose import ColumnTransformer, make_column_selector
from sklearn.preprocessing import FunctionTransformer

# 封装预处理逻辑为类
class PreprocessA:
    def __call__(self, x):
        return x ** 2

# 构建管道（传入类实例）
pipeline = ColumnTransformer([
    ('test', FunctionTransformer(PreprocessA()), make_column_selector())
])

# 拟合并保存
df = pd.DataFrame({'col1': [1, 2, 4], 'col2': [3, 6, 9]})
pipeline.fit(df)

with open('test_class.pkl', 'wb') as f:
    pickle.dump(pipeline, f)

# 修改类逻辑用于新实验
class PreprocessA:
    def __call__(self, x):
        return x ** 4

# 加载管道验证行为
with open('test_class.pkl', 'rb') as f:
    loaded_pipeline = pickle.load(f)

print(loaded_pipeline.transform(df))
# 输出：[[ 1  9]
#        [ 4 36]
#        [16 81]] （依然保留旧的平方逻辑）

方法3：手动序列化函数源码（无第三方依赖） #

通过inspect模块获取函数源代码，将代码与管道一起序列化，加载时重新执行代码还原原始函数。适合无法引入第三方库的场景。

代码示例：

import pickle
import pandas as pd
import inspect
from sklearn.compose import ColumnTransformer, make_column_selector
from sklearn.preprocessing import FunctionTransformer

# 定义初始预处理函数
def preprocess_A(x):
    return x ** 2

# 获取函数源代码
func_source = inspect.getsource(preprocess_A)

# 包装管道和函数源码
class PipelineWithFunc:
    def __init__(self, pipeline, func_source):
        self.pipeline = pipeline
        self.func_source = func_source
        # 还原原始函数并替换管道中的引用
        exec(func_source, globals())
        self.pipeline.named_transformers_['test'].func = locals()['preprocess_A']

# 构建管道
pipeline = ColumnTransformer([
    ('test', FunctionTransformer(preprocess_A), make_column_selector())
])

df = pd.DataFrame({'col1': [1, 2, 4], 'col2': [3, 6, 9]})
pipeline.fit(df)

# 保存包装后的对象
with open('test_func_source.pkl', 'wb') as f:
    pickle.dump(PipelineWithFunc(pipeline, func_source), f)

# 修改原函数用于新实验
def preprocess_A(x):
    return x ** 4

# 加载并验证
with open('test_func_source.pkl', 'rb') as f:
    wrapper = pickle.load(f)
    loaded_pipeline = wrapper.pipeline

print(loaded_pipeline.transform(df))
# 输出：[[ 1  9]
#        [ 4 36]
#        [16 81]]

总结 #

• 优先选择cloudpickle：无需修改原有代码结构，简单高效；
• 类封装方案：无第三方依赖，符合面向对象设计，可靠性高；
• 手动序列化源码：适合特殊受限场景，但需注意函数的依赖问题。

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