执行摘要

• 一句话：准备 SchedulerKvEventsPublisher 组件并迁移 KV 事件状态
• 推荐动作：建议精读此 PR 以理解调度器组件化重构的手法：分离状态、静态方法适配、内联计算消除循环依赖。可从中学习渐进式重构的技巧。

功能与动机

作为调度器职责分离重构（refactor chain: introduce-kv-events-publisher）的预备步骤，将 KV 事件相关的状态和逻辑从合集中提取，为后续将方法实际迁移到独立组件铺平道路，提升代码可维护性。

实现拆解

1. 创建新组件文件 (scheduler_components/kv_events_publisher.py)：将原 scheduler_metrics_mixin.py 中的 KvMetrics 数据类逐字节复制到新文件；添加 SchedulerKvEventsPublisher 类骨架，其 __post_init__ 委托给旧的 mixin 初始化方法，以便在不移动方法的情况下完成注入。
2. 重构 mixin 方法 (scheduler_metrics_mixin.py)：删除本地的 KvMetrics 定义，改为从新模块导入；将 init_kv_events、emit_kv_metrics（原 _emit_kv_metrics）、publish_kv_events（原 _publish_kv_events）转换为 @staticmethod，第一个参数类型标注为 SchedulerKvEventsPublisher；emit_kv_metrics 内部将 self.stats.X 读取替换为 self.get_stats().X。
3. 调度器依赖注入 (scheduler.py)：导入 SchedulerKvEventsPublisher；在 Scheduler.__init__ 末尾（is_initializing 之前）创建 self.kv_events_publisher 实例，注入所有必要依赖（包括 get_stats=lambda: self.stats）；将 on_idle 中对 _publish_kv_events 的调用改为 self.publish_kv_events(self.kv_events_publisher)；同时将 kv_cache_builder.build_kv_cache 的 enable_kv_cache_events 参数从预先计算的属性改为内联计算，避免对未初始化 publisher 的依赖。

关键文件：

• python/sglang/srt/managers/scheduler_components/kv_events_publisher.py（模块 调度器组件；类别 source；类型 core-logic；符号 SchedulerStats, KvMetrics, SchedulerKvEventsPublisher, post_init）: 新文件，定义 KvMetrics 数据类和 SchedulerKvEventsPublisher 组件骨架，是本次重构的核心产出。
• python/sglang/srt/observability/scheduler_metrics_mixin.py（模块 调度指标；类别 source；类型 core-logic；符号 KvMetrics, init_kv_events, _emit_kv_metrics, emit_kv_metrics）: 核心修改点：迁移 KvMetrics 定义、将三个方法转换为静态方法并调整内部实现，是逻辑适配的关键文件。
• python/sglang/srt/managers/scheduler.py（模块 调度器；类别 source；类型 dependency-wiring）: 调度器主文件，完成依赖注入并更新调用逻辑，包含对 enable_kv_cache_events 计算的内联调整。

关键符号：SchedulerKvEventsPublisher.post_init, SchedulerMetricsMixin.init_kv_events, SchedulerMetricsMixin.emit_kv_metrics, SchedulerMetricsMixin.publish_kv_events, Scheduler.init, Scheduler.on_idle

关键源码片段

python/sglang/srt/managers/scheduler_components/kv_events_publisher.py

新文件，定义 KvMetrics 数据类和 SchedulerKvEventsPublisher 组件骨架，是本次重构的核心产出。

# 文件 : python/sglang/srt/managers/scheduler_components/kv_events_publisher.py
​
from __future__ import annotations
​
import dataclasses
from dataclasses import dataclass
from typing import TYPE_CHECKING, Any, Callable, Optional
​
import zmq
​
if TYPE_CHECKING:
    from sglang.srt.distributed.parallel_state_wrapper import ParallelState
    from sglang.srt.mem_cache.base_prefix_cache import BasePrefixCache
​
# 临时占位定义，实际类型从 metrics_collector 导入（避免循环导入）
class SchedulerStats: ... # type: ignore[no-redef]
​
​
@dataclasses.dataclass
class KvMetrics:
    """KV 缓存事件相关的度量数据，原定义于 scheduler_metrics_mixin，现迁移至独立组件。"""
    request_active_slots: int = 0
    request_total_slots: int = 0
    kv_active_blocks: int = 0
    kv_total_blocks: int = 0
    num_requests_waiting: int = 0
    gpu_cache_usage_perc: float = 0.0
    gpu_prefix_cache_hit_rate: float = 0.0
    data_parallel_rank: int = 0
​
​
@dataclass(kw_only=True, slots=True)
class SchedulerKvEventsPublisher:
    """KV 事件发布器组件，当前仅包含骨架和初始化委托，后续提交会将方法移入。"""
    kv_events_config: Optional[str]
    ps: "ParallelState"
    attn_tp_rank: int
    attn_cp_rank: int
    attn_dp_rank: int
    dp_rank: Optional[int]
    tree_cache: "BasePrefixCache"
    send_metrics_from_scheduler: Optional["zmq.Socket"]
    max_running_requests: int
    max_total_num_tokens: int
    get_stats: Callable # 运行时可变 scheduler stats 的 getter
    enable_kv_cache_events: bool = False
    kv_event_publisher: Any = None
​
    def __post_init__(self) -> None:
        # 委托给旧 mixin 完成初始化，方法内容后续会搬到这里
        from sglang.srt.observability.scheduler_metrics_mixin import (
            SchedulerMetricsMixin,
        )
        SchedulerMetricsMixin.init_kv_events(self, self.kv_events_config)

python/sglang/srt/observability/scheduler_metrics_mixin.py

核心修改点：迁移 KvMetrics 定义、将三个方法转换为静态方法并调整内部实现，是逻辑适配的关键文件。

# 文件 : python/sglang/srt/observability/scheduler_metrics_mixin.py ( 部分摘录 )
​
# ... 导入部分新增：
from sglang.srt.managers.scheduler_components.kv_events_publisher import KvMetrics
if TYPE_CHECKING:
    from sglang.srt.managers.scheduler_components.kv_events_publisher import (
        SchedulerKvEventsPublisher,
    )
​
# 删除了本地的 KvMetrics 类定义
​
class SchedulerMetricsMixin:
    # ... 其他方法
​
    @staticmethod
    def init_kv_events(
        self: "SchedulerKvEventsPublisher", kv_events_config: Optional[str]
    ):
        """初始化 KV 事件相关配置，现在通过 publisher 实例调用。"""
        self.enable_kv_cache_events = bool(
            kv_events_config and self.ps.attn_tp_rank == 0 and self.ps.attn_cp_rank == 0
        )
        # ... 后续初始化
​
    # 方法已重命名为 emit_kv_events ( 去掉下划线前缀 )，并变为静态方法
    @staticmethod
    def emit_kv_metrics(self: "SchedulerKvEventsPublisher"):
        if not self.enable_kv_cache_events:
            return
        kv_metrics = KvMetrics()
        # 原为 self.stats.X，现在通过 get_stats() callable 获取
        kv_metrics.request_active_slots = self.get_stats().num_running_reqs.total
        # ... 其余填充逻辑不变
​
    @staticmethod
    def publish_kv_events(self: "SchedulerKvEventsPublisher"):
        # 实现与原 _publish_kv_events 相同，但通过 self 操作 publisher 属性
        ...
​
    def report_prefill_stats(self: Scheduler):
        # ... 调用方式改为：
        self.emit_kv_metrics(self.kv_events_publisher)
        self.publish_kv_events(self.kv_events_publisher)
​
    def report_decode_stats(self: Scheduler):
        # ... 同理
        self.emit_kv_metrics(self.kv_events_publisher)
        self.publish_kv_events(self.kv_events_publisher)

python/sglang/srt/managers/scheduler.py

调度器主文件，完成依赖注入并更新调用逻辑，包含对 enable_kv_cache_events 计算的内联调整。

# 文件 : python/sglang/srt/managers/scheduler.py ( 部分摘录 )
​
# 新增导入
from sglang.srt.managers.scheduler_components.kv_events_publisher import (
    SchedulerKvEventsPublisher,
)
​
# 在 __init__ 方法中，在 self.is_initializing = False 之前创建 publisher
self.kv_events_publisher = SchedulerKvEventsPublisher(
    kv_events_config=self.server_args.kv_events_config,
    ps=self.ps,
    attn_tp_rank=self.ps.attn_tp_rank,
    attn_cp_rank=self.ps.attn_cp_rank,
    attn_dp_rank=self.ps.attn_dp_rank,
    dp_rank=self.ps.dp_rank,
    tree_cache=self.tree_cache,
    send_metrics_from_scheduler=self.send_metrics_from_scheduler,
    max_running_requests=self.max_running_requests,
    max_total_num_tokens=self.max_total_num_tokens,
    get_stats=lambda: self.stats, # 运行时可变 stats
)
​
# 在 kv_cache_builder 调用处，将 enable_kv_cache_events 改为内联计算
result = kv_cache_builder.build_kv_cache(
    ...
    # 原为 enable_kv_cache_events=self.enable_kv_cache_events,
    enable_kv_cache_events=bool(
        self.server_args.kv_events_config
        and self.ps.attn_tp_rank == 0
        and self.ps.attn_cp_rank == 0
    ),
    ...
)
​
# on_idle 方法中的调用
# 原 : self._publish_kv_events()
# 现 :
self.publish_kv_events(self.kv_events_publisher)

评论区精华

无实质 review 讨论；仅有一条 gemini-code-assist 机器人的配额提示。

• 暂无高价值评论线程

风险与影响

• 风险：行为等价风险：虽然方法被静态化且调用方式改变，但经过参数透传后逻辑不变。enable_kv_cache_events 的内联计算与 init_kv_events 中的计算完全一致。但仍需关注：init_kv_events 在 __post_init__ 中被调用，而 __post_init__ 时 self.ps.attn_tp_rank 等属性可能尚未赋值？实际上在 SchedulerKvEventsPublisher 的 __post_init__ 中，这些属性已经通过 kw_only 传入，所以安全。
  回归风险：涉及调度器核心初始化路径，但无测试覆盖（本 PR 未增加测试）。

• 影响：影响范围：仅影响调度器内部，对用户无功能变化，不影响现有 API。
  影响程度：低。属于内部重构，但为后续大规模组件迁移奠定基础。

• 风险标记：核心路径变更, 缺少测试覆盖, 依赖循环调整

关联脉络

• PR #25626 Move KV-cache event emission to SchedulerKvEventsPublisher: 此PR是该重构链的下一步，将方法实际迁移到新组件，与本PR的预备工作直接衔接。
• PR #25629 Add SchedulerMetricsReporter and route metrics state through it: 属于同一组件化重构系列，将指标报告逻辑拆分为独立组件，与本PR手法相似。