FastAPI/Uvicorn集成Simple Transformers Roberta模型后服务异常终止

我来帮你搞定这个问题！结合你提供的信息——单独推理正常但集成FastAPI后Uvicorn直接崩溃，Flask却能正常运行——核心问题大概率出在模型加载时机、Uvicorn运行模式的兼容性或者资源占用过载上。下面一步步来排查解决：

1. 先抓出具体错误（这步最关键！） #

你的错误日志只显示服务器关闭，完全看不到崩溃原因，根本没法定位问题。先给接口加上异常捕获，把详细错误打出来：

@app.get("/article_classify")
def classification(text:str):
    """用深度学习模型分类文章"""
    try:
        _,_,result = inference(text)
        return result
    except Exception as e:
        import traceback
        traceback.print_exc()  # 把完整异常栈打印到控制台
        return {"error": str(e), "details": traceback.format_exc()}

重新启动服务再调用接口，就能看到到底是内存爆了、线程冲突还是路径问题导致的崩溃。

2. 调整模型加载时机，适配FastAPI的生命周期 #

你现在是把模型直接全局加载的（model = MultiLabelClassificationModel(...)写在脚本最外面），在Uvicorn的多进程/多线程模式下，很可能触发重复加载或者资源竞争。建议用FastAPI的启动事件统一加载模型，确保整个服务生命周期里只初始化一次：

from fastapi import FastAPI

app = FastAPI()
model = None  # 先声明全局变量
classes = [...]  # 这里放你的分类标签列表

@app.on_event("startup")
def load_model():
    global model
    model_name = "checkpoint-3380-epoch-20"
    model = MultiLabelClassificationModel("roberta", f"src/outputs/{model_name}")
    print("模型加载完成！")

def inference(input_text):
    """对单条/多条文本做推理"""
    global model
    all_tags =[]
    if isinstance(input_text,str):
        print("before")
        result ,output = model.predict([input_text])
        tags=[]
        for idx,each in enumerate(result[0]):
            if each==1:
                tags.append(classes[idx])
        all_tags.append(tags)
    elif isinstance(input_text,list):
        result ,output = model.predict(input_text)
        for res in result :
            tags=[]
            for idx,each in enumerate(res):
                if each==1:
                    tags.append(classes[idx])
            all_tags.append(tags)
    return result,output,all_tags

@app.get("/article_classify")
def classification(text:str):
    try:
        _,_,result = inference(text)
        return result
    except Exception as e:
        import traceback
        traceback.print_exc()
        return {"error": str(e)}

3. 强制Uvicorn用单进程运行 #

Uvicorn默认可能会启动多进程（尤其是生产环境配置），如果你的模型体积大，多进程加载会直接把内存/GPU显存占满，系统就直接把进程kill了。修改启动命令，强制只用一个进程：

uvicorn --host 0.0.0.0 --port 5000 --workers 1 src.main:app

如果是GPU环境，还可以加上--limit-concurrency 1限制并发请求，避免显存过载：

uvicorn --host 0.0.0.0 --port 5000 --workers 1 --limit-concurrency 1 src.main:app

4. 排查模型推理的线程安全问题 #

Simple Transformers底层依赖Hugging Face的Transformers库，部分模型在多线程环境下可能存在线程不安全的情况（比如权重缓存、内部状态更新冲突）。如果是这个问题，可以试试两种办法：

• 给推理过程加锁，确保同一时间只有一个请求调用模型：

import threading
model_lock = threading.Lock()

def inference(input_text):
    global model, model_lock
    all_tags =[]
    with model_lock:  # 加锁避免多线程竞争
        if isinstance(input_text,str):
            print("before")
            result ,output = model.predict([input_text])
            tags=[]
            for idx,each in enumerate(result[0]):
                if each==1:
                    tags.append(classes[idx])
            all_tags.append(tags)
        elif isinstance(input_text,list):
            result ,output = model.predict(input_text)
            for res in result :
                tags=[]
                for idx,each in enumerate(res):
                    if each==1:
                        tags.append(classes[idx])
                all_tags.append(tags)
    return result,output,all_tags

• 改用异步接口+单独线程执行推理：把接口改成异步的，用asyncio.to_thread把推理逻辑放到单独线程，避免干扰主事件循环：

import asyncio

@app.get("/article_classify")
async def classification(text:str):
    try:
        _,_,result = await asyncio.to_thread(inference, text)
        return result
    except Exception as e:
        import traceback
        traceback.print_exc()
        return {"error": str(e)}

5. 检查资源占用情况 #

如果上面的步骤都没用，就得看看服务器的内存/GPU显存是不是不够用了：

• CPU环境：用top命令看内存占用，推理时是不是直接把内存跑满了；
• GPU环境：用nvidia-smi看显存占用，确认模型推理时有没有超出显存限制（可以先试试加device="cpu"强制用CPU跑，看会不会崩溃，排除GPU的问题）。

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