Django框架中如何将OpenAI调用等耗时任务移至后台执行？

方案一：使用Celery实现异步任务（生产环境推荐） #

Celery是Python生态中成熟的异步任务队列工具，专门处理这类耗时后台任务，能有效解耦请求与执行流程，步骤如下：

1. 安装依赖 #

pip install celery redis  # 用Redis做消息代理，也可替换为RabbitMQ

2. 配置Celery（项目根目录新建celery.py） #

import os
from celery import Celery

os.environ.setdefault('DJANGO_SETTINGS_MODULE', '你的项目名.settings')

app = Celery('你的项目名')
app.config_from_object('django.conf:settings', namespace='CELERY')
app.autodiscover_tasks()

3. 在settings.py中添加Celery配置 #

CELERY_BROKER_URL = 'redis://localhost:6379/0'  # Redis服务地址
CELERY_RESULT_BACKEND = 'redis://localhost:6379/0'
CELERY_ACCEPT_CONTENT = ['json']
CELERY_TASK_SERIALIZER = 'json'
CELERY_RESULT_SERIALIZER = 'json'

4. 定义异步任务（示例在app/tasks.py） #

import json
import requests
from celery import shared_task
from django.conf import settings
from django.contrib.auth.models import User
from .models import UserProfile, GeneratedX  # 需提前创建存储结果的模型

@shared_task
def generate_x_task(user_id, position, company, description):
    # 获取用户及信息
    my_user = User.objects.get(id=user_id)
    my_user_profile = UserProfile.objects.get(user=my_user)
    
    # 构建请求Prompt
    user_prompt = createPrompt(my_user_profile, position, company, description)
    system_prompt = "You are a helpful assistant."
    
    # 调用OpenAI API
    url = "https://api.openai.com/v1/chat/completions"
    headers = {
        "Authorization": f"Bearer {settings.OPEN_AI_API_KEY}",
        "Content-Type": "application/json"
    }
    payload = {
        "model": "gpt-3.5-turbo",
        "messages": [
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": user_prompt}
        ],
        "temperature": 0.7,
    }
    response = requests.post(url, headers=headers, json=payload)
    data = json.loads(response.text)
    reply = data["choices"][0]["message"]["content"]
    
    # 存储结果到数据库，供用户后续查看
    GeneratedX.objects.create(
        user=my_user,
        x_content=reply,
        user_prompt=user_prompt,
        system_prompt=system_prompt,
        position=position,
        company=company
    )
    
    # 可选：添加通知逻辑，比如发送邮件/站内信
    # send_notification(my_user, "你的资源已生成")
    return reply

5. 修改原视图，触发异步任务 #

@login_required  
def createx(request):
    if request.method == "GET":
        return render(request, 'x/x_create.html', {})
    else:
        request_body = json.loads(request.body)
        my_user = request.user
        if not my_user.is_premium_user():
            messages.error(request, '你必须是付费用户才能创建X')
            return redirect('home')
        
        # 触发异步任务，获取任务ID
        task = generate_x_task.delay(
            my_user.id,
            request_body.get('position'),
            request_body.get('company'),
            request_body.get('description')
        )
        
        # 返回任务ID，供前端轮询状态
        return JsonResponse({'success': True, 'task_id': task.id})

6. 添加任务状态查询视图 #

from celery.result import AsyncResult
from django.http import JsonResponse

def check_task_status(request, task_id):
    result = AsyncResult(task_id)
    if result.ready():
        return JsonResponse({
            'status': 'completed',
            'task_id': task_id
        })
    else:
        return JsonResponse({'status': 'pending'})

7. 前端处理逻辑 #

用户提交请求后，拿到task_id，定期调用check_task_status接口轮询状态；任务完成后，提示用户跳转至资源查看页面。

方案二：使用Django异步视图（轻量场景） #

若你的Django版本≥3.1，可直接用异步视图处理IO密集请求，适合无需持久化任务状态的场景（注意：请求中断时任务可能终止）：

修改视图为异步函数 #

import asyncio
import aiohttp
import json
from django.http import JsonResponse
from django.contrib.auth.decorators import login_required
from django.utils.decorators import sync_to_async
from django.conf import settings

@login_required  
async def createx(request):
    if request.method == "GET":
        return render(request, 'x/x_create.html', {})
    else:
        request_body = json.loads(request.body)
        my_user = request.user
        if not my_user.is_premium_user():
            messages.error(request, '你必须是付费用户才能创建X')
            return redirect('home')
        
        # ORM操作需用sync_to_async包装
        my_user_profile = await sync_to_async(UserProfile.objects.get)(user=my_user)
        user_prompt = createPrompt(my_user_profile, request_body.get('position'), request_body.get('company'), request_body.get('description'))
        system_prompt = "You are a helpful assistant."
        
        # 用aiohttp发送异步请求
        url = "https://api.openai.com/v1/chat/completions"
        headers = {
            "Authorization": f"Bearer {settings.OPEN_AI_API_KEY}",
            "Content-Type": "application/json"
        }
        payload = {
            "model": "gpt-3.5-turbo",
            "messages": [
                {"role": "system", "content": system_prompt},
                {"role": "user", "content": user_prompt}
            ],
            "temperature": 0.7,
        }
        
        async with aiohttp.ClientSession() as session:
            async with session.post(url, headers=headers, json=payload) as response:
                data = await response.json()
                reply = data["choices"][0]["message"]["content"]
                
                # 可选：存储结果到数据库
                await sync_to_async(GeneratedX.objects.create)(
                    user=my_user,
                    x_content=reply,
                    user_prompt=user_prompt,
                    system_prompt=system_prompt
                )
                
                return JsonResponse({'success':True, 'x':reply})

注意：需用支持异步的服务器（如Uvicorn）运行项目，确保环境兼容。

方案三：使用Django Q（轻量替代Celery） #

Django Q是轻量级异步任务工具，配置更简单，适合小型项目：

1. 安装依赖 #

pip install django-q

2. 配置settings.py #

INSTALLED_APPS = [
    ...
    'django_q',
]

Q_CLUSTER = {
    'name': 'DjangORM',
    'workers': 4,
    'timeout': 60,
    'retry': 120,
    'queue_limit': 50,
    'bulk': 10,
    'orm': 'default'
}

3. 任务定义与视图修改 #

逻辑与Celery类似，用async_task触发任务，前端轮询状态即可。

关键注意事项 #

• 结果持久化：无论用哪种方案，建议将生成结果存入数据库，方便用户后续查看
• 错误处理：在异步任务中添加异常捕获，记录日志并通知用户任务失败情况
• 资源控制：根据服务器配置调整异步任务并发数，避免资源耗尽
• 用户通知：任务完成后，可通过邮件、站内信、WebSocket推送等方式主动通知用户

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