[Python] Redis를 활용한 중복 추천 방지 로직 상세 분석
이전 글: [Python] Redis를 활용한 중복 추천 방지 캐싱 서비스 개발
이전 포스팅에서는 Redis를 활용한 장소 캐싱 시스템(SpotRedisService)을 설계한 이유와 개요를 설명했다.
이번 글에서는 추천 시스템에서 Redis를 활용하는 구체적인 코드를 살펴보겠다.
여기서는 create_recommendation_restaurant 함수를 중심으로 Redis와 DB를 활용한 중복 추천 방지 로직을 분석하고, 각 부분별로 어떤 역할을 하는지 설명하겠다.
create_recommendation_restaurant 함수 개요
이 함수는 에이전트가 사용자에게 추천할 식당 목록을 생성하는 핵심 로직을 담당한다.
Redis 조회 흐름도
DB 조회 흐름도
주요 기능
- 사용자 정보 및 기존 추천 데이터 확인 (Redis & DB 조회)
- 입력 데이터 전처리
- Crew AI 실행 (추천 생성)
- 결과 처리 및 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 기반 추천 시스템이 보다 효율적으로 동작할 수 있도록 최적화되었으며, 중복된 데이터 문제를 방지하면서 일정 수정 기능도 안정적으로 동작할 수 있도록 개선되었다.