--- title: "CandidatePipeline trait（流水线执行器）" description: "`CandidatePipeline` 执行器 trait：getter 与钩子、execute 与 execute_traced、十阶段走读、结果结构与统计" --- > Documentation Index > Fetch the complete documentation index at: https://x-algorithm.vergego.ai/llms.txt > Use this file to discover all available pages before exploring further. # CandidatePipeline trait（流水线执行器） ## 概述 `CandidatePipeline` 是 `candidate-pipeline` 框架的执行器 trait。它定义在 `candidate-pipeline/candidate_pipeline.rs:81-462`。它接收一个查询，按固定顺序调度组件，产出排好序的候选和执行追踪。 它与 [七个组件 trait](https://x-algorithm.vergego.ai/concepts/candidate-pipeline-framework/) 分工：组件 trait 定义每类工作怎么做；执行器定义这些工作的顺序、并行方式与数据传递。业务流水线实现这个 trait 时，只实现返回组件列表的 getter。框架以默认方法提供编排逻辑，业务代码不重写。home-mixer 有六个实现，见 [home-mixer 编排层](https://x-algorithm.vergego.ai/concepts/home-mixer-orchestration/)。 ## trait 定义 ```rust // candidate-pipeline/candidate_pipeline.rs:81-103 #[async_trait] pub trait CandidatePipeline: Send + Sync where Q: PipelineQuery, C: PipelineCandidate, { fn query_hydrators(&self) -> &[Box>]; fn dependent_query_hydrators(&self) -> &[Box>] { &[] } fn sources(&self) -> &[Box>]; fn hydrators(&self) -> &[Box>]; fn filters(&self) -> &[Box>]; fn scorers(&self) -> &[Box>]; fn selector(&self) -> &dyn Selector; fn post_selection_hydrators(&self) -> &[Box>]; fn post_selection_filters(&self) -> &[Box>]; fn side_effects(&self) -> Arc>>>; fn result_size(&self) -> usize; fn result_size_for(&self, _query: &Q) -> usize { self.result_size() } fn finalize(&self, _query: &Q, _candidates: &mut Vec) {} // 其余方法都是默认实现 } ``` ## 完整方法集 ### 业务必须实现的 getter（10 个） | 方法 | 返回 | 说明 | |------|------|------| | `query_hydrators` | `&[Box>]` | 第一批 Query 水合器 | | `sources` | `&[Box>]` | 候选源 | | `hydrators` | `&[Box>]` | 候选水合器 | | `filters` | `&[Box>]` | 预打分过滤器，按列表顺序执行 | | `scorers` | `&[Box>]` | 打分器，按列表顺序执行 | | `selector` | `&dyn Selector` | 唯一的选择器 | | `post_selection_hydrators` | `&[Box>]` | 选后水合器 | | `post_selection_filters` | `&[Box>]` | 选后过滤器 | | `side_effects` | `Arc>>>` | 以 `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`。 ## 调用链 ```mermaid 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：十阶段走读 ```rust // 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 枚举 ```rust // 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() ```rust // candidate-pipeline/candidate_pipeline.rs:51-54 pub struct PipelineComponents { pub stage: PipelineStage, pub components: Vec, } ``` `components()` 按 Query 水合、依赖型 Query 水合、源、水合、过滤、打分、选择、选后水合、选后过滤、side effect 的顺序返回每阶段的组件名（`:181-229`）。 ## PipelineResult 结构 ```rust // candidate-pipeline/candidate_pipeline.rs:56-61 pub struct PipelineResult { pub retrieved_candidates: Vec, // 候选水合后、过滤前的全部候选 pub filtered_candidates: Vec, // 两道过滤删掉的候选 pub selected_candidates: Vec, // 最终结果 pub query: Arc, // 水合后的查询 } ``` `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 约束 ```rust // 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 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 实现 ## 相关页面 - [candidate-pipeline 框架与七个组件 trait](https://x-algorithm.vergego.ai/concepts/candidate-pipeline-framework/) —— 组件 trait 与执行模型 - [home-mixer 编排层](https://x-algorithm.vergego.ai/concepts/home-mixer-orchestration/) —— 六个实现 - [候选选择与成型](https://x-algorithm.vergego.ai/concepts/candidate-selection/) —— ⑦ 到截断的细节 - [过滤流水线](https://x-algorithm.vergego.ai/concepts/filtering-pipeline/) —— ⑤ 与 ⑨ 的过滤器 - [Under the Hood 透明度工具](https://x-algorithm.vergego.ai/concepts/under-the-hood/) —— 追踪的对外用途 - [系统架构总览](https://x-algorithm.vergego.ai/concepts/system-architecture/) —— 执行器在系统中的位置 Source: https://x-algorithm.vergego.ai/entities/candidate-pipeline/index.mdx