[FastAPI + AI] 8GB RAM 서버에서 OOM 없이 경량화 RAG 파이프라인 구축하기

새로운 AI 프로젝트를 시작할 때, 백엔드 개발자가 가장 먼저 마주하는 벽은 메모리 초과(OOM, Out Of Memory)와 응답 지연(Latency)입니다. 특히 8GB RAM이라는 제한된 하드웨어에서 무거운 이미지 처리와 AI 추론을 동시에 처리하려면 뼈대부터 극단적인 최적화가 필요합니다.

이 글에서는 FastAPI, ONNX Runtime, 인메모리 ChromaDB를 활용해 디스크 I/O 없이 빠르고 안전하게 동작하는 AI 비전 서버 구축 과정을 정리하고자 합니다.

1. 응답 스펙 강제하기 (Pydantic)

LLM은 JSON 형식을 깨뜨리기 일쑤입니다. 클라이언트(Spring Boot) 서버가 파싱 에러로 뻗는 것을 막기 위해 가장 먼저 응답 규격을 강제합니다.

schemas.py

from pydantic import BaseModel, Field

class FraudResponse(BaseModel):
    status: str = Field(description="FRAUD, NORMAL, 또는 SUSPICIOUS")
    fraudScore: float = Field(description="0.0에서 100.0 사이의 사기 확률 점수")
    description: str = Field(description="판단 근거 및 상세 리포트")

FastAPI 엔드포인트에 response_model=FraudResponse를 걸어두면, 이 규격에 맞지 않는 데이터는 프레임워크 단에서 알아서 걸러줍니다.

2. OOM 방어를 위한 2가지 핵심 아키텍처

① 동시성 제어 게이트 (Semaphore)

요청이 들어오는 대로 다 메모리에 올리면 8GB RAM은 1초 만에 터집니다. 한 번에 처리할 요청 수를 제한하고, 나머지는 대기열(Queue)에 세워두는 안전장치가 필수입니다.

② 글로벌 초기화

수백 MB짜리 AI 모델과 DB 클라이언트를 API가 호출될 때마다 로드하면 지연 시간이 폭발하고 메모리가 파편화됩니다. 반드시 엔드포인트 바깥(전역)에서 단 한 번만 로드해야 합니다.

main.py

import asyncio
from fastapi import FastAPI
import chromadb
from chromadb.utils import embedding_functions
import onnxruntime as ort

app = FastAPI()

# 1. 락 설정: 최대 5개의 요청만 동시 처리
concurrency_gate = asyncio.Semaphore(5)

# 2. 전역 모델 로드 (PyTorch 대신 가벼운 ONNX 및 경량 임베딩 사용)
sentence_transformer_ef = embedding_functions.SentenceTransformerEmbeddingFunction(model_name="all-MiniLM-L6-v2")
chroma_client = chromadb.Client() # 디스크를 안 쓰는 휘발성 In-memory DB
fraud_collection = chroma_client.get_or_create_collection(name="fraud_cases", embedding_function=sentence_transformer_ef)
ocr_session = ort.InferenceSession("lightweight_ocr_model.onnx")

3. 디스크 I/O 배제와 이미지 전처리

이미지를 하드디스크에 저장했다가 다시 읽는 과정(with open(...))은 엄청난 병목을 유발합니다. 들어온 파일은 io.BytesIO를 통해 무조건 RAM 위에만 띄워놓고 처리합니다.

AI 모델에 이미지를 넣기 전, 전처리 과정이 가장 중요합니다.

# API 내부 반복문 중 일부
original_img = Image.open(buffer).convert("RGB") # 투명도(Alpha) 제거, 3채널 고정
resized_img = original_img.resize((224, 224)) # AI 모델 표준 해상도. 메모리 압축
img_array = np.array(resized_img, dtype=np.float32) # 32비트 소수점 타입 고정

💡 메모리 계산 공식: $224 \times 224 \times 3 \text{(RGB)} \times 4 \text{(float32 바이트)} \approx 600\text{KB}$. 이렇게 최적화해야 여러 장을 처리해도 안전합니다.

img_array /= 255.0
img_array = np.transpose(img_array, (2, 0, 1)) # HWC -> CHW 차원 변경
input_tensor = np.expand_dims(img_array, axis=0) # Batch 차원 추가

• np.expand_dims: AI 모델은 묶음 단위로만 데이터를 받습니다. [3, 224, 224]인 3차원 데이터를 [1, 3, 224, 224]인 4차원으로 감싸서 '1장짜리 배치(Batch)'로 만들어줍니다.

4. 인메모리 벡터 DB 구축 및 ChromaDB RAG 검색

방금 이미지들에서 추출해 하나로 합친 텍스트(combined_text)를 과거 사기 수법과 비교하는 단계입니다. 이 과정은 RAG(검색 증강 생성)의 핵심 두뇌 역할을 합니다.

① 번역기와 임시 데이터베이스

서버가 켜질 때 세팅한 sentence_transformer_ef는 사람이 쓰는 자연어 텍스트를 컴퓨터가 이해할 수 있는 수학적인 좌표(벡터)로 번역해 주는 AI 모델입니다. 여기서 all-MiniLM-L6-v2 모델을 쓴 이유는, 성능이 준수하면서도 용량이 수십 메가바이트 수준으로 가벼워 8GB 램 환경에 가장 최적화되어 있기 때문입니다.

chromadb.Client()를 통해 무거운 외부 DB 대신 휘발성 메모리인 RAM 위에 빠르고 가벼운 임시 데이터베이스를 띄우고, get_or_create_collection으로 보관함을 만들어 방금 띄운 번역기를 전담으로 붙여주었습니다. 덕분에 우리가 일일이 코드를 짜지 않아도 알아서 텍스트를 좌표로 변환해 저장하고 비교해 줍니다.

② 메모리를 지키는 n_results=1

rag_results = fraud_collection.query(
    query_texts=[combined_text],
    n_results=1
)

fraud_collection.query를 실행하면, 방금 들어온 combined_text가 번역기를 거쳐 좌표로 변환된 뒤, DB 안의 과거 사기 사례들과 거리를 계산합니다. 이때 n_results=1이라고 못을 박아둔 것이 핵심입니다. 유사 사례를 여러 개 찾을 수도 있지만, 사례가 많아지면 다음 단계에서 LLM이 읽어야 할 컨텍스트(입력 토큰)가 기하급수적으로 길어집니다. 이는 곧 8GB RAM 환경에서의 메모리 터짐(OOM)과 심각한 추론 속도 저하로 직결되므로, 가장 확실한 판례 딱 1건만 가져오도록 강제했습니다.

③ 방어적 프로그래밍

크로마DB는 검색 결과를 돌려줄 때 텍스트 딱 하나만 덜렁 주는 게 아니라, 한 번에 여러 검색어를 처리하기 위해 '리스트 안의 리스트'라는 겹겹이 포장된 상자 형태로 데이터를 줍니다. 결과물 구조를 뜯어보면 {'documents': [['가장 비슷한 사기 사례 텍스트']]} 형태입니다. 그래서 진짜 글자만 빼내려면 제일 바깥 상자인 documents를 열고, 첫 번째 검색어 상자인 [0]을 열고, 마지막으로 그 안에서 1등 판례인 [0]을 열어야 하므로 [0][0]이라는 어색한 형태가 붙게 됩니다.

reference_case = ""
if rag_results['documents'] and rag_results['documents'][0]:
    reference_case = rag_results['documents'][0][0]

만약 데이터베이스가 갓 생성되어 텅 빈 상태라면, 크로마DB는 {'documents': []} 같은 빈 상자를 덜렁 돌려줍니다. 이때 다짜고짜 rag_results['documents'][0][0]으로 안쪽 상자에 손을 집어넣으려고 하면, 파이썬은 "비어있는 상자인데 없는 번호를 부르냐"며 IndexError를 발생시키고 서버를 다운시킵니다.

이를 원천 차단하기 위해 조건문으로 안전장치를 두 겹 쳤습니다. 가장 바깥쪽 상자가 있는지(rag_results['documents']), 그리고 첫 번째 결과 상자가 제대로 존재하는지(rag_results['documents'][0])를 검증한 뒤에야 안전하게 텍스트를 꺼냅니다. 파이썬 백엔드 개발 시 다차원 리스트로 인한 서버 의문사를 막는 필수 방어 습관입니다.

5. 파이프라인 완성: OCR 추출부터 RAG 검색까지

모든 로직이 조립된 최종 엔드포인트 코드입니다. 다중 이미지가 들어와도 순차적으로 처리하여 메모리를 방어하고, 추출된 텍스트를 하나로 뭉쳐 벡터 DB에 유사도를 검색합니다.

import io
import asyncio
from typing import List
from fastapi import FastAPI, File, Form, UploadFile
from schemas import FraudResponse
import chromadb
from chromadb.utils import embedding_functions
import onnxruntime as ort
from PIL import Image
import numpy as np

# ... [앞서 작성한 전역 초기화 코드] ...

@app.post("/analyze", response_model=FraudResponse)
async def analyze_images(
    files: List[UploadFile] = File(...),
    scamType: str = Form(...),
    imageType: str = Form(...)
):
    async with concurrency_gate: # 게이트 통과 대기
        image_buffers = []
        for file in files: # 디스크 I/O 없이 메모리에 적재
            image_bytes = await file.read()
            image_buffers.append(io.BytesIO(image_bytes))
            
        extracted_texts = []
        for buffer in image_buffers:
            buffer.seek(0)
            
            original_img = Image.open(buffer).convert("RGB")
            resized_img = original_img.resize((224, 224))
            img_array = np.array(resized_img, dtype=np.float32)
            
            img_array /= 255.0
            img_array = np.transpose(img_array, (2, 0, 1))
            input_tensor = np.expand_dims(img_array, axis=0)
            
            # 초경량 ONNX 추론
            ocr_result = ocr_session.run(None, {"input": input_tensor})
            extracted_texts.append(str(ocr_result[0]))
            
        # 여러 장의 결과 텍스트를 하나로 병합
        combined_text = " ".join(extracted_texts)

        # ChromaDB 유사 사례 검색 (RAG)
        rag_results = fraud_collection.query(
            query_texts=[combined_text],
            n_results=1
        )
        
        reference_case = ""
        if rag_results['documents'] and rag_results['documents'][0]:
            reference_case = rag_results['documents'][0][0]
            
        # Pydantic 모델로 감싸서 안전하게 반환
        return FraudResponse(
            status="SUSPICIOUS", 
            fraudScore=75.0, 
            description=f"유사 사례 참조: {reference_case[:20]}" if reference_case else "분석 완료"
        )