Django中Celery异步任务报错：参数数量不匹配问题排查

已完成Django与Celery的配置，基于OOP编写异步任务task_run_test_optimization，调用delay(data, target, type_model)时Celery worker报错，提示该函数需3个参数但传入了4个。

相关代码文件 #

helpers.py #

class MachineLearniaXHelpers(MagicModels):
    def __init__(self, data_dict, target_name, type_model, test_size = 0.33, random_state = 42, **kwargs) -> None:
        super().__init__()
        self.test_size = test_size
        self.random_state = random_state
        self.process = DataProcessor(
            data=data_dict,
            target_name=target_name
        )
        self.metrics = MetricsScorer(type_model=type_model)
        self.model = self.several_algorythme(type_model, **kwargs)
    
    @staticmethod
    def split_data(data_dict):
        return train_test_split(*data_dict)
        
    @staticmethod   
    def train_model(model, X, y):
        model.fit(X, y)
    
    def run_test_optimization(self):
        dict_result = []
        for model in self.model:
            feature_model, values = self.process.transform_dict_to_array_structure()
            x_train, x_test, y_train, y_test = self.split_data(values)
            self.train_model(model, x_train, y_train)
            dict_metrics_train = self.metrics.choice_metrics_by_type_model(y_train, model.predict(x_train))
            dict_metrics_test = self.metrics.choice_metrics_by_type_model(y_test, model.predict(x_test))
            dict_result.append({
                "name_model" : model.__class__.__name__,
                "features_model" : feature_model,
                "train_performances" : dict_metrics_train,
                "test_performances" : dict_metrics_test
            })
        return dict_result

tasks.py #

from celery import shared_task
from .helpers import MachineLearniaXHelpers

@shared_task
def task_run_test_optimization(data_dict, target_name, type_model):
    constructor = MachineLearniaXHelpers(
        data_dict,
        target_name,
        type_model
        )
    dict_result = constructor.run_test_optimization()
    return dict_result

views.py #

class MachineLearningXView(APIView):
    serializer_class = BuildModelMLSerializer
    permission_classes = [IsAuthenticated, UserPermissionMachineLearniaX]


    def post(self, request):
        if not self.request.session.exists(self.request.session.session_key):
            self.request.session.create()
        
        serializer = self.serializer_class(data=request.data)
        if serializer.is_valid():
            data = request.data.get('data')
            target = serializer.data.get('target_name')
            type_model = serializer.data.get('test_type')
            result = task_run_test_optimization.delay(data, target, type_model) # 问题出现位置
            print(result)
            return Response(status=status.HTTP_200_OK)

报错信息 #

POST请求正常返回，控制台输出任务ID：

System check identified no issues (0 silenced).
February 08, 2023 - 17:25:28
Django version 4.1.5, using settings 'ialab.settings'
Starting development server at http://127.0.0.1:8000/
Quit the server with CTRL-BREAK.
829e66ee-925f-4aff-8c42-49420e5758ce <-- 输出的任务ID
[08/Feb/2023 17:25:47] "POST /api/machinelearniax HTTP/1.1" 200 0

Celery worker报错：

[2023-02-08 17:25:47,813: INFO/MainProcess] Task api.tasks.task_run_test_optimization[829e66ee-925f-4aff-8c42-49420e5758ce] received
[2023-02-08 17:25:47,897: ERROR/MainProcess] Task api.tasks.task_run_test_optimization[829e66ee-925f-4aff-8c42-49420e5758ce] raised unexpected: TypeError('task_run_test_optimization() takes 3 positional arguments but 4 were given')
Traceback (most recent call last):
  File "C:\Users\basti\Desktop\IALab\ialab\.ialabenv\lib\site-packages\celery\app\trace.py", line 451, in trace_task
    R = retval = fun(*args, **kwargs)
  File "C:\Users\basti\Desktop\IALab\ialab\.ialabenv\lib\site-packages\celery\app\trace.py", line 734, in __protected_call__
    return self.run(*args, **kwargs)
TypeError: task_run_test_optimization() takes 3 positional arguments but 4 were given

原因分析 #

错误核心是Celery worker加载的任务函数定义与当前代码不一致，常见场景：

1. 修改task_run_test_optimization的参数后未重启Celery worker，导致worker仍使用旧版本的函数定义（比如旧版本有4个参数，当前改为3个，但worker未更新）。
2. Celery的任务签名被序列化机制缓存，导致参数传递不匹配。

解决步骤 #

1. 重启Celery worker：
   停止当前运行的Celery进程，重新启动worker以加载最新的tasks.py代码。示例命令（根据项目调整）：

   # Windows按Ctrl+C停止worker，Linux用kill命令终止进程
   celery -A ialab worker --loglevel=info  # 重新启动worker
   

2. 检查任务函数历史修改：
   确认之前的task_run_test_optimization是否存在4个参数（比如曾包含self或其他额外参数），确保代码修改已同步。
3. 清理任务缓存（可选）：
   若重启worker后问题仍存在，可清除Celery结果后端的缓存数据，或使用--purge参数启动worker清除旧任务：

   celery -A ialab worker --loglevel=info --purge
   

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