如需完整工程文件（含所有模板），可回复获取详细模板代码。
面向人群：自动化测试工程师、企业中后台开发人员、希望提升效率的 AI 业务从业者
核心收获：掌握 Django 三表关系设计、Celery 异步任务实践、基础 Web 交互与前后端分离思路，源码可直接落地，方便二次扩展
一、系统功能全览
本系统以大模型自动化任务管理为核心，涵盖了三大业务场景：
1. 大模型任务管理
- 场景：测试工程师或业务人员通过 Web 页面向企业内已接入的大模型（如 GPT、阿里百川等）发起任务请求，系统后台异步执行任务并实时反馈进度与结果。
- 功能：
- 任务列表：清晰展示任务名称、创建时间、发起人、提问内容、所用大模型、进度
- 任务详情：一键查看输入与大模型返回详情
- 一键新建任务：弹窗表单，选择发起人/模型、填写提示词，任务异步下发
2. 大模型管理
- 场景：平台支持多模型配置、维护，可随时增删编辑，灵活应对不同业务需求或供应商切换。
- 功能：
- 大模型列表：展示所有已接入大模型的关键信息
- 新建/编辑/删除大模型：表单支持 API Key、Base URL、模型名、维护人，快速扩展和维护
3. 人员管理
- 场景：平台中的“发起人”、“维护人”都从人员表下拉选取，方便团队协作与权限追踪。
- 功能：
- 列表、创建、编辑、删除人员，数据基础一目了然
二、系统架构与扩展性说明
- 异步任务：Celery+Redis 组合，任务下发即返回，处理高并发和长耗时 AI 推理毫无压力。
- 表单校验：所有关键字段必填，表单错误友好提示。
- 数据解耦：大模型、人员、任务三表设计，方便后续增加权限、标签等扩展字段。
- 易于二次开发：
- 任务可轻松拓展为定时任务、批量任务、API 任务
- 大模型表可加“模型类型”、“状态”等新字段
- Celery 任务内可真实调用大模型接口，支持异步回调、失败重试等企业级需求
- 权限与安全：可直接结合 Django 用户系统，扩展为企业级权限管理后台
环境准备
安装依赖
pip install django celery redis确保本地 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 Path
BASE_DIR = Path(__file__).resolve().parent.parent
SECRET_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 = True
STATIC_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 Celery
os.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 models
class 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.name
class 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_name
class 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_detail
task.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