使用 Transformers Pipeline 构建 Web 服务器推理引擎：Starlette 队列架构与动态批处理实战
【免费下载链接】transformers 🤗 Transformers: the model-definition framework for state-of-the-art machine learning models in text, vision, audio, and multimodal models, for both inference and training. 
项目地址: https://gitcode.com/GitHub_Trending/tra/transformers
本文围绕 🤗 Transformers 的 [Pipeline] 推理 API 与 Web 服务器的架构差异，讲解如何用 Starlette + uvicorn 搭建一个"接收请求与执行推理分离"的推理服务：由异步 Web 框架负责轻量的收发负载，由单一工作协程独占模型资源串行推理，并通过 asyncio.Queue 实现请求排队、动态批处理与过载保护。读完本文，你将掌握一份可直接落地的推理服务器骨架代码、队列与动态批处理的正确姿势，以及错误检查、熔断（503/504）、主线程阻塞等生产化关键考量。
为什么 Web 服务器不能直接"多线程跑模型"
一个 Web 服务器本质上是一个"等待请求、随到随处理"的系统。这意味着我们完全可以像在数据集上迭代推理那样，用一个迭代器逐个消费请求，把 [Pipeline] 当作推理引擎挂在 Web 服务器后面。
但两者在本质上存在冲突：
- Web 服务器天然是多路复用（multiplexed）的：多线程、异步等机制让它能同时处理大量并发请求，这是它的核心价值；
- [Pipeline] 和它底层的模型却天生不适合并行：模型推理是计算密集任务，且占用大量 RAM。如果为每个并发请求都加载一份模型副本，内存会迅速耗尽；即使共享同一模型，多线程并发执行推理也会互相争抢计算资源，导致吞吐反而下降。
因此本文的核心解决思路是分工：让 Web 服务器只做"轻活"——接收请求、排队、返回响应；让**单一的工作协程（单线程）**做"重活"——独占模型、串行推理。这样既保留了 Web 框架的高并发收发能力，又让模型在所有可用资源上高效运行。
需要说明的是：构建推理引擎是一个非常复杂的话题，不存在"唯一最佳解"。你是 CPU 还是 GPU？追求最低延迟、最高吞吐、多模型支持，还是把某个特定模型优化到极致？不同诉求对应不同方案。本文给出的是一套适合起步的优良默认值，不保证是你的最优解。
快速开始：搭建一个最小可用的推理服务器
示例选用轻量级 Web 框架 [Starlette]，理由纯粹是它足够简单；换成 FastAPI、aiohttp 等任意异步框架也完全可以，只是代码需要相应调整。先安装依赖：
pip install starlette uvicorn
然后在当前目录创建 server.py（完整代码见下）。关键原则：模型只加载一次，挂在应用级对象上，避免在服务器中产生多余副本浪费 RAM：
from starlette.applications import Starlette
from starlette.responses import JSONResponse
from starlette.routing import Route
from transformers import pipeline
import asyncio
async def homepage(request):
payload = await request.body()
string = payload.decode("utf-8")
response_q = asyncio.Queue()
await request.app.model_queue.put((string, response_q))
output = await response_q.get()
return JSONResponse(output)
async def server_loop(q):
pipe = pipeline(model="google-bert/bert-base-uncased")
while True:
(string, response_q) = await q.get()
out = pipe(string)
await response_q.put(out)
app = Starlette(
routes=[
Route("/", homepage, methods=["POST"]),
],
)
@app.on_event("startup")
async def startup_event():
q = asyncio.Queue()
app.model_queue = q
asyncio.create_task(server_loop(q))
启动服务器：
uvicorn server:app
用 POST 请求验证（示例模型是掩码语言模型，[MASK] 会被模型自动填充）：
curl -X POST -d "test [MASK]" http://localhost:8000/
#[{"score":0.7742936015129089,"token":1012,"token_str":".","sequence":"test."},…]
返回结果里 token_str 就是模型预测的填充词，sequence 是完整序列，score 是置信度。如果用更经典的句子 curl -X POST -d "Paris is the [MASK] of France." http://localhost:8000/，模型会给出 capital、heart、center、centre、city 等候选及其分数。到这里，一个"单线程独占模型 + 异步收发请求"的推理服务器骨架就完成了。
逐段拆解 server.py：三个关键部件
上面这 30 来行代码其实包含了三个各司其职的部件，理解它们才能在生产中灵活改造：
1. homepage —— 请求入口（轻负载）
async def homepage(request):
payload = await request.body()
string = payload.decode("utf-8")
response_q = asyncio.Queue()
await request.app.model_queue.put((string, response_q))
output = await response_q.get()
return JSONResponse(output)
它只做四件事：读请求体 → 解码为字符串 → 把 (输入文本, 本次请求专属的响应队列) 塞进全局任务队列 → 阻塞等待响应队列返回结果并序列化为 JSON。这里的关键技巧是为每个请求创建一个独立的 response_q 响应队列：多个并发请求可以同时挂在任务队列上等待，推理完成后各回各家，互不串扰。
2. server_loop —— 唯一的工作协程（重负载）
async def server_loop(q):
pipe = pipeline(model="google-bert/bert-base-uncased")
while True:
(string, response_q) = await q.get()
out = pipe(string)
await response_q.put(out)
它是一个永不停歇的消费者循环：从任务队列取一个请求 → 同步调用 pipe(string) 推理 → 把结果放回对应请求的响应队列。注意 pipe 在循环外只创建一次，这正是"模型只加载一次"原则的落点。它所在的协程是唯一执行推理的地方，模型资源被它独占。
3. startup_event —— 启动编排
@app.on_event("startup")
async def startup_event():
q = asyncio.Queue()
app.model_queue = q
asyncio.create_task(server_loop(q))
Starlette 应用启动时创建任务队列并挂到 app.model_queue 上（这样 homepage 才能通过 request.app.model_queue 访问到），同时用 asyncio.create_task 把 server_loop 作为后台任务拉起。app 是在模块顶层直接实例化的 Starlette 对象，因此 uvicorn server:app 才能通过模块名 + 属性名定位到它。
请求队列与动态批处理（Dynamic Batching）
队列机制的价值不止于"串行排队"，它还能实现一个高级玩法：动态批处理——先把一段时间内到达的若干请求攒起来，再一次性喂给模型做批推理。批推理借助 [Pipeline] 的 batch_size 参数完成，例如 pipe(strings, batch_size=len(strings)) 会把这批输入整体打包送入模型。
下面这段代码在原文档中特意以伪代码形式给出，为的是可读性而非性能，请先理解其思路，切勿不假思索直接上线运行（尤其要结合自身系统资源评估）：
(string, rq) = await q.get()
strings = []
queues = []
while True:
try:
(string, rq) = await asyncio.wait_for(q.get(), timeout=0.001) # 1ms
except asyncio.exceptions.TimeoutError:
break
strings.append(string)
queues.append(rq)
strings
outs = pipe(strings, batch_size=len(strings))
for rq, out in zip(queues, outs):
await rq.put(out)
这段伪代码存在几个明显问题，文档也逐一指出了：
更好的做法是设置一个单一的 1ms 截止时间：先取走第一笔请求立即入批，然后用同一个截止时间去收拢后续请求，超时即开始推理：
async def server_loop(q):
pipe = pipeline(task="fill-mask", model="google-bert/bert-base-uncased")
while True:
(string, rq) = await q.get()
strings = []
queues = []
strings.append(string)
queues.append(rq)
while True:
try:
(string, rq) = await asyncio.wait_for(q.get(), timeout=1)
except asyncio.exceptions.TimeoutError:
break
strings.append(string)
queues.append(rq)
outs = pipe(strings, batch_size=len(strings))
for rq, out in zip(queues, outs):
await rq.put(out)
即便如此，这个版本仍然没有批大小上限，实际落地时建议补充 if len(strings) >= MAX_BATCH: break 之类的约束。文档反复强调：不存在放之四海而皆准的批处理方案，一切以你的真实负载测量为准。
从源码看 Pipeline 的批处理语义
为什么 pipe(strings, batch_size=len(strings)) 能把一批字符串整体送入模型？在 src/transformers/pipelines/base.py 中，[Pipeline.__call__] 接收 batch_size 参数：当 batch_size 未显式指定时，若实例化时也未配置 self._batch_size，则默认 batch_size = 1（见该文件约第 1254-1258 行）——这也正是文档所说"Pipeline 默认不做动态批处理"的源码依据。显式传入 batch_size 后，内部会通过 DataLoader + pad_collate_fn 对输入做动态 padding 并打包成 batch 送入模型（同文件约第 1211-1213 行）。此外，pipeline 工厂函数（见 src/transformers/pipelines/init.py）还支持 device、device_map、dtype、model_kwargs 等参数，分别用于指定运行设备（如 device=0 使用 GPU、device="cpu" 强制 CPU）、借助 Accelerate 自动分配设备、加载半精度权重、以及传入量化配置——这些都是在生产推理服务器中为模型"省内存、提速度"的常用手段。
生产化必须考虑的四件事
1. 错误检查（Error Checking）
生产环境出问题的可能性远超想象：内存不足（OOM）、磁盘空间耗尽、模型加载失败、模型配置错误、请求格式非法…… 文档建议大量使用 try..except 把这些错误回显给用户，方便排查。但要注意：在安全敏感的场景下，把内部错误细节全部暴露给外部请求方本身也是一种安全风险，需要在"可调试性"与"信息最小化"之间做权衡。
2. 熔断（Circuit Breaking）
健康的 Web 服务器应该在过载时返回恰当的 4xx/5xx 错误，而不是让请求无限期挂起：例如过载时尽早返回 503（Service Unavailable），超时后返回 504（Gateway Timeout），而不是让用户一直等。由于本方案中只有一个全局队列，熔断实现起来非常容易——检查队列长度：当 q.qsize() 超过阈值时直接拒绝新请求、提前返回错误，就能避免服务器在重压下被拖垮。这是最基础的熔断起点，足以应对大多数场景。
3. 阻塞主线程（Block the Main Thread）
PyTorch 目前并不感知异步（async），它的计算会阻塞所在线程/事件循环。这意味着本文的 server_loop 在执行 pipe(string) 期间，整个事件循环的调度都会被卡住。因此文档建议：把 PyTorch 推理放到独立线程或独立进程中运行。本文没有这样做，是因为"线程 + 异步 + 队列"组合在一起会让代码复杂很多（本质上是同一件事的另一种实现方式）。但这个点很重要：当**单次推理耗时较长（>1s）**时，阻塞意味着推理期间到达的每个请求都要等 1 秒以上才能收到哪怕是一个错误响应，体验会非常糟糕。
4. 动态批处理的取舍
批处理并不天然优于逐条推理（更多关于批处理适用场景的讨论见 Pipeline 教程中的批处理推理）。它的收益强依赖具体环境：硬件、数据、模型都会影响批处理是否提速。默认情况下，Pipeline 是不开启动态批处理的（源码层面默认 batch_size=1，如前所述），因为批处理给了太多"反而变慢"的机会。但对 BLOOM 这类超大模型而言，动态批处理几乎是必须的——否则每个用户单独独占这么大的模型做推理，服务体验对所有人来说都不可接受。
总结
这份基于 Starlette 的推理服务器骨架，把"异步收发请求"与"单线程独占模型"两种范式通过一个 asyncio.Queue 优雅地缝合在一起：Web 层保持多路复用的高并发能力，推理层享受独占资源的确定性。在此基础上，你还可以按需叠加动态批处理、队列长度熔断、独立推理进程与 device/dtype/量化等内存优化手段，一步步把它打磨成符合自己场景的生产级推理服务。
进一步阅读：
- 本文英文原版：docs/source/en/pipeline_webserver.md
- Pipeline 入门与全部参数讲解：docs/source/en/pipeline_tutorial.md
- Pipeline 核心实现（__call__、批处理与迭代器支持）：src/transformers/pipelines/base.py
- 掩码填充 Pipeline 的初始化参数（top_k、targets 等）：src/transformers/pipelines/fill_mask.py
【免费下载链接】transformers 🤗 Transformers: the model-definition framework for state-of-the-art machine learning models in text, vision, audio, and multimodal models, for both inference and training. 项目地址: https://gitcode.com/GitHub_Trending/tra/transformers
创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
