[Python] Redis를 활용한 중복 추천 방지 로직 상세 분석

이전 글에서는 Redis를 활용한 장소 캐싱 시스템(SpotRedisService)을 설계한 이유와 개요를 설명했다.
이번 글에서는 추천 시스템에서 Redis를 활용하는 구체적인 코드를 살펴보겠다.

여기서는 create_recommendation_restaurant 함수를 중심으로 Redis와 DB를 활용한 중복 추천 방지 로직을 분석하고,
각 부분별로 어떤 역할을 하는지 설명하겠다.

 

create_recommendation_restaurant 함수 개요

이 함수는 에이전트가 사용자에게 추천할 식당 목록을 생성하는 핵심 로직을 담당한다.

Redis 조회 흐름도

 

---

DB 조회 흐름도

 

주요 기능

1. 사용자 정보 및 기존 추천 데이터 확인 (Redis & DB 조회)
2. 입력 데이터 전처리
3. Crew AI 실행 (추천 생성)
4. 결과 처리 및 Redis 캐싱 (중복 방지 저장)

 

코드 설명

 콘솔에 출력되는 로그가 많아 필요한 정보를 빠르게 찾기 어려울 수 있기 때문에, 각 로그 메시지에 이모티콘을 추가하여 가독성을 높이고 특정 정보를 쉽게 식별할 수 있도록 했다.

# logging 아이콘
# 🔵: 전달받은 데이터 유무 확인
# 🟢: 새로 생성된 일정이거나 plan_id 없는 경우(redis)
# 🟡: 기존 일정 수정(DB)
# 🟣: redis

 

1. 데이터 유효성 검사 및 초기 변수 설정

먼저, existing_spot_names 리스트를 초기화하여 중복 추천을 방지할 장소 목록을 저장할 준비를 한다.
또한, member_id 변수를 설정해 사용자의 고유 ID를 저장할 수 있도록 한다.

@time_check
async def create_recommendation_restaurant(
    self,
    input_data: dict,
    session: AsyncSession = None,
    redis_client: Redis = None,
    prompt: Optional[str] = None,
) -> dict:
    try:
        existing_spot_names = []
        member_id = None

        # 데이터 유무 확인
        print(f"🔵 email 존재: {bool(input_data.get('email'))}")
        print(f"🔵 session 존재: {bool(session)}")
        print(f"🔵 redis 존재: {bool(redis_client)}")
        print(f"🔵 plan_id 없음: {not input_data.get('plan_id')}")

• input_data → 사용자가 입력한 요청 데이터
• session → 데이터베이스(DB) 세션
• redis_client → Redis 클라이언트
• prompt → AI 모델이 참고할 프롬프트

 

2. 사용자 정보 및 기존 추천 데이터 확인

사용자의 고유 ID(member_id)를 찾고, 이를 활용해 Redis 및 DB에서 추천 이력을 조회할 준비를 한다.

# member_id 조회 및 Redis/DB 로직 실행
if input_data.get("email") and session:
    member_id = await get_memberId_by_email(input_data["email"], session)
    print(f"💥💥 member_id 조회됨: {bool(member_id)}")

• 사용자의 이메일을 기반으로 member_id를 조회한다.
• 이를 통해 개별 사용자별로 추천 기록을 관리할 수 있다.

 

3. 기존 추천된 장소 조회 (Redis & DB 활용)

(1) 새로운 일정이면 Redis에서 조회

이 단계에서는 사용자가 새로운 여행 일정을 요청한 경우(plan_id가 없는 경우), Redis에서 이전에 추천된 장소 목록을 조회하여 중복 추천을 방지한다.
이를 위해 SpotRedisService.get_spots() 함수를 사용하여 Redis에 저장된 데이터를 가져온다.

if not input_data.get("plan_id"):
    # 새로 생성된 일정이거나 plan_id 없는 경우 - Redis 사용
    logger.info("🟢 새로 생성된 일정: Redis 사용 로직 실행 시작")
    try:
        redis_service = SpotRedisService(redis_client)
        redis_excluded_spots = await redis_service.get_spots(
            member_id=member_id,
            main_location=input_data["main_location"],
            category=SpotCategory.RESTAURANT,
        )
        if redis_excluded_spots:
            existing_spot_names = redis_excluded_spots
            logger.info(f"🟢 Redis에서 가져온 제외 식당 목록: {redis_excluded_spots}")
    except Exception as e:
        logger.error(f"Redis 조회 중 오류 발생: {e}")

 

(2) 기존 일정 수정이면 DB에서 조회

사용자가 기존 일정을 수정하려는 경우, DB에서 해당 일정에 추천된 장소 목록을 가져온다.
이때, get_member_plan_spots() 함수를 사용하여 특정 일정(plan_id)에 대한 장소 데이터를 조회한다.
또한, DB에서도 중복된 장소가 추천되지 않도록 기존 추천 데이터를 필터링하여 반영한다.

else:
    # 기존 일정 수정의 경우 - DB 사용
    current_plan_id = input_data.get("plan_id")

    try:
        # 현재 plan이 해당 member의 것인지 확인
        plan_spots_with_spot_info = await get_member_plan_spots(
            current_plan_id, member_id, session
        )

        if not plan_spots_with_spot_info:
            latest_plan = await get_latest_plan(member_id, session)
            if latest_plan:
                plan_spots_with_spot_info = await get_member_plan_spots(
                    latest_plan.id, member_id, session
                )
                logger.info(f"🟡 최신 plan_id 사용: {latest_plan.id}")
        else:
            logger.info(f"🟡 전달받은 plan_id 사용: {current_plan_id}")

        if plan_spots_with_spot_info and "detail" in plan_spots_with_spot_info:
            existing_spot_names = [
                item["spot"].kor_name for item in plan_spots_with_spot_info["detail"]
            ]
            logger.info(f"🟡 DB에서 가져온 기존 장소들: {existing_spot_names}")
    except Exception as e:
        logger.error(f"🟡 DB 장소 조회 중 오류 발생: {e}")
        traceback.print_exc()

 

4. AI 추천 생성 및 데이터 전처리

기존에 추천된 장소(existing_spot_names)를 AI 추천 시스템에 전달하여 중복 추천을 방지한다.

# 1. 입력 데이터 전처리
processed_input, prompt_text = self._process_input(input_data, prompt)
processed_input["existing_spot_names"] = existing_spot_names
processed_input["prompt_text"] = prompt_text

 

5. 에이전트 실행 및 추천 결과 처리

에이전트가 새로운 장소를 추천하는 로직을 실행한 후, 생성된 결과를 _process_result() 함수를 통해 사용자가 확인할 수 있도록 후처리한다.

# Crew 실행 (AI 추천 시스템)
crew = Crew(
    agents=list(self.agents.values()),
    tasks=list(self.tasks.values()),
    process=Process.sequential,
    verbose=True,
    memory=True,
)

# 결과 실행 및 처리
result = await crew.kickoff_async(inputs=processed_input)
processed_result = self._process_result(result, processed_input)
print(f"⭐️ processed_result: {processed_result}")

 

6. 추천 결과를 Redis에 저장 (중복 방지 적용)

새로운 일정(plan_id가 없음)일 경우, 추천된 장소를 Redis에 저장하여 다음 추천 요청 시 중복을 방지한다.

# plan_id가 없는 경우, 결과를 Redis에 저장
if not input_data.get("plan_id") and redis_client:
    try:
        redis_service = SpotRedisService(redis_client)
        restaurants_to_save = [
            spot["kor_name"] for spot in processed_result.get("spots", [])
        ]
        print(f"🟢 spots to save: {restaurants_to_save}")

        await redis_service.add_spots(
            member_id=member_id,
            main_location=input_data["main_location"],
            category=SpotCategory.RESTAURANT,
            spots=restaurants_to_save,
        )
    except Exception as e:
        logger.error(f"Redis 저장 중 오류 발생: {e}")
        traceback.print_exc()

 

 

마무리

기존 DB에서 일정 수정 시 plan_id를 기준으로 조회하는 로직을 최신 plan_id까지 고려해 조건을 나눈 이유는, 팀원이 일정 수정을 할 때 기존 plan_id를 유지한 채 업데이트해야 했지만, 수정 과정에서 예상과 다르게 새로운 plan_id가 생성되면서 데이터가 계속 삽입되는 구조로 구현되었기 때문이다.

이러한 문제를 해결하기 위해 여러 방안을 고민한 끝에, 현재 개발된 방식처럼 plan_id와 최신 plan_id를 나눠서 처리하는 로직을 적용하게 되었다.

이를 통해 AI 기반 추천 시스템이 보다 효율적으로 동작할 수 있도록 최적화되었으며, 중복된 데이터 문제를 방지하면서 일정 수정 기능도 안정적으로 동작할 수 있도록 개선되었다.