SNS 피드 시스템 설계 가이드

engagement_score = (likes × 1) + (comments × 2) + (shares × 3)

def generate_candidates(user_id, limit=1000):
    # 팔로우하는 사용자들의 최근 게시물 조회
    following_users = get_following_list(user_id)
    
    candidates = []
    for author_id in following_users:
        # 최근 7일 이내 게시물만
        recent_posts = db.query("""
            SELECT * FROM posts 
            WHERE author_id = ? 
            AND created_at > NOW() - INTERVAL 7 DAY
            AND visibility IN ('public', 'friends')
            ORDER BY created_at DESC
            LIMIT 10
        """, author_id)
        candidates.extend(recent_posts)
    
    # 블록/뮤트 사용자 제외
    blocked_users = get_blocked_users(user_id)
    candidates = [p for p in candidates if p.author_id not in blocked_users]
    
    return candidates[:limit]

# 사용자 특성
user_features = {
    'age': 25,
    'gender': 'F',
    'location': 'Seoul',
    'avg_session_time_minutes': 15.3,
    'posts_liked_last_week': 42,
    'interests': ['tech', 'travel', 'food'],  # 원핫 인코딩
}
 
# 게시물 특성
post_features = {
    'post_id': 9876,
    'author_id': 555,
    'created_hours_ago': 2.5,
    'likes_count': 150,
    'comments_count': 23,
    'shares_count': 5,
    'is_video': 1,  # boolean
    'has_link': 0,
    'hashtags_count': 3,
    'text_length': 280,
}
 
# 관계 특성
relationship_features = {
    'is_following': 1,
    'past_interactions_count': 5,  # 이 작성자와 과거 상호작용
    'social_distance': 1,  # 1촌, 2촌, ...
    'last_interaction_days_ago': 3,
}
 
# 맥락 특성
context_features = {
    'time_of_day': 14,  # 14시
    'day_of_week': 3,  # 수요일
    'is_mobile': 1,
}

# 모든 features를 하나의 벡터로 결합
feature_vector = flatten([
    user_features,
    post_features, 
    relationship_features,
    context_features
])
# feature_vector = [25, 0, ..., 1]  # 300개 정도
 
# Gradient Boosting (XGBoost/LightGBM)로 예측
model = load_model('feed_ranking_gbdt.pkl')
score = model.predict([feature_vector])[0]
# score = 0.76 (상호작용 확률)

# 1. User Tower: 사용자를 128차원 벡터로 변환
user_tower = NeuralNetwork([
    Embedding(user_id, dim=64),
    Dense(past_liked_posts, dim=32),
    Dense(interests, dim=32),
    Concatenate(),  # 64 + 32 + 32 = 128
    Dense(128, activation='relu'),
])
 
user_embedding = user_tower.predict(user_data)
# user_embedding = [0.2, -0.5, 0.8, ..., 0.3]  # 128차원
 
# 2. Item Tower: 게시물을 128차원 벡터로 변환
item_tower = NeuralNetwork([
    Embedding(post_id, dim=64),
    TextEmbedding(content_text, dim=32),
    ImageEmbedding(content_image, dim=32),
    Concatenate(),  # 64 + 32 + 32 = 128
    Dense(128, activation='relu'),
])
 
post_embedding = item_tower.predict(post_data)
# post_embedding = [0.1, -0.3, 0.9, ..., 0.2]  # 128차원
 
# 3. 코사인 유사도 계산
def cosine_similarity(vec1, vec2):
    dot_product = np.dot(vec1, vec2)
    norm1 = np.linalg.norm(vec1)
    norm2 = np.linalg.norm(vec2)
    return dot_product / (norm1 * norm2)
 
score = cosine_similarity(user_embedding, post_embedding)
# score = 0.82

# 두 방식의 점수를 가중 평균
embedding_score = cosine_similarity(user_emb, post_emb)  # 0.82
feature_score = gbdt_model.predict(features)              # 0.76
 
final_score = 0.6 * embedding_score + 0.4 * feature_score
# final_score = 0.6 × 0.82 + 0.4 × 0.76 = 0.796

def rank_posts(user_id, candidates):
    scored_posts = []
    
    for post in candidates:  # 1000개 반복
        # Features 생성
        features = extract_features(user_id, post)
        
        # Embedding 점수
        user_emb = get_user_embedding(user_id)
        post_emb = get_post_embedding(post.id)
        emb_score = cosine_similarity(user_emb, post_emb)
        
        # Feature 점수
        feat_score = gbdt_model.predict(features)
        
        # 최종 점수
        final_score = 0.6 * emb_score + 0.4 * feat_score
        scored_posts.append((post, final_score))
    
    # 점수 기준 내림차순 정렬
    scored_posts.sort(key=lambda x: x[1], reverse=True)
    
    # 상위 50개만 반환
    return scored_posts[:50]

def rerank_feed(ranked_posts):
    final_feed = []
    author_count = {}  # 작성자별 노출 횟수 추적
    
    for post, score in ranked_posts:
        # 규칙 1: 같은 작성자 연속 3개 제한
        if author_count.get(post.author_id, 0) >= 3:
            continue
        
        # 규칙 2: 이미 본 게시물 제외
        if user_has_seen(post.id):
            continue
        
        # 규칙 3: 최소 품질 점수 (0.3 이상)
        if score < 0.3:
            break  # 이미 정렬되어 있으므로 break
        
        final_feed.append(post)
        author_count[post.author_id] = author_count.get(post.author_id, 0) + 1
        
        if len(final_feed) >= 20:
            break
    
    # 광고 삽입 (5번째, 15번째 위치)
    final_feed = insert_ads(final_feed, positions=[4, 14])
    
    return final_feed

시나리오: 1억 개 게시물, ML 모델 추론 시간 = 0.3ms

❌ 전체에 ML 적용:
1억 × 0.3ms = 30,000초 = 8.3시간 (불가능!)

✅ 3단계 파이프라인:
├─ Stage 1 (SQL 필터링): 1억 → 1000개 (0.001초)
├─ Stage 2 (ML 랭킹): 1000 × 0.3ms = 0.3초
└─ Stage 3 (재정렬): 50개 규칙 적용 (0.01초)

총 소요 시간: ~0.31초 ✓

GET /feed?cursor=eyJpZCI6MTIzNDU2fQ==&limit=20

// 마지막 게시물의 ID와 timestamp를 cursor로 인코딩
cursor = base64encode({id: last_post_id, timestamp: last_timestamp})

// 클라이언트
const eventSource = new EventSource('/feed/updates');
eventSource.onmessage = (event) => {
  showNewPostsBadge(event.data.count);
};

Client (Mobile/Web)
    ↓
Load Balancer
    ↓
API Gateway
    ↓
┌─────────────┬──────────────┬─────────────┐
│  Feed       │  Post        │  User       │
│  Service    │  Service     │  Service    │
└─────────────┴──────────────┴─────────────┘
    ↓              ↓               ↓
┌─────────────┬──────────────┬─────────────┐
│  Redis      │  Cassandra   │  PostgreSQL │
│  (Cache)    │  (Posts)     │  (Users)    │
└─────────────┴──────────────┴─────────────┘

def get_feed(user_id, cursor=None, limit=20):
    """
    사용자 피드 조회 API
    
    Args:
        user_id: 사용자 ID
        cursor: 페이지네이션 커서
        limit: 반환할 게시물 수 (기본 20개)
    
    Returns:
        피드 게시물 리스트 + 다음 커서
    """
    
    # ========== 단계 0: 캐시 확인 ==========
    cache_key = f"feed:{user_id}:{cursor}"
    cached_feed = redis.get(cache_key)
    
    if cached_feed:
        print("Cache hit!")
        return json.loads(cached_feed)
    
    print("Cache miss - generating feed...")
    
    # ========== Stage 1: Candidate Generation ==========
    # 1억 개 → 1000개로 축소 (0.001초)
    
    candidates = generate_candidates(user_id, limit=1000)
    print(f"Stage 1: Generated {len(candidates)} candidates")
    
    # ========== Stage 2: Ranking ==========
    # 1000개 → 50개로 정밀 필터링 (0.3초)
    
    # 사용자 임베딩 조회 (미리 계산되어 있음)
    user_embedding = embedding_cache.get(f"user_emb:{user_id}")
    
    scored_posts = []
    for post in candidates:
        # Feature 추출
        features = extract_features(user_id, post)
        
        # Embedding 점수
        post_embedding = embedding_cache.get(f"post_emb:{post.id}")
        emb_score = cosine_similarity(user_embedding, post_embedding)
        
        # Feature 점수 (GBDT)
        feat_score = ranking_model.predict([features])[0]
        
        # Hybrid 점수
        final_score = 0.6 * emb_score + 0.4 * feat_score
        
        scored_posts.append({
            'post': post,
            'score': final_score,
            'emb_score': emb_score,
            'feat_score': feat_score,
        })
    
    # 점수 기준 정렬
    scored_posts.sort(key=lambda x: x['score'], reverse=True)
    top_posts = scored_posts[:50]
    
    print(f"Stage 2: Ranked to top {len(top_posts)} posts")
    
    # ========== Stage 3: Re-ranking ==========
    # 50개 → 20개 + 비즈니스 규칙 (0.01초)
    
    final_feed = []
    author_count = {}
    seen_posts = get_seen_posts(user_id)
    
    for item in top_posts:
        post = item['post']
        score = item['score']
        
        # 규칙 1: 이미 본 게시물 제외
        if post.id in seen_posts:
            continue
        
        # 규칙 2: 같은 작성자 연속 3개 제한
        if author_count.get(post.author_id, 0) >= 3:
            continue
        
        # 규칙 3: 최소 품질 점수
        if score < 0.3:
            break
        
        final_feed.append(post)
        author_count[post.author_id] = author_count.get(post.author_id, 0) + 1
        
        if len(final_feed) >= limit:
            break
    
    # 광고 삽입
    final_feed = insert_sponsored_posts(final_feed, positions=[4, 14])
    
    print(f"Stage 3: Re-ranked to final {len(final_feed)} posts")
    
    # ========== 캐싱 및 반환 ==========
    
    # 다음 페이지 커서 생성
    next_cursor = None
    if len(final_feed) >= limit:
        last_post = final_feed[-1]
        next_cursor = base64encode({
            'id': last_post.id,
            'timestamp': last_post.created_at
        })
    
    result = {
        'posts': final_feed,
        'next_cursor': next_cursor,
    }
    
    # Redis 캐싱 (TTL 10분)
    redis.setex(cache_key, 600, json.dumps(result))
    
    return result

전체 프로세스 타임라인:

0ms    - API 요청 수신
1ms    - Redis 캐시 확인 (miss)
2ms    - Stage 1 시작: Candidate Generation
3ms    - DB/Redis에서 1000개 조회 완료
303ms  - Stage 2 완료: 1000개 랭킹 (각 0.3ms)
313ms  - Stage 3 완료: 재정렬 및 규칙 적용
315ms  - Redis 캐싱 및 응답 반환

총 소요 시간: ~315ms (P95 < 500ms 목표 달성 ✓)

캐시 hit 시: ~1ms (99% 빠름!)