CodeGeeX配置周期性的Celery任务调度【任务调度】

需通过Celery Beat配置周期性任务调度，包括配置Redis为Broker与Backend、定义CodeGeeX预热与健康检查任务、启动Beat与Worker协同运行、用crontab精确控制执行时间，并验证任务注册与触发。

☞☞☞AI 智能聊天, 问答助手, AI 智能搜索, 多模态理解力帮你轻松跨越从0到1的创作门槛☜☜☜

如果您希望在Django项目中让CodeGeeX模型定期执行代码生成、健康检查或预热加载等操作，则需通过Celery Beat配置周期性任务调度。以下是实现此目标的步骤：

一、配置Celery Beat并启用Redis作为Broker与Backend

该方法确保调度器能可靠地将定时任务推入队列，并由Worker持续拉取执行，避免因进程重启导致任务丢失。

1、确认Redis服务已在本地或远程运行，端口为6379，数据库索引未被其他服务占用。

2、在Django项目的settings.py中添加以下配置：

CELERY_BROKER_URL = 'redis://127.0.0.1:6379/0'

CELERY_RESULT_BACKEND = 'redis://127.0.0.1:6379/1'

CELERY_TIMEZONE = 'Asia/Shanghai'

CELERY_ENABLE_UTC = False

3、在项目根目录（与manage.py同级）创建celeryconfig.py文件，写入周期性任务定义：

from celery.schedules import crontab

CELERY_BEAT_SCHEDULE = {

'codigeeex-warmup-hourly': {

'task': 'myapp.tasks.warmup_codigeeex_model',

'schedule': 3600.0,

'args': ()

},

'codigeeex-healthcheck-daily': {

'task': 'myapp.tasks.check_codigeeex_health',

'schedule': crontab(hour=2, minute=0),

'args': ()

}

}

二、定义可被周期调度的CodeGeeX相关任务

该方法将模型加载、轻量推理或状态校验逻辑封装为独立Celery任务，确保其可被Beat按计划触发，且不依赖HTTP请求上下文。

1、在应用目录（如myapp/tasks.py）中定义两个任务函数：

from celery import shared_task

from transformers import AutoModelForSeq2SeqLM, AutoTokenizer

import torch

@shared_task

def warmup_codigeeex_model():

model = AutoModelForSeq2SeqLM.from_pretrained("THUDM/codegeex-4b", low_cpu_mem_usage=True)

tokenizer = AutoTokenizer.from_pretrained("THUDM/codegeex-4b")

inputs = tokenizer("def hello():", return_tensors="pt")

with torch.no_grad(): model.generate(**inputs, max_length=64, do_sample=False)

return "CodeGeeX model warmed up successfully"

@shared_task

def check_codigeeex_health():

try:

音剪

喜马拉雅旗下的一站式AI音频创作平台，强大的在线剪辑能力，帮你轻松创作优秀的音频作品

下载

from myapp.utils import is_codigeeex_ready

return {"status": "healthy" if is_codigeeex_ready() else "unavailable"}

except Exception as e:

return {"status": "error", "message": str(e)}

三、启动Celery Beat与Worker协同运行

该方法确保调度器与执行单元分离部署，满足生产环境高可用要求，同时避免单进程阻塞导致定时任务失效。

1、在终端中启动Celery Worker，指定任务模块并启用日志：

celery -A myproject worker -l info

2、另启一个终端，启动Celery Beat进程，显式指向配置模块：

celery -A myproject beat -l info --scheduler django_celery_beat.schedulers:DatabaseScheduler

3、若使用django_celery_beat扩展（推荐用于动态管理周期任务），需先执行迁移并安装：

pip install django-celery-beat

python manage.py migrate django_celery_beat

python manage.py createsuperuser

4、登录Django Admin，在Periodic Tasks中手动添加warmup_codigeeex_model任务，设置Interval为“every 1 hour”。

四、使用crontab语法精确控制每日执行时间点

该方法适用于需要严格对齐业务低峰期（如凌晨2点系统维护窗口）执行模型健康检查的场景，避免资源争抢。

1、修改celeryconfig.py中CELERY_BEAT_SCHEDULE条目，替换为：

'codigeeex-daily-prefetch': {

'task': 'myapp.tasks.prefetch_codigeeex_examples',

'schedule': crontab(hour=3, minute=15, day_of_week='mon,tue,wed,thu,fri'),

'args': ('fibonacci',)

}

2、在myapp/tasks.py中新增对应任务：

@shared_task

def prefetch_codigeeex_examples(topic: str):

prompt = f"Write a Python function to calculate {topic}:"

# 调用已封装的CodeGeeX生成逻辑（非阻塞方式）

result = generate_code_with_codigeeex(prompt)

return {"prompt": prompt, "generated": result[:200]}

五、验证周期任务是否成功注册并触发

该方法通过命令行工具直接探查调度器状态与任务分发记录，无需等待实际执行时间点，提升调试效率。

1、执行以下命令查看当前活跃的Beat调度器是否识别到配置：

celery -A myproject inspect scheduled

2、检查Worker是否已订阅对应队列：

celery -A myproject inspect active_queues

3、手动触发一次任务以验证执行路径无异常：

from myapp.tasks import warmup_codigeeex_model; warmup_codigeeex_model.apply()

4、观察Worker终端输出是否包含"CodeGeeX model warmed up successfully"字样及时间戳。