譯者 | 李睿

審校 | 重樓

本文介紹了如何使用FastAPI和Redis緩存加速機(jī)器學(xué)習(xí)模型服務(wù)。FastAPI作為高性能Web框架用于構(gòu)建API，Redis作為內(nèi)存中的數(shù)據(jù)結(jié)構(gòu)存儲(chǔ)系統(tǒng)作為緩存層。通過(guò)集成FastAPI和Redis，系統(tǒng)能快速響應(yīng)重復(fù)請(qǐng)求，避免冗余計(jì)算，顯著降低延遲和CPU負(fù)載。此外還詳細(xì)闡述了實(shí)現(xiàn)步驟，包括加載模型、創(chuàng)建FastAPI端點(diǎn)、設(shè)置Redis緩存及測(cè)試性能提升。

你是否因?yàn)?/span>等待機(jī)器學(xué)習(xí)模型返回預(yù)測(cè)結(jié)果而耗費(fèi)過(guò)長(zhǎng)時(shí)間？很多人都有過(guò)這樣的經(jīng)歷。機(jī)器學(xué)習(xí)模型在實(shí)時(shí)服務(wù)時(shí)可能會(huì)非常緩慢，尤其是那些大型且復(fù)雜的機(jī)器學(xué)習(xí)模型。另一方面，用戶(hù)希望得到即時(shí)反饋。因此，這使得延遲問(wèn)題愈發(fā)凸顯。從技術(shù)層面來(lái)看，最主要的問(wèn)題之一是當(dāng)相同的輸入反復(fù)觸發(fā)相同的緩慢過(guò)程時(shí)，會(huì)出現(xiàn)冗余計(jì)算。本文將展示如何解決這個(gè)問(wèn)題，因此將構(gòu)建一個(gè)基于FastAPI的機(jī)器學(xué)習(xí)服務(wù)，并集成Redis緩存，以便在毫秒級(jí)的時(shí)間內(nèi)迅速返回重復(fù)的預(yù)測(cè)結(jié)果。

什么是FastAPI？

FastAPI是一個(gè)基于Python的現(xiàn)代Web框架，用于構(gòu)建 API。它使用Python的類(lèi)型提示進(jìn)行數(shù)據(jù)驗(yàn)證，并使用Swagger UI和ReDoc自動(dòng)生成交互式API文檔。FastAPI基于Starlette和Pydantic構(gòu)建，支持異步編程，使其性能可與Node.js和Go相媲美。其設(shè)計(jì)有助于快速開(kāi)發(fā)健壯的、生產(chǎn)就緒的API，使其成為將機(jī)器學(xué)習(xí)模型部署為可擴(kuò)展的RESTful服務(wù)的絕佳選擇。

什么是Redis？

Redis（Remote Dictionary Server）是一個(gè)開(kāi)源的內(nèi)存數(shù)據(jù)結(jié)構(gòu)存儲(chǔ)系統(tǒng)，其功能包括數(shù)據(jù)庫(kù)、緩存和消息代理。通過(guò)將數(shù)據(jù)存儲(chǔ)在內(nèi)存中，Redis為讀寫(xiě)操作提供了超低延遲，使其成為緩存頻繁或計(jì)算密集型任務(wù)（例如機(jī)器學(xué)習(xí)模型預(yù)測(cè)）的理想選擇。它支持各種數(shù)據(jù)結(jié)構(gòu)，包括字符串、列表、集合和散列，并提供密鑰過(guò)期（TTL）等功能，以實(shí)現(xiàn)高效的緩存管理。

為什么要結(jié)合FastAPI和Redis？

將FastAPI與Redis集成，可以創(chuàng)建一個(gè)響應(yīng)速度快、效率高的系統(tǒng)。FastAPI作為處理API請(qǐng)求的快速且可靠的接口，而Redis則作為緩存層可以存儲(chǔ)之前計(jì)算的結(jié)果。當(dāng)再次接收到相同的輸入時(shí)，可以立即從Redis檢索結(jié)果，無(wú)需重新計(jì)算。這種方法降低了延遲，減輕了計(jì)算負(fù)載，并提高了應(yīng)用程序的可擴(kuò)展性。在分布式環(huán)境中，Redis充當(dāng)可由多個(gè)FastAPI實(shí)例訪問(wèn)的集中式緩存，使其非常適用于生產(chǎn)級(jí)機(jī)器學(xué)習(xí)部署。

接下來(lái)，深入了解如何實(shí)現(xiàn)一個(gè)使用Redis緩存提供機(jī)器學(xué)習(xí)模型預(yù)測(cè)的FastAPI應(yīng)用程序。這種設(shè)置能夠確保針對(duì)相同輸入的重復(fù)請(qǐng)求能夠迅速?gòu)木彺嬷蝎@取服務(wù)，從而大幅減少計(jì)算時(shí)間，并縮短響應(yīng)時(shí)間。其實(shí)現(xiàn)步驟如下：

• 加載預(yù)訓(xùn)練模型
• 創(chuàng)建FastAPI預(yù)測(cè)端點(diǎn)
• 設(shè)置Redis緩存
• 測(cè)試和衡量性能提升

以下詳細(xì)地了解這些步驟。

步驟1：加載預(yù)訓(xùn)練模型

首先，假設(shè)已經(jīng)擁有一個(gè)訓(xùn)練有素的機(jī)器學(xué)習(xí)模型，并準(zhǔn)備將其投入部署。在實(shí)際應(yīng)用中，大多數(shù)機(jī)器學(xué)習(xí)模型都是離線訓(xùn)練的（例如scikit-learn模型，TensorFlow/Pytorch模型等），并保存到磁盤(pán)中，然后加載到服務(wù)應(yīng)用程序中。在這個(gè)示例中，將創(chuàng)建一個(gè)簡(jiǎn)單的scikit-learn分類(lèi)器，它將在著名的Iris flower數(shù)據(jù)集上進(jìn)行訓(xùn)練，并使用joblib庫(kù)保存。如果已經(jīng)保存了一個(gè)模型文件，可以跳過(guò)訓(xùn)練步驟直接加載它。以下介紹如何訓(xùn)練一個(gè)模型，然后加載它進(jìn)行服務(wù)：

from sklearn.datasets import load_iris
from sklearn.ensemble import RandomForestClassifier
import joblib
# Load example dataset and train a simple model (Iris classification)
X, y = load_iris(return_X_y=True)
# Train the model
model = RandomForestClassifier().fit(X, y)
# Save the trained model to disk
joblib.dump(model, "model.joblib")
# Load the pre-trained model from disk (using the saved file)
model = joblib.load("model.joblib")
print("Model loaded and ready to serve predictions.")

在以上的代碼中，使用了scikit-learn的內(nèi)置Iris數(shù)據(jù)集訓(xùn)練了一個(gè)隨機(jī)森林分類(lèi)器，然后將該模型保存到一個(gè)名為model.joblib的文件中。之后，使用joblib.load方法將其重新加載。joblib庫(kù)在保存scikit-learn模型時(shí)非常常見(jiàn)，主要是因?yàn)樗瞄L(zhǎng)處理模型內(nèi)的NumPy數(shù)組。隨后，就有了一個(gè)可以預(yù)測(cè)新數(shù)據(jù)的模型對(duì)象。不過(guò)需要注意的是，在這里使用任何預(yù)訓(xùn)練的模型，使用FastAPI提供服務(wù)的方式以及緩存的結(jié)果或多或少是相同的。唯一的問(wèn)題是，模型應(yīng)該有一個(gè)預(yù)測(cè)方法，該方法接受一些輸入并產(chǎn)生結(jié)果。此外，確保每次輸入相同的數(shù)據(jù)時(shí)，都能給出一致的預(yù)測(cè)結(jié)果（即模型需具備確定性）。如果不是這樣，緩存對(duì)于非確定性模型來(lái)說(shuō)將會(huì)出現(xiàn)問(wèn)題，因?yàn)樗鼘⒎祷夭徽_的結(jié)果。

步驟2：創(chuàng)建FastAPI預(yù)測(cè)端點(diǎn)

現(xiàn)在已經(jīng)有了一個(gè)訓(xùn)練好的模型，可以通過(guò)API來(lái)使用它。我們將使用FASTAPI創(chuàng)建一個(gè)Web服務(wù)器來(lái)處理預(yù)測(cè)請(qǐng)求。FASTAPI可以很容易地定義端點(diǎn)并將請(qǐng)求參數(shù)映射到Python函數(shù)參數(shù)。在這個(gè)示例中，將假設(shè)模型需要四個(gè)特征作為輸入。并將創(chuàng)建一個(gè)GET端點(diǎn)/預(yù)測(cè)，該端點(diǎn)/預(yù)測(cè)接受這些特征作為查詢(xún)參數(shù)并返回模型的預(yù)測(cè)。

from fastapi import FastAPI
import joblib
app = FastAPI()
# Load the trained model at startup (to avoid re-loading on every request)
model = joblib.load("model.joblib") # Ensure this file exists from the training step
@app.get("/predict")
def predict(sepal_length: float, sepal_width: float, petal_length: float, petal_width: float):
 """ Predict the Iris flower species from input measurements. """
 # Prepare the features for the model as a 2D list (model expects shape [n_samples, n_features])
 features = [[sepal_length, sepal_width, petal_length, petal_width]]
 # Get the prediction (in the iris dataset, prediction is an integer class label 0,1,2 representing the species)
 prediction = model.predict(features)[0] # Get the first (only) prediction
 return {"prediction": str(prediction)}

在以上代碼中，成功創(chuàng)建了一個(gè)FastAPI應(yīng)用程序，并在執(zhí)行該文件之后啟動(dòng)API服務(wù)器。FastAPI對(duì)于Python來(lái)說(shuō)非常快，因此它可以輕松地處理大量請(qǐng)求。為避免在每次請(qǐng)求時(shí)都重復(fù)加載模型（這一操作會(huì)顯著降低性能），在程序啟動(dòng)時(shí)就將模型加載到內(nèi)存中，以便隨時(shí)調(diào)用。隨后使用@app創(chuàng)建了一個(gè)/predict端點(diǎn)。GET使測(cè)試變得簡(jiǎn)單，因?yàn)榭梢栽赨RL中傳遞內(nèi)容，但在實(shí)際項(xiàng)目中，可能會(huì)想要使用POST，特別是在發(fā)送大型或復(fù)雜的輸入（如圖像或JSON）時(shí)。

這個(gè)函數(shù)接受4個(gè)輸入?yún)?shù)：sepal_length、sepal_width、petal_length和petal_width， FastAPI會(huì)自動(dòng)從URL中讀取它們。在函數(shù)內(nèi)部，將所有輸入放入一個(gè)2D列表中（因?yàn)閟cikit-learn只接受二維數(shù)組作為輸入），然后調(diào)用model.predict()，它會(huì)返回一個(gè)列表。然后將其作為JSON返回，例如{ “prediction”: “...”}。

該系統(tǒng)現(xiàn)在已經(jīng)能正常運(yùn)行，可以使用uvicorn main:app–reload命令運(yùn)行它，然后訪問(wèn) /predict 端點(diǎn)并獲取結(jié)果。然而，再次發(fā)送相同的輸入，它仍然會(huì)再次運(yùn)行模型，這顯然不夠高效，所以下一步是添加Redis來(lái)緩存之前的結(jié)果，從而避免重復(fù)計(jì)算。

步驟3：設(shè)置Redis緩存

為了緩存模型輸出，將使用Redis。首先，確保Redis服務(wù)器正在運(yùn)行。你可以在本地安裝，或者直接運(yùn)行Docker容器；在默認(rèn)情況下，它通常運(yùn)行在端口6379上，并使用Python Redis庫(kù)與服務(wù)器通信。

所以，其思路很簡(jiǎn)單：當(dāng)請(qǐng)求進(jìn)來(lái)時(shí)，創(chuàng)建一個(gè)表示輸入的唯一鍵。然后檢查該鍵是否存在于Redis中；如果那個(gè)鍵已經(jīng)存在，這意味著之前已經(jīng)緩存了這個(gè)，所以只返回保存的結(jié)果，不需要再次調(diào)用模型。如果沒(méi)有，則執(zhí)行model.predict，獲得輸出，將其保存在Redis中，并返回預(yù)測(cè)。

現(xiàn)在更新FastAPI應(yīng)用程序來(lái)添加這個(gè)緩存邏輯。

!pip install redis
import redis # New import to use Redis
# Connect to a local Redis server (adjust host/port if needed)
cache = redis.Redis(host="localhost", port=6379, db=0)
@app.get("/predict")
def predict(sepal_length: float, sepal_width: float, petal_length: float, petal_width: float):
 """
 Predict the species, with caching to speed up repeated predictions.
 """
 # 1. Create a unique cache key from input parameters
 cache_key = f"{sepal_length}:{sepal_width}:{petal_length}:{petal_width}"
 # 2. Check if the result is already cached in Redis
 cached_val = cache.get(cache_key)
 if cached_val:
 # If cache hit, decode the bytes to a string and return the cached prediction
 return {"prediction": cached_val.decode("utf-8")}
 # 3. If not cached, compute the prediction using the model
 features = [[sepal_length, sepal_width, petal_length, petal_width]]
 prediction = model.predict(features)[0]
 # 4. Store the result in Redis for next time (as a string)
 cache.set(cache_key, str(prediction))
 # 5. Return the freshly computed prediction
 return {"prediction": str(prediction)}

在以上的代碼中添加了Redis。首先，使用redis.Redis（）創(chuàng)建了一個(gè)客戶(hù)端，它連接到Redis服務(wù)器。在默認(rèn)情況下使用db=0。然后，通過(guò)連接輸入值來(lái)創(chuàng)建一個(gè)緩存鍵。在這里，它之所以有效，是因?yàn)檩斎胧呛?jiǎn)單的數(shù)字，但對(duì)于復(fù)雜的數(shù)字，最好使用散列或JSON字符串。每個(gè)輸入的鍵必須是唯一的。因此使用了cache.get（cache_key）。如果它找到相同的鍵，它就返回這個(gè)鍵，這使其速度更快，并且不需要重新運(yùn)行模型。但是如果在緩存中沒(méi)有找到，需要運(yùn)行模型并獲得預(yù)測(cè)結(jié)果。最后，使用cache.set（）保存在Redis中。而當(dāng)相同的輸入在下次到來(lái)時(shí)，因?yàn)樗呀?jīng)存在，因?yàn)榫彺鎸?huì)很快。

步驟4：測(cè)試和衡量性能提升

現(xiàn)在，F(xiàn)astAPI應(yīng)用程序正在運(yùn)行并連接到Redis，現(xiàn)在是測(cè)試緩存如何提高響應(yīng)時(shí)間的時(shí)候了。在這里，演示如何使用Python的請(qǐng)求庫(kù)使用相同的輸入兩次調(diào)用API，并衡量每次調(diào)用所花費(fèi)的時(shí)間。此外，需要確保在運(yùn)行測(cè)試代碼之前啟動(dòng)FastAPI：

import requests, time
# Sample input to predict (same input will be used twice to test caching)
params = {
"sepal_length": 5.1,
"sepal_width": 3.5,
"petal_length": 1.4,
"petal_width": 0.2
}
# First request (expected to be a cache miss, will run the model)
start = time.time()
response1 = requests.get("http://localhost:8000/predict", params=params)
elapsed1 = time.time() - start
print("First response:", response1.json(), f"(Time: {elapsed1:.4f} seconds)")

# Second request (same params, expected cache hit, no model computation)
start = time.time()
response2 = requests.get("http://localhost:8000/predict", params=params)
elapsed2 = time.time() - start
print("Second response:", response2.json(), f"(Time: {elapsed2:.6f}seconds)")

當(dāng)運(yùn)行這個(gè)命令時(shí)，應(yīng)該看到第一個(gè)請(qǐng)求返回一個(gè)結(jié)果。然后第二個(gè)請(qǐng)求返回相同的結(jié)果，但明顯速度更快。例如，可能會(huì)發(fā)現(xiàn)第一次調(diào)用花費(fèi)了幾十毫秒的時(shí)間（取決于模型的復(fù)雜性），而第二次調(diào)用可能只有幾毫秒或更少的時(shí)間。在使用輕量級(jí)模型的簡(jiǎn)單演示中，差異可能很小（因?yàn)槟Ｐ捅旧硭俣群芸欤珜?duì)于更大的模型來(lái)說(shuō)，其效果非常顯著。

比較

為了更好地理解這一點(diǎn)，可以了解一下取得的成果：

• 無(wú)緩存：每個(gè)請(qǐng)求，即使是相同的請(qǐng)求，都會(huì)命中模型。如果模型每次預(yù)測(cè)需要100毫秒，那么10個(gè)相同的請(qǐng)求仍然需要約1000毫秒。
• 使用緩存：第一個(gè)請(qǐng)求需要全部命中（100毫秒），但接下來(lái)的9個(gè)相同的請(qǐng)求可能每個(gè)需要1～2毫秒（只是一個(gè)Redis查找和返回?cái)?shù)據(jù)）。因此，這10個(gè)請(qǐng)求可能總共120毫秒，而不是1000毫秒，在這種情況下，速度提高了8倍。

在實(shí)際實(shí)驗(yàn)中，緩存可以帶來(lái)數(shù)量級(jí)的性能提升。例如，在電子商務(wù)領(lǐng)域中，使用Redis意味著在微秒內(nèi)返回重復(fù)請(qǐng)求的建議，而不必使用完整的模型服務(wù)管道重新計(jì)算它們。性能提升將取決于模型推理的成本。模型越復(fù)雜，從緩存重復(fù)調(diào)用中的收益越大。這也取決于請(qǐng)求模式：如果每個(gè)請(qǐng)求都是唯一的，緩存將無(wú)法發(fā)揮作用（沒(méi)有重復(fù)請(qǐng)求可以從內(nèi)存中提供服務(wù)），但是許多應(yīng)用程序確實(shí)會(huì)看到重疊的請(qǐng)求（例如，流行的搜索查詢(xún)，推薦的項(xiàng)目等）。

為了驗(yàn)證Redis緩存是否正常存儲(chǔ)鍵值對(duì)可以直接對(duì)Redis緩存進(jìn)行檢查。

結(jié)論

本文展示了FastAPI和Redis如何協(xié)同工作以加速機(jī)器學(xué)習(xí)模型服務(wù)。FastAPI提供了一個(gè)快速且易于構(gòu)建的API層用于提供預(yù)測(cè)服務(wù)，Redis添加了一個(gè)緩存層，可以顯著減少重復(fù)計(jì)算的延遲和CPU負(fù)載。通過(guò)避免重復(fù)的模型調(diào)用，提高了響應(yīng)速度，并使系統(tǒng)能夠使用相同的資源處理更多的請(qǐng)求。

原文標(biāo)題：Accelerate Machine Learning Model Serving With FastAPI and Redis Caching，作者：Janvi Kumari