Pipeline System

Pipeline System#

The pipeline module provides multi-stage recommendation orchestration following the industry-standard Retrieval → Ranking → Reranking pattern.

Quick Start#

from corerec.pipelines import RecommendationPipeline, PipelineConfig

pipeline = RecommendationPipeline(
    config=PipelineConfig(retrieval_k=200, ranking_k=50, final_k=10)
)
pipeline.add_retriever(my_retriever, weight=1.0)
pipeline.set_ranker(my_ranker)
pipeline.add_reranker(my_reranker)

result = pipeline.recommend(user_id=123, top_k=10)

Orchestrator#

Recommendation Pipeline

Orchestrates the multi-stage recommendation process: 1. Retrieval (multiple sources) 2. Ranking (scoring) 3. Reranking (post-processing)

class corerec.pipelines.orchestrator.PipelineConfig(retrieval_k: int = 500, ranking_k: int = 100, final_k: int = 10, fusion_strategy: str = 'rrf')[source]#

Bases: object

Configuration for a recommendation pipeline.

retrieval_k: int = 500#: candidates to retrieve per retriever

ranking_k: int = 100#: candidates to pass to ranking

final_k: int = 10#: final recommendations to return

fusion_strategy: str = 'rrf'#: how to merge retrieval results

class corerec.pipelines.orchestrator.PipelineResult(items: ~typing.List[~typing.Any], scores: ~typing.List[float], retrieval_candidates: int = 0, ranking_candidates: int = 0, final_candidates: int = 0, retrieval_ms: float = 0.0, ranking_ms: float = 0.0, reranking_ms: float = 0.0, total_ms: float = 0.0, stage_results: ~typing.Dict[str, ~typing.Any] = <factory>)[source]#

Bases: object

Output from a pipeline recommendation.

Includes the final recommendations plus debugging info about what happened at each stage.

items: List[Any]#

scores: List[float]#

retrieval_candidates: int = 0#

ranking_candidates: int = 0#

final_candidates: int = 0#

retrieval_ms: float = 0.0#

ranking_ms: float = 0.0#

reranking_ms: float = 0.0#

total_ms: float = 0.0#

stage_results: Dict[str, Any]#

to_list() → List[Tuple[Any, float]][source]#: Return as list of (item_id, score) tuples.

class corerec.pipelines.orchestrator.RecommendationPipeline(config: PipelineConfig | None = None, name: str = 'pipeline')[source]#

Bases: object

Multi-stage recommendation pipeline.

Combines multiple retrievers, a ranker, and multiple rerankers into a single cohesive pipeline. Handles data flow, timing, and provides debugging info.

Example:

pipeline = RecommendationPipeline()

# add retrieval sources (multiple is fine)
pipeline.add_retriever(collaborative_retriever, weight=1.0)
pipeline.add_retriever(semantic_retriever, weight=0.5)

# set the ranker (only one)
pipeline.set_ranker(pointwise_ranker)

# add rerankers (applied in order)
pipeline.add_reranker(diversity_reranker)
pipeline.add_reranker(business_rules_reranker)

# get recommendations
result = pipeline.recommend(
    user_id=123,
    context={'session_id': 'abc'},
    top_k=10
)

__init__(config: PipelineConfig | None = None, name: str = 'pipeline')[source]#

Parameters:

config – pipeline configuration
name – identifier for this pipeline

add_retriever(retriever: BaseRetriever, weight: float = 1.0) → RecommendationPipeline[source]#

Add a retrieval source.

Parameters:

retriever – the retriever to add
weight – weight for ensemble fusion (higher = more influence)

set_ranker(ranker: BaseRanker) → RecommendationPipeline[source]#

Set the ranking model.

Parameters:: ranker – the ranker to use for Stage 2

add_reranker(reranker: BaseReranker) → RecommendationPipeline[source]#

Add a reranking step.

Rerankers are applied in the order they’re added.

Parameters:: reranker – the reranker to add

recommend(query: Any, context: Dict[str, Any] | None = None, top_k: int | None = None, **kwargs) → PipelineResult[source]#

Generate recommendations.

Parameters:

query – user_id, text query, or embedding
context – user features, session info, etc.
top_k – number of final recommendations (overrides config)

Returns:

PipelineResult with recommendations and timing info

recommend_batch(queries: List[Any], contexts: List[Dict[str, Any]] | None = None, top_k: int | None = None, **kwargs) → List[PipelineResult][source]#

Batch recommendation for multiple queries.

Default implementation loops; can be optimized in subclasses.

corerec.pipelines.orchestrator.PipelineOrchestrator#: alias of RecommendationPipeline

Stage Abstractions#

Modern RecSys pipeline - mimics industry standard approach.

This handles the full flow: retrieval -> ranking -> reranking. Each stage filters candidates progressively for speed+accuracy.

class corerec.pipelines.recommendation_pipeline.PipelineStage(name: str, config: Dict)[source]#

Bases: ABC

Base for stages in the rec pipeline.

abstractmethod process(user_data: Dict, item_pool: List, context: Dict) → List[source]#

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

class corerec.pipelines.recommendation_pipeline.RetrievalStage(model, index, config: Dict)[source]#

Bases: PipelineStage

Fast candidate generation - turn millions into thousands.

Uses vector similarity (ANN search) for speed. Goal: recall over precision.

process(user_data: Dict, item_pool: List, context: Dict) → List[Tuple[Any, float]][source]#: Pull top-k candidates using vector similarity.

class corerec.pipelines.recommendation_pipeline.RankingStage(model, config: Dict)[source]#

Bases: PipelineStage

Heavy scoring - turn thousands into tens.

Uses complex feature interactions (DCN, DeepFM, etc). Goal: precision.

process(user_data: Dict, candidates: List[Tuple[Any, float]], context: Dict) → List[Tuple[Any, float]][source]#: Score candidates with complex model.

class corerec.pipelines.recommendation_pipeline.RerankingStage(rules: List, config: Dict)[source]#

Bases: PipelineStage

Final polish - apply business rules, diversity, freshness.

Can also use RL or LLM for context-aware adjustment. Goal: user satisfaction + business objectives.

process(user_data: Dict, candidates: List[Tuple[Any, float]], context: Dict) → List[Tuple[Any, float]][source]#: Apply rules to adjust ranking.

class corerec.pipelines.recommendation_pipeline.DiversityRule(max_similar: int = 3)[source]#

Bases: object

Prevent showing too many similar items.

apply(candidates: List[Tuple[Any, float]], user_data: Dict, context: Dict) → List[Tuple[Any, float]][source]#: Enforce diversity by clustering candidates.

class corerec.pipelines.recommendation_pipeline.FreshnessRule(boost_factor: float = 1.2)[source]#

Bases: object

Boost newer items.

apply(candidates: List[Tuple[Any, float]], user_data: Dict, context: Dict) → List[Tuple[Any, float]][source]#: Apply recency boost.

class corerec.pipelines.recommendation_pipeline.StagePipeline(config: Dict)[source]#

Bases: object

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.

add_stage(stage: PipelineStage)[source]#: Add a processing stage to the pipeline.

recommend(user_data: Dict, item_pool: List, top_k: int = 10, context: Dict | None = None) → List[Tuple[Any, float]][source]#

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

explain(user_data: Dict, item_id: Any, context: Dict | None = None) → Dict[source]#: Explain why an item was/wasn’t recommended. Useful for debugging and transparency.

Configuration#

Pipeline Configuration

Load pipeline configurations from YAML or dict. Enables config-driven pipeline construction.

corerec.pipelines.config.load_pipeline_config(source: str | Path | Dict) → Dict[str, Any][source]#

Load pipeline configuration from YAML file or dict.

Parameters:: source – path to YAML file or config dict
Returns:: Configuration dictionary

Example YAML:

pipeline:
  name: production_v1

  retrieval:
    k: 500
    fusion: rrf
    sources:
      - type: collaborative
        model: sar
        weight: 1.0
      - type: semantic
        encoder: all-MiniLM-L6-v2
        weight: 0.5

  ranking:
    k: 100
    type: pointwise
    model_path: models/ranker.pkl

  reranking:
    - type: diversity
      lambda: 0.7
    - type: business
      boost:
        - item: 123
          multiplier: 2.0

corerec.pipelines.config.build_pipeline_from_config(config: Dict[str, Any]) → RecommendationPipeline[source]#

Build a pipeline from configuration dict.

This is a factory function that instantiates all components based on the config specification.

Parameters:: config – pipeline configuration
Returns:: Configured RecommendationPipeline

Data Pipeline#

Data Pipeline for Recommendation Data

Composable pipeline for data transformation and preprocessing.

Author: Vishesh Yadav (mail: sciencely98@gmail.com)

class corerec.pipelines.data_pipeline.DataTransformer[source]#

Bases: ABC

Base class for data transformers.

All transformers should inherit from this and implement transform().

Author: Vishesh Yadav (mail: sciencely98@gmail.com)

abstractmethod transform(data: DataFrame) → DataFrame[source]#

Transform data.

Parameters:: data – Input DataFrame
Returns:: Transformed DataFrame

Author: Vishesh Yadav (mail: sciencely98@gmail.com)

fit(data: DataFrame) → DataTransformer[source]#

Fit transformer to data (optional).

Override this if your transformer needs to learn from data.

Author: Vishesh Yadav (mail: sciencely98@gmail.com)

fit_transform(data: DataFrame) → DataFrame[source]#: Fit and transform in one step.

class corerec.pipelines.data_pipeline.DataPipeline[source]#

Bases: object

Composable data transformation pipeline.

Chains multiple transformers together for clean data processing.

Architecture:

Raw Data → [Transformer 1] → [Transformer 2] → ... → Clean Data
     ↓           ↓                ↓                     ↓
Missing Values  Encoding      Scaling            Ready for Model

Example:

from corerec.pipelines import DataPipeline
from corerec.pipelines import MissingValueHandler, CategoryEncoder

pipeline = DataPipeline()
pipeline.add(MissingValueHandler('mean'))
pipeline.add(CategoryEncoder())

# Transform data
clean_data = pipeline.transform(raw_data)

# Or fit then transform
pipeline.fit(train_data)
clean_train = pipeline.transform(train_data)
clean_test = pipeline.transform(test_data)  # Uses fitted params

Author: Vishesh Yadav (mail: sciencely98@gmail.com)

__init__()[source]#

Initialize empty pipeline.

Author: Vishesh Yadav (mail: sciencely98@gmail.com)

add(transformer: DataTransformer) → DataPipeline[source]#

Add a transformer to the pipeline.

Parameters:: transformer – Data transformer to add
Returns:: self for method chaining

Example:

pipeline.add(MissingValueHandler()).add(CategoryEncoder())

Author: Vishesh Yadav (mail: sciencely98@gmail.com)

fit(data: DataFrame) → DataPipeline[source]#

Fit all transformers to data.

Parameters:: data – Training data to fit transformers on
Returns:: self for method chaining

Author: Vishesh Yadav (mail: sciencely98@gmail.com)

transform(data: DataFrame) → DataFrame[source]#

Apply all transformers to data.

Parameters:: data – Data to transform
Returns:: Transformed data

Author: Vishesh Yadav (mail: sciencely98@gmail.com)

fit_transform(data: DataFrame) → DataFrame[source]#

Fit and transform in one step.

Parameters:: data – Data to fit and transform
Returns:: Transformed data

Author: Vishesh Yadav (mail: sciencely98@gmail.com)

get_transformers() → List[DataTransformer][source]#: Get list of transformers in pipeline.

is_fitted() → bool[source]#: Check if pipeline is fitted.

Pipeline System

Contents

Pipeline System#

Quick Start#

Orchestrator#

Stage Abstractions#

Configuration#

Data Pipeline#