# PR #42433 完整报告

- 仓库：`vllm-project/vllm`
- 标题：[EC Connector] Add EC Transfer Params
- 合并时间：2026-07-12 19:35
- 原文链接：http://prhub.com.cn/vllm-project/vllm/pull/42433

---

# 执行摘要

- 一句话：为 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()，是参数传递的关键环节。

```python
# 关键改动集中在 _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 后端一致。

```rust
/// 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