Django + Celery 打造企业级大模型异步任务管理平台 —— 从需求到完整实践（含全模板源码）

如需完整工程文件（含所有模板），可回复获取详细模板代码。

面向人群：自动化测试工程师、企业中后台开发人员、希望提升效率的 AI 业务从业者
核心收获：掌握 Django 三表关系设计、Celery 异步任务实践、基础 Web 交互与前后端分离思路，源码可直接落地，方便二次扩展

---

一、系统功能全览

本系统以大模型自动化任务管理为核心，涵盖了三大业务场景：

1. 大模型任务管理

• 场景：测试工程师或业务人员通过 Web 页面向企业内已接入的大模型（如 GPT、阿里百川等）发起任务请求，系统后台异步执行任务并实时反馈进度与结果。
• 功能：
  • 任务列表：清晰展示任务名称、创建时间、发起人、提问内容、所用大模型、进度
  • 任务详情：一键查看输入与大模型返回详情
  • 一键新建任务：弹窗表单，选择发起人/模型、填写提示词，任务异步下发

2. 大模型管理

• 场景：平台支持多模型配置、维护，可随时增删编辑，灵活应对不同业务需求或供应商切换。
• 功能：
  • 大模型列表：展示所有已接入大模型的关键信息
  • 新建/编辑/删除大模型：表单支持 API Key、Base URL、模型名、维护人，快速扩展和维护

3. 人员管理

• 场景：平台中的“发起人”、“维护人”都从人员表下拉选取，方便团队协作与权限追踪。
• 功能：
  • 列表、创建、编辑、删除人员，数据基础一目了然

---

二、系统架构与扩展性说明

• 异步任务：Celery+Redis 组合，任务下发即返回，处理高并发和长耗时 AI 推理毫无压力。
• 表单校验：所有关键字段必填，表单错误友好提示。
• 数据解耦：大模型、人员、任务三表设计，方便后续增加权限、标签等扩展字段。
• 易于二次开发：
  • 任务可轻松拓展为定时任务、批量任务、API 任务
  • 大模型表可加“模型类型”、“状态”等新字段
  • Celery 任务内可真实调用大模型接口，支持异步回调、失败重试等企业级需求
• 权限与安全：可直接结合 Django 用户系统，扩展为企业级权限管理后台

环境准备

1. 安装依赖

   pip install django celery redis
   

2. 确保本地 Redis 已启动（Windows 下可用）

---

项目结构

bigmodel_proj/
├── bigmodel_proj/
│   ├── __init__.py
│   ├── celery.py
│   ├── settings.py
│   ├── urls.py
│   └── wsgi.py
├── app/
│   ├── __init__.py
│   ├── admin.py
│   ├── apps.py
│   ├── models.py
│   ├── tasks.py
│   ├── views.py
│   ├── forms.py
│   ├── urls.py
│   └── templates/
│       └── app/
│           └── ...  # 所有前端模板
└── manage.py

---

四、核心代码实现

1. bigmodel_proj/settings.py（核心配置）

import os
from pathlib import PathBASE_DIR = Path(__file__).resolve().parent.parentSECRET_KEY = 'your-secret-key'
DEBUG = True
ALLOWED_HOSTS = []INSTALLED_APPS = ['django.contrib.admin','django.contrib.auth','django.contrib.contenttypes','django.contrib.sessions','django.contrib.messages','django.contrib.staticfiles','app',
]MIDDLEWARE = ['django.middleware.security.SecurityMiddleware','django.contrib.sessions.middleware.SessionMiddleware','django.middleware.common.CommonMiddleware','django.middleware.csrf.CsrfViewMiddleware','django.contrib.auth.middleware.AuthenticationMiddleware','django.contrib.messages.middleware.MessageMiddleware','django.middleware.clickjacking.XFrameOptionsMiddleware',
]ROOT_URLCONF = 'bigmodel_proj.urls'TEMPLATES = [{'BACKEND': 'django.template.backends.django.DjangoTemplates','DIRS': [BASE_DIR / 'app' / 'templates'],'APP_DIRS': True,'OPTIONS': {'context_processors': ['django.template.context_processors.debug','django.template.context_processors.request','django.contrib.auth.context_processors.auth','django.contrib.messages.context_processors.messages',],},},
]WSGI_APPLICATION = 'bigmodel_proj.wsgi.application'DATABASES = {'default': {'ENGINE': 'django.db.backends.sqlite3','NAME': BASE_DIR / 'db.sqlite3',}
}LANGUAGE_CODE = 'zh-hans'
TIME_ZONE = 'Asia/Shanghai'
USE_I18N = True
USE_L10N = True
USE_TZ = TrueSTATIC_URL = '/static/'# Celery 配置
CELERY_BROKER_URL = 'redis://localhost:6379/0'
CELERY_RESULT_BACKEND = 'redis://localhost:6379/0'
CELERY_ACCEPT_CONTENT = ['json']
CELERY_TASK_SERIALIZER = 'json'
CELERY_RESULT_SERIALIZER = 'json'

---

2. bigmodel_proj/celery.py（Celery 启动入口）

import os
from celery import Celeryos.environ.setdefault('DJANGO_SETTINGS_MODULE', 'bigmodel_proj.settings')
app = Celery('bigmodel_proj')
app.config_from_object('django.conf:settings', namespace='CELERY')
app.autodiscover_tasks()

bigmodel_proj/__init__.py：

from .celery import app as celery_app
__all__ = ('celery_app',)

---

3. app/models.py（三大核心数据模型）

from django.db import modelsclass Person(models.Model):name = models.CharField(max_length=50, verbose_name="姓名")created_at = models.DateTimeField(auto_now_add=True, verbose_name="创建时间")def __str__(self):return self.nameclass BigModel(models.Model):api_key = models.CharField(max_length=128, verbose_name="API Key")base_url = models.URLField(verbose_name="Base URL")model_name = models.CharField(max_length=100, verbose_name="模型名")maintainer = models.ForeignKey(Person, on_delete=models.SET_NULL, null=True, verbose_name="维护人")created_at = models.DateTimeField(auto_now_add=True, verbose_name="创建时间")updated_at = models.DateTimeField(auto_now=True, verbose_name="修改时间")def __str__(self):return self.model_nameclass BigModelTask(models.Model):STATUS_CHOICES = (('PENDING', '待处理'),('RUNNING', '进行中'),('SUCCESS', '成功'),('FAILURE', '失败'),)name = models.CharField(max_length=100, verbose_name="任务名称")creator = models.ForeignKey(Person, on_delete=models.SET_NULL, null=True, verbose_name="发起人")prompt = models.TextField(verbose_name="用户提问信息")big_model = models.ForeignKey(BigModel, on_delete=models.SET_NULL, null=True, verbose_name="大模型")status = models.CharField(max_length=20, choices=STATUS_CHOICES, default='PENDING', verbose_name="任务进度")created_at = models.DateTimeField(auto_now_add=True, verbose_name="创建时间")detail = models.TextField(blank=True, null=True, verbose_name="大模型请求详情")def __str__(self):return self.name

---

4. app/tasks.py（Celery 异步任务）

from celery import shared_task
import time
from .models import BigModelTask@shared_task(bind=True)
def execute_bigmodel_task(self, task_id):try:task = BigModelTask.objects.get(id=task_id)task.status = 'RUNNING'task.save(update_fields=['status'])# 模拟大模型请求耗时time.sleep(5)# 假装请求大模型并返回结果result_detail = f"请求大模型[{task.big_model.model_name}]完成，提示词：{task.prompt}"task.status = 'SUCCESS'task.detail = result_detailtask.save(update_fields=['status', 'detail'])except Exception as e:if task:task.status = 'FAILURE'task.detail = f"异常：{str(e)}"task.save(update_fields=['status', 'detail'])raise

---

5. app/forms.py（前端表单）

from django import forms
from