执行摘要

• 一句话：为 EC Connector 添加 ec_transfer_params 传输管道
• 推荐动作：值得精读。该 PR 展示了如何在 vLLM 中新增一种跨节点传输参数（transfer params）的完整链路，从请求协议到调度、响应输出，并保持与已有 kv_transfer_params 模式一致。对于需要扩展相似功能的读者有很强的参考价值。review 中提出的高优先级问题应尽快修复。

功能与动机

PR Body 指出：目的是为 OpenAI 兼容协议添加 ec_transfer_params 顶级字段（请求输入路由与响应输出），镜像已有的 kv_transfer_params 模式；同时将 ECConnectorBase.request_finished() 接入调度器的 _free_request()，使得 EC Connector 可在请求完成时发出传输参数。

实现拆解

1. 协议层：在 ChatCompletionRequest、ChatCompletionResponse 以及 Completion 和 Responses 对应协议模型中添加 ec_transfer_params: dict[str, Any] | None 字段，并在 to_sampling_params() 中将其注入 extra_args。
2. Request 构造：在 vllm/v1/request.py 的 __init__ 中从 sampling_params.extra_args 读取 ec_transfer_params，赋值给 req.ec_transfer_params。
3. 调度器集成：在 Scheduler._free_request() 中将返回类型从单值元组改为 (kv_params, ec_params)，并调用 self.ec_connector.request_finished(request) 获取 EC 参数；同时添加 SchedulerInterface.get_ec_connector() 抽象方法以便外部访问。
4. 引擎输出：在 EngineCoreOutput 和 RequestOutput 中添加 ec_transfer_params 字段，在 make_request_output() 中传入并最终在响应中体现。
5. Rust 前端：在 rust/src/server/src/utils.rs 增加 merge_ec_transfer_params() 函数，将请求中的 ec_transfer_params 合并到 vllm_xargs 供引擎核心消费。
6. 环境变量：新增 VLLM_EC_SIDE_CHANNEL_HOST 和 VLLM_EC_SIDE_CHANNEL_PORT 环境变量（通过 vllm/envs.py）。
7. 测试：新增专用单元测试文件，覆盖请求路由、Request 读取、RequestOutput 聚合、调度器 _free_request 以及无 connector 的默认情况；同时更新现有测试 mock 以适配新的返回类型。

关键文件：

• tests/v1/ec_connector/unit/test_ec_transfer_params.py（模块 传输参数；类别 test；类型 test-coverage；符号 test_ec_transfer_params_routed_to_sampling_params_extra_args, test_request_output_add_propagates_ec_transfer_params, _out, test_request_reads_ec_transfer_params_from_extra_args）: 新增专用测试文件，覆盖请求路由、Request 读取、RequestOutput 聚合、调度器 _free_request 及无 connector 默认情况，是验证 ec_transfer_params 流水线的核心测试。
• vllm/v1/core/sched/scheduler.py（模块 调度器；类别 source；类型 core-logic；符号 get_ec_connector）: 调度器核心修改：_free_request 返回元组并调用 ECConnector.request_finished()，是参数传递的关键环节。
• rust/src/server/src/utils.rs（模块 Rust 前端；类别 source；类型 core-logic；符号 merge_ec_transfer_params）: Rust 前端新增 merge_ec_transfer_params 函数，将请求中的 ec_transfer_params 注入 vllm_xargs，保持与 Python 后端一致。
• vllm/v1/core/sched/interface.py（模块 调度器接口；类别 source；类型 core-logic；符号 get_ec_connector）: 在 SchedulerInterface 中添加 get_ec_connector 抽象方法，提供统一的 EC Connector 访问接口。
• vllm/entrypoints/openai/chat_completion/protocol.py（模块 协议层；类别 source；类型 core-logic）: 在请求和响应模型中添加 ec_transfer_params 字段，并在 to_sampling_params 中注入到 extra_args，是参数传递的入口。

关键符号：ECConnectorBase.request_finished, Scheduler._free_request, Scheduler.get_ec_connector, SchedulerInterface.get_ec_connector, merge_ec_transfer_params, ChatCompletionRequest.to_sampling_params, RequestOutput.add, Request.init, make_request_output

关键源码片段

vllm/v1/core/sched/scheduler.py

调度器核心修改：_free_request 返回元组并调用 ECConnector.request_finished()，是参数传递的关键环节。

# 关键改动集中在 _free_request 方法和 get_ec_connector 方法
​
def _free_request(
    self, request: Request, delay_free_blocks: bool = False
) -> tuple[dict[str, Any] | None, dict[str, Any] | None]:
    assert request.is_finished()
    self._inflight_prefills.discard(request)
​
    # 先获取 kv connector 的 finished 参数
    connector_delay_free_blocks, kv_xfer_params = self._connector_finished(request)
​
    # EC Connector：镜像 KV hook，必须在释放编码器缓存前调用，
    # 以便 connector 检查 per-request state 并发出 ec_transfer_params 给响应体
    ec_xfer_params: dict[str, Any] | None = None
    if self.ec_connector is not None:
        ec_delay_free, ec_xfer_params = self.ec_connector.request_finished(request)
        connector_delay_free_blocks |= ec_delay_free
​
    self.encoder_cache_manager.free(request)
    request_id = request.request_id
    self.finished_req_ids.add(request_id)
​
    # ... 延迟释放块逻辑
    if not delay_free_blocks:
        self._free_blocks(request)
​
    # 返回 (kv_params, ec_params) 元组
    return kv_xfer_params, ec_xfer_params
​
​
def get_ec_connector(self) -> ECConnectorBase | None:
    """返回当前调度器使用的 EC Connector 实例。"""
    return self.ec_connector

rust/src/server/src/utils.rs

Rust 前端新增 merge_ec_transfer_params 函数，将请求中的 ec_transfer_params 注入 vllm_xargs，保持与 Python 后端一致。

/// Merge `ec_transfer_params` into the `vllm_xargs` map, mirroring the Python
/// vLLM behavior where `ec_transfer_params` is injected into `extra_args` for
/// engine-core consumption.
pub fn merge_ec_transfer_params(
    mut xargs: Option<HashMap<String, Value>>,
    ec_transfer_params: Option<&HashMap<String, Value>>,
) -> Option<HashMap<String, Value>> {
    // 如果 ec_transfer_params 存在，将其序列化为 JSON Value 并插入到 xargs 映射中
    if let Some(ec_params) = ec_transfer_params {
        let map = xargs.get_or_insert_with(HashMap::new);
        map.insert(
            "ec_transfer_params".to_string(),
            // 这里假定 ec_params 已经是合法 JSON，unwrap 是安全的
            serde_json::to_value(ec_params).unwrap(),
        );
    }
    xargs
}

评论区精华

@gemini-code-assist 指出 RequestOutput.add 中 ec_transfer_params 被无条件覆盖，可能导致流式输出时之前保存的参数丢失，建议仅在不为 None 时更新。
@orozery 对是否在响应中包含 ec_transfer_params 提出讨论，作者说明该字段用于通知 orchestrator 请求完成及编码大小等信息，直到 EC Events 上游实现。
@orozery 还要求将测试改为使用 create_scheduler 并将新测试目录加入 CI 配置。

• RequestOutput.add 中 ec_transfer_params 覆盖问题 (correctness): 建议仅在不为 None 时更新
• extra_args None 访问异常 (correctness): 建议使用 (sampling_params.extra_args or {}).get() 安全访问
• 是否在响应中包含 ec_transfer_params (design): 保留响应字段，后续可能通过 EC Events 重构

风险与影响

• 风险：
  1. 数据丢失风险：当前 RequestOutput.add() 无条件将 ec_transfer_params 设为新输出块的值，若后续块为 None 则会覆盖前序非 None 值，导致参数丢失。这是前一个 review 指出的高优先级问题。
  2. 空值异常：在 vllm/v1/request.py 中直接调用 sampling_params.extra_args.get()，但 extra_args 默认可为 None，将引发 AttributeError。
  3. 跨语言一致：Rust 前端与 Python 后端需保持字段名及类型一致，若后续修改需同步两边。
  4. 测试覆盖：新增测试虽覆盖主要路径，但缺少对异常流程（如 EC Connector 抛出异常）的处理。
     - 影响：对使用 EC Connector 进行 encoder-cache 分布式服务的用户，该 PR 使得跨节点传输参数能从请求端传递至引擎调度并通过响应返回，是 EC Connector 功能完整性的关键一步。改动涉及协议层、调度核心、Rust 前端和输出处理器，影响面较广，但均为对现有 kv_transfer_params 模式的镜像扩展，风险可控。
     - 风险标记：数据丢失风险（streaming）, 空值异常风险, 跨语言同步风险, 测试覆盖待完善

关联脉络

• 暂无明显关联 PR