能否同时训练多spaCy NER模型？FastAPI训练时API阻塞如何解决？

一、能否同时训练多个spaCy NER模型？ #

可以同时训练多个spaCy NER模型，但需注意两个核心问题：

• spaCy训练属于CPU/GPU密集型任务，并行训练会占用大量计算资源，可能导致所有训练进程速度下降，甚至服务器资源耗尽，建议根据服务器CPU核心数、显存大小合理控制并行数量。
• 原生spaCy训练逻辑是同步阻塞的，即使在FastAPI的async接口中调用，若不做异步处理，仍会阻塞主事件循环，导致其他API请求排队。

二、训练时让多API正常运行的解决方案 #

核心思路是将同步训练任务从FastAPI主事件循环中剥离，避免阻塞其他请求处理，具体方案如下：

1. 使用asyncio.to_thread将训练任务放到后台线程 #

Python 3.9+提供的asyncio.to_thread可将同步函数放到单独线程执行，不阻塞主事件循环，适合轻量到中等负载的训练任务。
示例代码：

from fastapi import FastAPI
import asyncio
import spacy
from spacy.training import Example

app = FastAPI()

# 同步的spaCy训练函数
def train_ner(train_data, base_model="en_core_web_sm"):
    nlp = spacy.load(base_model)
    ner = nlp.get_pipe("ner")
    
    # 向NER组件添加训练数据中的标签
    for _, annotations in train_data:
        for ent in annotations["entities"]:
            ner.add_label(ent[2])
    
    # 禁用其他非必要管道，提升训练速度
    with nlp.disable_pipes(*[pipe for pipe in nlp.pipe_names if pipe != "ner"]):
        optimizer = nlp.begin_training()
        for epoch in range(10):
            losses = {}
            for text, annotations in train_data:
                doc = nlp.make_doc(text)
                example = Example.from_dict(doc, annotations)
                nlp.update([example], sgd=optimizer, losses=losses)
        print(f"训练完成，损失值: {losses}")
    nlp.to_disk("./trained_ner_model")

# 训练API
@app.post("/start-training")
async def initiate_training():
    # 模拟训练数据
    train_sample = [
        ("特斯拉计划在上海新建超级工厂", {"entities": [(0, 3, "ORG"), (6, 8, "GPE")]}),
        ("苹果发布新款iPhone 15", {"entities": [(0, 2, "ORG"), (6, 15, "PRODUCT")]})
    ]
    # 将训练任务放到后台线程
    await asyncio.to_thread(train_ner, train_sample)
    return {"status": "训练任务已完成"}

# 其他API示例
@app.get("/health")
async def health_check():
    return {"status": "服务正常"}

2. 使用进程池处理CPU密集型训练任务 #

如果训练任务负载极高，或需要同时训练多个模型，建议用concurrent.futures.ProcessPoolExecutor创建进程池，绕过Python GIL限制，充分利用多核CPU资源。
示例代码：

from concurrent.futures import ProcessPoolExecutor
from fastapi import FastAPI
import asyncio
import spacy
from spacy.training import Example

app = FastAPI()
# 根据服务器CPU核心数设置最大进程数，建议为核心数的1-2倍
executor = ProcessPoolExecutor(max_workers=2)

def train_ner(train_data, base_model="en_core_web_sm"):
    # 同上述训练函数逻辑
    ...

@app.post("/train-multiple-models")
async def train_multiple():
    # 准备多组训练数据
    data_set1 = [...]
    data_set2 = [...]
    
    loop = asyncio.get_running_loop()
    # 提交多个训练任务到进程池
    task1 = loop.run_in_executor(executor, train_ner, data_set1)
    task2 = loop.run_in_executor(executor, train_ner, data_set2)
    
    # 等待所有任务完成
    await asyncio.gather(task1, task2)
    return {"status": "多模型训练完成"}

3. 优化Uvicorn服务器配置 #

启动Uvicorn时设置多工作进程，让多个进程同时处理请求，即使某个进程被训练任务占用，其他进程仍能响应API请求。
启动命令示例：

uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4

注：--workers数值建议设置为服务器CPU核心数的2倍，避免进程过多导致资源竞争。

4. 引入任务队列实现完全异步训练（进阶方案） #

如果需要更完善的任务管理（如任务状态查询、失败重试、资源调度），可以将训练任务提交到独立的任务队列（如基于Redis的简单队列），用单独的worker进程处理训练，FastAPI仅负责接收任务和返回任务ID，完全不阻塞API请求。

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