Skip to content
X 推荐算法

CandidatePipeline trait（流水线执行器）

`CandidatePipeline` 执行器 trait：getter 与钩子、execute 与 execute_traced、十阶段走读、结果结构与统计

Updated

概述

CandidatePipeline<Q, C> 是 candidate-pipeline 框架的执行器 trait。它定义在 candidate-pipeline/candidate_pipeline.rs:81-462。它接收一个查询，按固定顺序调度组件，产出排好序的候选和执行追踪。

它与 七个组件 trait 分工：组件 trait 定义每类工作怎么做；执行器定义这些工作的顺序、并行方式与数据传递。业务流水线实现这个 trait 时，只实现返回组件列表的 getter。框架以默认方法提供编排逻辑，业务代码不重写。home-mixer 有六个实现，见 home-mixer 编排层。

trait 定义

// candidate-pipeline/candidate_pipeline.rs:81-103
#[async_trait]
pub trait CandidatePipeline<Q, C>: Send + Sync
where
    Q: PipelineQuery,
    C: PipelineCandidate,
{
    fn query_hydrators(&self) -> &[Box<dyn QueryHydrator<Q>>];
    fn dependent_query_hydrators(&self) -> &[Box<dyn QueryHydrator<Q>>] { &[] }
    fn sources(&self) -> &[Box<dyn Source<Q, C>>];
    fn hydrators(&self) -> &[Box<dyn Hydrator<Q, C>>];
    fn filters(&self) -> &[Box<dyn Filter<Q, C>>];
    fn scorers(&self) -> &[Box<dyn Scorer<Q, C>>];
    fn selector(&self) -> &dyn Selector<Q, C>;
    fn post_selection_hydrators(&self) -> &[Box<dyn Hydrator<Q, C>>];
    fn post_selection_filters(&self) -> &[Box<dyn Filter<Q, C>>];
    fn side_effects(&self) -> Arc<Vec<Box<dyn SideEffect<Q, C>>>>;
    fn result_size(&self) -> usize;
    fn result_size_for(&self, _query: &Q) -> usize { self.result_size() }
    fn finalize(&self, _query: &Q, _candidates: &mut Vec<C>) {}
    // 其余方法都是默认实现
}

完整方法集

业务必须实现的 getter（10 个）

方法 返回 说明
query_hydrators &[Box<dyn QueryHydrator<Q>>] 第一批 Query 水合器
sources &[Box<dyn Source<Q, C>>] 候选源
hydrators &[Box<dyn Hydrator<Q, C>>] 候选水合器
filters &[Box<dyn Filter<Q, C>>] 预打分过滤器，按列表顺序执行
scorers &[Box<dyn Scorer<Q, C>>] 打分器，按列表顺序执行
selector &dyn Selector<Q, C> 唯一的选择器
post_selection_hydrators &[Box<dyn Hydrator<Q, C>>] 选后水合器
post_selection_filters &[Box<dyn Filter<Q, C>>] 选后过滤器
side_effects Arc<Vec<Box<dyn SideEffect<Q, C>>>> 以 Arc 返回，便于移进后台任务
result_size usize 最终结果条数上限

业务可选重写的钩子（3 个）

方法 默认 用途
dependent_query_hydrators 空 需要读取第一批 Query 水合结果的水合器
result_size_for(query) 返回 result_size() 新增：按请求决定结果条数（:100-102）
finalize(query, candidates) 空 截断后对最终列表做最后调整

在 2e2295b 中，home-mixer 的六个实现都没有重写这三个钩子。

框架提供的默认方法

方法 源码 作用
execute :105-107 调用 execute_traced，丢弃追踪
execute_traced :109-119 添加 pipeline 指标标签，在追踪作用域里执行，返回结果与追踪
execute_instrumented :121-124 带 #[receive_stats(name = "execute", latency = Bucket500To2500)] 的包装
execute_stages :126-179 真正的阶段编排
components :181-229 按阶段列出全部组件名
name :231-233 类型名最后一段
hydrate_query / hydrate_dependent_query :239-284 两批 Query 水合
fetch_candidates :291-306 候选源
hydrate / hydrate_post_selection / run_hydrators :312-348 候选水合与选后水合
filter / filter_post_selection / run_filters :358-397 两道过滤
score :403-414 打分
select :416-431 选择
run_side_effects :433-445 后台 side effect
stat_result_size :447-461 结果条数指标

旧版的 record_enabled_components、log_stage、log_stage_size、log_filters 已删除。它们的职责并入 pipeline_summary 模块的 StageStats。

调用链

flowchart LR
    A["execute(query)"] --> B["execute_traced"]
    B --> C["pipeline_summary::scope"]
    C --> D["execute_instrumented"]
    D --> E["execute_stages"]
    B -.->|"返回"| R["PipelineResult 与可选的 PipelineTrace"]

home-mixer 的服务层调用 execute_traced，并在 Under the Hood 请求里把追踪返回给客户端（home-mixer/scored_posts_server.rs:60-61、:96-98，home-mixer/for_you_server.rs:41-48）。嵌套执行时，内层流水线的追踪挂到外层当前阶段下，内层的 execute_traced 因此返回 None（candidate-pipeline/pipeline_summary.rs:46-53）。

execute_stages：十阶段走读

// candidate-pipeline/candidate_pipeline.rs:126-179(节选)
let hydrated_query = self.hydrate_query(query).await;                       // ①
let hydrated_query = self.hydrate_dependent_query(hydrated_query).await;    // ②
let candidates = self.fetch_candidates(&hydrated_query).await;              // ③
let hydrated_candidates = self.hydrate(&hydrated_query, candidates).await;  // ④
let (kept_candidates, mut filtered_candidates) =
    self.filter(&hydrated_query, hydrated_candidates.clone());              // ⑤
let scored_candidates = self.score(&hydrated_query, kept_candidates).await; // ⑥
let SelectResult { selected, non_selected } =
    self.select(&hydrated_query, scored_candidates);                        // ⑦
let post = self.hydrate_post_selection(&hydrated_query, selected).await;    // ⑧
let (mut final_candidates, post_filtered) =
    self.filter_post_selection(&hydrated_query, post);                      // ⑨
// 截断到 result_size_for,finalize,统计,⑩ 后台 side effect
# 阶段 执行方式 要点
① Query 水合 并行 先按 enable() 筛选；只对 Ok 结果调用 update()（:239-257）
② 依赖型 Query 水合 并行 列表为空时直接返回，不记阶段统计（:263-284）
③ 候选源 并行 成功结果按源顺序拼接；失败源贡献零条（:291-306）
④ 候选水合 并行 每个水合器读取同一份候选切片；结果逐个写回（:331-348）
⑤ 预打分过滤 顺序 上一个过滤器的 kept 是下一个的输入；删除项累计（:379-397）
⑥ 打分 顺序 每个打分器写回后下一个才开始（:403-414）
⑦ 选择 单次 选择器未启用时全部视为选中（:416-431）
⑧ 选后水合 并行 只作用于入选候选
⑨ 选后过滤 顺序 删除项并入 filtered_candidates（:150-152）
⑩ side effect 后台 tokio::spawn，不等待完成（:433-445）

⑨ 之后的收尾（:154-171）：

  1. 按 result_size_for(query) 截断。截掉的候选并入 non_selected。
  2. 调用 finalize。
  3. stat_result_size 记录 {name}.execute 的结果条数直方图；结果为空时额外计数 result_empty。
  4. pipeline_summary::emit 输出一行汇总日志。
  5. 把查询包进 Arc，构造 SideEffectInput（选中候选克隆一份，落选候选直接移入），启动 side effect。

⑤ 之前，框架对 hydrated_candidates 做了一次 clone()（:137）。原件作为 retrieved_candidates 返回，供调试接口展示“水合后、过滤前”的完整候选集。

PipelineStage 枚举

// candidate-pipeline/candidate_pipeline.rs:23-35
pub enum PipelineStage {
    QueryHydrator, DependentQueryHydrator, Source, Hydrator, PostSelectionHydrator,
    Filter, PostSelectionFilter, Scorer, Selector, SideEffect,
}

枚举有 10 个值，其顺序与执行顺序不同：选后水合、选后过滤排在较前位置。各组件的 run() 现在都接收 PipelineStage 参数，用于给统计加 stage 标签（stat_labels，:37-49）与写入追踪。追踪里的阶段名是 snake_case，例如 post_selection_filters（pipeline_summary.rs:9-24）。

PipelineComponents 与 components()

// candidate-pipeline/candidate_pipeline.rs:51-54
pub struct PipelineComponents {
    pub stage: PipelineStage,
    pub components: Vec<String>,
}

components() 按 Query 水合、依赖型 Query 水合、源、水合、过滤、打分、选择、选后水合、选后过滤、side effect 的顺序返回每阶段的组件名（:181-229）。

PipelineResult 结构

// candidate-pipeline/candidate_pipeline.rs:56-61
pub struct PipelineResult<Q, C> {
    pub retrieved_candidates: Vec<C>,  // 候选水合后、过滤前的全部候选
    pub filtered_candidates: Vec<C>,   // 两道过滤删掉的候选
    pub selected_candidates: Vec<C>,   // 最终结果
    pub query: Arc<Q>,                 // 水合后的查询
}

PipelineResult::empty()（:63-72）返回全空结果和默认查询。home-mixer 对测试账号直接返回它（home-mixer/scored_posts_server.rs:47-54）。落选与截断的候选不在 PipelineResult 里，框架只把它们交给 side effect。

追踪结构

execute_traced 返回的 PipelineTrace（pipeline_summary.rs:232-264）：

结构 字段
PipelineTrace 流水线名、类型名、总耗时、阶段列表
StageTrace 阶段名、组件总数、启用数、相对起点偏移、耗时、条数、保留数、删除数、组件列表、嵌套流水线
ComponentTrace 组件名、类型名、偏移、耗时、输入数、拉取数（源）、保留数与删除数（过滤器）

Display 实现把追踪格式化成一行。例如 filters{total=21 enabled=14 … kept=… removed=… removed_per_filter=[…]}（pipeline_summary.rs:266-303）。

PipelineQuery / PipelineCandidate 约束

// candidate-pipeline/candidate_pipeline.rs:73-79
pub trait PipelineQuery: Clone + Send + Sync + 'static {
    fn params(&self) -> &xai_feature_switches::Params;
    fn decider(&self) -> Option<&xai_decider::Decider>;
}
pub trait PipelineCandidate: Clone + Send + Sync + 'static {}
impl<T> PipelineCandidate for T where T: Clone + Send + Sync + 'static {}

ScoredPostsQuery 实现 PipelineQuery（home-mixer/models/query.rs:296-304）。PostCandidate 与 FeedItem 经 blanket impl 成为候选类型。

关键事实速查

问题 答案
业务要写哪些方法 10 个 getter；3 个钩子可选
哪些阶段并行 Query 水合、依赖型 Query 水合、源、候选水合、选后水合、side effect
哪些阶段顺序 预打分过滤、打分、选后过滤
组件失败会怎样 框架跳过该组件或该条目；条数不对时，水合器或打分器的整批结果作废
截断发生在哪 选后过滤之后，按 result_size_for
side effect 能否影响结果 不能。side effect 在后台运行，框架丢弃其结果
追踪怎么拿 调用 execute_traced

设计决策

决策 理由
编排逻辑放在默认方法 所有流水线共享同一套顺序、并行与错误应对逻辑
新增 execute_traced 而保留 execute 需要追踪的调用方拿追踪，其余调用方不受影响
截断放在选后过滤之后 选后过滤会删除候选；先过滤再截断，才能保证条数
返回 retrieved_candidates 的副本 调试接口能看到过滤前的全集
side effect 拿 Arc 输入 多个 side effect 共享一份数据，不必逐个克隆

源码锚点

  • candidate-pipeline/candidate_pipeline.rs:23-79 —— 阶段枚举、结果结构、泛型约束
  • candidate-pipeline/candidate_pipeline.rs:81-124 —— trait 定义与执行入口
  • candidate-pipeline/candidate_pipeline.rs:126-179 —— execute_stages()
  • candidate-pipeline/candidate_pipeline.rs:181-461 —— 各阶段方法与统计
  • candidate-pipeline/pipeline_summary.rs:30-327 —— 追踪收集与格式化
  • home-mixer/candidate_pipeline/phoenix_candidate_pipeline.rs:1112-1152 —— 内层流水线的 trait 实现
  • home-mixer/candidate_pipeline/for_you_candidate_pipeline.rs:326-367 —— 外层流水线的 trait 实现

相关页面

X 推荐算法

Type to search…

↑↓ navigate↵ selectEsc close