使用管道进行 Web 服务器推理
需要理解的关键点是,我们可以使用迭代器,就像您在数据集上一样,因为 Web 服务器基本上是一个等待请求并按顺序处理它们的系统。
通常,Web 服务器是多路复用的(多线程、异步等)以并发处理各种请求。另一方面,管道(以及大多数底层模型)并不真正适合并行处理;它们占用大量 RAM,因此在运行时最好为它们提供所有可用资源,或者它是一个计算密集型作业。
我们将通过让 Web 服务器处理接收和发送请求的轻负载,并让单个线程处理实际工作来解决这个问题。此示例将使用 `starlette`。实际的框架并不重要,但如果您使用的是另一个框架,则可能需要调整或更改代码以实现相同的效果。
创建 `server.py`
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
您可以查询它:
curl -X POST -d "test [MASK]" http://localhost:8000/
#[{"score":0.7742936015129089,"token":1012,"token_str":".","sequence":"test."},...]
好了,现在你对如何创建一个 Web 服务器有了一个很好的了解!
真正重要的是我们只加载模型**一次**,这样 Web 服务器上就不会有多个模型副本。这样可以避免不必要的 RAM 使用。然后,排队机制允许你做一些花哨的事情,比如在推断之前累积一些项目以使用动态批处理。
下面的代码示例有意以伪代码的形式编写,以提高可读性。在运行此代码之前,请务必检查它是否适合你的系统资源!
(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)
同样,建议的代码是为了可读性而优化的,而不是为了成为最佳代码。首先,它没有批大小限制,这通常不是一个好主意。其次,超时在每次队列获取时都会重置,这意味着你可能需要等待超过 1 毫秒才能运行推理(将第一个请求延迟这么多)。
最好设置一个单一的 1 毫秒截止时间。
即使队列为空,它也会始终等待 1 毫秒,这可能不是最好的,因为你可能希望在队列中没有任何内容时就开始进行推理。但如果批处理对你用例至关重要,那么它可能是有意义的。同样,真的没有一个最好的解决方案。
一些你可能需要考虑的事情
错误检查
在生产环境中,可能会发生很多错误:内存不足、空间不足、模型加载失败、查询错误、查询正确但由于模型配置错误而无法运行等等。
通常,如果服务器将错误输出给用户,那将是很好的,因此添加大量 try..except
语句来显示这些错误是一个好主意。但请记住,根据你的安全环境,公开所有这些错误也可能存在安全风险。
断路器
Web 服务器在进行断路器操作时通常看起来更好。这意味着当它们过载时,它们会返回适当的错误,而不是无限期地等待查询。返回 503 错误而不是长时间等待,或者在长时间后返回 504 错误。
在建议的代码中,这相对容易实现,因为它只有一个队列。查看队列大小是开始在 Web 服务器在负载下出现故障之前返回错误的基本方法。
阻塞主线程
目前 PyTorch 并没有异步感知能力,计算会在运行时阻塞主线程。这意味着如果 PyTorch 被强制在它自己的线程/进程中运行会更好。这里没有这样做,因为代码会变得更加复杂(主要是因为线程、异步和队列无法很好地协同工作)。但最终它实现了相同的功能。
如果单个项目的推理时间很长(> 1 秒),这将非常重要,因为在这种情况下,这意味着在推理期间的每个查询都必须等待 1 秒才能收到错误。
动态批处理
通常,批处理不一定比一次传递一个项目更好(有关更多信息,请参阅 批处理详细信息)。但在正确的设置中,它可以非常有效。在 API 中,默认情况下没有动态批处理(太多导致速度下降的机会)。但是对于 BLOOM 推理(这是一个非常大的模型),动态批处理对于为每个人提供良好的体验**至关重要**。
< > GitHub 上的更新