Web伺服器推理

Web伺服器是一種等待請求並按需提供服務的系統。這意味著你可以將Pipeline用作web伺服器上的推理引擎，因為你可以使用迭代器（類似於你遍歷資料集的方式）來處理每個傳入的請求。

然而，用Pipeline設計web伺服器是獨特的，因為它們本質上是不同的。Web伺服器是多路複用（多執行緒、非同步等）以併發處理多個請求的。Pipeline及其底層模型並非為並行設計，因為它們佔用大量記憶體。最好在Pipeline執行時或進行計算密集型任務時為其提供所有可用資源。

本指南展示瞭如何透過使用web伺服器處理接收和傳送請求的較輕負載，並使用單個執行緒處理執行Pipeline的較重負載來解決這種差異。

建立伺服器

Starlette是一個用於構建web伺服器的輕量級框架。你可以使用任何其他框架，但可能需要對下面的程式碼進行一些更改。

在開始之前，請確保已安裝Starlette和uvicorn。

!pip install starlette uvicorn

現在你可以在server.py檔案中建立一個簡單的web伺服器。關鍵是隻載入一次模型，以防止不必要的副本佔用記憶體。

建立一個pipeline來填充掩碼標記[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." https://: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伺服器也會始終等待1毫秒，這是低效的，因為這些時間可以用於開始推理。但如果批處理對你的用例至關重要，那可能就有意義。

最好設定一個1毫秒的單一截止時間，而不是每次獲取時都重置，如下所示。

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個請求時，它並不是必需的（更多詳細資訊請參閱批處理推理）。

< > 在 GitHub 上更新