Best Practices & Checklist

This final lesson consolidates everything into actionable checklists, common pitfalls to avoid, a debugging framework for recommendation quality issues, and answers to the most frequently asked questions by recommendation system engineers.

Production Deployment Checklist

Use this checklist before launching or updating any recommendation system in production.

Data Pipeline

• Training data validation: Check for data freshness (is the latest training data less than 24 hours old?), label quality (are click labels joined correctly?), and feature coverage (what percentage of users/items have all features populated?).
• Feature drift monitoring: Set up alerts for when feature distributions shift significantly from training time. A mean shift of more than 2 standard deviations should trigger investigation.
• Data deduplication: Ensure the same user-item interaction is not counted multiple times (e.g., from retry logic, duplicate events).
• Negative sampling: Verify your negative sampling strategy. Random negatives are biased — consider using in-batch negatives or hard negatives for better training signal.

Model Quality

• Offline metrics baseline: Record NDCG@10, Recall@50, and MAP@10 on a held-out test set. Any new model must beat this baseline.
• Segment-level evaluation: Evaluate separately for power users (50+ interactions), casual users (5–50), and cold-start users (<5). A model that improves average NDCG might degrade cold-start quality.
• Latency budget: Candidate generation <10ms, feature fetching <5ms, ranking <15ms, post-processing <5ms. Total end-to-end p99 <100ms.
• Fallback strategy: What happens when the model service is down? Pre-computed recommendations or popularity-based fallbacks must be ready.

Business Rules

• Out-of-stock filtering: Items with zero stock must never appear in recommendations.
• Already-purchased filtering: For non-consumable items, filter out items the user already owns.
• Content policy: Age-restricted, recalled, or policy-violating items must be filtered before serving.
• Category diversity: No single category should dominate more than 40% of recommendations.
• Sponsored content disclosure: Boosted or sponsored items must be clearly labeled in the UI.

Monitoring

• Serving latency: p50, p95, p99 latency dashboards with alerts at 2x baseline.
• Recommendation quality: Daily CTR, conversion rate, and revenue per session tracked per experiment group.
• Coverage and diversity: Weekly report on catalog coverage and intra-list diversity.
• Error rates: Feature fetch failures, model inference errors, cache miss rates.
• Cold start metrics: Separate dashboards for new user and new item recommendation quality.

Common Pitfalls

Debugging Recommendation Quality

When recommendation quality degrades, follow this systematic debugging framework to identify the root cause.

class RecommendationDebugger:
    """Systematic debugger for recommendation quality issues.

    Follow this top-down debugging flow:
    1. Is the issue with candidate generation or ranking?
    2. Is it a data issue, model issue, or serving issue?
    3. Is it affecting all users or a specific segment?
    """
    def diagnose(self, user_id: str, request_context: dict) -> dict:
        report = {}

        # Step 1: Check candidate generation
        candidates = self.candidate_generator.generate(user_id, request_context)
        report["candidate_count"] = len(candidates)
        report["candidate_sources"] = self._source_breakdown(candidates)

        # Are relevant items in the candidate set?
        known_good_items = self.get_known_good_items(user_id)
        candidates_set = {c["item_id"] for c in candidates}
        report["known_good_in_candidates"] = len(known_good_items & candidates_set)
        report["known_good_total"] = len(known_good_items)

        if report["known_good_in_candidates"] == 0:
            report["diagnosis"] = "CANDIDATE_GENERATION_FAILURE"
            report["action"] = "Relevant items not in candidate set. Check ANN index freshness, embedding quality, and candidate source diversity."
            return report

        # Step 2: Check feature quality
        features = self.feature_service.get_features(user_id, candidates[:10])
        report["feature_coverage"] = self._check_feature_coverage(features)
        report["feature_freshness"] = self._check_feature_freshness(features)

        if report["feature_coverage"] < 0.8:
            report["diagnosis"] = "FEATURE_QUALITY_ISSUE"
            report["action"] = f"Only {report['feature_coverage']*100:.0f}% feature coverage. Check feature pipeline for missing data."
            return report

        # Step 3: Check ranking scores
        ranked = self.ranker.rank_with_debug(user_id, candidates, request_context)
        report["top_10_scores"] = [r["score"] for r in ranked[:10]]
        report["score_spread"] = max(report["top_10_scores"]) - min(report["top_10_scores"])

        if report["score_spread"] < 0.01:
            report["diagnosis"] = "RANKING_DIFFERENTIATION_FAILURE"
            report["action"] = "Ranking scores are too similar. Model may be undertrained or features may not be discriminative."
            return report

        # Step 4: Check business rules impact
        pre_rules = [r["item_id"] for r in ranked[:20]]
        post_rules = self.business_rules.apply(ranked[:50], request_context)
        post_rules_ids = [r["item_id"] for r in post_rules[:20]]
        report["items_removed_by_rules"] = len(set(pre_rules) - set(post_rules_ids))

        if report["items_removed_by_rules"] > 10:
            report["diagnosis"] = "BUSINESS_RULES_TOO_AGGRESSIVE"
            report["action"] = f"{report['items_removed_by_rules']} of top 20 removed by business rules. Review rule configuration."
            return report

        # Step 5: Check for position/diversity changes
        report["diagnosis"] = "NO_OBVIOUS_ISSUE"
        report["action"] = "Pipeline looks healthy. Check if user preferences changed, or if the issue is subjective. Review A/B test metrics."

        return report

    def _source_breakdown(self, candidates):
        from collections import Counter
        return dict(Counter(c.get("source", "unknown") for c in candidates))

    def _check_feature_coverage(self, features):
        total_features = sum(len(f) for f in features.values())
        non_null = sum(1 for f in features.values() for v in f.values() if v is not None)
        return non_null / total_features if total_features > 0 else 0

    def _check_feature_freshness(self, features):
        import time
        stale_count = 0
        for f in features.values():
            last_updated = f.get("_last_updated", 0)
            if time.time() - last_updated > 86400:  # More than 1 day old
                stale_count += 1
        return {"stale_features": stale_count, "total": len(features)}

Frequently Asked Questions

Course Summary

You now have the knowledge to design, build, and operate a production-grade recommendation system. Here is the roadmap from zero to production: