본문 바로가기
Project/🚀 AI Finance 2026

#8 [모듈1-2] 금융 BERT 모델 메모리 관리

by 로띠 2026. 2. 19.
Design Pattern

Lazy Loading + Singleton — 300MB BERT 모델 메모리 관리

SentimentAnalyzer 설계 패턴 해부

Data Collect → NLP Sentiment → Feature Eng. → ML Signal

이 글은 AI_Finance(ALCHEMETRIC) 프로젝트에서 ~300MB 크기의 한국어 금융 BERT 모델(snunlp/KR-FinBert-SC)을 효율적으로 관리하기 위해 적용한 Lazy Loading + Singleton 패턴을 다룬다. 서버 시작은 100ms 이내로 유지하면서, 첫 추론 요청이 들어올 때만 모델을 로드하고, 이후에는 동일 인스턴스를 재사용하는 설계를 상세히 해부한다.

1. 문제 정의 — 300MB 모델의 딜레마

ALCHEMETRIC은 뉴스 기사의 감성을 분석하기 위해 HuggingFace의 snunlp/KR-FinBert-SC 모델을 사용한다. 한국어 금융 텍스트에 특화된 3-class(positive/negative/neutral) 분류 모델로, 파라미터 크기가 약 300MB이며 최초 로드에 3~5초가 소요된다.

이 모델을 서버에 얹는 방식에는 크게 두 가지 선택지가 있고, 둘 다 문제가 있다.

전략 동작 문제점
Eager Loading 서버 시작 시 즉시 모델 로드 startup 3~5초 지연. 뉴스 분석을 쓰지 않는 API 요청에도 300MB 메모리 점유
Per-Request Loading 매 요청마다 모델 로드 매 요청 3~5초 오버헤드. 동시 요청 시 메모리 폭발 (N x 300MB)
핵심 질문: 서버 시작은 빠르게, 모델은 딱 한 번만, 필요한 시점에 로드할 수 없을까?
→ Lazy Loading + Singleton 패턴이 정확히 이 문제를 해결한다.

2. Singleton 패턴 — 인스턴스 하나로 충분하다

센티먼트 분석기는 상태를 갖지 않는 순수 추론 도구다. 모든 요청이 동일한 모델 가중치를 사용하므로 인스턴스가 하나면 충분하다. 클래스 변수 _instance에 싱글턴을 저장하고, get_instance() 클래스 메서드로 접근한다.

class SentimentAnalyzer:
    # 한국어 금융 뉴스 센티먼트 분석기 (lazy loading singleton)

    _instance: Optional["SentimentAnalyzer"] = None

    @classmethod
    def get_instance(cls) -> "SentimentAnalyzer":
        if cls._instance is None:
            cls._instance = cls()
        return cls._instance

왜 모듈-레벨 전역 변수가 아닌 클래스 변수인가?

전역 변수 analyzer = SentimentAnalyzer()를 모듈 상단에 두면 import 시점에 인스턴스가 생성된다. 이것은 Eager Loading과 동일한 문제를 발생시킨다. 클래스 변수 + classmethod 조합은 호출 시점까지 생성을 지연시킬 수 있다.

방식 인스턴스 생성 시점 인스턴스 수
analyzer = SentimentAnalyzer() 모듈 import 시 1 (사실상 eager)
SentimentAnalyzer() 직접 호출 호출할 때마다 N (매번 새로 생성)
get_instance() Singleton 첫 호출 시 1회 항상 1

3. Lazy Loading — 필요한 순간까지 미룬다

Singleton으로 인스턴스가 하나임을 보장했지만, __init__에서 모델을 로드하면 get_instance() 호출 시점에 3~5초가 소요된다. 진짜 필요한 시점은 analyze()가 호출될 때다.

Lazy Loading은 모델 로드를 _load_model() 내부 메서드로 분리하고, analyze() 진입 시 최초 1회만 실행되도록 하는 것이다.

def __init__(self, model_name: str = DEFAULT_MODEL):
    self.model_name = model_name
    self._pipeline = None   # 아직 모델 로드 안 함

def _load_model(self):
    # 첫 호출 시 모델 로드
    if self._pipeline is not None:
        return            # 이미 로드됨 → 즉시 리턴

    try:
        from transformers import pipeline as hf_pipeline

        logger.info(f"센티먼트 모델 로딩: {self.model_name}", "_load_model")
        self._pipeline = hf_pipeline(
            "sentiment-analysis",
            model=self.model_name,
            tokenizer=self.model_name,
            max_length=512,
            truncation=True,
        )
        logger.info("센티먼트 모델 로드 완료", "_load_model")
    except Exception as e:
        raise ModelLoadError(f"센티먼트 모델 로드 실패: {e}")

Guard Clause 패턴

if self._pipeline is not None: return은 단순하지만 핵심적인 Guard Clause다. 이 한 줄이 두 번째 이후의 모든 호출에서 모델 로드를 건너뛰게 만든다. 첫 호출에서 _pipeline에 모델이 할당되면, 이후에는 즉시 리턴하므로 오버헤드가 사실상 0이다.

4. 왜 import를 지연하는가

코드를 자세히 보면 from transformers import pipeline이 파일 상단이 아니라 _load_model() 함수 내부에 위치한다. 이것은 의도적인 설계다.

transformers 라이브러리의 import 체인:
import transformers → import torch → import numpy → C extensions ...

이 체인만으로 수백 MB의 메모리와 1~2초의 시간이 소요된다. 모듈 레벨에 두면 sentiment_analyzer.py를 import하는 순간 이 비용이 발생한다.

ALCHEMETRIC 서버는 뉴스 분석뿐 아니라 주가 조회, 스케줄러 관리, ML 예측 등 다양한 API를 제공한다. 서버가 시작될 때 모든 모듈이 import되는데, transformers가 모듈 레벨에 있으면 뉴스 API를 사용하지 않더라도 torch/transformers 로드 비용을 지불해야 한다.

# sentiment_analyzer.py 상단
from typing import Optional              # 가벼움 (stdlib)
from core import get_logger, ModelLoadError  # 가벼움 (자체 모듈)
# from transformers import pipeline       <-- 여기에 없다!

# 대신 _load_model() 내부에서:
def _load_model(self):
    from transformers import pipeline as hf_pipeline  # 여기!
    ...

import 위치에 따른 비용 비교

import 위치 비용 발생 시점 서버 시작 영향
모듈 상단 (PEP 8 권장) 서버 시작 시 +1~2초 (torch 체인 로드)
함수 내부 (지연 import) 첫 _load_model() 호출 시 0초 (영향 없음)
PEP 8은 import를 파일 상단에 두라고 권장하지만, 무거운 라이브러리의 지연 import는 널리 인정되는 예외다. Django, FastAPI, pytest 등 주요 프레임워크도 동일한 패턴을 사용한다.

5. 배치 추론 — batch_size=32

뉴스 수집은 종목당 최대 50건, 시장 전체 뉴스 100건을 한 번에 가져온다. 기사마다 analyze_single()을 호출하면 파이프라인 오버헤드가 누적된다. analyze()는 텍스트 리스트를 받아 batch_size=32로 한번에 처리한다.

def analyze(self, texts: list[str], batch_size: int = 32) -> list[dict]:
    self._load_model()       # lazy load (첫 호출 시만 실행)

    if not texts:
        return []

    try:
        outputs = self._pipeline(texts, batch_size=batch_size)

        results = []
        for output in outputs:
            label = output["label"].lower()
            confidence = output["score"]
            sentiment_score = _LABEL_SCORE.get(label, 0.0) * confidence
            results.append({
                "label": label,
                "confidence": round(confidence, 4),
                "sentiment_score": round(sentiment_score, 4),
            })
        return results
    except Exception as e:
        raise InferenceError(f"센티먼트 추론 실패: {e}")

_LABEL_SCORE 매핑

KR-FinBert-SC의 3-class 출력을 단일 수치 점수로 변환하는 매핑이다. confidence를 곱해서 확신도가 낮은 예측은 0에 가깝게, 높은 예측은 극단값에 가깝게 만든다.

# label → normalized score 매핑
_LABEL_SCORE = {
    "positive":  1.0,   # 긍정 → +1.0 × confidence
    "negative": -1.0,   # 부정 → -1.0 × confidence
    "neutral":   0.0,   # 중립 →  0.0 (confidence 무관)
}
모델 출력 confidence sentiment_score 해석
positive 0.95 +0.95 강한 긍정
positive 0.55 +0.55 약한 긍정
neutral 0.90 0.0 중립 (confidence 무관)
negative 0.88 -0.88 강한 부정

6. 성능 비교 — Eager vs Lazy

세 가지 전략의 성능 수치를 비교한다. Lazy Singleton이 모든 시나리오에서 최적의 균형을 제공한다.

측정 항목 Eager Loading Per-Request Lazy Singleton
서버 시작 시간 3~5초 <100ms <100ms
첫 추론 요청 <5ms ~3초 ~3초 (모델 로드 포함)
이후 추론 요청 <5ms ~3초 <5ms
메모리 (유휴 시) ~300MB 0MB 0MB
메모리 (활성 시) ~300MB N x 300MB ~300MB
동시 요청 안전 O X (OOM 위험) O

타임라인 다이어그램

시간축으로 세 전략의 차이를 시각화하면 아래와 같다.

Eager Loading
startup: 모델 로드 3~5s
req1 <5ms
req2 <5ms
Per-Request Loading
start <0.1s
req1: load+infer ~3s
req2: load+infer ~3s
Lazy Singleton
start <0.1s
req1: load+infer ~3s (1회)
req2 <5ms

7. 시퀀스 다이어그램 — 첫 호출 vs 이후 호출

NewsService가 SentimentAnalyzer를 호출하는 전체 흐름을 시퀀스 다이어그램으로 표현한다. 첫 호출과 이후 호출의 경로 차이에 주목하자.

첫 번째 호출 (Cold Start)

NewsService SentimentAnalyzer transformers HuggingFace
get_instance() → _instance is None → cls()
analyze(texts) → _load_model()
_pipeline is None
from transformers import → import torch (~1s)
hf_pipeline(...) → model.from_pretrained() → download ~300MB
self._pipeline = ... ← pipeline ready
_pipeline(texts, batch_size=32)
results ← [{label, confidence, score}, ...]
총 소요: ~3초 (import 1s + model load 2s + inference <5ms)

두 번째 이후 호출 (Warm)

NewsService SentimentAnalyzer transformers HuggingFace
get_instance() → _instance exists → return
analyze(texts) → _load_model()
_pipeline is not None → return skip skip
_pipeline(texts, batch_size=32)
results ← [{label, confidence, score}, ...]
총 소요: <5ms (guard clause 2개 + inference만)
두 번째 호출부터는 guard clause 2개(_instance is not None + _pipeline is not None)만 통과하면 바로 추론에 진입한다. transformers와 HuggingFace에는 아예 접근하지 않는다.

8. 전체 코드 — SentimentAnalyzer

app/data_collector/sentiment_analyzer.py의 전체 코드다. 97줄에 Singleton, Lazy Loading, Batch Inference, Error Handling이 모두 담겨 있다.

"""
한국어 뉴스 센티먼트 분석기 (Phase 4)
====================================

snunlp/KR-FinBert-SC (Korean Financial Sentiment Classification)
- 3-class: positive / negative / neutral
- transformers pipeline, lazy loading, CPU inference
"""

from typing import Optional

from core import get_logger, ModelLoadError, InferenceError

logger = get_logger("sentiment_analyzer")

DEFAULT_MODEL = "snunlp/KR-FinBert-SC"

# label → normalized score 매핑
_LABEL_SCORE = {
    "positive": 1.0,
    "negative": -1.0,
    "neutral": 0.0,
}


class SentimentAnalyzer:
    # 한국어 금융 뉴스 센티먼트 분석기 (lazy loading singleton)

    _instance: Optional["SentimentAnalyzer"] = None

    @classmethod
    def get_instance(cls) -> "SentimentAnalyzer":
        if cls._instance is None:
            cls._instance = cls()
        return cls._instance

    def __init__(self, model_name: str = DEFAULT_MODEL):
        self.model_name = model_name
        self._pipeline = None

    def _load_model(self):
        # 첫 호출 시 모델 로드
        if self._pipeline is not None:
            return

        try:
            from transformers import pipeline as hf_pipeline

            logger.info(f"센티먼트 모델 로딩: {self.model_name}", "_load_model")
            self._pipeline = hf_pipeline(
                "sentiment-analysis",
                model=self.model_name,
                tokenizer=self.model_name,
                max_length=512,
                truncation=True,
            )
            logger.info("센티먼트 모델 로드 완료", "_load_model")
        except Exception as e:
            raise ModelLoadError(f"센티먼트 모델 로드 실패: {e}")

    def analyze(self, texts: list[str], batch_size: int = 32) -> list[dict]:
        """
        텍스트 배치 분석

        Returns:
            [{"label": "positive", "confidence": 0.95,
              "sentiment_score": 0.95}, ...]
        """
        self._load_model()

        if not texts:
            return []

        try:
            outputs = self._pipeline(texts, batch_size=batch_size)

            results = []
            for output in outputs:
                label = output["label"].lower()
                confidence = output["score"]
                sentiment_score = _LABEL_SCORE.get(label, 0.0) * confidence
                results.append({
                    "label": label,
                    "confidence": round(confidence, 4),
                    "sentiment_score": round(sentiment_score, 4),
                })
            return results
        except Exception as e:
            raise InferenceError(f"센티먼트 추론 실패: {e}")

    def analyze_single(self, text: str) -> dict:
        # 단일 텍스트 분석
        results = self.analyze([text])
        return results[0] if results else {
            "label": "neutral",
            "confidence": 0.0,
            "sentiment_score": 0.0,
        }

9. 서비스 계층 통합 — NewsService

NewsService.collect()는 뉴스 수집 → 센티먼트 분석 → DB 저장을 오케스트레이션하는 메서드다. SentimentAnalyzer를 어떻게 사용하는지 확인하자.

class NewsService:

    def collect(self, market, codes, ...):
        # Singleton 인스턴스 획득 (첫 호출이면 생성)
        analyzer = SentimentAnalyzer.get_instance()

        for code, name in codes:
            records = self.fetcher.fetch_stock_news(code, name, ...)
            if records:
                # 배치 분석 → 레코드에 점수 주입
                records = self._analyze_records(analyzer, records)
            ...

    @staticmethod
    def _analyze_records(analyzer, records):
        # title + description 결합 → 배치 분석
        texts = [
            r["title"] + ". " + (r.get("description") or "")
            for r in records
        ]
        sentiments = analyzer.analyze(texts)  # 배치 추론
        for rec, sent in zip(records, sentiments):
            rec["sentiment_score"] = sent["sentiment_score"]
            rec["sentiment_label"] = sent["label"]
        return records

호출 흐름 요약

Scheduler → NewsService.collect() → SentimentAnalyzer.get_instance() → analyzer.analyze(texts) → DB upsert

이 구조에서 SentimentAnalyzer는 NewsService에 의존성으로 주입되지 않고, get_instance()로 자체 관리한다. 서비스 계층은 모델 로드 타이밍을 전혀 신경 쓸 필요가 없다. analyze()를 호출하면 내부에서 알아서 로드되기 때문이다.

10. 정리 — 설계 원칙 요약

원칙 구현 효과
Singleton _instance 클래스 변수 + get_instance() 300MB 모델이 메모리에 단 1개만 존재
Lazy Loading _load_model() guard clause 서버 시작 <100ms, 첫 사용 시에만 3초 투자
Deferred Import 함수 내부 from transformers import torch 체인 로딩 비용 완전 지연
Batch Inference batch_size=32 파이프라인 오버헤드 최소화, 처리량 극대화
Error Isolation ModelLoadError / InferenceError 로드 실패와 추론 실패를 분리, 디버깅 용이
Graceful Fallback analyze_single() neutral 기본값 빈 입력/예외 시에도 안전한 기본값 반환
결론: 97줄의 sentiment_analyzer.py 하나로 300MB BERT 모델의 로드 타이밍, 메모리 점유, 에러 처리를 모두 제어한다. Lazy Singleton은 무거운 ML 모델을 웹 서버에 올릴 때 가장 먼저 고려해야 할 패턴이다.

소스 코드: app/data_collector/sentiment_analyzer.py (97 lines)  |  app/services/news_service.py (180 lines)

'Project > 🚀 AI Finance 2026' 카테고리의 다른 글

#10 [모듈1-4] 종목 뉴스 vs 시장 뉴스 분리 전략  (0) 2026.02.26
#9 [모듈1-3] 뉴스 센티먼트 분석 & 일간 리포트 시스템  (0) 2026.02.19
#7 [모듈1-1] 한국어 금융 뉴스의 감성 sentiment 분석  (0) 2026.02.19
#6 [모듈1] 데이터 분석 - 개발문서  (0) 2026.02.19
#5 [모듈0] 데이터 수집 - 개발문서  (0) 2026.02.19

관련글

티스토리툴바