如何实现运行时每日更新sklearn、PyTorch模型/向量，规避文件打开过多报错

核心思路 #

• 用全局变量缓存已加载的模型实例，避免每次请求重复加载，从根源解决文件描述符溢出问题
• 加读写锁保证并发场景下模型读取、切换的安全性，读多写少场景下性能损失极低
• 版本切换时优先加载当日版本，加载失败自动降级到已有版本，完全匹配可用性约束

具体实现步骤 #

1. 依赖准备 #

除你已有的fastapi、joblib依赖外，额外用到两个工具库：readerwriterlock处理读写锁，APScheduler做定时检测（如果用主动触发更新的方式可以不装后者）

pip install readerwriterlock apscheduler

2. 全局缓存与服务初始化 #

from fastapi import FastAPI
from readerwriterlock import rwlock
from datetime import date
import joblib
import os

app = FastAPI()

# 全局模型缓存结构
MODEL_CACHE = {
    "current_version": None,
    "model_instance": None
}
# 读写锁，读请求不互斥，仅切换模型时加写锁
RW_LOCK = rwlock.RWLockWrite()
# 模型文件路径模板，按实际存储规则修改
MODEL_PATH_TPL = "./models/model_{date_str}.joblib"

def load_model_by_date(target_date: date):
    """按指定日期加载模型，文件不存在/加载失败返回None"""
    date_str = target_date.strftime("%Y%m%d")
    model_path = MODEL_PATH_TPL.format(date_str=date_str)
    if not os.path.exists(model_path):
        return None
    try:
        return joblib.load(model_path)
    except Exception:
        return None

# 服务启动时初始化最近可用模型
def init_model():
    # 优先加载当日模型，失败则加载昨日模型
    today = date.today()
    model = load_model_by_date(today)
    if model:
        MODEL_CACHE["current_version"] = today
        MODEL_CACHE["model_instance"] = model
        return
    yesterday = today.replace(day=today.day-1)
    model = load_model_by_date(yesterday)
    if not model:
        raise RuntimeError("无可用模型文件，服务启动失败")
    MODEL_CACHE["current_version"] = yesterday
    MODEL_CACHE["model_instance"] = model

init_model()

3. 模型更新逻辑 #

两种触发方式二选一即可：

方式一：主动触发（最优，无额外开销）

应用内负责生成模型的模块，写完当日模型文件后直接调用更新函数即可，时效性最高

def update_model():
    """尝试切换到当日版本模型，加载失败自动保留旧版本"""
    today = date.today()
    # 已经是当日版本无需重复更新
    if MODEL_CACHE["current_version"] == today:
        return
    new_model = load_model_by_date(today)
    if not new_model:
        return
    # 加写锁替换模型，避免并发读取异常
    with RW_LOCK.gen_wlock():
        MODEL_CACHE["current_version"] = today
        MODEL_CACHE["model_instance"] = new_model

方式二：定时轮询（低耦合，适合模块拆分场景）

启动后台定时任务，定期检查当日模型是否就绪，无需和模型生成模块做逻辑耦合

from apscheduler.schedulers.background import BackgroundScheduler

@app.on_event("startup")
def start_check_scheduler():
    scheduler = BackgroundScheduler()
    # 每10分钟检查一次模型更新，可根据实际需求调整间隔
    scheduler.add_job(update_model, "interval", minutes=10)
    scheduler.start()

4. 接口调用逻辑 #

接口请求时加读锁获取模型，读锁不互斥，不会影响高并发场景下的接口性能

@app.post("/predict")
def predict(input_text: str):
    # 加读锁获取模型，多个请求可同时持有读锁
    with RW_LOCK.gen_rlock():
        model = MODEL_CACHE["model_instance"]
    # 执行推理逻辑
    result = model.predict([input_text])
    return {"result": result[0]}

额外注意事项 #

• 生成模型文件时做原子操作：先写入后缀为.tmp的临时文件，写完再重命名为正式的.joblib文件，避免更新逻辑读取到未写完的损坏文件
• 可定期清理超过7天的历史模型文件，避免磁盘占用过高
• 若模型体积较大加载时间长，可以先异步加载完成后再替换全局缓存，全程不会阻塞线上请求

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