vLLM Request 全生命周期

Architecture Overview

全局概览

API Server 进程 (FastAPI + Uvicorn)

1 API 接收与解析

2 渲染与分词

3 多模态处理

4 提交到引擎

↓ZMQ IPC (ROUTER → DEALER)

Engine Core 进程 — step() 编排者

P1 Schedule

P4 State Update

↓model_executor 分发到 Worker

GPU Worker 进程 (被 EngineCore 编排)

P2 模型执行 (S6-7)

P3 GPU 采样 (S8)

↓ZMQ IPC (PUSH → PULL)

API Server 进程 (output handler)

10 输出处理与反分词

11 流式返回

↻ Phase 1-4 在 EngineCore.step() 中每轮执行，Worker 不自主驱动

阅读路径：总图建立进程边界 → Stage 1-4 请求解析 → Iteration Loop 理解 step() 四阶段 → Stage 10-11 输出返回

Stage 1

API 接收与解析

ServingChat

ServingRender

Renderer

AsyncLLM

请求对象

ChatCompletionRequest (Pydantic model) model, messages, temperature, max_tokens stop, stop_token_ids, logprobs response_format (结构化输出) image_url, audio_url, video_url (多模态)

API服务启动调用栈

run_server() └── run_server_worker() ├── build_async_engine_client() → AsyncLLM ├── build_app() → FastAPI routes └── uvicorn.run()

sequenceDiagram
    autonumber
    participant C as Client
    participant R as API Router
    participant S as ServingChat
    participant SR as ServingRender
    participant Ren as Renderer
    participant A as AsyncLLM

    C->>R: POST /v1/chat/completions
    R->>S: create_chat_completion()
    Note over S: _check_model() / engine health
    S->>SR: openai_serving_render.render_chat()
    Note over SR: validate_chat_template
preprocess_chat()
    SR->>Ren: renderer.render_chat_async()
    Note over Ren: Step 1: render_messages
Step 2: tokenize_prompts
Step 3: apply_prompt_extras
Step 4: process_for_engine
    Ren-->>SR: EngineInput
    SR-->>S: engine_inputs
    Note over S: to_sampling_params()
    S->>A: engine_client.generate()
    Note over A: 返回 async generator
    A-->>S: AsyncGenerator[RequestOutput]
    S-->>R: StreamingResponse / JSON
    R-->>C: SSE / JSON Response

entrypoints/openai/api_server.py → chat_completion/api_router.py → chat_completion/serving.py:229 → serve/render/serving.py:184

Stage 2-3

渲染、分词与多模态处理

Serving

Renderer

MultiModal

Renderer 四步主流程 — render_chat() 的骨架

1

render_messages()

解析 chat messages + Jinja template → 文本 + mm_data

2

tokenize_prompts()

文本 → token IDs (HF tokenizer.encode)

3

_apply_prompt_extras()

cache_salt 等附加字段

4

process_for_engine()

多模态处理 → EngineInput (纯文本直接返回 tokens_input)

Serving 层调用

OpenAIServingChat.create_chat_completion() ├─ tokenizer / reasoning_parser 初始化 ├─ render_chat_request(request) │ ├─ _check_model(request) │ └─ openai_serving_render.render_chat(request) │ ├─ validate_chat_template(...) │ └─ preprocess_chat(...) │ ├─ 构造 ChatParams (template, tools, mm_processor_kwargs, ...) │ └─ renderer.render_chat_async(...) ← 进入四步主流程 ├─ 得到 conversation, engine_inputs ├─ request.to_sampling_params() └─ engine_client.generate(...)

Renderer.render_chat_async()

Step 1: HfRenderer.render_messages_async() │ ├─ parse_chat_messages_async() │ ├─ 规范化 OpenAI message content │ ├─ 下载/读取 image/audio/video parts │ ├─ 输出 conversation │ ├─ 输出 mm_data: MultiModalDataDict │ └─ 输出 mm_uuids │ ├─ safe_apply_chat_template() │ └─ HF tokenizer.apply_chat_template / Jinja │ ├─ prompt dict: │ ├─ prompt / prompt_token_ids │ ├─ multi_modal_data = mm_data │ └─ multi_modal_uuids = mm_uuids

Step 2: BaseRenderer.tokenize_prompts() └─ _tokenize_singleton_prompt() └─ tokenizer.encode() （除非已有 prompt_token_ids）

Step 3: BaseRenderer._apply_prompt_extras() └─ 写入 cache_salt 等附加字段

Step 4: BaseRenderer.process_for_engine_async() ├─ 无 MM: tokens_input(prompt_token_ids) → 直接返回 └─ 有 MM: _process_multimodal_async() ├─ mm_processor.info.parse_mm_data() │ └─ 原始 image/audio/video → MultiModalDataItems ├─ parse_mm_uuids() + _process_mm_uuids() ├─ 构造 ProcessorInputs └─ BaseMultiModalProcessor.apply() ├─ _cached_apply_hf_processor() │ └─ _call_hf_processor() │ └─ HF ProcessorMixin (text+media → tensors) ├─ _maybe_apply_prompt_updates() │ ├─ HF 已更新 prompt: _find_mm_placeholders() │ └─ HF 未更新: _apply_prompt_updates() └─ mm_input(prompt_token_ids, mm_kwargs, mm_hashes, mm_placeholders)

vLLM vs HF Processor 分工

vLLM BaseMultiModalProcessor

解析 MultiModalDataItems
处理 processor cache
计算 mm_hashes
定义 prompt updates / placeholder 规则
HF BatchFeature → MultiModalKwargsItems

HF ProcessorMixin

图像 resize / normalize / patch
音频 feature extraction
视频 frame / vision 前处理
text + media 联合 tokenization
→ BatchFeature (pixel_values, ...)

renderers/base.py:969 | renderers/hf.py:622 | multimodal/processing/processor.py

Stage 4

提交到引擎

提交与回收的边界

提交路径 (Request → EngineCore)

OpenAI serving task │ engine_client.generate(...) ▼ AsyncLLM.generate() ├─ add_request() │ ├─ InputProcessor.process_inputs() │ │ └─ EngineInput → EngineCoreRequest │ ├─ OutputProcessor.add_request() │ │ └─ request_id → RequestState + collector │ └─ EngineCoreClient.add_request_async() │ └─ ZMQ ROUTER send ├─ generate() 不读 ZMQ │ └─ 等待 RequestOutputCollector └─ yield RequestOutput → serving 层

回收路径 (EngineCore → Request)

AsyncLLM output_handler task │ ├─ await engine_core.get_output_async() │ <── AsyncMPClient.outputs_queue.get() │ ↑ ZMQ PULL recv EngineCoreOutputs │ ├─ process_outputs() │ ├─ Detokenizer.update() → 增量解码 │ ├─ LogprobsProcessor.update() │ ├─ stop strings check │ └─ collector.put(RequestOutput) │ └─ 必要时 abort stop-string 命中请求

generate() 不直接从 ZMQ 读输出；它只消费 per-request collector，经 OutputProcessor 处理后的结果

队列 / 缓冲一览

位置	队列	生产者	消费者
API 进程	`RequestOutputCollector`	OutputProcessor	当前 request 的 generate()
API 进程	`outputs_queue`	process_outputs_socket()	output_handler
API → Core	`ZMQ ROUTER → DEALER`	AsyncMPClient	EngineCoreProc
Core 进程	`input_queue`	input socket thread	run_busy_loop()
Core 进程	`aborts_queue`	输入线程 (ABORT 时写入)	step() 后的 _process_aborts_queue()
Core 进程	`output_queue`	run_busy_loop()	output socket thread
Core → API	`ZMQ PUSH → PULL`	output socket thread	AsyncMPClient

三个并发循环

1. Per-Request 循环

AsyncLLM.generate():
q.get_nowait() or await q.get()
yield RequestOutput
→ 直到 finished=True

2. Output Handler 循环

_run_output_handler():
await get_output_async()
process_outputs() → collectors
→ 处理所有 request 的 batch 输出

3. Engine Core 循环

run_busy_loop():
_process_input_queue() → _process_engine_step() → output_queue.put()
→ 只要 scheduler 有 unfinished requests 就继续 step

v1/engine/async_llm.py:521 generate | v1/engine/async_llm.py:280 add_request | v1/engine/core_client.py:1058

EngineCore.step()

The Iteration Loop

EngineCore (CPU)

Worker (GPU)

并行重叠

↻ Phase 1-4 在每次 step() 调用中顺序执行，Worker 不自主驱动

EngineCore.step() 四阶段总览

P1

Schedule (Stage 5)

Scheduler.schedule() → SchedulerOutput [CPU, EngineCore]

P2

Model Execution (Stage 6-7)

execute_model() → logits 暂存, 返回 None [GPU, Worker, non-blocking]

P3

Sample (Stage 8)

grammar mask (与 P2 并行) → sample_tokens() [GPU, Worker]

P4

State Update (Stage 9)

update_from_output() → append tokens, check stop, free KV [CPU, EngineCore]

单轮执行时序

sequenceDiagram
    participant EC as EngineCore
    participant S as Scheduler
    participant W as GPUWorker

    Note over EC,W: Phase 1: Schedule
    EC->>S: schedule()
    Note right of S: RUNNING decode
WAITING → RUNNING
allocate KV blocks
    S-->>EC: SchedulerOutput

    Note over EC,W: Phase 2: Model Execution (non-blocking)
    EC->>W: execute_model(scheduler_output)
    Note right of W: _update_states()
_prepare_inputs()
model forward → logits

    Note over EC: get_grammar_bitmask()
(与 GPU forward 并行)

    W-->>EC: future: None (logits 暂存)
    EC->>EC: future.result() [等待 GPU]

    Note over EC,W: Phase 3: Sample
    EC->>W: sample_tokens(grammar_output)
    Note right of W: apply grammar mask [GPU]
Sampler: top-k/top-p [GPU]
bookkeeping sync [GPU→CPU]
    W-->>EC: ModelRunnerOutput

    Note over EC,W: Phase 4: State Update
    EC->>S: update_from_output()
    Note right of S: append tokens
check stop
free KV if done
    S-->>EC: EngineCoreOutputs

Phase 1: Schedule — Stage 5

请求到达 EngineCore

EngineCoreProc.run_busy_loop() (行 1164) ├── _process_input_queue() │ ├── 空闲时阻塞等待 input_queue │ ├── 有 work 后 drain 当前 input_queue │ └── _handle_client_request() │ ├── ADD → Scheduler.add_request() → waiting 队列 │ ├── ABORT → finish_requests() │ └── UTILITY └── _process_engine_step() → step()

Scheduler.schedule() 调用栈

Scheduler.schedule() (行 352) │ ├─ 调度 RUNNING (行 387) │ ├─ 计算 num_new_tokens = num_tokens_with_spec - num_computed_tokens │ ├─ 长预填充分块 (long_prefill_token_threshold) │ ├─ 分配 KV cache blocks │ └─ 内存不足时 → 抢占低优先级请求 │ ├─ 调度 WAITING (行 567) │ ├─ 从 waiting 队列取出请求 │ ├─ 检查 prefix cache 命中 │ ├─ 准入控制 (can_fit_full_sequence) │ ├─ 分配 KV cache blocks │ └─ 移入 running 列表 │ └─ 返回 SchedulerOutput ├─ num_scheduled_tokens: {req_id: int} ├─ scheduled_new_reqs / scheduled_cached_reqs ├─ block_ids 映射 └─ finished_req_ids

v1/engine/core.py:402 step | v1/core/sched/scheduler.py:352 schedule

Phase 2: Model Execution — Stage 6-7

Worker / ModelRunner / Model 三层架构

Worker (gpu_worker.py) ├─ 绑定 rank / local_rank, 管理 accelerator device ├─ 初始化分布式通信和 GPU 内存 └─ 持有 ModelRunner GPUModelRunner (gpu_model_runner.py) ├─ 持久化批次状态: InputBatch (所有活跃请求的 GPU tensor 汇总) ├─ 接收 SchedulerOutput → 更新 InputBatch ├─ _prepare_inputs() → 构建 GPU 输入 (positions, attn metadata) ├─ _preprocess() → 多模态编码器 + embedding 合并 ├─ model forward → logits → 暂存到 execute_model_state └─ 后续被 sample_tokens() 调用: grammar + 采样 + bookkeeping Model (torch.nn.Module) ├─ Embedding → N × TransformerLayer → logits ├─ 每层: RMSNorm → Attention (PagedAttention) → RMSNorm → FFN/MoE └─ [多模态] model.embed_multimodal() → vision/audio encoder → MM embeddings

execute_model() 调用栈

GPUModelRunner.execute_model(scheduler_output) │ ├─ _update_states() (行 1061) │ ├─ 移除 finished 请求的 cached state / InputBatch 条目 │ ├─ 释放 encoder cache, 处理新 KV blocks │ ├─ 写入 new / resumed / running 状态 │ └─ 更新 block table, 采样元数据, LoRA 状态 │ ├─ _prepare_inputs() (行 1776) │ ├─ 排序 (decode 在前, prefill 在后) │ ├─ 构建 idx_mapping, query_start_loc │ ├─ [Prefill] Triton kernel 填 input_ids │ └─ [Decode] 合并上次 token + drafts │ ├─ _preprocess() (行 3211) │ ├─ [有 MM] _execute_mm_encoder() (行 2733) │ │ ├─ _batch_mm_inputs_from_scheduler() → mm_kwargs │ │ ├─ 按 modality 分组 batch │ │ ├─ model.embed_multimodal(**mm_kwargs) → MM embeddings │ │ └─ 缓存到 encoder_cache[mm_hash] │ ├─ [有 MM] _gather_mm_embeddings() → 从 cache 取出并合并 │ └─ [有 MM] embed_input_ids(token_ids, multimodal_embeddings) → inputs_embeds │ ├─ [CUDA Graph] → graph.replay() ├─ [Eager] → model(**model_inputs) │ ├─ Embedding → N × TransformerLayer → logits │ └─ [有 MM] inputs_embeds 已含 MM embeddings, 跳过 token embedding │ └─ 暂存 logits → execute_model_state, 返回 None

v1/worker/gpu_model_runner.py:3787 execute | v1/worker/gpu_model_runner.py:1776 prepare

Phase 3: Sample — Stage 8

GPUModelRunner.sample_tokens(grammar_output) (行 4140) │ ├─ 从 execute_model_state 取出 logits (GPU tensor) │ ├─ apply_grammar_bitmask() → 不允许 token → -inf [GPU] │ ├─ _sample(logits) → Sampler.forward() │ ├─ apply_logits_processors(): │ │ ├─ logit_bias / allowed_token_ids │ │ ├─ frequency / presence / repetition penalty │ │ ├─ temperature 缩放 │ │ └─ min-p / top-k / top-p 过滤 │ └─ top-k/top-p sampler (FlashInfer CUDA kernel) / greedy (argmax) │ 全程 GPU, 避免 torch.multinomial 的同步 │ ├─ _update_states_after_model_execute() ├─ _bookkeeping_sync() → GPU tensor → Python list[int] │ ├─ [Spec Decoding] propose_draft_token_ids() │ └─ 返回 ModelRunnerOutput (CPU, 含 sampled ids, logprobs)

v1/worker/gpu_model_runner.py:4140 sample | v1/sample/sampler.py:68 Sampler.forward

Phase 4: State Update — Stage 9

Scheduler.update_from_output(scheduler_output, model_output) (行 1303) │ ├─ 遍历所有被调度的请求： │ ├─ [Spec Decoding] 计算被拒绝的 draft tokens │ ├─ _update_request_with_output() │ │ ├─ request.append_output_token_ids(token) │ │ └─ check_stop() → EOS / max_tokens / stop tokens │ ├─ [结构化输出] grammar.accept_tokens() │ └─ 如果 stopped: │ ├─ _handle_stopped_request() → 释放 KV cache │ └─ 创建 EngineCoreOutput │ └─ 返回 EngineCoreOutputs → output_queue → ZMQ → API

v1/core/sched/scheduler.py:1303 update_from_output | v1/core/sched/utils.py:94 check_stop

Stage 10-11

输出处理与流式返回

EngineCore 输出 → Client 返回路径

sequenceDiagram
    participant W as GPUWorker
    participant S as Scheduler
    participant E as EngineCore
    participant M as AsyncMPClient
    participant A as AsyncLLM
    participant API as API Server
    participant C as Client

    Note over W,C: ─── Phase 4: State Update ───
    W->>S: sampled token ids
    Note right of S: append tokens
check EOS / max_tokens
build EngineCoreOutput
    S-->>E: EngineCoreOutput

    Note over W,C: ─── ZMQ 返回 ───
    Note right of E: output_queue.put()
    E->>M: ZMQ PUSH msgpack
    M->>M: decode → outputs_queue

    Note over W,C: ─── Stage 10: 输出处理 ───
    M->>A: outputs_queue
    Note right of A: process_outputs()
├ Detokenizer.update()
├ LogprobsProcessor
└ stop strings check
    A->>A: collector.put()

    Note over W,C: ─── Stage 11: 流式返回 ───
    A-->>API: generate() yield RequestOutput
    API->>API: build ChatCompletionStreamResponse
    API-->>C: SSE: data: {...}
    Note over W,C: 如果 finished=False → 重复 step()
    API-->>C: SSE: data: [DONE]

OutputProcessor 处理流

process_outputs(engine_core_outputs) ├─ Detokenizer.update() │ ├─ 增量解码 token → 文本 │ │ (FastIncrementalDetokenizer: Rust) │ └─ 检查 stop strings ├─ LogprobsProcessor.update() ├─ RequestState.make_request_output() │ ├─ stream_interval 控制 │ └─ 创建 RequestOutput └─ 放入 per-request collector (DELTA 模式: 聚合输出)

SSE 流式响应

chat_completion_stream_generator() ├─ for request_output in result_generator: │ ├─ 构建 ChatCompletionStreamResponse │ │ ├─ delta.content (增量文本) │ │ ├─ delta.reasoning_content │ │ └─ usage 信息 │ └─ yield SSE: "data: {json}\n\n" └─ yield "data: [DONE]\n\n"

反分词

FastIncrementalDetokenizer — Rust DecodeStream
SlowIncrementalDetokenizer — Python 回退
增量解码，避免每步对全部历史重新解码

v1/engine/output_processor.py:572 | v1/engine/detokenizer.py:95 | chat_completion/serving.py:525 stream

Appendix

压缩完整时序图

sequenceDiagram
    participant C as Client
    participant API as API Server
    participant A as AsyncLLM
    participant E as EngineCore
    participant S as Scheduler
    participant W as GPUWorker

    C->>API: POST /v1/chat/completions
    Note right of API: Stage 1: API 接收与解析
Stage 2-3: 渲染分词 / 多模态
    API->>A: generate()
    Note right of A: Stage 4: InputProcessor
+ OutputProcessor.add + ZMQ send

    A->>E: ZMQ: ADD
    E->>S: add_request() → waiting

    rect rgba(63, 185, 80, 0.05)
    Note over S,W: ── EngineCore.step() 循环 ──
    Note over E: Phase 1: Schedule
    E->>S: schedule()
    Note right of S: RUNNING decode
WAITING → RUNNING
allocate KV blocks
    S-->>E: SchedulerOutput
    Note over E: Phase 2: Model Execution
    E->>W: execute_model()
    Note right of W: _update_states
_prepare_inputs
model forward → logits
    W-->>E: None (logits 暂存)
    Note over E: Phase 3: Sample
    E->>W: sample_tokens()
    Note right of W: grammar mask [GPU]
Sampler [GPU]
bookkeeping sync
    W-->>E: ModelRunnerOutput
    Note over E: Phase 4: State Update
    E->>S: update_from_output()
    Note right of S: append tokens
check stop
free KV if done
    end

    E->>A: ZMQ PUSH EngineCoreOutputs
    Note right of A: Stage 10: detokenize
+ logprobs + stop strings
    A-->>API: yield RequestOutput
    Note right of API: Stage 11: SSE
    API-->>C: data: {...}

    Note over S,W: 重复 step() 直到 finished
    API-->>C: data: [DONE]

Appendix

关键数据对象流转

flowchart TD
    A["ChatCompletionRequest
messages, temperature, max_tokens"]
    B["prompt_token_ids
list[int] + mm_kwargs"]
    C["EngineInput
token / multimodal / embeds"]
    D["EngineCoreRequest
prompt + SamplingParams + mm_features"]
    E["Request
num_computed, output_ids, status, kv_blocks"]
    F["SchedulerOutput
scheduled reqs, tokens, block_ids"]
    G["InputBatch
input_ids, positions, seq_lens (GPU)"]
    H["Hidden States
GPU tensor"]
    I["Sampled Token IDs
GPU"]
    J["EngineCoreOutput
new_token_ids, finish_reason, logprobs"]
    K["RequestOutput
text, token_ids, logprobs, finished"]
    L["SSE / JSON Response"]

    A -->|render + tokenize| B
    B -->|process_for_engine| C
    C -->|InputProcessor| D
    D -.->|ZMQ IPC| E
    E -->|schedule| F
    F -->|prepare inputs| G
    G -->|model forward| H
    H -->|sample| I
    I -->|update_from_output| J
    J -.->|ZMQ IPC| K
    K -->|SSE stream| L

    style A fill:#1c2128,stroke:#58a6ff,color:#e6edf3
    style B fill:#1c2128,stroke:#58a6ff,color:#e6edf3
    style C fill:#1c2128,stroke:#58a6ff,color:#e6edf3
    style D fill:#1c2128,stroke:#58a6ff,color:#e6edf3
    style E fill:#1c2128,stroke:#3fb950,color:#e6edf3
    style F fill:#1c2128,stroke:#3fb950,color:#e6edf3
    style G fill:#1c2128,stroke:#f0883e,color:#e6edf3
    style H fill:#1c2128,stroke:#f0883e,color:#e6edf3
    style I fill:#1c2128,stroke:#f0883e,color:#e6edf3
    style J fill:#1c2128,stroke:#3fb950,color:#e6edf3
    style K fill:#1c2128,stroke:#bc8cff,color:#e6edf3
    style L fill:#1c2128,stroke:#bc8cff,color:#e6edf3

API Server

Engine Core

GPU Worker

Output

ZMQ IPC

Appendix

横切关注点

结构化输出 (Structured Output)

S1

response_format → SamplingParams

S4

InputProcessor → StructuredOutputRequest

S5

grammar_init() 异步编译 grammar

S8

grammar bitmask → logits 屏蔽

S9

grammar.accept_tokens() 推进状态机

Speculative Decoding

S8

propose_draft_token_ids() → N 个候选

S5

draft tokens 计入 num_tokens_with_spec

S7

一次 forward 验证所有 draft tokens

S9

接受/拒绝 → 回退 num_computed_tokens

请求 Abort

API 进程: generate() CancelledError → abort()
EngineCore: _process_aborts_queue() step 间处理
Worker: 下一个 _update_states() 中移除

LoRA

S1: ChatCompletionRequest 指定 adapter
S5: Scheduler 跟踪 active LoRAs (max_loras)
S6: Worker 加载 LoRA weights → GPU
S8: CUDA Graph 按 num_active_loras 特化

DP / TP / PP 下的 Request 路由

TP/PP：一个 EngineCore 调度一组 worker；
这些 worker 共同完成同一个模型 forward。

DP：多个 EngineCore，每个 DP rank
管一组 worker；前端 EngineCoreClient
或外部负载均衡选择发到哪个 rank。

进程数量公式：
API Server: A (常随 DP 扩展)
EngineCore: DP
GPU Worker: N = DP × PP × TP
DP Coordinator: DP > 1 时为 1

client_index + request_id→engine 路由表
确保输出回到发起该 request 的 frontend