"""
Modern RecSys pipeline - mimics industry standard approach.
This handles the full flow: retrieval -> ranking -> reranking.
Each stage filters candidates progressively for speed+accuracy.
"""
import torch
import numpy as np
from typing import Dict, List, Tuple, Any, Optional
import logging
from abc import ABC, abstractmethod
[docs]
class PipelineStage(ABC):
"""Base for stages in the rec pipeline."""
def __init__(self, name: str, config: Dict):
self.name = name
self.config = config
self.log = logging.getLogger(f"Pipeline.{name}")
[docs]
@abstractmethod
def process(self, user_data: Dict, item_pool: List, context: Dict) -> List:
"""
Transform input to output candidates.
user_data: info about the user (features, history, etc)
item_pool: candidate items to score/filter
context: metadata that might be useful
Returns: filtered/scored list of items
"""
pass
[docs]
class RetrievalStage(PipelineStage):
"""
Fast candidate generation - turn millions into thousands.
Uses vector similarity (ANN search) for speed.
Goal: recall over precision.
"""
def __init__(self, model, index, config: Dict):
super().__init__("Retrieval", config)
self.model = model # two-tower or similar
self.index = index # FAISS/similar for ANN search
self.k = config.get("num_candidates", 1000)
[docs]
def process(self, user_data: Dict, item_pool: List, context: Dict) -> List[Tuple[Any, float]]:
"""Pull top-k candidates using vector similarity."""
# encode user into vector space
user_vec = self.model.encode_user(user_data)
if self.index is not None:
# fast ANN search in pre-built index
scores, item_ids = self.index.search(user_vec, self.k)
candidates = [(item_ids[i], scores[i]) for i in range(len(item_ids))]
else:
# fallback: score all items (slow for large catalogs)
candidates = []
for item in item_pool:
item_vec = self.model.encode_item(item)
score = self.model.score(user_vec, item_vec)
candidates.append((item, score))
# take top-k
candidates.sort(key=lambda x: x[1], reverse=True)
candidates = candidates[:self.k]
self.log.debug(f"Retrieved {len(candidates)} candidates from pool of {len(item_pool)}")
return candidates
[docs]
class RankingStage(PipelineStage):
"""
Heavy scoring - turn thousands into tens.
Uses complex feature interactions (DCN, DeepFM, etc).
Goal: precision.
"""
def __init__(self, model, config: Dict):
super().__init__("Ranking", config)
self.model = model # DLRM, Wide&Deep, DCN, etc
self.k = config.get("top_k", 100)
[docs]
def process(self, user_data: Dict, candidates: List[Tuple[Any, float]], context: Dict) -> List[Tuple[Any, float]]:
"""Score candidates with complex model."""
if len(candidates) == 0:
return []
# extract just the items
items = [c[0] for c in candidates]
# batch scoring for efficiency
batch_user = self._prepare_user_batch(user_data, len(items))
batch_items = self._prepare_item_batch(items)
scores = self.model.score_batch(batch_user, batch_items)
# pair items with new scores
ranked = [(items[i], scores[i]) for i in range(len(items))]
ranked.sort(key=lambda x: x[1], reverse=True)
self.log.debug(f"Ranked {len(candidates)} candidates, keeping top {self.k}")
return ranked[:self.k]
def _prepare_user_batch(self, user_data: Dict, batch_sz: int):
"""Repeat user features for each candidate item."""
# simple implementation - override for custom logic
return {k: v.repeat(batch_sz, 1) if isinstance(v, torch.Tensor) else [v] * batch_sz
for k, v in user_data.items()}
def _prepare_item_batch(self, items: List):
"""Convert item list to batch format."""
# override based on item representation
return items
[docs]
class RerankingStage(PipelineStage):
"""
Final polish - apply business rules, diversity, freshness.
Can also use RL or LLM for context-aware adjustment.
Goal: user satisfaction + business objectives.
"""
def __init__(self, rules: List, config: Dict):
super().__init__("Reranking", config)
self.rules = rules # list of reranking rules
[docs]
def process(self, user_data: Dict, candidates: List[Tuple[Any, float]], context: Dict) -> List[Tuple[Any, float]]:
"""Apply rules to adjust ranking."""
if len(candidates) == 0:
return []
# apply each rule in sequence
result = candidates
for rule in self.rules:
result = rule.apply(result, user_data, context)
self.log.debug(f"Reranked {len(candidates)} candidates")
return result
[docs]
class DiversityRule:
"""Prevent showing too many similar items."""
def __init__(self, max_similar: int = 3):
self.max_similar = max_similar
[docs]
def apply(self, candidates: List[Tuple[Any, float]], user_data: Dict, context: Dict) -> List[Tuple[Any, float]]:
"""Enforce diversity by clustering candidates."""
# simplified - real impl would use embeddings
return candidates # TODO: implement clustering logic
[docs]
class FreshnessRule:
"""Boost newer items."""
def __init__(self, boost_factor: float = 1.2):
self.boost = boost_factor
[docs]
def apply(self, candidates: List[Tuple[Any, float]], user_data: Dict, context: Dict) -> List[Tuple[Any, float]]:
"""Apply recency boost."""
# check if items have timestamp
result = []
for item, score in candidates:
if hasattr(item, 'created_at'):
# newer = higher boost (simplified)
age_hours = (context.get('current_time', 0) - item.created_at) / 3600
if age_hours < 24:
score *= self.boost
result.append((item, score))
result.sort(key=lambda x: x[1], reverse=True)
return result
[docs]
class StagePipeline:
"""
Stage-based pipeline orchestrator.
A simpler pipeline that chains PipelineStage instances sequentially.
For the full-featured orchestrator with retrieval/ranking/reranking
integration, use RecommendationPipeline from corerec.pipelines.orchestrator.
"""
def __init__(self, config: Dict):
self.config = config
self.stages = []
self.log = logging.getLogger("RecPipeline")
[docs]
def add_stage(self, stage: PipelineStage):
"""Add a processing stage to the pipeline."""
self.stages.append(stage)
self.log.info(f"Added stage: {stage.name}")
[docs]
def recommend(self, user_data: Dict, item_pool: List, top_k: int = 10,
context: Optional[Dict] = None) -> List[Tuple[Any, float]]:
"""
Run full pipeline to generate recommendations.
user_data: dict with user info (id, features, history)
item_pool: list of all available items (or None to use indexed items)
top_k: how many final recommendations to return
context: extra info (time, location, etc)
Returns: list of (item, score) tuples
"""
if context is None:
context = {}
self.log.info(f"Starting recommendation for user {user_data.get('user_id', 'unknown')}")
# initial pool
candidates = item_pool if item_pool is not None else []
# run through each stage
for stage in self.stages:
start = context.get('current_time', 0)
candidates = stage.process(user_data, candidates, context)
elapsed = context.get('current_time', 0) - start
self.log.debug(f"{stage.name} took {elapsed:.3f}s, output: {len(candidates)} items")
if len(candidates) == 0:
self.log.warning(f"No candidates after {stage.name}, stopping early")
break
# final trim to top_k
final = candidates[:top_k]
self.log.info(f"Pipeline complete: returning {len(final)} recommendations")
return final
[docs]
def explain(self, user_data: Dict, item_id: Any, context: Optional[Dict] = None) -> Dict:
"""
Explain why an item was/wasn't recommended.
Useful for debugging and transparency.
"""
# track item through pipeline
explanation = {
'item_id': item_id,
'stages': []
}
# TODO: implement stage-by-stage tracking
# this would involve running pipeline with logging enabled
return explanation