006d3d058a
- Score and boost contributions now captured per candidate in signal_snapshot:
score_by_sort() returns (f64, Vec<(String, f64)>) with named signal values
(e.g. view_velocity, share_velocity, view, like). Boost contributions are
appended as `{signal}_boost`. compute_raw_score() threads the snapshot
through; score_inner() and score_personalized() write it onto ScoredCandidate
instead of always initialising to vec![].
- Quickstart switched from `trending` to `hot` profile. The trending profile
scores by 24h velocity, which requires signals to arrive over real elapsed
time to populate hour-level bucket aggregates — impossible in a self-contained
demo. The hot profile (AllTime view count + age decay) works with any timestamp
and produces clearly differentiated scores. Signals now use different counts
per item to drive meaningful ranking output.
434 lines
17 KiB
Rust
434 lines
17 KiB
Rust
//! Profile executor: scores and ranks candidates according to a `RankingProfile`.
|
||
//!
|
||
//! The executor implements a five-stage pipeline:
|
||
//! 1. **Gate**: filter out candidates that fail any gate threshold
|
||
//! 2. **Score**: compute raw scores per candidate using the profile's `Sort` mode
|
||
//! 3. **Sort**: order candidates by descending score
|
||
//! 4. **Normalize**: min-max normalize scores to `[0.0, 1.0]`
|
||
//! 5. **Diversity**: apply per-creator and format-mix constraints via `DiversitySelector`
|
||
|
||
pub mod context;
|
||
pub mod formulas;
|
||
pub mod helpers;
|
||
|
||
// Re-export all public items so that `crate::ranking::executor::Foo` paths continue to work.
|
||
pub use context::{ScoredCandidate, UserContext};
|
||
|
||
use std::collections::HashMap;
|
||
|
||
use crate::load::DegradationLevel;
|
||
use crate::schema::{EntityId, Timestamp};
|
||
use crate::session::SessionContext;
|
||
use crate::signals::SignalLedger;
|
||
|
||
use super::profile::RankingProfile;
|
||
use crate::entities::{UserSignalIndex, UserStateIndex};
|
||
use formulas::INTERACTION_BOOST_WEIGHT;
|
||
use helpers::{normalize, passes_gates, read_agg};
|
||
|
||
// -- Executor -----------------------------------------------------------------
|
||
|
||
/// Scores and ranks candidates according to a `RankingProfile`.
|
||
///
|
||
/// Holds a reference to the `SignalLedger` for reading decay scores, windowed
|
||
/// counts, and velocities. The executor itself is stateless -- all configuration
|
||
/// comes from the profile passed to `score()`.
|
||
pub struct ProfileExecutor<'a> {
|
||
ledger: &'a SignalLedger,
|
||
/// Optional reference to the `created_at` range index for age-aware hot scoring.
|
||
/// When `None`, `score_hot` falls back to a 24-hour age assumption.
|
||
///
|
||
/// TODO: Wire up a proper reverse lookup (`EntityId` -> `created_at_ns`) for O(1)
|
||
/// age resolution. The `RangeIndex<u64>` maps `(value, entity_id)` for range queries,
|
||
/// but does not support reverse lookup by entity ID. A separate `HashMap<u32, u64>`
|
||
/// would be needed. Deferred to M4.
|
||
#[allow(dead_code)]
|
||
pub created_at_index: Option<&'a crate::storage::indexes::range::RangeIndex<u64>>,
|
||
/// M6: co-engagement index for `related` profile scoring.
|
||
co_engagement: Option<&'a crate::entities::CoEngagementIndex>,
|
||
/// M6: seed item for co-engagement scoring (used with `similar_to`).
|
||
seed_item: Option<EntityId>,
|
||
/// M6: per-user signal index for social-graph-scoped trending.
|
||
user_signal_index: Option<&'a UserSignalIndex>,
|
||
/// M6: resolved social subgraph user IDs for trending scope.
|
||
social_graph_users: Option<Vec<u64>>,
|
||
/// M6p3: item metadata for metadata-based sorts (alphabetical, duration).
|
||
item_metadata: Option<&'a HashMap<u64, HashMap<String, String>>>,
|
||
/// M6p3: user state + user ID for `DateSaved` sort.
|
||
user_state_for_date_saved: Option<(&'a UserStateIndex, u64)>,
|
||
/// M7p2: degradation level for coarse-aggregate window substitution.
|
||
degradation_level: DegradationLevel,
|
||
}
|
||
|
||
impl<'a> ProfileExecutor<'a> {
|
||
#[must_use]
|
||
pub const fn new(ledger: &'a SignalLedger) -> Self {
|
||
Self {
|
||
ledger,
|
||
created_at_index: None,
|
||
co_engagement: None,
|
||
seed_item: None,
|
||
user_signal_index: None,
|
||
social_graph_users: None,
|
||
item_metadata: None,
|
||
user_state_for_date_saved: None,
|
||
degradation_level: DegradationLevel::Full,
|
||
}
|
||
}
|
||
|
||
/// Set the degradation level for coarse-aggregate window substitution.
|
||
///
|
||
/// Under `CoarseAggregates`, `SignalAgg::Value` reads use `Window::AllTime`
|
||
/// and `SignalAgg::Velocity` reads use `Window::TwentyFourHours`, reducing
|
||
/// the number of fine-grained bucket scans during high-load scoring.
|
||
#[must_use]
|
||
pub(crate) const fn with_degradation_level(mut self, level: DegradationLevel) -> Self {
|
||
self.degradation_level = level;
|
||
self
|
||
}
|
||
|
||
/// Attach a `created_at` index for age-aware hot scoring.
|
||
#[must_use]
|
||
pub const fn with_created_at_index(
|
||
mut self,
|
||
index: &'a crate::storage::indexes::range::RangeIndex<u64>,
|
||
) -> Self {
|
||
self.created_at_index = Some(index);
|
||
self
|
||
}
|
||
|
||
/// Attach a co-engagement index for `related` profile scoring.
|
||
#[must_use]
|
||
pub const fn with_co_engagement(
|
||
mut self,
|
||
index: &'a crate::entities::CoEngagementIndex,
|
||
) -> Self {
|
||
self.co_engagement = Some(index);
|
||
self
|
||
}
|
||
|
||
/// Set the seed item for co-engagement scoring (from `similar_to`).
|
||
#[must_use]
|
||
pub const fn with_seed(mut self, seed: EntityId) -> Self {
|
||
self.seed_item = Some(seed);
|
||
self
|
||
}
|
||
|
||
/// Attach a per-user signal index for social-graph-scoped trending.
|
||
#[must_use]
|
||
pub const fn with_user_signal_index(mut self, index: &'a UserSignalIndex) -> Self {
|
||
self.user_signal_index = Some(index);
|
||
self
|
||
}
|
||
|
||
/// Set the resolved social subgraph user IDs for trending scope.
|
||
///
|
||
/// When set together with `user_signal_index`, `score_trending` uses
|
||
/// aggregate velocity across these users instead of the global ledger.
|
||
#[must_use]
|
||
pub fn with_social_graph_users(mut self, users: Vec<u64>) -> Self {
|
||
self.social_graph_users = Some(users);
|
||
self
|
||
}
|
||
|
||
/// Attach item metadata for metadata-based sorts (alphabetical, duration).
|
||
#[must_use]
|
||
pub const fn with_item_metadata(
|
||
mut self,
|
||
metadata: &'a HashMap<u64, HashMap<String, String>>,
|
||
) -> Self {
|
||
self.item_metadata = Some(metadata);
|
||
self
|
||
}
|
||
|
||
/// Attach user state context for `DateSaved` sort.
|
||
#[must_use]
|
||
pub const fn with_date_saved_context(
|
||
mut self,
|
||
user_state: &'a UserStateIndex,
|
||
user_id: u64,
|
||
) -> Self {
|
||
self.user_state_for_date_saved = Some((user_state, user_id));
|
||
self
|
||
}
|
||
|
||
/// Score and rank a set of candidate entities according to the given profile.
|
||
///
|
||
/// Executes the five-stage pipeline: gate -> score -> sort -> normalize -> diversity.
|
||
///
|
||
/// - `candidates`: entity IDs to consider; gate-failing candidates are excluded.
|
||
/// - `profile`: the ranking profile controlling sort mode, boosts, gates, and diversity.
|
||
/// - `now`: the current timestamp, used for decay calculations.
|
||
///
|
||
/// Returns candidates sorted by descending normalized score in `[0.0, 1.0]`.
|
||
#[must_use]
|
||
pub fn score(
|
||
&self,
|
||
candidates: &[EntityId],
|
||
profile: &RankingProfile,
|
||
now: Timestamp,
|
||
) -> Vec<ScoredCandidate> {
|
||
self.score_inner(candidates, profile, now, None, &HashMap::new())
|
||
}
|
||
|
||
/// Score and rank candidates with an optional FOR SESSION boost applied.
|
||
///
|
||
/// When `session_ctx` is `Some`, applies an additive boost per candidate based
|
||
/// on keyword hint matching: keywords from session annotations are matched against
|
||
/// each item's metadata values.
|
||
///
|
||
/// The session boost is applied after base scoring and before normalization.
|
||
#[must_use]
|
||
pub fn score_with_session(
|
||
&self,
|
||
candidates: &[EntityId],
|
||
profile: &RankingProfile,
|
||
now: Timestamp,
|
||
session_ctx: Option<&SessionContext>,
|
||
item_metadata: &HashMap<u64, HashMap<String, String>>,
|
||
) -> Vec<ScoredCandidate> {
|
||
self.score_inner(candidates, profile, now, session_ctx, item_metadata)
|
||
}
|
||
|
||
/// Score and rank candidates with user-specific personalization.
|
||
///
|
||
/// In addition to the base scoring and optional session boost, this method
|
||
/// applies an additive creator-interaction boost: items from creators the
|
||
/// user has interacted with receive a positive boost proportional to the
|
||
/// interaction weight times `INTERACTION_BOOST_WEIGHT`.
|
||
///
|
||
/// The interaction boost is applied after base scoring and session boost,
|
||
/// but before normalization, so it participates in the min-max range.
|
||
#[must_use]
|
||
pub fn score_personalized(
|
||
&self,
|
||
candidates: &[EntityId],
|
||
profile: &RankingProfile,
|
||
now: Timestamp,
|
||
session_ctx: Option<&SessionContext>,
|
||
user_ctx: &UserContext,
|
||
item_metadata: &HashMap<u64, HashMap<String, String>>,
|
||
) -> Vec<ScoredCandidate> {
|
||
let empty: HashMap<String, String> = HashMap::new();
|
||
let mut scored: Vec<ScoredCandidate> = candidates
|
||
.iter()
|
||
.filter(|&&entity_id| passes_gates(entity_id, &profile.gates, self.ledger))
|
||
.map(|&entity_id| {
|
||
let (raw, snapshot) = self.compute_raw_score(entity_id, profile, now);
|
||
let metadata = item_metadata.get(&entity_id.as_u64()).map_or(&empty, |m| m);
|
||
let session_boost = session_ctx.map_or(0.0, |ctx| {
|
||
Self::session_boost(entity_id.as_u64(), ctx, metadata)
|
||
});
|
||
|
||
// Creator-interaction boost: look up the per-item boost from
|
||
// the pre-computed map. Items from highly-interacted creators
|
||
// get a positive additive boost before normalization.
|
||
#[allow(clippy::cast_possible_truncation)]
|
||
let interaction_boost = user_ctx
|
||
.creator_interaction_boosts
|
||
.get(&(entity_id.as_u64() as u32))
|
||
.map_or(0.0, |&weight| weight * INTERACTION_BOOST_WEIGHT);
|
||
|
||
ScoredCandidate {
|
||
entity_id,
|
||
score: raw + session_boost + interaction_boost,
|
||
signal_snapshot: snapshot,
|
||
creator_id: None,
|
||
format: None,
|
||
}
|
||
})
|
||
.collect();
|
||
|
||
// `sort_unstable_by` is safe here: ranking score ordering is not stability-sensitive.
|
||
// `select_nth_unstable_by` optimization requires knowing `limit` at this site;
|
||
// truncation is applied by the caller after normalization.
|
||
scored.sort_unstable_by(|a, b| {
|
||
b.score
|
||
.partial_cmp(&a.score)
|
||
.unwrap_or(std::cmp::Ordering::Equal)
|
||
});
|
||
|
||
// Normalize to [0, 1].
|
||
normalize(&mut scored);
|
||
|
||
// Apply profile diversity constraints if any are set.
|
||
if profile.diversity.max_per_creator.is_some()
|
||
|| profile.diversity.format_mix_max_fraction.is_some()
|
||
{
|
||
use super::diversity::{DiversityConstraints, DiversitySelector};
|
||
let constraints = DiversityConstraints {
|
||
max_per_creator: profile.diversity.max_per_creator,
|
||
format_mix_max_fraction: profile.diversity.format_mix_max_fraction,
|
||
..DiversityConstraints::new()
|
||
};
|
||
let result = DiversitySelector::select(&scored, &constraints, scored.len());
|
||
scored = result.selected;
|
||
}
|
||
|
||
scored
|
||
}
|
||
|
||
/// Shared five-stage scoring pipeline: gate -> score -> sort -> normalize -> diversity.
|
||
///
|
||
/// When `session_ctx` is `Some`, an additive session boost is applied to each
|
||
/// candidate's raw score before normalization. `item_metadata` provides the
|
||
/// metadata for keyword hint matching; pass an empty map when not needed.
|
||
fn score_inner(
|
||
&self,
|
||
candidates: &[EntityId],
|
||
profile: &RankingProfile,
|
||
now: Timestamp,
|
||
session_ctx: Option<&SessionContext>,
|
||
item_metadata: &HashMap<u64, HashMap<String, String>>,
|
||
) -> Vec<ScoredCandidate> {
|
||
let empty: HashMap<String, String> = HashMap::new();
|
||
let mut scored: Vec<ScoredCandidate> = candidates
|
||
.iter()
|
||
.filter(|&&entity_id| passes_gates(entity_id, &profile.gates, self.ledger))
|
||
.map(|&entity_id| {
|
||
let (raw, snapshot) = self.compute_raw_score(entity_id, profile, now);
|
||
let metadata = item_metadata.get(&entity_id.as_u64()).map_or(&empty, |m| m);
|
||
let boost = session_ctx.map_or(0.0, |ctx| {
|
||
Self::session_boost(entity_id.as_u64(), ctx, metadata)
|
||
});
|
||
ScoredCandidate {
|
||
entity_id,
|
||
score: raw + boost,
|
||
signal_snapshot: snapshot,
|
||
creator_id: None,
|
||
format: None,
|
||
}
|
||
})
|
||
.collect();
|
||
|
||
// `sort_unstable_by` is safe here: ranking score ordering is not stability-sensitive.
|
||
// `select_nth_unstable_by` optimization requires knowing `limit` at this site;
|
||
// truncation is applied by the caller after normalization.
|
||
scored.sort_unstable_by(|a, b| {
|
||
b.score
|
||
.partial_cmp(&a.score)
|
||
.unwrap_or(std::cmp::Ordering::Equal)
|
||
});
|
||
|
||
// Normalize to [0, 1].
|
||
normalize(&mut scored);
|
||
|
||
// Apply profile diversity constraints if any are set.
|
||
if profile.diversity.max_per_creator.is_some()
|
||
|| profile.diversity.format_mix_max_fraction.is_some()
|
||
{
|
||
use super::diversity::{DiversityConstraints, DiversitySelector};
|
||
let constraints = DiversityConstraints {
|
||
max_per_creator: profile.diversity.max_per_creator,
|
||
format_mix_max_fraction: profile.diversity.format_mix_max_fraction,
|
||
..DiversityConstraints::new()
|
||
};
|
||
let result = DiversitySelector::select(&scored, &constraints, scored.len());
|
||
scored = result.selected;
|
||
}
|
||
|
||
scored
|
||
}
|
||
|
||
/// Compute the session-specific score boost for a single entity.
|
||
///
|
||
/// - Keyword hint score: fraction of annotation keywords that match any metadata value.
|
||
/// `hint_score = hits / total_keywords * 0.3` (capped via the [0,1] fraction).
|
||
/// - Velocity boost: `vel / (vel + 1) * 0.2` (Michaelis-Menten saturation).
|
||
fn session_boost(
|
||
_entity_id: u64,
|
||
ctx: &SessionContext,
|
||
metadata: &HashMap<String, String>,
|
||
) -> f64 {
|
||
let total_kw: usize = ctx
|
||
.keywords
|
||
.iter()
|
||
.flat_map(|h| h.split_whitespace())
|
||
.count();
|
||
let hits = ctx
|
||
.keywords
|
||
.iter()
|
||
.flat_map(|h| h.split_whitespace())
|
||
.filter(|kw| {
|
||
let kw_lower = kw.to_lowercase();
|
||
metadata
|
||
.values()
|
||
.any(|v| v.to_lowercase().contains(&kw_lower))
|
||
})
|
||
.count();
|
||
#[allow(clippy::cast_precision_loss)]
|
||
let hint_score = if total_kw == 0 {
|
||
0.0
|
||
} else {
|
||
hits as f64 / total_kw as f64
|
||
};
|
||
let vel_norm = ctx.reward_velocity / (ctx.reward_velocity + 1.0);
|
||
hint_score * 0.3 + vel_norm * 0.2
|
||
}
|
||
|
||
/// Compute raw score and signal snapshot for a single candidate.
|
||
///
|
||
/// Returns `(score, snapshot)` where `snapshot` lists the raw signal values
|
||
/// that contributed, for explain-ability in API responses.
|
||
fn compute_raw_score(
|
||
&self,
|
||
entity_id: EntityId,
|
||
profile: &RankingProfile,
|
||
now: Timestamp,
|
||
) -> (f64, Vec<(String, f64)>) {
|
||
let (base, mut snapshot) = self.score_by_sort(entity_id, profile.sort.as_ref(), now);
|
||
|
||
// Apply boosts and capture their contributions in the snapshot.
|
||
let boost_sum: f64 = profile
|
||
.boosts
|
||
.iter()
|
||
.map(|b| {
|
||
let val = read_agg(
|
||
entity_id,
|
||
&b.signal,
|
||
&b.agg,
|
||
b.window,
|
||
self.ledger,
|
||
self.degradation_level,
|
||
);
|
||
let weighted = b.weight * val;
|
||
if weighted != 0.0 {
|
||
snapshot.push((format!("{}_boost", b.signal), weighted));
|
||
}
|
||
weighted
|
||
})
|
||
.sum();
|
||
|
||
// M6: co-engagement blend for `related`-style queries.
|
||
//
|
||
// The spec states: `final = embedding_sim × 0.6 + co_engagement × 0.3 + signal_score × 0.1`
|
||
// (a normalized convex combination). The actual formula is additive rather than
|
||
// a strict convex combination because embedding similarity is not available as
|
||
// a per-candidate scoring signal in the executor: ANN retrieval (Stage 1) uses
|
||
// the embedding for candidate *selection*, not for re-scoring. The 0.3 co-engagement
|
||
// weight is correct; incorporating embedding_sim as a multiplicative factor requires
|
||
// storing per-candidate ANN distances and plumbing them through the executor context
|
||
// (deferred to a future milestone).
|
||
//
|
||
// Effective formula: base_signal_score + boost_sum + co_eng_score × 0.3
|
||
let co_eng_boost = if let (Some(co_eng), Some(seed)) = (self.co_engagement, self.seed_item)
|
||
{
|
||
let boost = f64::from(co_eng.score(seed, entity_id)) * 0.3;
|
||
if boost != 0.0 {
|
||
snapshot.push(("co_engagement".to_string(), boost));
|
||
}
|
||
boost
|
||
} else {
|
||
0.0
|
||
};
|
||
|
||
(base + boost_sum + co_eng_boost, snapshot)
|
||
}
|
||
}
|
||
|
||
mod scoring;
|
||
|
||
#[cfg(test)]
|
||
#[allow(clippy::unwrap_used, clippy::float_cmp, clippy::cast_precision_loss)]
|
||
mod tests;
|