모르는 상태로 하는 RSS&분석&RAG 프로젝트(3) Redis

2일동안 Docker 활성화를 위해 고생한 뒤로 다시 Redis를 살펴보고있다. 

 

Redis는 단순한 캐시를 넘어서 다양한 실전 시나리오에서 강력한 도구로 활용할 수 있다. 

이 프로젝트는 Django 기반의 트렌드 분석 플랫폼으로, 다음과 같은 Redis 서비스를 구현했다:

1. RAG 질의응답 캐싱 - LLM 응답 결과를 캐싱하여 비용 절감
2. 중복 데이터 수집 방지 - 이미 수집한 데이터를 추적하여 중복 방지
3. 검색 결과 캐싱 - 대시보드 검색 결과를 캐싱하여 성능 향상
4. API Rate Limiting - 외부 API 호출 제한 관리
5. 실시간 통계 집계 - 실시간 카운터 및 이벤트 추적

모든 서비스는 `RedisService` 기본 클래스를 상속받아 Redis 연결을 공유한다.

 기본 구조: RedisService 클래스

모든 Redis 서비스의 기반이 되는 기본 클래스다. 이 클래스를 상속받아 공통 Redis 연결을 사용한다.

class RedisService:
    """Redis 기본 서비스 클래스"""
    
    def __init__(self):
        self.client = redis.Redis(
            host=settings.REDIS_HOST,
            port=int(settings.REDIS_PORT),
            db=int(settings.REDIS_DB),
            decode_responses=True   응답을 자동으로 문자열로 변환
        )

설계 포인트:- `decode_responses=True`: Redis 응답을 자동으로 문자열로 변환하여 사용 편의성 향상
- Django settings에서 설정값을 읽어 유연한 환경 구성 지원

 

 

 1. RAG 질의응답 캐싱 (RAGCacheService)

 문제 상황

LLM을 사용한 RAG(Retrieval-Augmented Generation) 질의응답 시스템에서 동일한 질문에 대해 반복적으로 LLM을 호출하면 비용이 급증한다. 특히 인기 있는 질문의 경우 수백, 수천 번 호출될 수 있다.

 해결 방법

많이 호출 될 수록 트래픽이 상승하여 DB 혹은 Server가 버티지 못하고 터질 가능성이 있기에 Redis를 활용하여 처리 해야한다.
질의 텍스트를 해시화하여 캐시 키를 생성하고, LLM 응답 결과를 Redis에 저장한다.

 핵심 구현

 캐시 키 생성할 때 고정 지침

def get_cache_key(self, query: str) -> str:
 쿼리 정규화: 소문자 변환 및 공백 제거
    normalized = query.lower().strip()
    
 MD5 해시를 사용하여 고정 길이 키 생성
    query_hash = hashlib.md5(normalized.encode()).hexdigest()
    
    return f"{self.CACHE_PREFIX}{query_hash}"

설계 포인트
- 정규화: 소문자 변환과 공백 제거로 유사한 질문을 동일하게 처리
- 해시 사용 이유:
  1. 키 길이 제한 해결
  2. 특수문자 문제 해결
  3. 일관된 키 형식 보장

 

 캐시 조회 및 저장

def get_cached_response(self, query: str) -> Optional[Dict]:
    cache_key = self.get_cache_key(query)
    cached = self.client.get(cache_key)
    
    if cached:
        return json.loads(cached)
    return None

def cache_response(self, query: str, response: Dict, ttl: Optional[int] = None):
    cache_key = self.get_cache_key(query)
    ttl = ttl or self.CACHE_TTL   기본값: 24시간
    
    self.client.setex(
        cache_key,
        ttl,
        json.dumps(response, ensure_ascii=False)
    )

주요 특징:

- TTL: 기본 24시간 (86400초)
- JSON 직렬화: `ensure_ascii=False`로 한글 등 유니코드 문자 보존
- SETEX 사용: 키와 TTL을 함께 설정하여 자동 만료 관리

 

 효과

- 비용 절감: 동일 질문에 대한 LLM 호출 횟수 대폭 감소
- 응답 속도 향상: 캐시 히트 시 즉시 응답 가능
- 일관성: 동일 질문에 대해 항상 동일한 답변 제공

---

 2. 중복 데이터 수집 방지 (DuplicatePreventionService)

 문제 상황

뉴스 기사, 소셜 미디어 게시물 등을 수집할 때 같은 데이터를 여러 번 수집하면:
- 불필요한 DB 쓰기 발생
- 처리 시간 낭비
- 중복 데이터로 인한 분석 오류

 해결 방법

각 데이터의 고유 식별자(URL, 게시물 ID 등)를 Redis 키로 사용하여 수집 여부를 추적할 수 있다.

 핵심 구현

def is_already_collected(self, source: str, identifier: str) -> bool:
    key = f"{self.DUPLICATE_PREFIX}{source}:{identifier}"
    return self.client.exists(key) > 0

def mark_as_collected(self, source: str, identifier: str):
    key = f"{self.DUPLICATE_PREFIX}{source}:{identifier}"
    ttl_seconds = self.TTL_DAYS  24  3600   7일
    self.client.setex(key, ttl_seconds, "1")

설계 포인트:

- EXISTS 명령어: O(1) 시간 복잡도로 빠른 조회

시간복잡도 비교표

| 복잡도 | 속도 | n=1000일 때 연산 횟수 | 예시 |
|--------|------|----------------------|------|
| O(1) | 가장 빠름 | 1 | 배열 인덱스 접근 |
| O(log n) | 매우 빠름 | ~10 | 이진 탐색 |
| O(n) | 보통 | 1,000 | 배열 순회 |
| O(n log n) | 보통 | ~10,000 | 효율적인 정렬 |
| O(n²) | 느림 | 1,000,000 | 중첩 반복문 |
| O(2ⁿ) | 매우 느림 | 10³⁰⁰ | 재귀 피보나치 |
| O(n!) | 가장 느림 | 10²⁵⁶⁷ | 모든 순열 |

- TTL 자동 관리: 7일 후 자동 삭제로 메모리 효율성 확보
- 값은 중요하지 않음: 존재 여부만 확인하므로 "1"로 고정

 사용 예시

duplicate_check = DuplicatePreventionService()

 중복 체크
if duplicate_check.is_already_collected('news', article_url):
    continue   이미 수집했으므로 스킵

 데이터 수집
article = fetch_article(article_url)
save_to_db(article)

 수집 완료 표시
duplicate_check.mark_as_collected('news', article_url)

 효과

- 중복 방지: 동일 데이터의 반복 수집 완전 차단
- 성능 향상: DB 쓰기 횟수 대폭 감소
- 메모리 효율: 실제 데이터는 저장하지 않고 존재 여부만 표시

오늘은 이렇게 정리해 보았다. 

사실 AI가 만들어준거라 외우지도 이해하지도 못했지만 차근차근 살펴보면서 어떤식으로 활용하는지 살펴볼 것이다.