如何在Django中无需加载页面运行Python函数？

我正在开发一个项目，其中包含一个通过Google Gemini API循环自动生成响应的函数，生成的响应保存到数据库后需要同步到HTML页面。当前遇到的问题是：当函数循环生成多条响应时，页面会持续加载直到所有响应生成完成，才会渲染home.html页面。我希望这个函数能在后台运行，不阻塞页面加载。

现有代码 #

Python视图函数 #

def home(request):
    print("Bot: Hello how can i help you?")

    i = 0
    while i < 2:
        if len(history_global) == 0:
            user_input = input("You: ")
            chat_session = model.start_chat(
            history=history
            )

            response = chat_session.send_message(user_input)                        #1 Taking input from user
            
            model_response = response.text
            
            history.append({"role" : "user", "parts" : [user_input]})              #3 updating history - (user part)
            history.append({"role" : "model", "parts" : [model_response]})         #4 updating history - (response part)
            history_global = model_response                                        #5 updating value of history_global with the response created

            # Save the data to the database
            saveit = Data1(question=user_input, answer=history_global)
            saveit.save()
          
            print(i)
            print("Question : ", user_input)
            print()
            print("Answer : ", model_response)
            print()
            print("...............................................................................................1")
            print()

            random_interval = random.randint(10, 15)
            time.sleep(random_interval)

        else:
            chat_session = model.start_chat(
                history=history
            )

            new_prompt = "[make any one question from the following content] - " + history_global

            response = chat_session.send_message(new_prompt)                      #1 Taking input from previous response
                                                                                  # generating output result (basically question).......
            model_response = response.text                                        #2 Generated question
            history_global_question = model_response

            history.append({"role" : "user", "parts" : [new_prompt]})              #3 updating history - (user part)
            history.append({"role" : "model", "parts" : [model_response]})         #4 updating history - (response part)
            history_global = model_response                                        #5 updating value of history_global
           
            random_interval_01 = random.randint(10, 15)
            time.sleep(random_interval_01)

            response2 = chat_session.send_message(history_global_question)         #6 Taking question generated in #2 as a prompt
            model_response2 = response2.text                                       #7 Generated answer

            history.append({"role" : "user", "parts" : [user_input]})              #8 updating history - (user part)
            history.append({"role" : "model", "parts" : [model_response]})         #9 updating history - (response part)
            history_global = model_response2                                       #10 updating history_global with new response

            # Save the data to the database
            saveit = Data1(question=history_global_question, answer=model_response2)
            saveit.save()

            print(i)
            print("Question : ", history_global_question)
            print()
            print("Answer : ", model_response2)
            print()
            print("...............................................................................................2")
            print()
            
            random_interval = random.randint(10, 15)
            time.sleep(random_interval)
            
        i+=1

    data1 = Data1.objects.all()

    context = {"data_test" : data1}

    return render(request, 'home.html', context)

HTML模板（home.html） #

{% load static %}
<!-- main/templates/home.html -->
<!DOCTYPE html>
<html lang="en">

<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>Welcome</title>
    <link rel="stylesheet" type="text/css" href="{% static 'home.css' %}">
</head>

<body>
    <h1>Welcome to our website</h1>

    {% for data1 in data_test %}
    <div>
        <h1 style="color: antiquewhite;">{{data1.question}}</h1>
        <br>
        <p style="color: aqua;">{{data1.answer}}</p>
        <a href="contentpage/{{data1.id}}">click here</a>
    </div>
    {% endfor %}
</body>

</html>

核心思路是把耗时的Gemini API调用和循环逻辑从视图中剥离，放到后台异步执行；前端先加载页面，再通过轮询实时获取新生成的数据。

步骤1：用Celery实现后台异步任务 #

先安装celery和Redis（消息中间件），然后把循环生成响应的逻辑封装为Celery任务：

# tasks.py
from celery import Celery
from .models import Data1
import random
import time
# 导入你的Gemini model和对话历史相关变量

app = Celery('tasks', broker='redis://localhost:6379/0')

@app.task
def generate_responses():
    print("Bot: Hello how can i help you?")
    i = 0
    while i < 2:
        if len(history_global) == 0:
            # 注意：后台任务无法使用input()，这里改为预设测试输入或从缓存/数据库获取
            user_input = "Test user input"
            chat_session = model.start_chat(history=history)

            response = chat_session.send_message(user_input)
            model_response = response.text
            
            history.append({"role" : "user", "parts" : [user_input]})
            history.append({"role" : "model", "parts" : [model_response]})
            history_global = model_response

            saveit = Data1(question=user_input, answer=history_global)
            saveit.save()
        
            print(i)
            print("Question : ", user_input)
            print("Answer : ", model_response)
            print("...............................................................................................1")

            random_interval = random.randint(10, 15)
            time.sleep(random_interval)

        else:
            chat_session = model.start_chat(history=history)

            new_prompt = "[make any one question from the following content] - " + history_global

            response = chat_session.send_message(new_prompt)
            model_response = response.text
            history_global_question = model_response

            history.append({"role" : "user", "parts" : [new_prompt]})
            history.append({"role" : "model", "parts" : [model_response]})
            history_global = model_response
       
            random_interval_01 = random.randint(10, 15)
            time.sleep(random_interval_01)

            response2 = chat_session.send_message(history_global_question)
            model_response2 = response2.text

            # 修正原代码笔误：用当前问题更新历史，而非旧的user_input
            history.append({"role" : "user", "parts" : [history_global_question]})
            history.append({"role" : "model", "parts" : [model_response2]})
            history_global = model_response2

            saveit = Data1(question=history_global_question, answer=model_response2)
            saveit.save()

            print(i)
            print("Question : ", history_global_question)
            print("Answer : ", model_response2)
            print("...............................................................................................2")
            
            random_interval = random.randint(10, 15)
            time.sleep(random_interval)
            
        i+=1

步骤2：修改视图函数，先返回页面再触发任务 #

调整home视图，不再执行耗时逻辑，直接返回页面并触发后台任务：

# views.py
from .tasks import generate_responses
from .models import Data1

def home(request):
    # 触发后台异步任务
    generate_responses.delay()
    # 获取已有数据，先渲染页面
    data1 = Data1.objects.all()
    context = {"data_test" : data1}
    return render(request, 'home.html', context)

步骤3：前端添加轮询，实时更新内容 #

修改HTML模板，添加JavaScript轮询逻辑，定期获取新数据并更新页面：

{% load static %}
<!-- main/templates/home.html -->
<!DOCTYPE html>
<html lang="en">

<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>Welcome</title>
    <link rel="stylesheet" type="text/css" href="{% static 'home.css' %}">
</head>

<body>
    <h1>Welcome to our website</h1>

    <div id="responses-container">
        {% for data1 in data_test %}
        <div class="response-item" data-id="{{data1.id}}">
            <h1 style="color: antiquewhite;">{{data1.question}}</h1>
            <br>
            <p style="color: aqua;">{{data1.answer}}</p>
            <a href="contentpage/{{data1.id}}">click here</a>
        </div>
        {% endfor %}
    </div>

    <script>
        // 轮询间隔：5秒
        const POLL_INTERVAL = 5000;
        let lastId = {% if data_test %}{{data_test.last.id}}{% else %}0{% endif %};

        function fetchNewResponses() {
            fetch(`/get-new-responses/?last_id=${lastId}`)
                .then(response => response.json())
                .then(data => {
                    if (data.new_responses.length > 0) {
                        const container = document.getElementById('responses-container');
                        data.new_responses.forEach(item => {
                            // 创建新的响应元素
                            const div = document.createElement('div');
                            div.className = 'response-item';
                            div.dataset.id = item.id;
                            div.innerHTML = `
                                <h1 style="color: antiquewhite;">${item.question}</h1>
                                <br>
                                <p style="color: aqua;">${item.answer}</p>
                                <a href="contentpage/${item.id}">click here</a>
                            `;
                            container.appendChild(div);
                            // 更新lastId
                            lastId = item.id;
                        });
                    }
                })
                .catch(error => console.error('Error fetching new responses:', error));
        }

        // 初始加载后开始轮询
        setInterval(fetchNewResponses, POLL_INTERVAL);
    </script>
</body>

</html>

步骤4：添加获取新数据的API视图 #

在views.py中添加API接口，用于返回最新生成的响应：

# views.py
from django.http import JsonResponse
from .models import Data1

def get_new_responses(request):
    last_id = request.GET.get('last_id', 0)
    new_responses = Data1.objects.filter(id__gt=last_id).values('id', 'question', 'answer')
    return JsonResponse({
        'new_responses': list(new_responses)
    })

同时在urls.py中添加路由：

# urls.py
from django.urls import path
from . import views

urlpatterns = [
    path('', views.home, name='home'),
    path('get-new-responses/', views.get_new_responses, name='get_new_responses'),
    # 其他路由...
]

注意事项 #

• 原代码中的history_global和history是全局变量，在后台任务中使用会有并发问题，建议改用数据库或缓存存储对话历史。
• 后台任务无法使用input()，需调整为前端传入、预设值或其他方式获取用户输入。
• 启动Celery worker的命令：celery -A 你的项目名 worker --loglevel=info

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