执行摘要

• 一句话：按请求过滤 KV 卸载二级层
• 推荐动作：该 PR 设计成熟，经过了充分的设计讨论和评审迭代。强烈建议负责 KV 卸载和多层缓存的开发者仔细阅读 TierMatcher/TierFilter 的匹配逻辑以及 _parse_tier_filter 的健壮性处理。对于未来计划引入 locality 动态过滤的团队，该实现提供了良好的扩展基础。

功能与动机

为了支持灵活的多层 KV 卸载策略，需要对每个请求控制参与查找和提升的二级层，避免从不适用的介质或跨地域层加载数据。详细设计讨论见 PR #48123 和关联 Issue #48281。

实现拆解

1. 核心抽象定义：在 vllm/v1/kv_offload/base.py 中新增 Medium（CPU/STORAGE）和 Locality（LOCAL/REMOTE）枚举；定义 TierMatcher NamedTuple，包含可选的 medium 和 locality 字段及其 matches() 方法；定义冻结 TierFilter 数据类，持有 matchers 序列，提供 allows() 方法检查给定的 medium/locality 是否匹配任意 matcher。
2. 输入解析与验证：在 vllm/distributed/kv_transfer/kv_connector/v1/offloading/scheduler.py 中实现 _parse_tier_filter()，将用户传入的字典列表（如 [{"medium":"storage"}]）转换为 TierFilter 实例，未知介质 / 地域值记录警告并跳过该条目，空列表视为“拒绝全部”，完全无效输入回退为 TierFilter.ALL。
3. 请求上下文集成：扩展 ReqContext 增加 load_tier_filter 字段（默认 TierFilter.ALL），在 _create_req_context() 中从 req.kv_transfer_params 提取 kv_load_tiers 并解析为过滤器。
4. 过滤执行：在 TieringOffloadingManager.lookup() 的二级层循环中，对每个候选层调用 req_context.load_tier_filter.allows(tier.medium, tier.locality)，不匹配则跳过该层。主层不受过滤影响。
5. 事件类型强化：OffloadingEvent.medium 由 str 改为 Medium 枚举；在 OffloadingEventsTracker 中引入 _MEDIUM_TO_EVENT_STR 字典将 Medium 映射为 KV 事件字符串常量；移除 MEDIUM_FS/MEDIUM_OBJ 常量，新增 MEDIUM_STORAGE。

关键文件：

• vllm/v1/kv_offload/base.py（模块 核心抽象；类别 source；类型 core-logic；符号 Medium, Locality, TierMatcher, matches）: 定义了整个过滤系统的核心抽象：Medium、Locality、TierMatcher、TierFilter 以及 ReqContext 上的 load_tier_filter 字段。所有过滤逻辑依赖于此。
• vllm/distributed/kv_transfer/kv_connector/v1/offloading/scheduler.py（模块 调度器；类别 source；类型 core-logic；符号 _parse_tier_filter）: 实现 _parse_tier_filter 函数将请求参数解析为 TierFilter，包含验证和降级逻辑。同时修改 _create_req_context 在创建 ReqContext 时提取过滤配置。
• tests/v1/kv_offload/tiering/test_tiering_offloading.py（模块 测试；类别 test；类型 test-coverage；符号 test_tier_filter_skips_filtered_secondary, test_tier_filter_allows_matching_secondary, test_parse_tier_filter_valid, test_parse_tier_filter_invalid_returns_all）: 新增 5 个测试用例覆盖 TierFilter 解析与过滤行为，包括有效/无效输入、空列表、过滤跳过/放行等场景。
• vllm/distributed/kv_transfer/kv_connector/v1/offloading/events.py（模块 事件层；类别 source；类型 dependency-wiring）: OffloadingEvent.medium 类型由 str 改为 Medium，引入 _MEDIUM_TO_EVENT_STR 映射，确保与 kv_events 字符串常量兼容；_placeholder_stored 等函数相应调整。
• vllm/v1/kv_offload/tiering/base.py（模块 二级层基类；类别 source；类型 dependency-wiring）: 在 SecondaryTierManager 基类中添加 ClassVar medium 属性和实例属性 locality，使子类可以声明自身的介质和地域，供 TierFilter.allows 检查。
• vllm/v1/kv_offload/tiering/manager.py（模块 层级管理器；类别 source；类型 core-logic）: TieringOffloadingManager.lookup 中增加过滤逻辑：对每个二级层调用 allows()，不匹配则跳过。实现了过滤的核心执行点。

关键符号：Medium, Locality, TierMatcher, matches, TierFilter, allows, _parse_tier_filter, _create_req_context

关键源码片段

vllm/v1/kv_offload/base.py

定义了整个过滤系统的核心抽象：Medium、Locality、TierMatcher、TierFilter 以及 ReqContext 上的 load_tier_filter 字段。所有过滤逻辑依赖于此。

# vllm/v1/kv_offload/base.py
# 介质枚举：CPU 表示本地内存，STORAGE 表示远端存储（FS/OBJ 统一）
class Medium(Enum):
    CPU = 'CPU'
    STORAGE = 'STORAGE'
​
​
class Locality(Enum):
    """存储地域相对于发布实例"""
    LOCAL = 'LOCAL'
    REMOTE = 'REMOTE'
​
​
class TierMatcher(NamedTuple):
    """单个匹配条件，两个字段均为 None 时匹配任意层"""
    medium: Medium | None = None
    locality: Locality | None = None
​
    def matches(self, medium: Medium | None, locality: Locality | None) -> bool:
        # 自身字段为 None 或待匹配值为 None 时视为匹配（None 表示未约束）
        medium_matches = self.medium is None or medium is None or self.medium == medium
        locality_matches = (
            self.locality is None or locality is None or self.locality == locality
        )
        return medium_matches and locality_matches
​
​
@dataclass(frozen=True)
class TierFilter:
    """按请求的层级过滤器，包含一组 TierMatcher"""
    matchers: tuple[TierMatcher, ...] = ()
​
    ALL: ClassVar['TierFilter'] # 允许类变量声明
​
    def allows(self, medium: Medium | None, locality: Locality | None) -> bool:
        # 如果自身是全局 ALL 实例，直接放行
        if self is TierFilter.ALL:
            return True
        # 任意一个 matcher 匹配即可
        return any(m.matches(medium, locality) for m in self.matchers)
​
​
TierFilter.ALL = TierFilter(matchers=(TierMatcher(),))

vllm/distributed/kv_transfer/kv_connector/v1/offloading/scheduler.py

实现 _parse_tier_filter 函数将请求参数解析为 TierFilter，包含验证和降级逻辑。同时修改 _create_req_context 在创建 ReqContext 时提取过滤配置。

# vllm/distributed/kv_transfer/kv_connector/v1/offloading/scheduler.py
​
KV_LOAD_TIERS_KEY = 'kv_load_tiers'
MATCHER_MEDIUM_KEY = 'medium'
MATCHER_LOCALITY_KEY = 'locality'
​
​
def _parse_tier_filter(raw: Any) -> TierFilter:
    """将 `kv_load_tiers` 原始值解析为 `TierFilter`。
​
    支持格式：`[{'medium': 'cpu'}, {'medium': 'storage', 'locality': 'local'}]`
    无效条目会被跳过并记录警告，空列表视为拒绝全部。
    """
    if not isinstance(raw, list):
        logger.warning('_parse_tier_filter: 期望 list，但收到 %s；忽略', type(raw).__name__)
        return TierFilter.ALL
​
    matchers: list[TierMatcher] = []
    for entry in raw:
        if not isinstance(entry, dict):
            logger.warning('_parse_tier_filter: 条目不是 dict；跳过')
            continue
        medium: Medium | None = None
        locality: Locality | None = None
​
        raw_medium = entry.get(MATCHER_MEDIUM_KEY)
        if raw_medium is not None:
            try:
                medium = Medium(raw_medium.upper())
            except (ValueError, AttributeError):
                logger.warning(
                    '_parse_tier_filter: 未知 medium %r；跳过条目', raw_medium
                )
                continue
​
        raw_locality = entry.get(MATCHER_LOCALITY_KEY)
        if raw_locality is not None:
            try:
                locality = Locality(raw_locality.upper())
            except (ValueError, AttributeError):
                logger.warning(
                    '_parse_tier_filter: 未知 locality %r；跳过条目', raw_locality
                )
                continue
​
        matchers.append(TierMatcher(medium=medium, locality=locality))
​
    if not matchers:
        if not raw: # 显式传入空列表 []，用户想拒绝所有二级层
            return TierFilter(matchers=())
        # 所有条目均无效，回退为允许全部
        return TierFilter.ALL
​
    return TierFilter(matchers=tuple(matchers))
​
​
def _create_req_context(req: Request) -> ReqContext:
    params = req.kv_transfer_params
    load_filter = TierFilter.ALL
    if params:
        raw = params.get(KV_LOAD_TIERS_KEY)
        if raw is not None:
            load_filter = _parse_tier_filter(raw)
    return ReqContext(
        req_id=req.request_id,
        kv_transfer_params=params,
        load_tier_filter=load_filter,
    )

评论区精华

1. 过滤范围：orozery 提出 filter 是否应跳过主层，Change72 确认此 PR 先只过滤二级层，主层始终参与（主层是提升目标，需要持续推进）。
2. 介质粒度：orozery 和 Change72 澄清 Medium 应为 CPU/STORAGE 而非 FS/OBJ，后者在事件层归入 STORAGE。
3. None 语义：orozery 解释 None 在介质 / 地域上表示“动态/依赖上下文”（如 P2P 层的地域随请求变化），过滤器应放行并让层自决。
4. 空列表行为：orozery 指出空列表应拒绝全部而非回退，作者修复为返回空 matchers 的 TierFilter。
5. 日志频率：ronensc 最初使用 logger.warning_once() 避免热路径日志垃圾，orozery 建议改用普通 logger.warning 以便问题可见，最终使用 warning。

• Filter only secondary tiers? (design): 过滤仅应用于二级层，主层始终参与。后续可按需扩展主层过滤。
• Medium granularity: FS/OBJ vs STORAGE (design): Medium 枚举包含 CPU 和 STORAGE，FS/OBJ 在事件层映射为 STORAGE。
• None semantics in TierMatcher.matches (design): tier 端 None 表示未约束，通过任何 filter；matcher 端 None 也表示匹配任意。具体约束由 tier 自身决定。
• Empty list as deny-all (correctness): _parse_tier_filter 对空列表返回空 matchers 的 TierFilter，从而不会允许任何层。
• Unknown filter value handling (question): 使用 logger.warning 记录未知值，跳过该条目，保留有效条目；全部无效则回退 ALL。

风险与影响

• 风险：
  1. 热路径性能：lookup() 对每个 block 调用，新增的 allows() 执行入参检查和 any() 匹配，对于拥有多个 matcher 的场景可能增加微秒级延迟。建议在负载测试中对比开启/关闭过滤的吞吐。
  2. API 兼容性：OffloadingEvent.medium 从 str 改为 Medium 枚举，外部事件消费者（如日志分析、监控）如果直接读取字符串值需要适配。MEDIUM_FS/MEDIUM_OBJ 常量被移除，可能破坏遗留导入。
  3. 日志噪声：_parse_tier_filter 使用普通 logger.warning，若客户端持续发送无效 kv_load_tiers，可能产生大量警告日志。
  4. 配置扩散：Medium/Locality 枚举与 kv_events.py 中的字符串常量存在两套表示，需要维护映射关系，未来可能产生不一致。
• 影响：
  • 用户：新增 kv_load_tiers API，请求方可按介质和地域过滤二级层，适用于多数据中心多层次的 KV 卸载场景。默认行为不变（加载所有层）。
  • 系统：二级层查找可跳过不匹配层，减少不必要的跨域网络调用；主层不受影响，避免干扰提升路径。
  • 团队：需要在新版本发布时说明 API 变更和枚举迁移；潜在的高频日志需要监控。
  • 风险标记：热路径开销, API串兼容性, 日志噪声

关联脉络

• PR #49502 [3/N][Core][KV Connector] Support reliable partial-tail KV offload for sub-block prompts: 同属 KV offloading 功能线，涉及多层卸载的请求级别控制，本 PR 提供过滤机制可与之配合使用。
• PR #49843 [Bugfix][ROCm] Use batch DMA for CPU KV cache loads: CPU 层 KV 卸载的 bug 修复，与本 PR 的 CPU 介质过滤逻辑可能产生交互，确保过滤不影响基本卸载路径。