Transformers documentation
Web 服务器推理
Web 服务器推理
Web 服务器是一种等待请求并按顺序处理请求的系统。这意味着你可以将 Pipeline 用作 Web 服务器上的推理引擎，因为你可以使用迭代器（类似于你如何遍历数据集）来处理每个传入的请求。
使用 Pipeline 设计 Web 服务器有其独特之处，因为两者在本质上截然不同。Web 服务器通过多路复用（多线程、异步等）来并发处理多个请求。而 Pipeline 及其底层模型并非为并行处理而设计，因为它们占用大量内存。在运行 Pipeline 或执行计算密集型任务时，最好为其提供所有可用资源。
本指南介绍如何通过使用 Web 服务器来处理接收和发送请求的较轻负载，并由单个线程来处理运行 Pipeline 的较重负载，从而规避这一差异。
创建服务器
Starlette 是一个用于构建 Web 服务器的轻量级框架。您可以使用其他任何框架，但可能需要对以下代码进行一些修改。
在开始之前，请确保已安装 Starlette 和 uvicorn。
!pip install starlette uvicorn
现在你可以在 server.py 文件中创建一个简单的 Web 服务器。关键在于只加载模型 一次，以避免不必要的副本占用内存。
创建一个流水线，用于填充被掩码的标记 [MASK]。
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(task="fill-mask",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 请求查询服务器。
curl -X POST -d "Paris is the [MASK] of France." http://localhost:8000/这应该会返回以下输出。
[{'score': 0.9969332218170166,
'token': 3007,
'token_str': 'capital',
'sequence': 'paris is the capital of france.'},
{'score': 0.0005914849461987615,
'token': 2540,
'token_str': 'heart',
'sequence': 'paris is the heart of france.'},
{'score': 0.00043787318281829357,
'token': 2415,
'token_str': 'center',
'sequence': 'paris is the center of france.'},
{'score': 0.0003378340043127537,
'token': 2803,
'token_str': 'centre',
'sequence': 'paris is the centre of france.'},
{'score': 0.00026995912776328623,
'token': 2103,
'token_str': 'city',
'sequence': 'paris is the city of france.'}]请求排队
服务器的排队机制可用于一些有趣的应用，例如动态批处理。动态批处理会先累积多个请求，然后再使用 Pipeline 进行处理。
下面的示例以伪代码编写，注重可读性而非性能，特别是您会注意到：
没有批次大小限制。
每次从队列中获取数据时，超时时间都会重置，因此在处理请求之前，您等待的时间可能会远远超过
timeout的值。这也会使第一个推理请求的延迟增加相应的时间。即使队列为空，Web 服务器也始终会等待 1ms，这是低效的，因为这段时间本可以用来启动推理。不过，如果您的用例中批处理是必需的，那么这样做可能也是合理的。最好使用单一的 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)错误检查
在生产环境中，可能会出现许多问题。例如，内存不足、存储空间不足、模型加载失败、模型配置不正确、查询语句不正确，等等。
添加 try...except 语句有助于将这些错误返回给用户以便调试。但请注意，如果不应向用户透露某些信息，这样做可能存在安全风险。
熔断
当服务器过载时，请尝试返回 503 或 504 错误，而不是让用户无限期等待。
实现这些错误类型相对简单，因为只有一个队列。请查看队列大小，以确定在服务器因负载过高而崩溃之前何时开始返回错误。
阻塞主线程
PyTorch 不支持异步，因此计算会阻塞主线程的运行。
因此，最好让 PyTorch 在独立的线程或进程中运行。当单个请求的推理时间特别长（超过 1 秒）时，这一点尤为重要，因为这意味着推理期间的每个查询都必须等待 1 秒才能收到错误。
动态批处理
动态批处理在合适的场景下非常有效，但如果你每次只发送 1 个请求，则无需使用（有关更多详细信息，请参阅 批量推理）。
Update on GitHub