084 종합 프로젝트 - 추천 시스템 기초
키워드: 추천 시스템, recommendation, 협업 필터링
개요
추천 시스템은 사용자의 선호도를 예측하여 관련 아이템을 추천하는 시스템입니다. 이 글에서는 FLAML AutoML을 활용하여 기본적인 추천 모델을 구축하는 방법을 알아봅니다.
실습 환경
- Python 버전: 3.11 권장
- 필요 패키지:
flaml[automl]
pip install flaml[automl] pandas numpy scikit-learn
추천 시스템 유형
import numpy as np
import pandas as pd
rec_types = {
'유형': ['협업 필터링', '콘텐츠 기반', '하이브리드'],
'방법': [
'사용자/아이템 유사도',
'아이템 특성 매칭',
'두 방법 결합'
],
'장점': [
'숨겨진 패턴 발견',
'콜드 스타트 해결',
'균형잡힌 추천'
],
'FLAML 적용': [
'평점 예측 회귀',
'특성 기반 분류/회귀',
'앙상블'
]
}
print("추천 시스템 유형:")
print(pd.DataFrame(rec_types).to_string(index=False))
데이터 생성 (영화 평점)
np.random.seed(42)
# 084 사용자, 영화, 평점 데이터 생성
n_users = 500
n_movies = 200
n_ratings = 10000
# 084 사용자별 선호 장르 (잠재 요인)
user_preferences = np.random.randn(n_users, 5)
# 084 영화별 장르 특성 (잠재 요인)
movie_features = np.random.randn(n_movies, 5)
# 084 평점 데이터 생성
ratings_data = []
for _ in range(n_ratings):
user_id = np.random.randint(0, n_users)
movie_id = np.random.randint(0, n_movies)
# 잠재 요인 기반 평점 (1-5 범위)
latent_score = np.dot(user_preferences[user_id], movie_features[movie_id])
rating = np.clip(latent_score + np.random.randn() * 0.5 + 3, 1, 5)
ratings_data.append({
'user_id': user_id,
'movie_id': movie_id,
'rating': round(rating, 1)
})
ratings_df = pd.DataFrame(ratings_data).drop_duplicates(['user_id', 'movie_id'])
print(f"평점 데이터: {len(ratings_df)}개")
print(f"사용자 수: {ratings_df['user_id'].nunique()}")
print(f"영화 수: {ratings_df['movie_id'].nunique()}")
print(f"\n평점 분포:\n{ratings_df['rating'].describe()}")
사용자/아이템 특성 생성
# 084 사용자 특성
user_features = pd.DataFrame({
'user_id': range(n_users),
'age': np.random.randint(18, 65, n_users),
'gender': np.random.choice(['M', 'F'], n_users),
'avg_rating': np.random.uniform(2.5, 4.5, n_users),
'rating_count': np.random.randint(5, 100, n_users),
'preference_action': np.random.uniform(0, 1, n_users),
'preference_comedy': np.random.uniform(0, 1, n_users),
'preference_drama': np.random.uniform(0, 1, n_users),
})
# 084 영화 특성
movie_features_df = pd.DataFrame({
'movie_id': range(n_movies),
'year': np.random.randint(1990, 2024, n_movies),
'genre_action': np.random.choice([0, 1], n_movies),
'genre_comedy': np.random.choice([0, 1], n_movies),
'genre_drama': np.random.choice([0, 1], n_movies),
'avg_rating': np.random.uniform(2.0, 4.5, n_movies),
'popularity': np.random.randint(10, 1000, n_movies),
})
print("사용자 특성:")
print(user_features.head())
print("\n영화 특성:")
print(movie_features_df.head())
학습 데이터 구성
from sklearn.preprocessing import LabelEncoder
# 084 평점 데이터에 특성 결합
df = ratings_df.merge(user_features, on='user_id')
df = df.merge(movie_features_df, on='movie_id', suffixes=('_user', '_movie'))
# 084 범주형 인코딩
le_gender = LabelEncoder()
df['gender'] = le_gender.fit_transform(df['gender'])
# 084 특성과 타겟 분리
feature_cols = [c for c in df.columns if c not in ['user_id', 'movie_id', 'rating']]
X = df[feature_cols]
y = df['rating']
print(f"특성 수: {len(feature_cols)}")
print(f"데이터 크기: {X.shape}")
print(f"\n특성 목록: {feature_cols}")
데이터 분할
from sklearn.model_selection import train_test_split
# 084 학습/검증/테스트 분할
X_train, X_temp, y_train, y_temp = train_test_split(
X, y, test_size=0.3, random_state=42
)
X_val, X_test, y_val, y_test = train_test_split(
X_temp, y_temp, test_size=0.5, random_state=42
)
print(f"학습: {X_train.shape}")
print(f"검증: {X_val.shape}")
print(f"테스트: {X_test.shape}")
FLAML로 평점 예측 모델 학습
from flaml import AutoML
# 084 FLAML AutoML - 회귀
automl = AutoML()
automl.fit(
X_train, y_train,
task="regression",
metric="rmse", # 평점 예측이므로 RMSE
time_budget=60,
estimator_list=['lgbm', 'xgboost', 'rf'],
n_jobs=-1,
seed=42,
verbose=2
)
print(f"\n최적 모델: {automl.best_estimator}")
print(f"최적 설정: {automl.best_config}")
모델 평가
from sklearn.metrics import mean_squared_error, mean_absolute_error, r2_score
# 084 예측
y_train_pred = automl.predict(X_train)
y_val_pred = automl.predict(X_val)
y_test_pred = automl.predict(X_test)
def evaluate_regression(y_true, y_pred, name=""):
rmse = np.sqrt(mean_squared_error(y_true, y_pred))
mae = mean_absolute_error(y_true, y_pred)
r2 = r2_score(y_true, y_pred)
print(f"{name}: RMSE={rmse:.4f}, MAE={mae:.4f}, R2={r2:.4f}")
return {'RMSE': rmse, 'MAE': mae, 'R2': r2}
print("\n=== 평점 예측 성능 ===")
evaluate_regression(y_train, y_train_pred, "학습")
evaluate_regression(y_val, y_val_pred, "검증")
test_metrics = evaluate_regression(y_test, y_test_pred, "테스트")
추천 함수 구현
def recommend_movies(user_id, model, user_features, movie_features, top_n=10, exclude_watched=None):
"""사용자에게 영화 추천"""
# 사용자 특성 가져오기
user_data = user_features[user_features['user_id'] == user_id].copy()
if len(user_data) == 0:
print(f"사용자 {user_id} 정보 없음")
return None
# 모든 영화에 대해 예측
predictions = []
for _, movie in movie_features.iterrows():
movie_id = movie['movie_id']
# 이미 본 영화 제외
if exclude_watched and movie_id in exclude_watched:
continue
# 특성 결합
features = user_data.drop('user_id', axis=1).copy()
for col in movie_features.columns:
if col != 'movie_id':
features[col + '_movie'] = movie[col]
# 평점 예측
pred_rating = model.predict(features)[0]
predictions.append({
'movie_id': movie_id,
'predicted_rating': pred_rating
})
# 상위 N개 추천
recommendations = pd.DataFrame(predictions)
recommendations = recommendations.sort_values('predicted_rating', ascending=False)
return recommendations.head(top_n)
# 084 추천 테스트
test_user = 0
watched_movies = ratings_df[ratings_df['user_id'] == test_user]['movie_id'].tolist()
print(f"\n사용자 {test_user}의 시청 영화 수: {len(watched_movies)}")
print(f"\n추천 영화 (Top 10):")
recs = recommend_movies(test_user, automl, user_features, movie_features_df,
top_n=10, exclude_watched=watched_movies)
print(recs)
유사 사용자 기반 추천
from sklearn.metrics.pairwise import cosine_similarity
def find_similar_users(user_id, user_features, top_n=5):
"""유사한 사용자 찾기"""
# 수치형 특성만 사용
numeric_cols = user_features.select_dtypes(include=[np.number]).columns
numeric_cols = [c for c in numeric_cols if c != 'user_id']
user_matrix = user_features[numeric_cols].values
# 코사인 유사도
similarities = cosine_similarity(user_matrix)
# 해당 사용자의 유사도
user_similarities = similarities[user_id]
# 자기 자신 제외하고 상위 N명
similar_indices = np.argsort(user_similarities)[::-1][1:top_n+1]
similar_scores = user_similarities[similar_indices]
return list(zip(similar_indices, similar_scores))
# 084 유사 사용자 찾기
similar_users = find_similar_users(test_user, user_features)
print(f"\n사용자 {test_user}와 유사한 사용자:")
for user, score in similar_users:
print(f" 사용자 {user}: 유사도 {score:.4f}")
하이브리드 추천
def hybrid_recommend(user_id, model, user_features, movie_features, ratings_df,
top_n=10, alpha=0.7):
"""하이브리드 추천 (콘텐츠 + 협업 필터링)"""
# 1. 콘텐츠 기반 예측
content_recs = recommend_movies(user_id, model, user_features, movie_features, top_n=50)
# 2. 유사 사용자의 평점
similar_users = find_similar_users(user_id, user_features, top_n=10)
similar_user_ids = [u[0] for u in similar_users]
# 유사 사용자들의 평점
similar_ratings = ratings_df[ratings_df['user_id'].isin(similar_user_ids)]
collab_scores = similar_ratings.groupby('movie_id')['rating'].mean()
# 3. 하이브리드 점수
hybrid_scores = []
for _, row in content_recs.iterrows():
movie_id = row['movie_id']
content_score = row['predicted_rating']
collab_score = collab_scores.get(movie_id, content_score)
# 가중 평균
hybrid_score = alpha * content_score + (1 - alpha) * collab_score
hybrid_scores.append({
'movie_id': movie_id,
'content_score': content_score,
'collab_score': collab_score,
'hybrid_score': hybrid_score
})
result = pd.DataFrame(hybrid_scores).sort_values('hybrid_score', ascending=False)
return result.head(top_n)
# 084 하이브리드 추천
print("\n=== 하이브리드 추천 ===")
hybrid_recs = hybrid_recommend(test_user, automl, user_features, movie_features_df,
ratings_df, top_n=10)
print(hybrid_recs)
평가 지표
def evaluate_recommendations(model, ratings_df, user_features, movie_features, k=10):
"""추천 시스템 평가"""
# 일부 사용자로 평가
test_users = ratings_df['user_id'].unique()[:50]
precision_list = []
recall_list = []
for user_id in test_users:
# 실제 높게 평가한 영화 (4점 이상)
user_ratings = ratings_df[ratings_df['user_id'] == user_id]
relevant = set(user_ratings[user_ratings['rating'] >= 4]['movie_id'])
if len(relevant) == 0:
continue
# 추천 영화
recs = recommend_movies(user_id, model, user_features, movie_features, top_n=k)
if recs is None:
continue
recommended = set(recs['movie_id'])
# Precision@K, Recall@K
hits = len(relevant & recommended)
precision = hits / k
recall = hits / len(relevant)
precision_list.append(precision)
recall_list.append(recall)
print(f"\n=== 추천 평가 (K={k}) ===")
print(f"Precision@{k}: {np.mean(precision_list):.4f}")
print(f"Recall@{k}: {np.mean(recall_list):.4f}")
return np.mean(precision_list), np.mean(recall_list)
# 084 평가 실행
evaluate_recommendations(automl, ratings_df, user_features, movie_features_df, k=10)
모델 저장
import joblib
# 084 추천 모델 패키지 저장
rec_package = {
'model': automl,
'user_features': user_features,
'movie_features': movie_features_df,
'label_encoders': {'gender': le_gender}
}
joblib.dump(rec_package, 'recommendation_model.pkl')
print("\n추천 모델 저장: recommendation_model.pkl")
정리
- 평점 예측: FLAML 회귀로 사용자-영화 평점 예측
- 콘텐츠 기반: 사용자/영화 특성 활용
- 협업 필터링: 유사 사용자 기반 추천
- 하이브리드: 두 방법의 가중 결합
- 평가: Precision@K, Recall@K
다음 글 예고
다음 글에서는 이미지 특성 기반 분류 프로젝트를 진행합니다. 이미지에서 추출한 특성으로 FLAML 분류 모델을 구축합니다.
FLAML AutoML 마스터 시리즈 #084