x-pipe-rs

Composable recommendation and feed pipeline framework built on comp-cat-rs.

Inspired by the X algorithm's candidate-pipeline architecture, generalized into a typed, reusable framework where every pipeline stage is a Kleisli arrow in the Io monad.

Architecture

A pipeline assembles six kinds of stages:

1. Source -- produce candidate items from backends
2. Hydrator -- enrich candidates with additional data
3. Filter -- remove candidates that fail criteria
4. Scorer -- assign relevance scores
5. Selector -- choose final candidates (dedup, diversity, budget)
6. SideEffect -- observe without modifying (logging, metrics)

Each trait converts to a Stage<E, A, B>, the universal Kleisli arrow type. Stages compose via Stage::then (Kleisli composition). The final Pipeline is itself a single Stage from query to scored results.

Categorical justification

Stage forms a category whose morphisms are Kleisli arrows A -> Io<E, B>. Composition is associative with identity, and by comp-cat-rs's collapse::monad_is_kan, every such composition is a pair of Kan extensions. Score implements JoinSemilattice, making score merging a colimit (left Kan extension).

Quick start

use comp_cat_rs::effect::io::Io;
use x_pipe_rs::pipeline::Pipeline;
use x_pipe_rs::score::Score;
use x_pipe_rs::scorer::Scorer;
use x_pipe_rs::selector::{Budget, MaxItems, TopNSelector};
use x_pipe_rs::source::Source;

struct MySource;
impl Source<(), String, std::convert::Infallible> for MySource {
    fn candidates(&self, _q: &()) -> Io<std::convert::Infallible, Vec<String>> {
        Io::pure(vec!["hello".into(), "world".into()])
    }
}

struct LengthScorer;
impl Scorer<String, std::convert::Infallible> for LengthScorer {
    fn score(&self, item: &String) -> Io<std::convert::Infallible, Score> {
        Io::pure(Score::new(item.len() as f64).unwrap_or(Score::zero()))
    }
}

let pipeline = Pipeline::from_source(MySource)
    .score(LengthScorer)
    .select(TopNSelector, Budget::new(MaxItems::new(1)));

// Pipeline returns a lazy Io; call .run() only at the boundary.
let results = pipeline.execute(()).run();

Running the example

cargo run --example rss_ranker

License

MIT OR Apache-2.0