recommendation-system

レコメンデーションシステム

特徴量ストア、マルチティアキャッシュ、A/B テスト、包括的なモニタリングを備えた、スケーラブルなレコメンデーションシステム向けの本番環境対応アーキテクチャです。

このスキルを使用する場面

このスキルは、次の場合に読み込んでください。

• レコメンデーション API の構築：パーソナライズされたレコメンデーションを大規模に提供する
• キャッシュの実装：サブミリ秒のレイテンシを実現するマルチティアキャッシュ
• A/B テストの実施：レコメンデーションアルゴリズムを試す
• 品質のモニタリング：CTR、コンバージョン、多様性、網羅率を追跡する
• パフォーマンスの最適化：レイテンシを削減し、スループットを向上させる
• 特徴量エンジニアリング：特徴量ストアでユーザー／アイテムの特徴量を管理する

クイックスタート：5 ステップでレコメンデーション API を作成

# 1. Install dependencies
pip install "fastapi>=0.109.0" "redis>=5.0.0" "prometheus-client>=0.19.0"

# 2. Start Redis (for caching and feature store)
docker run -d -p 6379:6379 redis:alpine

# 3. Create recommendation service: app.py
cat > app.py << 'EOF'
from fastapi import FastAPI
from pydantic import BaseModel
from typing import List
import redis
import json

app = FastAPI()
cache = redis.Redis(host='localhost', port=6379, decode_responses=True)

class RecommendationResponse(BaseModel):
    user_id: str
    items: List[str]
    cached: bool

@app.post("/recommendations", response_model=RecommendationResponse)
async def get_recommendations(user_id: str, n: int = 10):
    # Check cache
    cache_key = f"recs:{user_id}:{n}"
    cached = cache.get(cache_key)

    if cached:
        return RecommendationResponse(
            user_id=user_id,
            items=json.loads(cached),
            cached=True
        )

    # Generate recommendations (simplified)
    items = [f"item_{i}" for i in range(n)]

    # Cache for 5 minutes
    cache.setex(cache_key, 300, json.dumps(items))

    return RecommendationResponse(
        user_id=user_id,
        items=items,
        cached=False
    )

@app.get("/health")
async def health():
    return {"status": "healthy"}
EOF

# 4. Run API
uvicorn app:app --host 0.0.0.0 --port 8000

# 5. Test
curl -X POST "http://localhost:8000/recommendations?user_id=user_123&n=10"

結果：キャッシュを備えた、動作するレコメンデーション API を 5 分以内に構築できます。

システムアーキテクチャ

┌─────────────┐     ┌─────────────┐     ┌─────────────┐
│ User Events │────▶│ Feature     │────▶│ Model       │
│ (clicks,    │     │ Store       │     │ Serving     │
│  purchases) │     │ (Redis)     │     │             │
└─────────────┘     └─────────────┘     └─────────────┘
                           │                    │
                           ▼                    ▼
                    ┌─────────────┐     ┌─────────────┐
                    │ Training    │     │ API         │
                    │ Pipeline    │     │ (FastAPI)   │
                    └─────────────┘     └─────────────┘
                                               │
                                               ▼
                                        ┌─────────────┐
                                        │ Monitoring  │
                                        │ (Prometheus)│
                                        └─────────────┘

コアコンポーネント

1. 特徴量ストア

ユーザーとアイテムの特徴量を一元管理するストレージです。

import redis
import json

class FeatureStore:
    """Fast feature access with Redis caching."""

    def __init__(self, redis_client):
        self.redis = redis_client
        self.ttl = 3600  # 1 hour

    def get_user_features(self, user_id: str) -> dict:
        cache_key = f"user_features:{user_id}"
        cached = self.redis.get(cache_key)

        if cached:
            return json.loads(cached)

        # Fetch from database
        features = fetch_from_db(user_id)

        # Cache
        self.redis.setex(cache_key, self.ttl, json.dumps(features))
        return features

2. モデルサービング

A/B テスト用に複数のモデルを提供します。

class ModelServing:
    """Serve multiple recommendation models."""

    def __init__(self):
        self.models = {}

    def register_model(self, name: str, model, is_default: bool = False):
        self.models[name] = model
        if is_default:
            self.default_model = name

    def predict(self, user_features: dict, item_features: list, model_name: str = None):
        model = self.models.get(model_name or self.default_model)
        return model.predict(user_features, item_features)

3. キャッシュレイヤー

低レイテンシを実現するマルチティアキャッシュです。

class TieredCache:
    """L1 (memory) -> L2 (Redis) -> L3 (database)."""

    def __init__(self, redis_client):
        self.l1_cache = {}  # In-memory
        self.redis = redis_client  # L2

    def get(self, key: str):
        # L1: In-memory (fastest)
        if key in self.l1_cache:
            return self.l1_cache[key]

        # L2: Redis
        cached = self.redis.get(key)
        if cached:
            value = json.loads(cached)
            self.l1_cache[key] = value  # Promote to L1
            return value

        # L3: Miss (fetch from database)
        return None

主要な指標

指標	説明	目標
CTR	クリックスルー率	>5%
コンバージョン率	レコメンデーション経由の購入	>2%
P95 レイテンシ	応答時間の 95 パーセンタイル	<200ms
キャッシュヒット率	キャッシュから提供された割合	>80%
網羅率	レコメンデーションされたカタログの割合	>50%
多様性	レコメンデーションの多様さ	>0.7

既知の問題を防ぐ

1. 新規ユーザーのコールドスタート

問題：履歴のないユーザーにレコメンデーションがなく、最初の利用体験が損なわれます。

解決策：人気度に基づくフォールバックを使用します。

def get_recommendations(user_id: str, n: int = 10):
    user_features = feature_store.get_user_features(user_id)

    # Check if new user (no purchase history)
    if user_features.get('total_purchases', 0) == 0:
        # Fallback to popular items
        return get_popular_items(n)

    # Personalized recommendations
    return generate_personalized_recs(user_id, n)

2. ユーザーの操作時にキャッシュを無効化する

問題：ユーザーが商品を購入しても、キャッシュに購入済みの商品がレコメンデーションとして表示され続けます。

解決策：関連する操作があったときにキャッシュを無効化します。

INVALIDATING_ACTIONS = {'purchase', 'rating', 'add_to_cart'}

def on_user_action(user_id: str, action: str):
    if action in INVALIDATING_ACTIONS:
        cache_key = f"recs:{user_id}:*"
        redis_client.delete(cache_key)
        logger.info(f"Invalidated cache for {user_id} due to {action}")

3. キャッシュ期限切れ時のサンダリングハード

問題：多数のユーザーのキャッシュが同時に期限切れになり、データベースやモデルに過剰な負荷がかかります。

解決策：TTL にランダムなジッターを追加します。

import random

def set_cache(key: str, value: dict, base_ttl: int = 300):
    # Add ±10% jitter
    jitter = random.uniform(-0.1, 0.1) * base_ttl
    ttl = int(base_ttl + jitter)
    redis_client.setex(key, ttl, json.dumps(value))

4. 多様性の不足によるフィルターバブル

問題：レコメンデーションが似通いすぎて、ユーザーに同じカテゴリばかりが表示されます。

解決策：多様性の制約を実装します。

def rank_with_diversity(items: list, scores: list, n: int = 10):
    selected = []
    category_counts = {}

    for item, score in sorted(zip(items, scores), key=lambda x: -x[1]):
        category = item['category']

        # Limit 3 items per category
        if category_counts.get(category, 0) >= 3:
            continue

        selected.append(item)
        category_counts[category] = category_counts.get(category, 0) + 1

        if len(selected) >= n:
            break

    return selected

5. モニタリング不足による気づかれない品質低下

問題：レコメンデーションの品質が低下しても、ユーザーから苦情が出るまで誰も気づきません。

解決策：アラートを設定して継続的にモニタリングします。

from prometheus_client import Counter, Histogram

recommendation_clicks = Counter('recommendation_clicks_total')
recommendation_latency = Histogram('recommendation_latency_seconds')

@app.post("/recommendations")
async def get_recommendations(user_id: str):
    start = time.time()

    recs = generate_recs(user_id)

    latency = time.time() - start
    recommendation_latency.observe(latency)

    return recs

@app.post("/track/click")
async def track_click(user_id: str, item_id: str):
    recommendation_clicks.inc()
    # Alert if CTR drops below 3%

6. 特徴量の陳腐化による古いレコメンデーション

問題：ユーザーの好みが変わっても特徴量が更新されず、レコメンデーションが適切でなくなります。

解決策：適切な TTL と更新トリガーを設定します。

class FeatureStore:
    def __init__(self, redis_client):
        self.redis = redis_client
        # Shorter TTL for frequently changing features
        self.user_ttl = 300  # 5 minutes
        self.item_ttl = 3600  # 1 hour

    def update_on_event(self, user_id: str, event: str):
        # Invalidate on important events
        if event in ['purchase', 'rating']:
            self.redis.delete(f"user_features:{user_id}")
            logger.info(f"Refreshed features for {user_id}")

7. A/B テストのサンプルサイズ不足

問題：勝者を早まって決定し、統計的に有意でない結果に基づいてしまいます。

解決策：まず必要なサンプルサイズを計算します。

def calculate_sample_size(
    baseline_rate: float,
    min_detectable_effect: float,
    alpha: float = 0.05,
    power: float = 0.8
) -> int:
    """Calculate required sample size per variant."""
    from scipy import stats

    z_alpha = stats.norm.ppf(1 - alpha/2)
    z_beta = stats.norm.ppf(power)

    p1 = baseline_rate
    p2 = baseline_rate * (1 + min_detectable_effect)
    p_avg = (p1 + p2) / 2

    n = (
        (z_alpha + z_beta)**2 * 2 * p_avg * (1 - p_avg) /
        (p2 - p1)**2
    )

    return int(n)

# Example: detect 10% lift with baseline CTR=5%
n_required = calculate_sample_size(
    baseline_rate=0.05,
    min_detectable_effect=0.10
)
print(f"Required sample size: {n_required} per variant")
# Wait until both variants reach this size before concluding

リファレンスを読み込む場面

本番環境向けの詳しい実装については、リファレンスファイルを読み込んでください。

• 本番環境アーキテクチャ：バッチ取得、キャッシュ統合、FastAPI のデプロイパターンを備えた FeatureStore、ModelServing、RecommendationService の完全な実装については、references/production-architecture.md を読み込んでください。

• キャッシュ戦略：マルチティアキャッシュ（L1/L2/L3）、キャッシュウォーミング、無効化戦略、確率的リフレッシュ、サンダリングハードの防止を実装する場合は、references/caching-strategies.md を読み込んでください。

• A/B テストフレームワーク：決定論的なバリアント割り当て、Thompson サンプリング（マルチアームドバンディット）、ベイズ統計および頻度論に基づく有意性検定、実験追跡については、references/ab-testing-framework.md を読み込んでください。

• モニタリングとアラート：Prometheus のメトリクス統合、ダッシュボードのエンドポイント、アラートルール、品質モニタリング（多様性、網羅率）については、references/monitoring-alerting.md を読み込んでください。

ベストプラクティス

1. 特徴量の事前計算：特徴量をオフラインで計算し、キャッシュから提供します
2. バッチ取得：複数のユーザー／アイテムには Redis MGET を使用します
3. 積極的なキャッシュ：ユーザーへのおすすめには 5～15 分の TTL を設定します
4. 適切なフォールバック：パーソナライズに失敗した場合は、人気アイテムを返します
5. すべてを監視：CTR、レイテンシ、多様性、カバレッジを追跡します
6. 継続的な A/B テスト：常に新しいアルゴリズムを試します
7. 多様性の制約：おすすめに多様性を持たせます
8. おすすめの理由を説明：「高評価」「人気」などの理由を提示します

よくあるパターン

おすすめサービス

class RecommendationService:
    def __init__(self, feature_store, model_serving, cache):
        self.feature_store = feature_store
        self.model_serving = model_serving
        self.cache = cache

    def get_recommendations(self, user_id: str, n: int = 10):
        # 1. Check cache
        cached = self.cache.get(f"recs:{user_id}:{n}")
        if cached:
            return cached

        # 2. Get features
        user_features = self.feature_store.get_user_features(user_id)
        candidates = self.get_candidates(user_id)

        # 3. Score candidates
        scores = self.model_serving.predict(user_features, candidates)

        # 4. Rank with diversity
        recommendations = self.rank_with_diversity(candidates, scores, n)

        # 5. Cache
        self.cache.set(f"recs:{user_id}:{n}", recommendations, ttl=300)

        return recommendations

A/B テスト

def assign_variant(user_id: str, experiment_id: str) -> str:
    """Deterministic assignment - same user always gets same variant."""
    import hashlib

    hash_input = f"{user_id}:{experiment_id}"
    hash_value = int(hashlib.md5(hash_input.encode()).hexdigest(), 16)

    # 50/50 split
    return 'control' if hash_value % 2 == 0 else 'treatment'

# Usage
variant = assign_variant('user_123', 'rec_algo_v2')
model_name = 'main' if variant == 'control' else 'experimental'
recs = get_recommendations(user_id, model_name=model_name)

監視

from prometheus_client import Counter, Histogram

requests_total = Counter('recommendation_requests_total', ['status'])
latency_seconds = Histogram('recommendation_latency_seconds')

@app.post("/recommendations")
async def get_recommendations(user_id: str):
    with latency_seconds.time():
        try:
            recs = generate_recs(user_id)
            requests_total.labels(status='success').inc()
            return recs
        except Exception as e:
            requests_total.labels(status='error').inc()
            raise