4편. 추천 알고리즘 설계 - 가중치 기반 개인화

 

이번 편에서 다루는 것

임베딩이 준비되었으니, 이제 실제로 "이 학생에게 어떤 강좌를 추천할것인가?"를 결정하는 알고리즘을 설계한다.

핵심은 학과 기반 추천과 관심분야 기반 추천을 어떻게 결합할 것인가였다.

---

추천 전략의 고민

옵션 1: 학과 기반만 (100%)

학생의 소속 학과에 개설된 강좌만 추천

 

장점:

• 전공 학점 취득에 유리
• 단순하고 명확한 로직

단점:

• 학생의 개인적 관심사 반영 불가
• 모든 학생에게 동일한 추천 결과
• 차별화된 가치 제공 불가

옵션 2: 관심분야 기반만 (100%)

학생이 선택한 관심분야와 가장 유사한 강좌만 추천

 

장점:

• 개인화된 추천
• 학생 만족도 높음

단점:

• 전공과 무관한 강좌가 추천될 수 있음
• 학점 인정 문제 발생 가능

옵션 3: 하이브리드 방식 (선택)

두 기준을 가중치로 결합

최종 점수 = (학과 점수 × W1) + (관심분야 점수 × W2)

이 방식을 선택했다.

---

가중치 결정: 왜 70:30인가?

고려한 요소

요소 설명

요소	설명
학사 제도	학생은 우선 자기 학과 강좌를 들어야 함
사용자 경험	완전히 동떨어진 추천은 혼란 유발
차별화	관심분야 없이는 모든 학생에게 동일한 결과

시뮬레이션 결과

여러 비율로 테스트해보았다:

학과:관심분야	결과 특성
100:0	동일 학과 학생에게 모두 같은 추천
80:20	관심분야 영향이 미미함
70:30	학과 우선 + 관심분야로 차별화
50:50	전공과 무관한 강좌가 상위에 등장
30:70	학과 기반이 무력화됨

최종 결정

BASE_WEIGHT = 0.7   # 학과 기반 70%
INTEREST_WEIGHT = 0.3  # 관심분야 30%

 

이 값은 설정으로 조정 가능하게 구현해서, 향후 피드백에 따라 변경할 수 있도록 했다.

---

알고리즘 상세 설계

전체 흐름

1. 학생 정보 로드 (학과, 관심분야)
2. 학과 기반 점수 계산
3. 관심분야 기반 점수 계산  
4. 가중치 적용하여 최종 점수 산출
5. 상위 K개 선별
6. 결과 반환

학과 기반 점수

학생의 학과와 강좌의 개설 학과가 일치하는지 확인:

def calculate_base_score(self, student_dept: str, course_dept: str) -> float:
    """학과 기반 점수 계산"""
    
    if student_dept == course_dept:
        return 1.0  # 완전 일치
    
    # 관련 학과 여부 확인 (예: IT계열 학과들)
    if self._is_related_dept(student_dept, course_dept):
        return 0.7  # 관련 학과
    
    return 0.3  # 무관한 학과

관심분야 기반 점수

학생의 관심분야와 강좌 설명 간의 코사인 유사도:

def calculate_interest_score(
    self, 
    student_interests: List[str], 
    course_embedding: np.ndarray
) -> float:
    """관심분야 기반 점수 계산"""
    
    if not student_interests:
        return 0.5  # 관심분야 미설정 시 중립
    
    # 학생의 관심분야들을 하나의 벡터로 결합
    interest_text = " ".join(student_interests)
    interest_embedding = self.engine.encode(interest_text)
    
    # 코사인 유사도 계산
    similarity = self.engine.similarity(interest_embedding, course_embedding)
    
    # 0~1 범위로 정규화 (코사인 유사도는 -1~1 범위)
    normalized = (similarity + 1) / 2
    
    return normalized

최종 점수 계산

def calculate_final_score(
    self,
    base_score: float,
    interest_score: float
) -> float:
    """가중치를 적용한 최종 점수"""
    
    final = (base_score * self.BASE_WEIGHT) + (interest_score * self.INTEREST_WEIGHT)
    
    return final

---

전체 추천 엔진 구현

from dataclasses import dataclass
from typing import List, Dict, Any
import numpy as np

@dataclass
class Recommendation:
    """ 추천 결과 """
    course_id: str
    course_name: str
    final_score: float
    base_score: float
    interest_score: float

class StudentRecommender:
    def __init__(
        self, 
        engine: EmbeddingEngine,
        base_weight: float = 0.7,
        interest_weight: float = 0.3
    ):
        self.engine = engine
        self.BASE_WEIGHT = base_weight
        self.INTEREST_WEIGHT = interest_weight
    
    def recommend_for_student(
        self,
        student: Dict[str, Any],
        courses: List[Dict[str, Any]],
        course_embeddings: Dict[str, np.ndarray],
        top_k: int = 5
    ) -> List[Recommendation]:
        """학생에게 강좌 추천"""
        
        student_dept = student['department']
        student_interests = student.get('interests', [])
        
        # 관심분야 임베딩 (사전 계산)
        interest_embedding = None
        if student_interests:
            interest_text = " ".join(student_interests)
            interest_embedding = self.engine.encode(interest_text)
        
        # 모든 강좌에 대해 점수 계산
        scored_courses = []
        
        for course in courses:
            course_id = course['id']
            course_dept = course['department']
            course_embedding = course_embeddings.get(course_id)
            
            if course_embedding is None:
                continue
            
            # 학과 기반 점수
            base_score = self._calculate_base_score(student_dept, course_dept)
            
            # 관심분야 기반 점수
            if interest_embedding is not None:
                interest_score = self._calculate_interest_score(
                    interest_embedding, 
                    course_embedding
                )
            else:
                interest_score = 0.5
            
            # 최종 점수
            final_score = self._calculate_final_score(base_score, interest_score)
            
            scored_courses.append(Recommendation(
                course_id=course_id,
                course_name=course['name'],
                final_score=final_score,
                base_score=base_score,
                interest_score=interest_score
            ))
        
        # 점수 내림차순 정렬 후 상위 K개
        scored_courses.sort(key=lambda x: x.final_score, reverse=True)
        
        return scored_courses[:top_k]
    
    def _calculate_base_score(self, student_dept: str, course_dept: str) -> float:
        if student_dept == course_dept:
            return 1.0
        return 0.3
    
    def _calculate_interest_score(
        self, 
        interest_embedding: np.ndarray, 
        course_embedding: np.ndarray
    ) -> float:
        similarity = self.engine.similarity(interest_embedding, course_embedding)
        return (similarity + 1) / 2
    
    def _calculate_final_score(self, base_score: float, interest_score: float) -> float:
        return (base_score * self.BASE_WEIGHT) + (interest_score * self.INTEREST_WEIGHT)

---

실제 추천 결과 예시

테스트 케이스

학생 정보:

• 학과: 컴퓨터공학과
• 관심분야: ["AI", "데이터 사이언스", "머신러닝"]

추천 결과:

순위	강좌명	최종점수	학과점수	관심점수
1	머신러닝 기초	0.934	1.0	0.847
2	데이터분석및실습	0.921	1.0	0.803
3	인공지능개론	0.908	1.0	0.760
4	파이썬프로그래밍	0.876	1.0	0.653
5	데이터베이스	0.862	1.0	0.607

 

학과가 같으므로 base_score는 모두 1.0, 관심분야에 따라 순위가 결정되었다.

다른 학과 학생과 비교

학생 B:

• 학과: 컴퓨터공학과
• 관심분야: ["웹 개발", "UI/UX", "프론트엔드"]

순위 강좌명 최종점수

순위	강좌명	최종점수
1	웹프로그래밍	0.941
2	UI/UX디자인	0.923
3	프론트엔드개발	0.912
4	자바스크립트	0.887
5	데이터베이스	0.862

 

같은 학과지만 관심분야가 다르므로 추천 결과가 다르다.

---

비교과 활동 추천 확장

강좌와 같은 로직으로 비교과 활동도 추천한다.

def recommend_activities(
    self,
    student: Dict[str, Any],
    activities: List[Dict[str, Any]],
    activity_embeddings: Dict[str, np.ndarray],
    top_k: int = 5
) -> List[Recommendation]:
    """학생에게 비교과 활동 추천"""
    
    # 비교과는 학과 제한이 없으므로 관심분야 위주로 추천
    student_interests = student.get('interests', [])
    
    if not student_interests:
        # 관심분야 없으면 인기 활동 추천
        return self._recommend_popular_activities(activities, top_k)
    
    interest_text = " ".join(student_interests)
    interest_embedding = self.engine.encode(interest_text)
    
    scored = []
    for activity in activities:
        activity_id = activity['id']
        activity_embedding = activity_embeddings.get(activity_id)
        
        if activity_embedding is None:
            continue
        
        similarity = self.engine.similarity(interest_embedding, activity_embedding)
        score = (similarity + 1) / 2
        
        scored.append(Recommendation(
            course_id=activity_id,
            course_name=activity['name'],
            final_score=score,
            base_score=0,  # 비교과는 학과 점수 없음
            interest_score=score
        ))
    
    scored.sort(key=lambda x: x.final_score, reverse=True)
    return scored[:top_k]

---

성능 최적화: 배치 추천

5,000명 학생을 한 명씩 처리하면 너무 느리다.

문제

# 느린 방식
for student in students:  # 5,000명
    recommendations = recommender.recommend_for_student(student, courses, embeddings)
    save_to_db(recommendations)

해결: 벡터 연산의 배치화

import numpy as np
from sklearn.metrics.pairwise import cosine_similarity

class BatchRecommender:
    def recommend_batch(
        self,
        students: List[Dict],
        courses: List[Dict],
        course_embeddings: np.ndarray  # (num_courses, 768)
    ) -> Dict[str, List[Recommendation]]:
        """모든 학생에 대해 일괄 추천"""
        
        results = {}
        
        # 학생 관심분야 임베딩을 일괄 계산
        interest_texts = [
            " ".join(s.get('interests', [])) or "기본 관심분야"
            for s in students
        ]
        student_embeddings = self.engine.encode(interest_texts)  # (num_students, 768)
        
        # 유사도 행렬 일괄 계산: O(students * courses)
        similarity_matrix = cosine_similarity(student_embeddings, course_embeddings)
        # shape: (num_students, num_courses)
        
        # 각 학생별로 상위 K개 선별
        for i, student in enumerate(students):
            student_id = student['id']
            similarities = similarity_matrix[i]  # 이 학생의 모든 강좌 유사도
            
            # 학과 점수 계산 및 최종 점수 산출
            recommendations = self._compute_final_scores(
                student, courses, similarities
            )
            
            results[student_id] = recommendations
        
        return results

성능 비교

방식 5,000명 처리 시간

방식	5,000명 처리시간
개별 처리	약 25분
배치 처리	약 3분

 

8배 이상 빨라졌다.

---

결과 저장

추천 결과를 DB에 저장해서 API로 제공한다.

from sqlalchemy.dialects.postgresql import insert

def save_recommendations(
    self,
    student_id: str,
    recommendations: List[Recommendation],
    rec_type: str  # 'course' or 'activity'
):
    """추천 결과 DB 저장"""
    
    records = [
        {
            "student_id": student_id,
            "item_id": rec.course_id,
            "item_type": rec_type,
            "rank": i + 1,
            "score": rec.final_score,
            "base_score": rec.base_score,
            "interest_score": rec.interest_score
        }
        for i, rec in enumerate(recommendations)
    ]
    
    # Upsert로 중복 방지
    stmt = insert(RecommendationTable).values(records)
    stmt = stmt.on_conflict_do_update(
        index_elements=['student_id', 'item_id', 'item_type'],
        set_={'rank': stmt.excluded.rank, 'score': stmt.excluded.score}
    )
    
    self.session.execute(stmt)
    self.session.commit()

---

성과 및 교훈

성과

지표 수치

추천 대상 학생	5,000명+
학생당 추천	강좌 5개 + 활동 5개
전체 추천 생성	약 3분
개인화율	100% (관심분야 설정 학생)

배운 점

1. 가중치는 실험으로 결정: 직관보다 시뮬레이션 결과로 판단
2. 설정 가능하게 설계: 향후 조정 가능성 열어두기
3. 배치 처리 필수: 대용량은 벡터 연산으로 처리
4. 점수 분해 저장: 디버깅과 분석을 위해 세부 점수 기록

---

다음 편 예고

추천 알고리즘이 완성되었다. 하지만 한 가지 문제가 있었다.

강좌 설명이 부족하면 임베딩 품질이 떨어진다.

 

다음 편에서는 LLM을 활용해 강좌 설명을 보강하는 방법을 다룬다.

---

시리즈 목차

1. 프로젝트 소개 - 왜 AI 추천이 필요했나
2. 추천을 위한 데이터 파이프라인 설계
3. XLM-R로 강좌 임베딩 구축하기
4. 추천 알고리즘 설계 - 가중치 기반 개인화 ← 현재 글
5. LLM으로 강좌 설명 보강하기
6. 성능 최적화 - CPU에서 27배 빠르게
7. API 서버 구축과 배포