graph TD
A[데이터 수집] --> B[데이터 전처리]
B --> C[탐색적 데이터 분석]
C --> D[특징 엔지니어링]
D --> E[모델 선택]
E --> F[모델 학습]
F --> G[모델 평가]
G --> H[배포 및 서빙]
H --> I[모니터링]
I --> J[모델 업데이트]
J --> F
개인화된 쇼핑 경험을 위한 AI 기반 추천 시스템 구축
5/2/23
graph TD
A[데이터 수집] --> B[데이터 전처리]
B --> C[탐색적 데이터 분석]
C --> D[특징 엔지니어링]
D --> E[모델 선택]
E --> F[모델 학습]
F --> G[모델 평가]
G --> H[배포 및 서빙]
H --> I[모니터링]
I --> J[모델 업데이트]
J --> F
import pandas as pd
import numpy as np
from sklearn.metrics.pairwise import cosine_similarity
# 사용자-아이템 매트릭스 생성
user_item_matrix = ratings.pivot_table(
index='user_id',
columns='item_id',
values='rating'
).fillna(0)
# 사용자 간 유사도 계산
user_similarity = cosine_similarity(user_item_matrix)
# 추천 점수 계산
def get_recommendations(user_id, num_recommendations=10):
user_idx = user_item_matrix.index.get_loc(user_id)
sim_scores = user_similarity[user_idx]
# 가중 평균으로 추천 점수 계산
recommendations = np.dot(sim_scores, user_item_matrix.values)
return recommendations.argsort()[::-1][:num_recommendations]from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.metrics.pairwise import linear_kernel
# 상품 설명 텍스트 벡터화
tfidf = TfidfVectorizer(
max_features=5000,
stop_words='english',
ngram_range=(1, 2)
)
# 상품 설명을 TF-IDF 매트릭스로 변환
tfidf_matrix = tfidf.fit_transform(products['description'])
# 코사인 유사도 계산
cosine_sim = linear_kernel(tfidf_matrix, tfidf_matrix)
def content_based_recommendations(product_id, num_recommendations=10):
idx = products[products['id'] == product_id].index[0]
sim_scores = list(enumerate(cosine_sim[idx]))
sim_scores = sorted(sim_scores, key=lambda x: x[1], reverse=True)
# 가장 유사한 상품들의 인덱스 반환
product_indices = [i[0] for i in sim_scores[1:num_recommendations+1]]
return products.iloc[product_indices]import tensorflow as tf
from tensorflow.keras import layers, Model
class NCF(Model):
def __init__(self, num_users, num_items, embedding_size=50):
super(NCF, self).__init__()
# 사용자 임베딩
self.user_embedding = layers.Embedding(num_users, embedding_size)
self.user_bias = layers.Embedding(num_users, 1)
# 아이템 임베딩
self.item_embedding = layers.Embedding(num_items, embedding_size)
self.item_bias = layers.Embedding(num_items, 1)
# 신경망 레이어
self.dense1 = layers.Dense(128, activation='relu')
self.dense2 = layers.Dense(64, activation='relu')
self.output_layer = layers.Dense(1, activation='sigmoid')
def call(self, inputs):
user_id, item_id = inputs
# 임베딩 벡터 추출
user_vec = self.user_embedding(user_id)
item_vec = self.item_embedding(item_id)
# 벡터 결합
concat_vec = tf.concat([user_vec, item_vec], axis=1)
# 신경망 통과
x = self.dense1(concat_vec)
x = self.dense2(x)
return self.output_layer(x)
graph LR
A[사용자 행동] --> B[실시간 데이터 수집]
B --> C[스트리밍 처리]
C --> D[특징 추출]
D --> E[모델 서빙]
E --> F[추천 결과]
F --> G[캐싱]
G --> H[API 응답]
I[배치 학습] --> J[모델 업데이트]
J --> E
K[A/B 테스트] --> L[성능 모니터링]
L --> M[모델 개선]
M --> I
import numpy as np
from scipy import stats
class ABTest:
def __init__(self, control_group, treatment_group):
self.control = control_group
self.treatment = treatment_group
def calculate_metrics(self):
# 전환율 계산
control_conversion = np.mean(self.control['converted'])
treatment_conversion = np.mean(self.treatment['converted'])
# 통계적 유의성 검정
t_stat, p_value = stats.ttest_ind(
self.control['converted'],
self.treatment['converted']
)
return {
'control_conversion': control_conversion,
'treatment_conversion': treatment_conversion,
'lift': (treatment_conversion - control_conversion) / control_conversion,
'p_value': p_value,
'significant': p_value < 0.05
}class FashionRecommender:
def __init__(self):
self.collaborative_model = CollaborativeFiltering()
self.content_model = ContentBasedFiltering()
self.visual_model = VisualSimilarity()
self.price_model = PriceBasedFiltering()
def get_recommendations(self, user_id, context=None):
# 다중 모델 결합
collab_recs = self.collaborative_model.recommend(user_id)
content_recs = self.content_model.recommend(user_id)
visual_recs = self.visual_model.recommend(user_id)
price_recs = self.price_model.recommend(user_id)
# 가중 결합
final_recs = self.combine_recommendations([
(collab_recs, 0.4),
(content_recs, 0.3),
(visual_recs, 0.2),
(price_recs, 0.1)
])
return final_recsclass RecommendationMonitor:
def __init__(self):
self.metrics_collector = MetricsCollector()
self.alerting_system = AlertingSystem()
def monitor_performance(self):
# 성능 지표 수집
metrics = self.metrics_collector.collect_metrics()
# 임계값 체크
if metrics['response_time'] > 200: # 200ms 초과
self.alerting_system.send_alert("High response time detected")
if metrics['error_rate'] > 0.01: # 1% 초과
self.alerting_system.send_alert("High error rate detected")
# 자동 스케일링
if metrics['cpu_usage'] > 80:
self.auto_scale_up()함께 나누고 싶은 질문들:
질문이 있으신가요?
유용한 리소스:
연락처:
E-commerce 상품추천 시스템 | 홍성학