コールドスタート問題

コールドスタートの3パターン

新規ユーザーへの対策

class ColdStartUserHandler:
    """新規ユーザーのコールドスタート対応"""

    def __init__(self, popularity_model, content_model, hybrid_model):
        self.popularity = popularity_model
        self.content = content_model
        self.hybrid = hybrid_model

    def recommend(self, user, n_items=10):
        """ユーザーの情報量に応じた推薦戦略"""
        interaction_count = user.get('interaction_count', 0)

        if interaction_count == 0:
            # Phase 1: 人気アイテム + デモグラフィック
            return self._phase1_recommend(user, n_items)
        elif interaction_count < 5:
            # Phase 2: コンテンツベース補完
            return self._phase2_recommend(user, n_items)
        else:
            # Phase 3: ハイブリッド推薦
            return self._phase3_recommend(user, n_items)

    def _phase1_recommend(self, user, n_items):
        """行動データなし: 人気 + デモグラフィック"""
        # ユーザーの属性（年齢層、性別）に応じた人気アイテム
        segment = self._get_segment(user)
        popular_items = self.popularity.get_popular_by_segment(
            segment, n_items
        )
        return {
            'strategy': 'popularity_demographic',
            'items': popular_items,
            'confidence': 'low',
        }

    def _phase2_recommend(self, user, n_items):
        """少量の行動データ: コンテンツベース"""
        content_items = self.content.recommend(user, n_items)
        return {
            'strategy': 'content_based',
            'items': content_items,
            'confidence': 'medium',
        }

    def _phase3_recommend(self, user, n_items):
        """十分な行動データ: ハイブリッド"""
        hybrid_items = self.hybrid.recommend(user, n_items)
        return {
            'strategy': 'hybrid',
            'items': hybrid_items,
            'confidence': 'high',
        }

    def _get_segment(self, user):
        """デモグラフィック情報からセグメント推定"""
        age = user.get('age', 30)
        gender = user.get('gender', 'unknown')
        if age < 25:
            return f"young_{gender}"
        elif age < 40:
            return f"middle_{gender}"
        else:
            return f"senior_{gender}"

新規アイテムへの対策

class ColdStartItemHandler:
    """新規アイテムのコールドスタート対応"""

    def __init__(self, item_feature_model):
        self.model = item_feature_model

    def compute_item_similarity(self, new_item, existing_items):
        """新規アイテムと既存アイテムの類似度を計算"""
        new_features = self._extract_features(new_item)

        similarities = []
        for item in existing_items:
            item_features = self._extract_features(item)
            sim = self._cosine_similarity(new_features, item_features)
            similarities.append({
                'item_id': item['id'],
                'similarity': sim,
            })

        return sorted(similarities, key=lambda x: -x['similarity'])

    def recommend_new_item_to_users(self, new_item, user_item_matrix, top_k=100):
        """新規アイテムを推薦すべきユーザーを特定"""
        # 類似アイテムを購入したユーザーを抽出
        similar_items = self.compute_item_similarity(new_item, existing_items)[:10]
        similar_item_ids = [s['item_id'] for s in similar_items]

        # 類似アイテムを好んだユーザーを集計
        candidate_users = {}
        for item_id in similar_item_ids:
            users = user_item_matrix.get_users_who_liked(item_id)
            for user_id in users:
                candidate_users[user_id] = candidate_users.get(user_id, 0) + 1

        # スコア順にソート
        ranked_users = sorted(
            candidate_users.items(), key=lambda x: -x[1]
        )[:top_k]

        return ranked_users

    def _extract_features(self, item):
        """アイテムの特徴量を抽出"""
        return np.concatenate([
            self.model.encode_category(item.get('category', '')),
            self.model.encode_text(item.get('description', '')),
            np.array([item.get('price', 0)]),
        ])

    def _cosine_similarity(self, a, b):
        return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b) + 1e-8)

オンボーディングによるデータ収集

class OnboardingRecommender:
    """初回登録時のオンボーディングで嗜好情報を収集"""

    def __init__(self, item_pool, n_questions=5):
        self.item_pool = item_pool
        self.n_questions = n_questions

    def generate_onboarding_items(self):
        """オンボーディング用アイテムを選定"""
        # 多様性を確保: カテゴリごとに代表的なアイテム
        categories = set(item['category'] for item in self.item_pool)
        selected = []

        for cat in categories:
            cat_items = [i for i in self.item_pool if i['category'] == cat]
            # カテゴリ内で人気の高いアイテムを選択
            top_item = max(cat_items, key=lambda x: x['popularity'])
            selected.append(top_item)

        # 上位N件を返す
        return sorted(selected, key=lambda x: -x['popularity'])[:self.n_questions]

    def process_responses(self, user_id, responses):
        """オンボーディング回答から初期プロファイルを構築"""
        liked = [r['item_id'] for r in responses if r['liked']]
        disliked = [r['item_id'] for r in responses if not r['liked']]

        profile = {
            'user_id': user_id,
            'liked_categories': self._extract_categories(liked),
            'disliked_categories': self._extract_categories(disliked),
            'price_preference': self._estimate_price_pref(liked),
        }

        return profile

    def _extract_categories(self, item_ids):
        categories = []
        for iid in item_ids:
            item = next(i for i in self.item_pool if i['id'] == iid)
            categories.append(item['category'])
        return categories

    def _estimate_price_pref(self, liked_ids):
        prices = []
        for iid in liked_ids:
            item = next(i for i in self.item_pool if i['id'] == iid)
            prices.append(item['price'])
        return np.mean(prices) if prices else None

まとめ

チェックリスト

• コールドスタート問題の3パターンを説明できる
• 新規ユーザーへの段階的推薦戦略を設計できる
• 新規アイテムの類似度ベース推薦を理解した
• オンボーディングによるデータ収集を設計できる

次のステップへ

コールドスタート問題への対処を理解した。次は演習でハイブリッド推薦システムを構築しよう。

推定読了時間: 30分