OpenAI Serving 与 AsyncLLM:HTTP API 与引擎的异步桥

源码版本v0.25.1

职责

vllm serve 启动后跑的就是一个 FastAPI 服务,监听 /v1/chat/completions、/v1/completions、/v1/responses、/v1/embeddings、/pooling/*、/speech_to_text/* 等 OpenAI 兼容端点。HTTP 层的入口是 build_app(api_server.py:157),它注册各 register_*_api_routers、装 CORS / 鉴权 / 异常处理器。引擎那一侧的入口是 AsyncLLM(async_llm.py:70),它把 EngineCore(独立进程,通过 ZMQ 通信)包成可 await 的 EngineClient,在 asyncio 事件循环里跑请求。

build_async_engine_client_from_engine_args(api_server.py:109)是装配点:它先 engine_args.create_engine_config() 得到 VllmConfig,然后 AsyncLLM.from_vllm_config(...)(async_llm.py:202)起一个 AsyncLLM,内部 EngineCoreClient.make_async_mp_client 起一个 AsyncMPClient(进程间 + asyncio)。init_app_state(api_server.py:297)把 engine_client、OpenAIServingModels、OnlineRenderer、ServingTokenization 这些挂到 app.state,然后 init_generate_state 进一步把 OpenAIServingChat、OpenAIServingCompletion、OpenAIServingResponses 这些 handler 也建好。

HTTP 请求进来后,比如 POST /v1/chat/completions,路由调 OpenAIServingChat.create_chat_completion(serving.py:233)。它用 OnlineRenderer 把消息列表渲染成 prompt,然后 engine_client.generate(prompt, sampling_params, request_id, ...)(serving.py:357)拿到一个 AsyncGenerator[RequestOutput],边 yield 边把流式响应写回 HTTP。

设计动机

• 多客户端支持:AsyncLLM.__init__ 接 client_addresses / client_count / client_index(async_llm.py:84-86),让一个 OpenAI server 能连多个 DP rank 的 EngineCore,做负载均衡。
• InputProcessor / OutputProcessor 分离:AsyncLLM 持 InputProcessor(把 prompt 加工成 EngineCoreRequest)和 OutputProcessor(把 EngineCoreOutputs 翻译回 RequestOutput)(async_llm.py:135-143),让 HTTP 层只看到 EngineInput 和 RequestOutput,不直接接触 EngineCore。
• 后台 output_handler 任务:_run_output_handler(async_llm.py:637)起一个 asyncio task,从 EngineCore 拉 outputs、output_processor.process_outputs 切片处理、把 RequestOutput 推到对应请求的 RequestOutputCollector。
• 分块处理防阻塞事件循环:output_handler 用 VLLM_V1_OUTPUT_PROC_CHUNK_SIZE 切片(async_llm.py:654),每块之间 await asyncio.sleep(0) 让其它请求有机会 yield。
• n>1 fan-out:add_request 在 params.n > 1 时把请求拆成 n 个子请求,各自独立 request_id,共享 ParentRequest 和同一个 RequestOutputCollector(async_llm.py:385-397)。
• 断开自动 abort:HTTP 客户端断开会让 generate 的 AsyncGenerator 抛 asyncio.CancelledError / GeneratorExit,handler 抓住后 await self.abort(request_id, internal=True)(async_llm.py:591-596)把请求从 EngineCore 撤掉,不浪费 GPU。
• 启动失败优雅:__init__ 把 _run_output_handler 推迟到第一次 add_request(async_llm.py:370-373),这样 AsyncLLM 可以在事件循环外构造,启动失败时能直接抛错退出。
• FastAPI 多端点共用 engine:init_app_state 把 engine_client 装进 state(api_server.py:336),chat / completion / responses / pooling / speech_to_text 各 handler 共享同一个 AsyncLLM。

关键文件

• build_app:157 — 构造 FastAPI app,挂所有 router、CORS、鉴权、异常 handler。
• build_async_engine_client_from_engine_args:109 — 装配点:EngineArgs → VllmConfig → AsyncLLM。
• init_app_state:297 — 把 engine_client、models、renderer、serving_tokenization 挂到 app.state。
• AsyncLLM 类:70 — EngineClient 的异步实现,持 InputProcessor / OutputProcessor / engine_core。
• AsyncLLM.from_vllm_config:202 — 类方法工厂,调 Executor.get_class 选执行器后构造 AsyncLLM。
• AsyncLLM.add_request:280 — 入口:把 prompt 喂给 InputProcessor、起 RequestOutputCollector、_add_request 丢到 EngineCore。
• AsyncLLM.generate:524 — add_request + while not finished: q.get() 的 AsyncGenerator 包装。
• _run_output_handler:637 — 后台 task:engine_core.get_output_async → output_processor.process_outputs → 推 queue。
• _add_request:400 — 两步:output_processor.add_request + engine_core.add_request_async。
• AsyncLLM.abort:709 — 取消请求,internal=True 是断连触发的内部 abort。
• OpenAIServingChat:106 — chat completions handler,继承 GenerateBaseServing。
• create_chat_completion:233 — HTTP 入口,渲染 prompt + 调 engine_client.generate。
• OpenAIServingCompletion:55 — completions API handler。
• BaseServing:29 — 所有 handler 的基类,提供 _check_model / create_error_response。

数据流

一次 POST /v1/chat/completions 的完整路径:FastAPI 路由 → OpenAIServingChat.create_chat_completion → OnlineRenderer.render_chat() 把消息列表渲染成 prompt token ids → engine_client.generate(prompt, sampling_params, request_id, ...) 拿到 AsyncGenerator。generate 内部走 add_request:

# vllm/v1/engine/async_llm.py L357-L373 (简化)
generator = self.engine_client.generate(
    engine_input,
    sampling_params,
    sub_request_id,
    lora_request=lora_request,
    trace_headers=trace_headers,
    priority=request.priority,
    data_parallel_rank=data_parallel_rank,
    reasoning_ended=reasoning_ended,
    reasoning_parser_kwargs={...} if parser is not None and parser.reasoning_parser is not None else None,
)

add_request 把 prompt 喂给 InputProcessor.process_inputs 加工成 EngineCoreRequest,然后调 _add_request:

# vllm/v1/engine/async_llm.py L400-L415
async def _add_request(
    self,
    request: EngineCoreRequest,
    prompt: str | None,
    parent_req: ParentRequest | None,
    index: int,
    queue: RequestOutputCollector,
):
    # Add the request to OutputProcessor (this process).
    self.output_processor.add_request(request, prompt, parent_req, index, queue)

    # Add the EngineCoreRequest to EngineCore (separate process).
    await self.engine_core.add_request_async(request)

    if self.log_requests:
        logger.info("Added request %s.", request.request_id)

请求进了 EngineCore 之后,后台 output_handler task 一直在 await engine_core.get_output_async() 拉输出,切片 process_outputs,把 RequestOutput 推到对应 collector:

# vllm/v1/engine/async_llm.py L656-L683
async def output_handler():
    try:
        while True:
            # 1) Pull EngineCoreOutputs from the EngineCore.
            outputs = await engine_core.get_output_async()
            num_outputs = len(outputs.outputs)

            iteration_stats = (
                IterationStats() if (log_stats and num_outputs) else None
            )

            # Split outputs into chunks of at most
            # VLLM_V1_OUTPUT_PROC_CHUNK_SIZE, so that we don't block the
            # event loop for too long.
            engine_core_outputs = outputs.outputs
            for start in range(0, num_outputs, chunk_size):
                end = start + chunk_size
                outputs_slice = engine_core_outputs[start:end]
                # 2) Process EngineCoreOutputs.
                processed_outputs = output_processor.process_outputs(
                    outputs_slice, outputs.timestamp, iteration_stats
                )
                # NOTE: RequestOutputs are pushed to their queues.
                assert not processed_outputs.request_outputs

                # Allow other asyncio tasks to run between chunks
                if end < num_outputs:
                    await asyncio.sleep(0)

generate 的调用方就在 while not finished: q.get_nowait() or await q.get() 里循环拿 RequestOutput,边拿边 yield 给 HTTP 流式响应。

边界与失败

• 客户端断连自动 abort:generate 抓 asyncio.CancelledError / GeneratorExit,调 await self.abort(request_id, internal=True)(async_llm.py:591-596),把请求从 EngineCore 队列撤掉。
• Engine 死了抛 EngineDeadError:generate 抓 EngineDeadError 不 abort 直接 raise(async_llm.py:598-602),因为已经没东西可 abort;HTTP 层把它转成 500。
• n>1 fan-out 共享 collector:add_request 在 params.n > 1 时把请求拆成 n 个子请求,共享同一个 RequestOutputCollector(async_llm.py:385-397),上层拿到的还是一条流。
• kv_sharing_fast_prefill 不支持 prompt_logprobs:add_request 检测到 kv_sharing_fast_prefill + prompt_logprobs 直接抛 ValueError(async_llm.py:305-314),因为这种 KV 共享路径下 prompt logprobs 算不准。
• 流式输入不支持 reasoning_ended:传 AsyncGenerator 形式的 streaming input 时,带 reasoning_ended / reasoning_parser_kwargs 直接 NotImplementedError(async_llm.py:316-318)。
• output_handler 循环引用防护:_run_output_handler 显式把 engine_core / output_processor / renderer 存成局部变量(async_llm.py:643-653),避免 task 持有 self 形成 circular ref 导致 GC 不掉。
• request_id 不匹配会 warning:add_request 传了 EngineCoreRequest 但 request_id 不一致时只 warning 不抛错(async_llm.py:342-347),以 EngineCoreRequest.request_id 为准。

小结

OpenAI serving 是 vLLM 的 HTTP 入口,由两层构成:FastAPI app + OpenAIServingChat / OpenAIServingCompletion 等 handler 处理 OpenAI 兼容协议;AsyncLLM 把 EngineCore 包成 EngineClient,在 asyncio 事件循环里把请求丢进 EngineCore、把输出拉回 HTTP 流。EngineCore 内部的调度和前向循环见 /engine/engine-core;HTTP 请求最终被 worker 怎么算出来,见 /worker/gpu-model-runner;采样最后一步见 /sampling/sampler。

对照官方资料:vLLM 文档 · README。