머신러닝 모델의 예측 결과를 기다리느라 답답했던 경험이 있으신가요? 특히 복잡한 모델일수록 실시간 서빙에서 느린 응답 속도는 사용자 경험을 크게 해치는 문제입니다. 사용자들은 즉각적인 반응을 기대하지만, 동일한 입력에 대해 매번 모델을 재실행하는 것은 비효율적입니다. 이번 글에서는 FastAPI와 Redis 캐싱을 활용해 이런 문제를 해결하고, 반복되는 예측 요청을 밀리초 단위로 처리하는 방법을 알아보겠습니다.
FastAPI와 Redis가 만나면 어떤 일이 일어날까요?
FastAPI는 Python으로 API를 구축할 때 뛰어난 성능과 개발 편의성을 제공하는 현대적인 웹 프레임워크입니다. 타입 힌트를 활용한 자동 검증과 Swagger UI 문서 자동 생성, 그리고 Node.js나 Go에 버금가는 비동기 처리 성능이 특징입니다.
Redis는 메모리 기반의 데이터 저장소로, 데이터베이스, 캐시, 메시지 브로커 역할을 모두 수행할 수 있습니다. 메모리에 데이터를 저장하기 때문에 읽기/쓰기 작업에서 극도로 낮은 지연시간을 보장하며, 키 만료(TTL) 기능으로 효율적인 캐시 관리가 가능합니다.
이 두 기술을 결합하면 FastAPI가 빠르고 안정적인 API 인터페이스를 제공하고, Redis가 이전 계산 결과를 저장하여 동일한 입력에 대해서는 즉시 결과를 반환할 수 있습니다. 실제로 이 방식은 응답 시간을 대폭 단축시키고, 연산 부하를 줄여 더 많은 요청을 처리할 수 있게 해줍니다.
단계별 구현 가이드
1단계: 머신러닝 모델 준비
먼저 서빙할 모델을 준비해야 합니다. 실제 환경에서는 미리 훈련된 모델을 디스크에서 불러오는 경우가 대부분입니다. 예시로 Iris 데이터셋을 사용한 간단한 분류 모델을 만들어보겠습니다.
from sklearn.datasets import load_iris
from sklearn.ensemble import RandomForestClassifier
import joblib
# Iris 데이터셋 로드 및 모델 훈련
X, y = load_iris(return_X_y=True)
# 모델 훈련
model = RandomForestClassifier().fit(X, y)
# 훈련된 모델을 디스크에 저장
joblib.dump(model, "model.joblib")
# 저장된 모델 로드 (실제 서빙 시에는 이 부분만 실행)
model = joblib.load("model.joblib")
print("모델이 로드되어 예측 준비가 완료되었습니다.")모델을 저장할 때 joblib을 사용하는 이유는 NumPy 배열을 효율적으로 처리할 수 있기 때문입니다. 중요한 점은 모델이 동일한 입력에 대해 항상 같은 결과를 반환해야 한다는 것입니다. 비결정적 모델의 경우 캐싱이 제대로 작동하지 않을 수 있습니다.
2단계: FastAPI 예측 엔드포인트 생성
이제 모델을 API로 서빙하기 위한 FastAPI 애플리케이션을 만들어보겠습니다.
from fastapi import FastAPI
import joblib
app = FastAPI()
# 애플리케이션 시작 시 모델 로드 (매 요청마다 로드하지 않도록)
model = joblib.load("model.joblib")
@app.get("/predict")
def predict(sepal_length: float, sepal_width: float, petal_length: float, petal_width: float):
""" Iris 꽃의 종류를 예측합니다. """
# 모델이 기대하는 2D 배열 형태로 특성 준비
features = [[sepal_length, sepal_width, petal_length, petal_width]]
# 예측 수행 (첫 번째 결과만 가져오기)
prediction = model.predict(features)[0]
return {"prediction": str(prediction)}이 코드에서 중요한 부분은 모델을 애플리케이션 시작 시에만 로드한다는 점입니다. 매 요청마다 모델을 다시 로드하면 성능이 크게 저하됩니다. FastAPI는 GET 방식으로 URL 파라미터를 자동으로 함수 인자로 변환해주므로 편리하게 사용할 수 있습니다.
3단계: Redis 캐싱 시스템 통합
이제 가장 중요한 캐싱 기능을 추가해보겠습니다. Redis 서버가 실행 중이어야 하므로, 먼저 Redis를 설치하고 실행해주세요.
import redis # Redis 클라이언트 임포트
# 로컬 Redis 서버에 연결 (필요시 호스트/포트 조정)
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):
"""
캐싱을 통해 반복되는 예측을 빠르게 처리합니다.
"""
# 1. 입력 파라미터로부터 고유한 캐시 키 생성
cache_key = f"{sepal_length}:{sepal_width}:{petal_length}:{petal_width}"
# 2. Redis에서 캐시된 결과 확인
cached_result = cache.get(cache_key)
if cached_result:
# 캐시 히트: 저장된 결과를 즉시 반환
return {"prediction": cached_result.decode("utf-8")}
# 3. 캐시 미스: 모델로 예측 수행
features = [[sepal_length, sepal_width, petal_length, petal_width]]
prediction = model.predict(features)[0]
# 4. 결과를 Redis에 저장 (다음 요청을 위해)
cache.set(cache_key, str(prediction))
# 5. 새로 계산된 예측 결과 반환
return {"prediction": str(prediction)}캐시 키는 입력값들을 결합하여 만들어집니다. 간단한 숫자 값들이므로 콜론으로 구분하여 연결했지만, 복잡한 입력의 경우 JSON 문자열이나 해시값을 사용하는 것이 좋습니다. 중요한 것은 동일한 입력에 대해서는 항상 같은 키가 생성되어야 한다는 점입니다.
4단계: 성능 테스트 및 측정
이제 캐싱이 실제로 얼마나 성능을 개선하는지 측정해보겠습니다. FastAPI 애플리케이션을 uvicorn main:app --reload 명령으로 실행한 후, 다음 테스트 코드를 실행해보세요.
import requests
import time
# 테스트할 샘플 입력값
params = {
"sepal_length": 5.1,
"sepal_width": 3.5,
"petal_length": 1.4,
"petal_width": 0.2
}
# 첫 번째 요청 (캐시 미스 예상, 모델 실행)
start = time.time()
response1 = requests.get("http://localhost:8000/predict", params=params)
elapsed1 = time.time() - start
print(f"첫 번째 응답: {response1.json()} (소요시간: {elapsed1:.4f}초)")
# 두 번째 요청 (캐시 히트 예상, 즉시 반환)
start = time.time()
response2 = requests.get("http://localhost:8000/predict", params=params)
elapsed2 = time.time() - start
print(f"두 번째 응답: {response2.json()} (소요시간: {elapsed2:.6f}초)")테스트를 실행하면 첫 번째 요청은 모델 실행 시간이 포함되어 상대적으로 느리고, 두 번째 요청은 캐시에서 즉시 결과를 가져와 훨씬 빠른 것을 확인할 수 있습니다.
성능 개선 효과 분석
실제 테스트 결과를 통해 캐싱의 효과를 구체적으로 살펴보겠습니다.
캐싱 없는 경우: 모든 요청이 모델을 거쳐야 하므로, 100ms가 걸리는 모델에 동일한 요청 10개가 들어오면 총 1000ms가 소요됩니다.
캐싱 적용 시: 첫 번째 요청만 100ms가 걸리고, 나머지 9개 요청은 각각 1-2ms씩만 걸리므로 총 120ms 정도로 약 8배의 성능 향상을 얻을 수 있습니다.
실제 e-commerce 환경에서는 추천 시스템에 동일한 사용자의 반복 요청이 자주 발생하는데, Redis 캐싱을 통해 마이크로초 단위의 응답 시간을 달성할 수 있었다는 사례가 있습니다. 성능 향상 폭은 모델의 복잡도와 요청 패턴에 따라 달라지지만, 복잡한 모델일수록 캐싱의 효과는 더욱 극대화됩니다.
실무 적용 시 고려사항
캐시 만료 시간 설정
실제 운영 환경에서는 데이터가 시간에 따라 변화할 수 있으므로 적절한 TTL(Time To Live) 설정이 중요합니다.
# 1시간 후 캐시 만료 설정
cache.setex(cache_key, 3600, str(prediction))메모리 관리
Redis는 메모리 기반이므로 캐시 크기를 모니터링하고 관리해야 합니다. LRU(Least Recently Used) 정책을 사용하여 메모리가 부족할 때 오래된 캐시를 자동으로 삭제할 수 있습니다.
분산 환경에서의 활용
여러 FastAPI 인스턴스가 동일한 Redis 서버를 공유할 수 있어 분산 환경에서도 효율적으로 캐싱 효과를 얻을 수 있습니다. 이는 마이크로서비스 아키텍처에서 특히 유용합니다.
보안 고려사항
민감한 데이터를 캐싱할 때는 Redis 인증 설정과 네트워크 보안을 반드시 고려해야 합니다. 또한 캐시 키에 개인정보가 포함되지 않도록 주의해야 합니다.
결론
FastAPI와 Redis의 조합은 머신러닝 모델 서빙에서 놀라운 성능 향상을 가져다줍니다. FastAPI는 빠르고 안정적인 API 인터페이스를 제공하고, Redis는 반복되는 계산을 효과적으로 캐싱하여 응답 시간을 대폭 단축시킵니다.
이 접근 방식의 핵심은 중복 계산을 피하는 것입니다. 동일한 입력에 대해 매번 모델을 실행하는 대신, 한 번 계산한 결과를 메모리에 저장해 놓고 재사용함으로써 시스템의 응답성과 효율성을 크게 개선할 수 있습니다.
특히 복잡한 딥러닝 모델이나 대용량 데이터를 처리하는 모델의 경우, 캐싱을 통한 성능 개선 효과는 더욱 극대화됩니다. 이는 단순히 응답 속도를 높이는 것을 넘어서, 동일한 하드웨어 자원으로 더 많은 사용자 요청을 처리할 수 있게 해주므로 인프라 비용 절감에도 기여합니다.
실제 프로덕션 환경에 적용할 때는 캐시 만료 시간, 메모리 관리, 보안 등을 종합적으로 고려해야 하지만, 기본적인 구현은 이 글에서 소개한 방법으로 충분히 시작할 수 있습니다. 여러분의 머신러닝 애플리케이션에도 이 기술을 적용해보시기 바랍니다.
참고자료:
Comments