candidate-pipeline 框架与七个组件 trait

这一页回答什么

candidate-pipeline 是 home-mixer 所有流水线共用的框架 crate。本页回答五个问题：它定义了哪些 trait；每个 trait 的职责与默认行为；框架怎样调度它们（哪些并行、哪些顺序）；组件出错时框架怎样应对；框架怎样记录每次执行的追踪。执行器 trait CandidatePipeline 的逐方法说明见 CandidatePipeline trait。

核心结论

1. 7 个组件 trait 加 1 个执行器 trait。每个组件 trait 负责一类工作：QueryHydrator、Source、Hydrator、Filter、Scorer、Selector、SideEffect。执行器 CandidatePipeline 把它们排成固定顺序。crate 根导出 10 个模块（candidate-pipeline/lib.rs:6-15），比旧版多了 pipeline_summary。
2. 框架刻意区分并行与顺序。Query 水合、候选源、候选水合、选后水合、side effect 并行执行；过滤器与打分器顺序执行（candidate-pipeline/candidate_pipeline.rs:239-445）。
3. 组件失败不会中断请求。Query 水合器、候选源、水合器、打分器返回 Err 时，框架跳过这份结果，其余组件照常生效。水合器与打分器返回的条数不对时，框架作废整批结果（hydrator.rs:42-58、scorer.rs:29-45）。
4. 每个组件有 enable() 门，默认值为 true。框架在每个阶段先筛掉未启用的组件，并记录总数与启用数（candidate_pipeline.rs:242-243）。
5. 新增流水线追踪。框架每次执行生成一份 PipelineTrace，记录各阶段与各组件的耗时、条数、删除数。嵌套调用的流水线挂在父流水线当前阶段下（candidate-pipeline/pipeline_summary.rs:30-55）。这份追踪供日志与 Under the Hood 使用。

框架在系统里的位置

flowchart TB
    subgraph FW["candidate-pipeline crate"]
        T["7 个组件 trait"]
        E["CandidatePipeline trait<br/>execute_stages 编排"]
        PS["pipeline_summary<br/>阶段与组件追踪"]
    end
    subgraph HM["home-mixer"]
        P1[PhoenixCandidatePipeline]
        P2[ForYouCandidatePipeline]
        P3["其余 4 条流水线"]
        C["约百个组件实现"]
    end
    C -->|实现| T
    P1 -->|实现| E
    P2 -->|实现| E
    P3 -->|实现| E
    E --> PS

业务代码只做两件事：实现组件 trait；在流水线结构体里把组件装进列表。编排、并行、错误应对、统计都由框架完成。

泛型参数 Q 与 C

每个组件 trait 都带泛型 Q（查询）与 C（候选）。约束定义在 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 {}

• PipelineQuery 要求查询能提供 feature switch 参数与 decider。组件的 enable() 通常读取这两项。
• PipelineCandidate 有 blanket impl，任何满足 Clone + Send + Sync + 'static 的类型都自动是候选。home-mixer 用了两种候选：PostCandidate 与 FeedItem。

七个组件 trait

所有组件 trait 都有 enable()（默认 true）和 name()（默认取类型名最后一段，util.rs:1-3）。每个 trait 的 run() 是框架调用的包装方法。它接收当前 PipelineStage，在业务方法前后记录组件统计。

1. QueryHydrator：补全查询

// candidate-pipeline/query_hydrator.rs:34-36
async fn hydrate(&self, query: &Q) -> Result<Q, String>;
fn update(&self, query: &mut Q, hydrated: Q);

hydrate() 返回一个新查询，只填本组件负责的字段。update() 把这些字段拷回主查询。多个水合器并行读取同一份查询，互不干扰。失败时，run() 打印错误并返回 Err（query_hydrator.rs:21-32），框架跳过它的 update()。

2. Source：拉候选

// candidate-pipeline/source.rs:37
async fn source(&self, query: &Q) -> Result<Vec<C>, String>;

run() 记录拉到的条数（source.rs:22-35）。框架把所有成功结果拼成一个列表。失败的源贡献零条（candidate_pipeline.rs:296-302）。

3. Hydrator：补全候选

// candidate-pipeline/hydrator.rs:21, 61
async fn hydrate(&self, query: &Q, candidates: &[C]) -> Vec<Result<C, String>>;
fn update(&self, candidate: &mut C, hydrated: C);

• 返回值是与输入一一对应的结果列表。条数不一致时，run() 把整批替换成 Err（hydrator.rs:42-58）。
• update_all() 只对 Ok 的条目调用 update()（hydrator.rs:63-69）。单条失败不影响其他候选。
• hydrate_for_stage()（hydrator.rs:23-30）是新增钩子，让同一个水合器在候选水合与选后水合两个阶段分别记统计。

CachedHydrator：带缓存的水合器

CachedHydrator（hydrator.rs:82-134）不是一个独立阶段，而是 Hydrator 的一种实现方式。框架为它提供 blanket impl（hydrator.rs:136-222），执行逻辑如下：

1. already_hydrated() 为真的候选直接从自身取值（新增，:105-107、:166-169）。
2. 其余候选用 cache_key_for(query, candidate) 查 CacheStore。cache_key_for 是新增钩子，缓存键可以包含查询字段（:97-99）。
3. 未命中的候选一次性交给 hydrate_from_client() 远程拉取。成功结果写回缓存。
4. 按阶段记录命中与未命中数（:115-133）。

CacheStore<K, V>（hydrator.rs:76-80）只有异步的 get 与 insert。业务代码提供具体存储。

4. Filter：过滤候选

// candidate-pipeline/filter.rs:9-12, 64
pub struct FilterResult<C> { pub kept: Vec<C>, pub removed: Vec<C> }
fn filter(&self, query: &Q, candidates: Vec<C>) -> FilterResult<C>;

• filter() 是同步方法，取得候选的所有权。
• 新增 FilterResult::partition(candidates, keep)（filter.rs:14-27）：它按谓词原地拆分候选，保留输入顺序，对每个候选只调用一次谓词。home-mixer 的多数过滤器使用它。
• run() 记录 kept_count、removed_count、filter_rate，并发出 {name}.run 指标（filter.rs:46-84）。

5. Scorer：打分

// candidate-pipeline/scorer.rs:48-50
async fn score(&self, query: &Q, candidates: &[C]) -> Vec<Result<C, String>>;
fn update(&self, candidate: &mut C, scored: C);

接口形状与 Hydrator 相同，也有条数校验（scorer.rs:29-45）。区别在执行方式：打分器顺序执行，后一个打分器能读到前一个写入的分数（candidate_pipeline.rs:408-411）。

6. Selector：选择

// candidate-pipeline/selector.rs:9-12, 52, 68, 80
pub struct SelectResult<C> { pub selected: Vec<C>, pub non_selected: Vec<C> }
fn select(&self, _query: &Q, candidates: Vec<C>) -> SelectResult<C>;  // 默认:sort 后按 size 截断
fn score(&self, candidate: &C) -> f64;                                // 必须实现
fn size(&self) -> Option<usize> { None }

默认 select() 调用 sort() 按分降序排序，再按 size() 截断（selector.rs:52-78）。SelectResult::is_empty() 要求两边都为空（:19-21）。选择器的 enable() 为 false 时，框架把全部候选视为选中（candidate_pipeline.rs:416-431）。用法见 候选选择与成型。

7. SideEffect：副作用

// candidate-pipeline/side_effect.rs:7-12, 33
pub struct SideEffectInput<Q, C> {
    pub query: Arc<Q>,
    pub selected_candidates: Vec<C>,
    pub non_selected_candidates: Vec<C>,
}
async fn side_effect(&self, input: Arc<SideEffectInput<Q, C>>) -> Result<(), String>;

enable() 接收 Arc<Q>（side_effect.rs:20-22）。框架用 tokio::spawn 把全部 side effect 放进后台任务，并丢弃 join_all 的结果（candidate_pipeline.rs:433-445）。side effect 失败不影响响应，也不回报给调用方。

七个 trait 对照表

执行顺序

execute_stages()（candidate_pipeline.rs:126-179）的顺序：

flowchart TB
    Q1["1 Query 水合 · 并行"] --> Q2["2 依赖型 Query 水合 · 并行"]
    Q2 --> S["3 候选源 · 并行"]
    S --> H["4 候选水合 · 并行"]
    H --> F["5 过滤 · 顺序"]
    F --> SC["6 打分 · 顺序"]
    SC --> SEL["7 选择 · 单次"]
    SEL --> PH["8 选后水合 · 并行"]
    PH --> PF["9 选后过滤 · 顺序"]
    PF --> TR["截断到 result_size_for · finalize"]
    TR --> SE["10 side effect · 后台"]

• 并行阶段使用 join_all。Query 水合器之间不能互相依赖。需要读取前序结果的水合器放进 dependent_query_hydrators()，这批水合器在第一批之后运行（:263-284）。home-mixer 的六条流水线都没有用这个阶段。
• 过滤按顺序执行：run_filters 把上一个过滤器的 kept 交给下一个（:390-394）。前序过滤器缩小集合，后序过滤器可以假定“某类候选已被清掉”。
• 打分按顺序执行：每个打分器 await 完成并写回后，下一个才开始（:408-411）。
• 候选水合与选后水合共用 run_hydrators（:331-348）。两者只有组件列表与阶段标签不同。
• 截断使用 result_size_for(query)，默认等于 result_size()（:100-102、:154-158）。截掉的候选并入 non_selected。

流水线追踪

pipeline_summary 模块是新增的可观测性层。

• execute_traced() 用 pipeline_summary::scope 包住整次执行，并给所有指标加上 pipeline 标签（candidate_pipeline.rs:109-119）。
• 各阶段开始时调用 StageStats::begin，结束时写入耗时、条数、保留与删除数（pipeline_summary.rs:66-141）。各组件的 run() 用 ComponentStats 写入自己的耗时与条数（:143-193）。
• 框架把追踪存放在 tokio 任务局部变量里。嵌套执行的流水线不单独输出追踪，而是把追踪挂到父流水线最后一个阶段的 nested 下（:30-55）。For You 的追踪因此在“候选源”阶段下包含整条 PhoenixCandidatePipeline 的追踪。
• emit() 在每次执行结束时输出一行汇总日志，包含各源拉取数、各过滤器删除数（:57-64、:266-303）。
• home-mixer 把追踪转成 proto，在 Under the Hood 请求里返回，并为每个组件附上 GitHub 源码链接（home-mixer/util/under_the_hood.rs:11-60）。

tracing span 的级别由 quiet-spans feature 控制：开启时为 DEBUG，否则为 INFO（lib.rs:1-4）。

具体流水线怎样插入组件

业务流水线是一个持有组件列表的结构体。它在 impl CandidatePipeline 中实现各 getter，返回这些列表。以 PhoenixCandidatePipeline 为例（home-mixer/candidate_pipeline/phoenix_candidate_pipeline.rs:224-234、:1112-1152）：

pub struct PhoenixCandidatePipeline {
    query_hydrators: Vec<Box<dyn QueryHydrator<ScoredPostsQuery>>>,
    sources: Vec<Box<dyn Source<ScoredPostsQuery, PostCandidate>>>,
    hydrators: Vec<Box<dyn Hydrator<ScoredPostsQuery, PostCandidate>>>,
    filters: Vec<Box<dyn Filter<ScoredPostsQuery, PostCandidate>>>,
    scorers: Vec<Box<dyn Scorer<ScoredPostsQuery, PostCandidate>>>,
    selector: TopKScoreSelector,
    post_selection_hydrators: Vec<Box<dyn Hydrator<ScoredPostsQuery, PostCandidate>>>,
    post_selection_filters: Vec<Box<dyn Filter<ScoredPostsQuery, PostCandidate>>>,
    side_effects: Arc<Vec<Box<dyn SideEffect<ScoredPostsQuery, PostCandidate>>>>,
}

不需要某个阶段时，getter 返回 &[]。ForYouCandidatePipeline 的候选水合、打分、选后水合采用这种做法（for_you_candidate_pipeline.rs:336-354）。components()（candidate_pipeline.rs:181-229）按阶段列出全部组件名，供自检与调试使用。

设计决策

FAQ

Q：一个 Query 水合器失败，下游会怎样？ A：该水合器负责的字段保持默认值，其他水合器的结果照常写入（candidate_pipeline.rs:250-254）。下游组件要自行应对字段缺失。例如， PhoenixSource 在缺少召回序列时返回错误（home-mixer/sources/phoenix_source.rs:113-116）。

Q：过滤器能不能并行？ A：框架不支持。run_filters 是同步循环（candidate_pipeline.rs:379-397）。过滤器本身多是内存计算，Filter trait 也没有异步方法。

源码锚点

• candidate-pipeline/lib.rs:1-15 —— crate 根，10 个模块与 span 级别
• candidate-pipeline/candidate_pipeline.rs:23-35 —— PipelineStage 十阶段
• candidate-pipeline/candidate_pipeline.rs:73-104 —— 泛型约束与执行器 getter
• candidate-pipeline/candidate_pipeline.rs:126-179 —— execute_stages()
• candidate-pipeline/candidate_pipeline.rs:239-445 —— 各阶段的执行方法
• candidate-pipeline/query_hydrator.rs:10-41 —— QueryHydrator
• candidate-pipeline/source.rs:10-42 —— Source
• candidate-pipeline/hydrator.rs:11-222 —— Hydrator、CacheStore、CachedHydrator
• candidate-pipeline/filter.rs:9-85 —— FilterResult 与 Filter
• candidate-pipeline/scorer.rs:9-63 —— Scorer
• candidate-pipeline/selector.rs:9-107 —— SelectResult 与 Selector
• candidate-pipeline/side_effect.rs:7-38 —— SideEffectInput 与 SideEffect
• candidate-pipeline/pipeline_summary.rs:30-327 —— 流水线追踪

相关页面

• CandidatePipeline trait —— 执行器方法逐个说明
• home-mixer 编排层 —— 六条流水线怎样装配组件
• 过滤流水线 —— Filter 的具体实现
• 打分与排序 —— Scorer 的具体实现
• 候选选择与成型 —— Selector 的具体实现
• Under the Hood 透明度工具 —— 流水线追踪的对外用途
• 系统架构总览 —— 框架在系统中的位置