为何我的NLP模型处理问题时会多次重复加载？

问题描述 #

程序接收用户问题后，会调用run_predict函数查找匹配段落，但每次处理请求时模型都会重新加载，尝试将模型放到单独文件中调用也无法解决该问题。

代码问题分析 #

• 全局模型未被正确复用：在run_predict函数中，重复调用load_model()并赋值给局部变量model。若SocketIO运行在多进程/多线程模式下，每个请求进程的全局model变量会被重新初始化，导致每次请求都触发模型重新加载。
• 变量冗余覆盖：run_predict函数内用question = data["question"]覆盖了传入的question参数，虽不直接引发模型重复加载，但属于代码冗余问题。

修复方案 #

核心修改点 #

1. 直接复用启动时已加载的全局模型，删除run_predict中重复调用load_model()的代码；
2. 配置SocketIO以单进程模式运行，避免多进程导致全局变量无法共享；
3. 修复参数覆盖问题，重命名传入参数避免冲突。

修复后的完整代码 #

from flask import Flask, render_template, request, jsonify
from flask_socketio import SocketIO, emit
import os
import json
import logging
from simpletransformers.question_answering import QuestionAnsweringModel, QuestionAnsweringArgs
from multiprocessing import freeze_support
from models.find_top_paragraphs import main as find_top_paragraphs

app = Flask(__name__)
app.config['SECRET_KEY'] = 'secret!'
# 配置SocketIO使用单线程+单进程模式，避免多进程导致模型重复加载
socketio = SocketIO(app, threaded=False)

# 配置模型参数
model_args = QuestionAnsweringArgs()
model_args.eval_batch_size = 16

# 获取当前目录绝对路径
current_dir = os.path.abspath(os.getcwd())
# 模型存储路径
outputs_dir = os.path.join(current_dir, "outputs", "best_model")

print(f"Model directory: {outputs_dir}")

# 全局模型变量
model = None

def load_model():
    global model
    if model is None:
        model = QuestionAnsweringModel(
            model_type="bert", 
            model_name=outputs_dir, 
            args=model_args,
            use_cuda=False  # 有GPU可改为True
        )
        print("Load model successfully!")
    else:
        print("Model is already loaded.")
    return model

# 启动时仅加载一次模型
model = load_model()

def run_predict(input_question):
    print("Running predict function")
    # 调用段落匹配函数
    find_top_paragraphs(input_question)

    # 读取匹配结果
    output_path = os.path.join(os.path.dirname(__file__), "models", "top_paragraphs.json")
    with open(output_path, "r", encoding='utf-8') as file:
        data = json.load(file)
    
    question = data["question"]
    top_paragraph = data["top_paragraphs"][0]  # 取最匹配的段落
    print(f"Top Paragraph: {top_paragraph}")

    # 构造预测数据
    to_predict = [
        {
            "context": top_paragraph,
            "qas": [
                {
                    "question": question,
                    "id": "0",
                }
            ],
        }
    ]

    # 直接使用全局已加载的模型
    answers, probabilities = model.predict(to_predict)

    # 整理预测结果
    all_answers = []
    for answer in answers:
        for a in answer['answer']:
            all_answers.append(a)
        
        # 筛选最优答案
        try:
            probability = probabilities[0]['probability']
            best_answer_idx = probability.index(max(probability))
            best_answer = answer['answer'][best_answer_idx]
            print(f"Best Answer: {best_answer}")
            return all_answers, best_answer
        except Exception as e:
            print(f"An error occurred while selecting the best answer: {e}")
            return all_answers, None

@app.route('/')
def index():
    return render_template('index.html')

@socketio.on('send_message')
def handle_message(data):
    question = data['message']
    all_answers, best_answer = run_predict(question)
    response = {
        'all_answers': all_answers,
        'best_answer': best_answer
    }
    emit('receive_message', response)

if __name__ == '__main__':
    freeze_support()
    # 指定单进程运行，确保全局模型只加载一次
    socketio.run(app, debug=False, processes=1)

额外说明 #

若业务需要多进程模式，可在每个进程的初始化阶段加载模型（比如通过@socketio.on('connect')事件检查模型状态），但会导致每个进程加载一次模型，适合多GPU分布式场景。

内容的提问来源于stack exchange，提问作者Ngọc Minh