当前位置：首页 > 文章列表 > 文章 > python教程 > Flask模型训练页面跳转与结果展示教程

Flask模型训练页面跳转与结果展示教程

2026-02-11 08:18:51 0浏览收藏

一分耕耘，一分收获！既然打开了这篇文章《Flask训练模型页面跳转与结果渲染教程》，就坚持看下去吧！文中内容包含等等知识点...希望你能在阅读本文后，能真真实实学到知识或者帮你解决心中的疑惑，也欢迎大佬或者新人朋友们多留言评论，多给建议！谢谢！

Flask后台训练模型时如何正确实现页面跳转与结果渲染

在Flask中使用线程异步执行耗时任务（如模型训练）时，不能直接在子线程中调用render_template()——该函数仅在请求上下文中有效，而子线程无请求上下文，也无法主动响应HTTP请求，因此visualize.html不会被发送给用户。

Flask 是同步单请求模型：每个 HTTP 请求由一个工作线程（或进程）处理，其生命周期从接收请求开始，到返回响应结束。render_template() 本质是生成 HTML 响应体并随当前请求一并返回；而你在 train_models() 中调用它时，该函数运行在独立的 Thread 中，既无 request 上下文，也无 app.app_context()（你虽加了 with app.app_context():，但这仅解决 Flask 全局对象（如 current_app, g）访问问题，并不能让 render_template() 触发一次新的 HTTP 响应）。

✅ 正确做法是：分离“执行”与“响应”

主请求线程快速返回 loading.html（前端可配合轮询或 WebSocket 等待结果）；
后台线程完成计算后，将结果（如图表路径、评估指标、HTML 片段等）安全存入共享存储（如全局字典 + threading.Lock、Redis、或内存缓存如 werkzeug.contrib.cache）；
用户通过另一个路由（如 /results）主动拉取结果，或前端定时 AJAX 请求该接口，成功后重定向至 visualize.html 或动态渲染内容。

? 示例修正（轻量级方案，使用线程安全全局缓存）：

import threading
from flask import Flask, request, render_template, jsonify, redirect, url_for

# 线程安全的结果存储
_results_cache = {}
_cache_lock = threading.Lock()

@app.route("/inputs", methods=["POST"])
def inputs():
    file = request.files.get("dataset")
    algo = request.form.getlist("algo")

    if not file:
        return render_template("input_form.html", error="Please upload a CSV file", algos=ALGO, selected_algo=algo)

    data = io.StringIO(file.stream.read().decode("UTF8"), newline=None)
    dataset = pd.read_csv(data)

    # 生成唯一任务ID（例如用时间戳+随机数）
    task_id = f"task_{int(time.time())}_{id(threading.current_thread())}"

    # 启动训练线程，传入 task_id 用于结果落库
    train_thread = threading.Thread(
        target=train_models, 
        args=(app, algo, dataset, task_id)
    )
    train_thread.daemon = True  # 避免阻塞服务退出
    train_thread.start()

    # 立即返回 loading 页面，并携带 task_id 供前端轮询
    return render_template("loading.html", task_id=task_id)

def train_models(app, algo, dataset, task_id):
    print("Starting training...")
    with app.app_context():
        plot_model = Gen_Plot()
        plots = {}
        selected_algo = {}

        for model in algo:
            # 训练逻辑...
            pass

        # ✅ 将结果存入全局缓存（注意线程安全）
        with _cache_lock:
            _results_cache[task_id] = {
                "status": "completed",
                "plots": plots,  # 可序列化数据，如 base64 图片、JSON 指标
                "selected_algo": selected_algo
            }
        print(f"Results saved for task {task_id}")

# 新增结果查询接口（供前端 AJAX 轮询）
@app.route("/api/task/<task_id>")
def get_task_result(task_id):
    with _cache_lock:
        result = _results_cache.get(task_id)

    if result and result.get("status") == "completed":
        return jsonify({
            "status": "success",
            "data": result
        })
    else:
        return jsonify({"status": "pending"}), 202  # HTTP 202 Accepted

# 可选：重定向入口（前端获取结果后跳转）
@app.route("/visualize/<task_id>")
def visualize(task_id):
    with _cache_lock:
        result = _results_cache.get(task_id)

    if not result or result.get("status") != "completed":
        return redirect(url_for("inputs"))  # 或返回错误页

    return render_template("visualize.html", **result)

? 前端 loading.html 可添加简单轮询逻辑（示例）：

<script>
const taskId = "{{ task_id }}";
function checkResult() {
  fetch(`/api/task/${taskId}`)
    .then(r => r.json())
    .then(data => {
      if (data.status === "success") {
        window.location.href = `/visualize/${taskId}`;
      } else {
        setTimeout(checkResult, 1000);
      }
    });
}
checkResult();
</script>

⚠️ 注意事项：