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