执行摘要
- 一句话：为RLHF dev API路由添加/abort_requests端点
- 推荐动作：值得精读，特别是跨语言实现相同语义的设计（Python直接收集内部ID，Rust通过空向量表示中止所有）。注意Rust端跟踪时机窗口问题，未来需关注是否引入锁或调整注册时机。
功能与动机
RL rollout框架需要在步骤边界丢弃过采样/长尾生成，同时保留部分输出。现有的/pause?mode=abort会暂停调度器需要/resume，使用不便。disagg路由器已有等效/abort_requests，现在为RLHF开发路由器添加相同能力。
实现拆解
- Python API路由层：在
vllm/entrypoints/serve/dev/rlhf/api_router.py中添加POST /abort_requests异步处理函数。解析请求体，若request_ids存在则直接调用engine.abort(request_ids)；若缺失则从AsyncLLM.output_processor收集所有内部请求ID（包括父请求ID），并调用engine.abort(request_ids, internal=True)以中止所有请求。同时处理JSONDecodeError返回400。
- Rust前端路由层：在
rust/src/server/src/routes/abort_requests.rs中将request_ids从必填改为可选，使用unwrap_or_default()替代ok_or_else错误返回，空body时传递空向量给后端。
- Rust LLM内核：在
rust/src/llm/src/lib.rs中扩展Llm::abort方法：当external_ids为空时调用self.inflight.all_internal_ids()收集所有跟踪的内部ID，否则使用原resolve逻辑。
- 跟踪机制：在
rust/src/llm/src/inflight.rs中新增all_internal_ids方法，从锁保护的HashMap中收集所有内部ID。
- 测试与文档：在
rust/src/server/src/routes/tests.rs中将空body预期从400改为200并重命名测试函数；同步更新三份文档文件添加端点说明。
关键文件：
vllm/entrypoints/serve/dev/rlhf/api_router.py（模块 API路由；类别 source；类型 entrypoint；符号 abort_requests）: 核心入口，添加POST /abort_requests端点，处理指定或全部请求中止，并包含关键修复（internal=True、parent IDs、JSON拒绝）。rust/src/server/src/routes/abort_requests.rs（模块 Rust路由；类别 source；类型 entrypoint）: Rust前端路由，修改request_ids从必填变为可选，实现空body转发。rust/src/llm/src/lib.rs（模块 Rust内核；类别 source；类型 core-logic）: Rust LLM核心，扩展abort方法以支持空external_ids时中止所有请求。rust/src/llm/src/inflight.rs（模块 请求跟踪；类别 source；类型 core-logic）: 新增all_internal_ids方法，支持获取所有在途请求ID。rust/src/server/src/routes/tests.rs（模块 测试套件；类别 source；类型 test；符号 abort_requests_route_rejects_missing_request_ids, abort_requests_route_aborts_all_when_request_ids_missing）: 更新测试用例，反映空body行为从400变为200。docs/serving/online_serving/README.md（模块 文档；类别 docs；类型 documentation）: 新增端点文档说明。docs/training/async_rl.md（模块 文档；类别 docs；类型 documentation）: 同步更新文档。docs/usage/security.md（模块 文档；类别 docs；类型 documentation）: 同步更新文档。
关键符号：abort_requests, abort_requests (Rust route), abort (Llm), all_internal_ids (InflightRequests), abort_requests_route_rejects_missing_request_ids (test), abort_requests_route_aborts_all_when_request_ids_missing (test)
关键源码片段
vllm/entrypoints/serve/dev/rlhf/api_router.py
核心入口，添加POST /abort_requests端点，处理指定或全部请求中止，并包含关键修复（internal=True、parent IDs、JSON拒绝）。
@router.post("/abort_requests")
async def abort_requests(raw_request: Request) -> JSONResponse:
"""Abort in-flight requests without pausing the scheduler.
Empty/missing `request_ids` aborts all in-flight requests.
"""
engine = engine_client(raw_request)
# Parse JSON body; reject malformed JSON explicitly
try:
body = await raw_request.json()
except json.JSONDecodeError as e:
raise HTTPException(status_code=400, detail="Invalid JSON format") from e
request_ids = body.get("request_ids")
try:
if request_ids:
# User-supplied external IDs
await engine.abort(request_ids)
else:
# Dev RL server uses AsyncLLM; gather all internal IDs
from vllm.v1.engine.async_llm import AsyncLLM
assert isinstance(engine, AsyncLLM)
op = engine.output_processor
# Include both child (request_states) and parent (parallel-sampling) IDs
request_ids = [
*op.request_states.keys(),
*op.parent_requests.keys(),
]
# Internal flag is required because these are internal suffixed IDs
await engine.abort(request_ids, internal=True)
return JSONResponse(
content={"status": "aborted", "aborted": len(request_ids)},
status_code=HTTPStatus.OK.value,
)
except Exception as err: # pragma: no cover - defensive
logger.exception("Failed to abort requests")
return JSONResponse(
content={"error": f"Failed to abort requests: {err}"},
status_code=HTTPStatus.INTERNAL_SERVER_ERROR.value,
)
rust/src/server/src/routes/abort_requests.rs
Rust前端路由，修改request_ids从必填变为可选，实现空body转发。
use std::sync::Arc;
use axum::Json;
use axum::extract::State;
use axum::extract::rejection::JsonRejection;
use axum::http::StatusCode;
use serde::Deserialize;
use crate::error::ApiError;
use crate::state::AppState;
use crate::utils::utility_call_error;
#[derive(Debug, Deserialize)]
pub(crate) struct AbortRequestsRequest {
// `request_ids` is now optional; missing/unwrapped defaults to empty vec
request_ids: Option<Vec<String>>,
}
pub async fn abort_requests(
State(state): State<Arc<AppState>>,
body: Result<Json<AbortRequestsRequest>, JsonRejection>,
) -> Result<StatusCode, ApiError> {
let Json(body) = body.map_err(|error| ApiError::json_parse_error(error.body_text()))?;
// Empty/missing `request_ids` aborts all in-flight requests.
let request_ids = body.request_ids.unwrap_or_default();
state
.chat
.abort(&request_ids)
.await
.map_err(|error| utility_call_error("abort_requests", error))?;
Ok(StatusCode::OK)
}
rust/src/llm/src/lib.rs
Rust LLM核心，扩展abort方法以支持空external_ids时中止所有请求。
/// Abort in-flight requests by their external (user-supplied) request ids.
///
/// External ids are resolved to the internal engine ids actually known to
/// engine-core (one external id may map to several internal ids). Unknown
/// or already-finished ids resolve to nothing and are a safe no-op. The
/// tracking entries themselves are removed when the corresponding output
/// streams are dropped, not here.
pub async fn abort(&self, external_ids: &[String]) -> Result<()> {
// Empty `external_ids` means abort every in-flight request.
let internal_ids = if external_ids.is_empty() {
self.inflight.all_internal_ids()
} else {
self.inflight.resolve(external_ids)
};
if internal_ids.is_empty() {
return Ok(());
}
self.client.abort(&internal_ids).await?;
Ok(())
}
评论区精华
风险与影响
- 风险：
- 空body语义可能误终止所有请求，尤其在Rust端跟踪时机窗口内请求未完全注册时。
- 并行采样父ID泄漏问题已修复，但返回的计数仍可能不准确。
- 依赖AsyncLLM内部结构（output_processor、parent_requests），未来重构可能引发兼容性问题。
- Rust端
all_internal_ids仅覆盖已注册请求，generate()中client.call之前发生的abort-all会遗漏。
- 影响：
- 用户：RL训练框架开发者获得更精细的请求控制，无需暂停/恢复调度器，简化rollout流程。
- 系统：新增一个端点，对整体性能无影响。
- 团队：需要维护Python和Rust两个实现，但逻辑一致，降低长期维护成本。
- 风险标记：空body误终止, 并行父ID泄漏（已修复）, Rust跟踪窗口遗漏, aborted计数不准确
关联脉络
- PR #46725 Runtime Draft Weight Update for Speculative Decoding: 在同一个RLHF开发路由器上添加了weight transfer相关端点，本PR是同一系列的延续。
参与讨论