---
title: "candidate-pipeline 框架与七个组件 trait"
description: "可复用流水线框架：7 个组件 trait、enable 门、并行与顺序执行模型、带缓存的水合器、流水线追踪"
---
> 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.
# candidate-pipeline 框架与七个组件 trait
## 这一页回答什么
`candidate-pipeline` 是 home-mixer 所有流水线共用的框架 crate。本页回答五个问题：它定义了哪些 trait；每个 trait 的职责与默认行为；框架怎样调度它们（哪些并行、哪些顺序）；组件出错时框架怎样应对；框架怎样记录每次执行的追踪。执行器 trait `CandidatePipeline` 的逐方法说明见 [CandidatePipeline trait](https://x-algorithm.vergego.ai/entities/candidate-pipeline/)。
## 核心结论
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](https://x-algorithm.vergego.ai/concepts/under-the-hood/) 使用。
## 框架在系统里的位置
```mermaid
flowchart TB
subgraph FW["candidate-pipeline crate"]
T["7 个组件 trait"]
E["CandidatePipeline trait
execute_stages 编排"]
PS["pipeline_summary
阶段与组件追踪"]
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`：
```rust
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 {}
```
- `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：补全查询
```rust
// candidate-pipeline/query_hydrator.rs:34-36
async fn hydrate(&self, query: &Q) -> Result;
fn update(&self, query: &mut Q, hydrated: Q);
```
`hydrate()` 返回一个新查询，只填本组件负责的字段。`update()` 把这些字段拷回主查询。多个水合器并行读取同一份查询，互不干扰。失败时，`run()` 打印错误并返回 `Err`（`query_hydrator.rs:21-32`），框架跳过它的 `update()`。
### 2. Source：拉候选
```rust
// candidate-pipeline/source.rs:37
async fn source(&self, query: &Q) -> Result, String>;
```
`run()` 记录拉到的条数（`source.rs:22-35`）。框架把所有成功结果拼成一个列表。失败的源贡献零条（`candidate_pipeline.rs:296-302`）。
### 3. Hydrator：补全候选
```rust
// candidate-pipeline/hydrator.rs:21, 61
async fn hydrate(&self, query: &Q, candidates: &[C]) -> Vec>;
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`（`hydrator.rs:76-80`）只有异步的 `get` 与 `insert`。业务代码提供具体存储。
### 4. Filter：过滤候选
```rust
// candidate-pipeline/filter.rs:9-12, 64
pub struct FilterResult { pub kept: Vec, pub removed: Vec }
fn filter(&self, query: &Q, candidates: Vec) -> FilterResult;
```
- `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：打分
```rust
// candidate-pipeline/scorer.rs:48-50
async fn score(&self, query: &Q, candidates: &[C]) -> Vec>;
fn update(&self, candidate: &mut C, scored: C);
```
接口形状与 `Hydrator` 相同，也有条数校验（`scorer.rs:29-45`）。区别在执行方式：打分器顺序执行，后一个打分器能读到前一个写入的分数（`candidate_pipeline.rs:408-411`）。
### 6. Selector：选择
```rust
// candidate-pipeline/selector.rs:9-12, 52, 68, 80
pub struct SelectResult { pub selected: Vec, pub non_selected: Vec }
fn select(&self, _query: &Q, candidates: Vec) -> SelectResult; // 默认:sort 后按 size 截断
fn score(&self, candidate: &C) -> f64; // 必须实现
fn size(&self) -> Option { None }
```
默认 `select()` 调用 `sort()` 按分降序排序，再按 `size()` 截断（`selector.rs:52-78`）。`SelectResult::is_empty()` 要求两边都为空（`:19-21`）。选择器的 `enable()` 为 false 时，框架把全部候选视为选中（`candidate_pipeline.rs:416-431`）。用法见 [候选选择与成型](https://x-algorithm.vergego.ai/concepts/candidate-selection/)。
### 7. SideEffect：副作用
```rust
// candidate-pipeline/side_effect.rs:7-12, 33
pub struct SideEffectInput {
pub query: Arc,
pub selected_candidates: Vec,
pub non_selected_candidates: Vec,
}
async fn side_effect(&self, input: Arc>) -> Result<(), String>;
```
`enable()` 接收 `Arc`（`side_effect.rs:20-22`）。框架用 `tokio::spawn` 把全部 side effect 放进后台任务，并丢弃 `join_all` 的结果（`candidate_pipeline.rs:433-445`）。side effect 失败不影响响应，也不回报给调用方。
## 七个 trait 对照表
| trait | 业务方法 | 同步 / 异步 | 执行方式 | 失败处理 |
|-------|----------|-------------|----------|----------|
| `QueryHydrator` | `hydrate` + `update` | 异步 | 并行 | 跳过该组件 |
| `Source` | `source` | 异步 | 并行 | 该源贡献零条 |
| `Hydrator` | `hydrate` + `update` | 异步 | 并行 | 跳过失败条目；条数不对整批作废 |
| `Filter` | `filter` | 同步 | 顺序 | 无错误通道 |
| `Scorer` | `score` + `update` | 异步 | 顺序 | 同 `Hydrator` |
| `Selector` | `score`（可重写 `select`） | 同步 | 单次 | 无错误通道 |
| `SideEffect` | `side_effect` | 异步 | 后台并行 | 结果丢弃 |
## 执行顺序
`execute_stages()`（`candidate_pipeline.rs:126-179`）的顺序：
```mermaid
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`）：
```rust
pub struct PhoenixCandidatePipeline {
query_hydrators: Vec>>,
sources: Vec>>,
hydrators: Vec>>,
filters: Vec>>,
scorers: Vec>>,
selector: TopKScoreSelector,
post_selection_hydrators: Vec>>,
post_selection_filters: Vec>>,
side_effects: Arc>>>,
}
```
不需要某个阶段时，getter 返回 `&[]`。`ForYouCandidatePipeline` 的候选水合、打分、选后水合采用这种做法（`for_you_candidate_pipeline.rs:336-354`）。`components()`（`candidate_pipeline.rs:181-229`）按阶段列出全部组件名，供自检与调试使用。
## 设计决策
| 决策 | 理由 |
|------|------|
| 组件 trait 与执行器分离 | 业务只写“做什么”，框架统一“怎么调度” |
| 独立工作并行，有依赖的工作顺序 | 水合彼此独立，并行省延迟；过滤与打分依赖前序结果 |
| 组件失败只跳过自己 | 单个下游服务故障不让整次请求失败 |
| 水合与打分校验返回条数 | 防止结果错位写到别的候选上 |
| side effect 放后台 | 写日志、写缓存不进入请求延迟 |
| 追踪按任务局部变量收集并支持嵌套 | 内外两层流水线的追踪合成一棵树，不需要显式传递上下文 |
| `result_size_for` 钩子 | 结果条数可按请求变化，无需改框架 |
## 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](https://x-algorithm.vergego.ai/entities/candidate-pipeline/) —— 执行器方法逐个说明
- [home-mixer 编排层](https://x-algorithm.vergego.ai/concepts/home-mixer-orchestration/) —— 六条流水线怎样装配组件
- [过滤流水线](https://x-algorithm.vergego.ai/concepts/filtering-pipeline/) —— `Filter` 的具体实现
- [打分与排序](https://x-algorithm.vergego.ai/concepts/scoring-and-ranking/) —— `Scorer` 的具体实现
- [候选选择与成型](https://x-algorithm.vergego.ai/concepts/candidate-selection/) —— `Selector` 的具体实现
- [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/concepts/candidate-pipeline-framework/index.mdx