api_server.py 66.4 KB
Newer Older
1
# SPDX-License-Identifier: Apache-2.0
2
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
3
import asyncio
4
import hashlib
5
6
import importlib
import inspect
7
import json
8
import multiprocessing
9
import multiprocessing.forkserver as forkserver
10
import os
11
import secrets
12
import signal
13
import socket
14
import tempfile
15
import uuid
16
from argparse import Namespace
17
from collections.abc import AsyncGenerator, AsyncIterator, Awaitable
18
from contextlib import asynccontextmanager
19
from http import HTTPStatus
20
from typing import Annotated, Any, Literal
21

22
import model_hosting_container_standards.sagemaker as sagemaker_standards
23
import prometheus_client
24
import pydantic
25
import regex as re
26
import uvloop
27
from fastapi import APIRouter, Depends, FastAPI, Form, HTTPException, Query, Request
Zhuohan Li's avatar
Zhuohan Li committed
28
29
from fastapi.exceptions import RequestValidationError
from fastapi.middleware.cors import CORSMiddleware
30
from fastapi.responses import JSONResponse, Response, StreamingResponse
31
32
from prometheus_client import make_asgi_app
from prometheus_fastapi_instrumentator import Instrumentator
33
from starlette.concurrency import iterate_in_threadpool
34
from starlette.datastructures import URL, Headers, MutableHeaders, State
35
from starlette.routing import Mount
36
from starlette.types import ASGIApp, Message, Receive, Scope, Send
37
from typing_extensions import assert_never
Zhuohan Li's avatar
Zhuohan Li committed
38

39
import vllm.envs as envs
40
from vllm.config import VllmConfig
Woosuk Kwon's avatar
Woosuk Kwon committed
41
from vllm.engine.arg_utils import AsyncEngineArgs
42
from vllm.engine.protocol import EngineClient
43
44
45
46
47
48
49
from vllm.entrypoints.anthropic.protocol import (
    AnthropicError,
    AnthropicErrorResponse,
    AnthropicMessagesRequest,
    AnthropicMessagesResponse,
)
from vllm.entrypoints.anthropic.serving_messages import AnthropicServingMessages
50
from vllm.entrypoints.launcher import serve_http
51
from vllm.entrypoints.logger import RequestLogger
52
from vllm.entrypoints.openai.cli_args import make_arg_parser, validate_parsed_serve_args
53
from vllm.entrypoints.openai.orca_metrics import metrics_header
54
55
56
57
58
59
60
61
62
from vllm.entrypoints.openai.protocol import (
    ChatCompletionRequest,
    ChatCompletionResponse,
    CompletionRequest,
    CompletionResponse,
    DetokenizeRequest,
    DetokenizeResponse,
    ErrorInfo,
    ErrorResponse,
63
64
    GenerateRequest,
    GenerateResponse,
65
66
67
68
69
70
71
72
73
74
    ResponsesRequest,
    ResponsesResponse,
    StreamingResponsesResponse,
    TokenizeRequest,
    TokenizeResponse,
    TranscriptionRequest,
    TranscriptionResponse,
    TranslationRequest,
    TranslationResponse,
)
75
76
from vllm.entrypoints.openai.serving_chat import OpenAIServingChat
from vllm.entrypoints.openai.serving_completion import OpenAIServingCompletion
77
from vllm.entrypoints.openai.serving_engine import OpenAIServing
78
79
80
81
from vllm.entrypoints.openai.serving_models import (
    BaseModelPath,
    OpenAIServingModels,
)
82
from vllm.entrypoints.openai.serving_responses import OpenAIServingResponses
83
from vllm.entrypoints.openai.serving_tokenization import OpenAIServingTokenization
84
from vllm.entrypoints.openai.serving_tokens import ServingTokens
85
from vllm.entrypoints.openai.serving_transcription import (
86
87
88
    OpenAIServingTranscription,
    OpenAIServingTranslation,
)
89
from vllm.entrypoints.openai.tool_parsers import ToolParserManager
90
91
92
93
94
from vllm.entrypoints.openai.utils import validate_json_request
from vllm.entrypoints.pooling.classify.serving import ServingClassification
from vllm.entrypoints.pooling.embed.serving import OpenAIServingEmbedding
from vllm.entrypoints.pooling.pooling.serving import OpenAIServingPooling
from vllm.entrypoints.pooling.score.serving import ServingScores
95
96
97
98
99
from vllm.entrypoints.tool_server import DemoToolServer, MCPToolServer, ToolServer
from vllm.entrypoints.utils import (
    cli_env_setup,
    load_aware_call,
    log_non_default_args,
100
101
    process_chat_template,
    process_lora_modules,
102
103
    with_cancellation,
)
104
from vllm.logger import init_logger
105
from vllm.reasoning import ReasoningParserManager
106
from vllm.tasks import POOLING_TASKS
yhu422's avatar
yhu422 committed
107
from vllm.usage.usage_lib import UsageContext
Cyrus Leung's avatar
Cyrus Leung committed
108
from vllm.utils.argparse_utils import FlexibleArgumentParser
109
from vllm.utils.gc_utils import freeze_gc_heap
110
from vllm.utils.network_utils import is_valid_ipv6_address
111
from vllm.utils.system_utils import decorate_logs, set_ulimit
112
from vllm.v1.engine.exceptions import EngineDeadError
113
from vllm.v1.metrics.prometheus import get_prometheus_registry
114
from vllm.version import __version__ as VLLM_VERSION
Zhuohan Li's avatar
Zhuohan Li committed
115

116
prometheus_multiproc_dir: tempfile.TemporaryDirectory
117

118
# Cannot use __name__ (https://github.com/vllm-project/vllm/pull/4765)
119
logger = init_logger("vllm.entrypoints.openai.api_server")
120

121
122
ENDPOINT_LOAD_METRICS_FORMAT_HEADER_LABEL = "endpoint-load-metrics-format"

123
_running_tasks: set[asyncio.Task] = set()
124

125

126
@asynccontextmanager
127
async def lifespan(app: FastAPI):
128
129
    try:
        if app.state.log_stats:
130
            engine_client: EngineClient = app.state.engine_client
131
132
133

            async def _force_log():
                while True:
134
                    await asyncio.sleep(envs.VLLM_LOG_STATS_INTERVAL)
135
                    await engine_client.do_log_stats()
136
137
138
139
140
141

            task = asyncio.create_task(_force_log())
            _running_tasks.add(task)
            task.add_done_callback(_running_tasks.remove)
        else:
            task = None
142
143
144

        # Mark the startup heap as static so that it's ignored by GC.
        # Reduces pause times of oldest generation collections.
145
        freeze_gc_heap()
146
147
148
149
150
151
152
153
        try:
            yield
        finally:
            if task is not None:
                task.cancel()
    finally:
        # Ensure app state including engine ref is gc'd
        del app.state
154
155


156
@asynccontextmanager
157
async def build_async_engine_client(
158
    args: Namespace,
159
160
    *,
    usage_context: UsageContext = UsageContext.OPENAI_API_SERVER,
161
162
    disable_frontend_multiprocessing: bool | None = None,
    client_config: dict[str, Any] | None = None,
163
) -> AsyncIterator[EngineClient]:
164
165
166
167
    if os.getenv("VLLM_WORKER_MULTIPROC_METHOD") == "forkserver":
        # The executor is expected to be mp.
        # Pre-import heavy modules in the forkserver process
        logger.debug("Setup forkserver with pre-imports")
168
        multiprocessing.set_start_method("forkserver")
169
170
171
172
        multiprocessing.set_forkserver_preload(["vllm.v1.engine.async_llm"])
        forkserver.ensure_running()
        logger.debug("Forkserver setup complete!")

173
    # Context manager to handle engine_client lifecycle
174
175
    # Ensures everything is shutdown and cleaned up on error/exit
    engine_args = AsyncEngineArgs.from_cli_args(args)
176
177
178
    if client_config:
        engine_args._api_process_count = client_config.get("client_count", 1)
        engine_args._api_process_rank = client_config.get("client_index", 0)
179

180
    if disable_frontend_multiprocessing is None:
181
        disable_frontend_multiprocessing = bool(args.disable_frontend_multiprocessing)
182

183
    async with build_async_engine_client_from_engine_args(
184
185
186
187
        engine_args,
        usage_context=usage_context,
        disable_frontend_multiprocessing=disable_frontend_multiprocessing,
        client_config=client_config,
188
    ) as engine:
189
190
191
192
193
194
        yield engine


@asynccontextmanager
async def build_async_engine_client_from_engine_args(
    engine_args: AsyncEngineArgs,
195
196
    *,
    usage_context: UsageContext = UsageContext.OPENAI_API_SERVER,
197
    disable_frontend_multiprocessing: bool = False,
198
    client_config: dict[str, Any] | None = None,
199
) -> AsyncIterator[EngineClient]:
200
    """
201
    Create EngineClient, either:
202
203
204
205
206
207
        - in-process using the AsyncLLMEngine Directly
        - multiprocess using AsyncLLMEngine RPC

    Returns the Client or None if the creation failed.
    """

208
209
210
    # Create the EngineConfig (determines if we can use V1).
    vllm_config = engine_args.create_engine_config(usage_context=usage_context)

211
    if disable_frontend_multiprocessing:
212
        logger.warning("V1 is enabled, but got --disable-frontend-multiprocessing.")
213

214
    from vllm.v1.engine.async_llm import AsyncLLM
215

216
    async_llm: AsyncLLM | None = None
217
218
219
220
221
222

    # Don't mutate the input client_config
    client_config = dict(client_config) if client_config else {}
    client_count = client_config.pop("client_count", 1)
    client_index = client_config.pop("client_index", 0)

223
224
225
226
227
    try:
        async_llm = AsyncLLM.from_vllm_config(
            vllm_config=vllm_config,
            usage_context=usage_context,
            enable_log_requests=engine_args.enable_log_requests,
228
            aggregate_engine_logging=engine_args.aggregate_engine_logging,
229
230
231
            disable_log_stats=engine_args.disable_log_stats,
            client_addresses=client_config,
            client_count=client_count,
232
233
            client_index=client_index,
        )
234
235

        # Don't keep the dummy data in memory
236
        assert async_llm is not None
237
238
239
240
241
242
        await async_llm.reset_mm_cache()

        yield async_llm
    finally:
        if async_llm:
            async_llm.shutdown()
243
244


Ethan Xu's avatar
Ethan Xu committed
245
router = APIRouter()
Zhuohan Li's avatar
Zhuohan Li committed
246

247

248
249
250
251
class PrometheusResponse(Response):
    media_type = prometheus_client.CONTENT_TYPE_LATEST


252
def mount_metrics(app: FastAPI):
253
254
255
    """Mount prometheus metrics to a FastAPI app."""

    registry = get_prometheus_registry()
256

257
258
259
260
    # `response_class=PrometheusResponse` is needed to return an HTTP response
    # with header "Content-Type: text/plain; version=0.0.4; charset=utf-8"
    # instead of the default "application/json" which is incorrect.
    # See https://github.com/trallnag/prometheus-fastapi-instrumentator/issues/163#issue-1296092364
261
262
263
264
265
266
267
268
269
270
    Instrumentator(
        excluded_handlers=[
            "/metrics",
            "/health",
            "/load",
            "/ping",
            "/version",
            "/server_info",
        ],
        registry=registry,
271
    ).add().instrument(app).expose(app, response_class=PrometheusResponse)
272
273
274

    # Add prometheus asgi middleware to route /metrics requests
    metrics_route = Mount("/metrics", make_asgi_app(registry=registry))
275

276
    # Workaround for 307 Redirect for /metrics
277
    metrics_route.path_regex = re.compile("^/metrics(?P<path>.*)$")
278
    app.routes.append(metrics_route)
279
280


281
282
283
284
285
def base(request: Request) -> OpenAIServing:
    # Reuse the existing instance
    return tokenization(request)


286
287
288
289
def models(request: Request) -> OpenAIServingModels:
    return request.app.state.openai_serving_models


290
def responses(request: Request) -> OpenAIServingResponses | None:
291
292
293
    return request.app.state.openai_serving_responses


294
295
296
297
def messages(request: Request) -> AnthropicServingMessages:
    return request.app.state.anthropic_serving_messages


298
def chat(request: Request) -> OpenAIServingChat | None:
299
300
301
    return request.app.state.openai_serving_chat


302
def completion(request: Request) -> OpenAIServingCompletion | None:
303
304
305
    return request.app.state.openai_serving_completion


306
307
def tokenization(request: Request) -> OpenAIServingTokenization:
    return request.app.state.openai_serving_tokenization
308
309


310
311
312
313
def transcription(request: Request) -> OpenAIServingTranscription:
    return request.app.state.openai_serving_transcription


314
315
316
317
def translation(request: Request) -> OpenAIServingTranslation:
    return request.app.state.openai_serving_translation


318
def engine_client(request: Request) -> EngineClient:
319
320
321
    return request.app.state.engine_client


322
323
324
325
def generate_tokens(request: Request) -> ServingTokens | None:
    return request.app.state.serving_tokens


326
327
@router.get("/health", response_class=Response)
async def health(raw_request: Request) -> Response:
328
    """Health check."""
329
330
331
332
333
    try:
        await engine_client(raw_request).check_health()
        return Response(status_code=200)
    except EngineDeadError:
        return Response(status_code=503)
334
335


336
337
338
339
340
341
342
@router.get("/load")
async def get_server_load_metrics(request: Request):
    # This endpoint returns the current server load metrics.
    # It tracks requests utilizing the GPU from the following routes:
    # - /v1/chat/completions
    # - /v1/completions
    # - /v1/audio/transcriptions
343
    # - /v1/audio/translations
344
345
    # - /v1/embeddings
    # - /pooling
346
    # - /classify
347
348
349
350
351
    # - /score
    # - /v1/score
    # - /rerank
    # - /v1/rerank
    # - /v2/rerank
352
    return JSONResponse(content={"server_load": request.app.state.server_load_metrics})
353
354


355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
@router.post("/pause")
async def pause_generation(
    raw_request: Request,
    wait_for_inflight_requests: bool = Query(False),
    clear_cache: bool = Query(True),
) -> JSONResponse:
    """Pause generation requests to allow weight updates.

    Args:
        wait_for_inflight_requests: When ``True`` waits for in-flight
            requests to finish before pausing. When ``False`` (default),
            aborts any in-flight requests immediately.
        clear_cache: Whether to clear KV/prefix caches after draining.
    """

    engine = engine_client(raw_request)

    try:
        await engine.pause_generation(
            wait_for_inflight_requests=wait_for_inflight_requests,
            clear_cache=clear_cache,
        )
        return JSONResponse(
            content={"status": "paused"},
            status_code=HTTPStatus.OK.value,
        )

    except ValueError as err:
        return JSONResponse(
            content={"error": str(err)},
            status_code=HTTPStatus.BAD_REQUEST.value,
        )
    except Exception as err:  # pragma: no cover - defensive
        logger.exception("Failed to pause generation")
        return JSONResponse(
            content={"error": f"Failed to pause generation: {err}"},
            status_code=HTTPStatus.INTERNAL_SERVER_ERROR.value,
        )


@router.post("/resume")
async def resume_generation(raw_request: Request) -> JSONResponse:
    """Resume generation after a pause."""

    engine = engine_client(raw_request)

    try:
        await engine.resume_generation()
        return JSONResponse(
            content={"status": "resumed"},
            status_code=HTTPStatus.OK.value,
        )
    except Exception as err:  # pragma: no cover - defensive
        logger.exception("Failed to resume generation")
        return JSONResponse(
            content={"error": f"Failed to resume generation: {err}"},
            status_code=HTTPStatus.INTERNAL_SERVER_ERROR.value,
        )


@router.get("/is_paused")
async def is_paused(raw_request: Request) -> JSONResponse:
    """Return the current pause status."""

    engine = engine_client(raw_request)

    try:
        paused = await engine.is_paused()
    except Exception as err:  # pragma: no cover - defensive
        logger.exception("Failed to fetch pause status")
        return JSONResponse(
            content={"error": f"Failed to fetch pause status: {err}"},
            status_code=HTTPStatus.INTERNAL_SERVER_ERROR.value,
        )

    return JSONResponse(content={"is_paused": paused})


433
434
435
436
437
438
439
440
441
442
@router.post(
    "/tokenize",
    dependencies=[Depends(validate_json_request)],
    responses={
        HTTPStatus.BAD_REQUEST.value: {"model": ErrorResponse},
        HTTPStatus.NOT_FOUND.value: {"model": ErrorResponse},
        HTTPStatus.INTERNAL_SERVER_ERROR.value: {"model": ErrorResponse},
        HTTPStatus.NOT_IMPLEMENTED.value: {"model": ErrorResponse},
    },
)
443
@with_cancellation
444
async def tokenize(request: TokenizeRequest, raw_request: Request):
445
446
    handler = tokenization(raw_request)

447
448
449
    try:
        generator = await handler.create_tokenize(request, raw_request)
    except NotImplementedError as e:
450
451
452
        raise HTTPException(
            status_code=HTTPStatus.NOT_IMPLEMENTED.value, detail=str(e)
        ) from e
453
    except Exception as e:
454
455
456
        raise HTTPException(
            status_code=HTTPStatus.INTERNAL_SERVER_ERROR.value, detail=str(e)
        ) from e
457

458
    if isinstance(generator, ErrorResponse):
459
460
461
        return JSONResponse(
            content=generator.model_dump(), status_code=generator.error.code
        )
462
    elif isinstance(generator, TokenizeResponse):
463
464
        return JSONResponse(content=generator.model_dump())

465
466
    assert_never(generator)

467

468
469
470
471
472
473
474
475
476
@router.post(
    "/detokenize",
    dependencies=[Depends(validate_json_request)],
    responses={
        HTTPStatus.BAD_REQUEST.value: {"model": ErrorResponse},
        HTTPStatus.NOT_FOUND.value: {"model": ErrorResponse},
        HTTPStatus.INTERNAL_SERVER_ERROR.value: {"model": ErrorResponse},
    },
)
477
@with_cancellation
478
async def detokenize(request: DetokenizeRequest, raw_request: Request):
479
480
    handler = tokenization(raw_request)

481
482
483
484
485
    try:
        generator = await handler.create_detokenize(request, raw_request)
    except OverflowError as e:
        raise RequestValidationError(errors=[str(e)]) from e
    except Exception as e:
486
487
488
        raise HTTPException(
            status_code=HTTPStatus.INTERNAL_SERVER_ERROR.value, detail=str(e)
        ) from e
489

490
    if isinstance(generator, ErrorResponse):
491
492
493
        return JSONResponse(
            content=generator.model_dump(), status_code=generator.error.code
        )
494
    elif isinstance(generator, DetokenizeResponse):
495
496
        return JSONResponse(content=generator.model_dump())

497
498
    assert_never(generator)

499

500
501
def maybe_register_tokenizer_info_endpoint(args):
    """Conditionally register the tokenizer info endpoint if enabled."""
502
    if getattr(args, "enable_tokenizer_info_endpoint", False):
503
504
505
506
507

        @router.get("/tokenizer_info")
        async def get_tokenizer_info(raw_request: Request):
            """Get comprehensive tokenizer information."""
            result = await tokenization(raw_request).get_tokenizer_info()
508
509
510
511
512
513
            return JSONResponse(
                content=result.model_dump(),
                status_code=result.error.code
                if isinstance(result, ErrorResponse)
                else 200,
            )
514
515


Ethan Xu's avatar
Ethan Xu committed
516
@router.get("/v1/models")
517
async def show_available_models(raw_request: Request):
518
    handler = models(raw_request)
519

520
521
    models_ = await handler.show_available_models()
    return JSONResponse(content=models_.model_dump())
Zhuohan Li's avatar
Zhuohan Li committed
522
523


Ethan Xu's avatar
Ethan Xu committed
524
@router.get("/version")
525
async def show_version():
526
    ver = {"version": VLLM_VERSION}
527
528
529
    return JSONResponse(content=ver)


530
async def _convert_stream_to_sse_events(
531
    generator: AsyncGenerator[StreamingResponsesResponse, None],
532
) -> AsyncGenerator[str, None]:
533
534
    """Convert the generator to a stream of events in SSE format"""
    async for event in generator:
535
        event_type = getattr(event, "type", "unknown")
536
        # https://developer.mozilla.org/en-US/docs/Web/API/Server-sent_events/Using_server-sent_events#event_stream_format
537
538
539
        event_data = (
            f"event: {event_type}\ndata: {event.model_dump_json(indent=None)}\n\n"
        )
540
541
542
        yield event_data


543
544
545
546
547
548
549
550
551
552
@router.post(
    "/v1/responses",
    dependencies=[Depends(validate_json_request)],
    responses={
        HTTPStatus.OK.value: {"content": {"text/event-stream": {}}},
        HTTPStatus.BAD_REQUEST.value: {"model": ErrorResponse},
        HTTPStatus.NOT_FOUND.value: {"model": ErrorResponse},
        HTTPStatus.INTERNAL_SERVER_ERROR.value: {"model": ErrorResponse},
    },
)
553
554
555
556
557
@with_cancellation
async def create_responses(request: ResponsesRequest, raw_request: Request):
    handler = responses(raw_request)
    if handler is None:
        return base(raw_request).create_error_response(
558
559
            message="The model does not support Responses API"
        )
560
561
562
    try:
        generator = await handler.create_responses(request, raw_request)
    except Exception as e:
563
564
565
        raise HTTPException(
            status_code=HTTPStatus.INTERNAL_SERVER_ERROR.value, detail=str(e)
        ) from e
566
567

    if isinstance(generator, ErrorResponse):
568
569
570
        return JSONResponse(
            content=generator.model_dump(), status_code=generator.error.code
        )
571
572
    elif isinstance(generator, ResponsesResponse):
        return JSONResponse(content=generator.model_dump())
573

574
575
576
    return StreamingResponse(
        content=_convert_stream_to_sse_events(generator), media_type="text/event-stream"
    )
577
578
579


@router.get("/v1/responses/{response_id}")
580
581
582
async def retrieve_responses(
    response_id: str,
    raw_request: Request,
583
584
    starting_after: int | None = None,
    stream: bool | None = False,
585
):
586
587
588
    handler = responses(raw_request)
    if handler is None:
        return base(raw_request).create_error_response(
589
590
            message="The model does not support Responses API"
        )
591

592
    try:
593
594
595
596
597
        response = await handler.retrieve_responses(
            response_id,
            starting_after=starting_after,
            stream=stream,
        )
598
    except Exception as e:
599
600
601
        raise HTTPException(
            status_code=HTTPStatus.INTERNAL_SERVER_ERROR.value, detail=str(e)
        ) from e
602
603

    if isinstance(response, ErrorResponse):
604
605
606
        return JSONResponse(
            content=response.model_dump(), status_code=response.error.code
        )
607
608
    elif isinstance(response, ResponsesResponse):
        return JSONResponse(content=response.model_dump())
609
610
611
    return StreamingResponse(
        content=_convert_stream_to_sse_events(response), media_type="text/event-stream"
    )
612
613
614
615
616
617
618


@router.post("/v1/responses/{response_id}/cancel")
async def cancel_responses(response_id: str, raw_request: Request):
    handler = responses(raw_request)
    if handler is None:
        return base(raw_request).create_error_response(
619
620
            message="The model does not support Responses API"
        )
621

622
623
624
    try:
        response = await handler.cancel_responses(response_id)
    except Exception as e:
625
626
627
        raise HTTPException(
            status_code=HTTPStatus.INTERNAL_SERVER_ERROR.value, detail=str(e)
        ) from e
628
629

    if isinstance(response, ErrorResponse):
630
631
632
        return JSONResponse(
            content=response.model_dump(), status_code=response.error.code
        )
633
634
635
    return JSONResponse(content=response.model_dump())


636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
@router.post(
    "/v1/messages",
    dependencies=[Depends(validate_json_request)],
    responses={
        HTTPStatus.OK.value: {"content": {"text/event-stream": {}}},
        HTTPStatus.BAD_REQUEST.value: {"model": AnthropicErrorResponse},
        HTTPStatus.NOT_FOUND.value: {"model": AnthropicErrorResponse},
        HTTPStatus.INTERNAL_SERVER_ERROR.value: {"model": AnthropicErrorResponse},
    },
)
@with_cancellation
@load_aware_call
async def create_messages(request: AnthropicMessagesRequest, raw_request: Request):
    def translate_error_response(response: ErrorResponse) -> JSONResponse:
        anthropic_error = AnthropicErrorResponse(
            error=AnthropicError(
                type=response.error.type,
                message=response.error.message,
            )
        )
        return JSONResponse(
            status_code=response.error.code, content=anthropic_error.model_dump()
        )

    handler = messages(raw_request)
    if handler is None:
        error = base(raw_request).create_error_response(
            message="The model does not support Messages API"
        )
        return translate_error_response(error)

    try:
        generator = await handler.create_messages(request, raw_request)
    except Exception as e:
        logger.exception("Error in create_messages: %s", e)
        return JSONResponse(
            status_code=HTTPStatus.INTERNAL_SERVER_ERROR.value,
            content=AnthropicErrorResponse(
                error=AnthropicError(
                    type="internal_error",
                    message=str(e),
                )
            ).model_dump(),
        )

    if isinstance(generator, ErrorResponse):
        return translate_error_response(generator)

    elif isinstance(generator, AnthropicMessagesResponse):
685
686
687
        resp = generator.model_dump(exclude_none=True)
        logger.debug("Anthropic Messages Response: %s", resp)
        return JSONResponse(content=resp)
688
689
690
691

    return StreamingResponse(content=generator, media_type="text/event-stream")


692
693
694
695
696
697
698
699
700
701
@router.post(
    "/v1/chat/completions",
    dependencies=[Depends(validate_json_request)],
    responses={
        HTTPStatus.OK.value: {"content": {"text/event-stream": {}}},
        HTTPStatus.BAD_REQUEST.value: {"model": ErrorResponse},
        HTTPStatus.NOT_FOUND.value: {"model": ErrorResponse},
        HTTPStatus.INTERNAL_SERVER_ERROR.value: {"model": ErrorResponse},
    },
)
702
@with_cancellation
703
@load_aware_call
704
async def create_chat_completion(request: ChatCompletionRequest, raw_request: Request):
705
706
707
    metrics_header_format = raw_request.headers.get(
        ENDPOINT_LOAD_METRICS_FORMAT_HEADER_LABEL, ""
    )
708
709
710
    handler = chat(raw_request)
    if handler is None:
        return base(raw_request).create_error_response(
711
712
            message="The model does not support Chat Completions API"
        )
713
714
715
    try:
        generator = await handler.create_chat_completion(request, raw_request)
    except Exception as e:
716
717
718
        raise HTTPException(
            status_code=HTTPStatus.INTERNAL_SERVER_ERROR.value, detail=str(e)
        ) from e
719
    if isinstance(generator, ErrorResponse):
720
721
722
        return JSONResponse(
            content=generator.model_dump(), status_code=generator.error.code
        )
723

724
    elif isinstance(generator, ChatCompletionResponse):
725
726
727
728
        return JSONResponse(
            content=generator.model_dump(),
            headers=metrics_header(metrics_header_format),
        )
729

730
731
    return StreamingResponse(content=generator, media_type="text/event-stream")

732

733
734
735
736
737
738
739
740
741
742
@router.post(
    "/v1/completions",
    dependencies=[Depends(validate_json_request)],
    responses={
        HTTPStatus.OK.value: {"content": {"text/event-stream": {}}},
        HTTPStatus.BAD_REQUEST.value: {"model": ErrorResponse},
        HTTPStatus.NOT_FOUND.value: {"model": ErrorResponse},
        HTTPStatus.INTERNAL_SERVER_ERROR.value: {"model": ErrorResponse},
    },
)
743
@with_cancellation
744
@load_aware_call
745
async def create_completion(request: CompletionRequest, raw_request: Request):
746
747
748
    metrics_header_format = raw_request.headers.get(
        ENDPOINT_LOAD_METRICS_FORMAT_HEADER_LABEL, ""
    )
749
750
751
    handler = completion(raw_request)
    if handler is None:
        return base(raw_request).create_error_response(
752
753
            message="The model does not support Completions API"
        )
754

755
756
757
    try:
        generator = await handler.create_completion(request, raw_request)
    except OverflowError as e:
758
759
760
        raise HTTPException(
            status_code=HTTPStatus.BAD_REQUEST.value, detail=str(e)
        ) from e
761
    except Exception as e:
762
763
764
        raise HTTPException(
            status_code=HTTPStatus.INTERNAL_SERVER_ERROR.value, detail=str(e)
        ) from e
765

766
    if isinstance(generator, ErrorResponse):
767
768
769
        return JSONResponse(
            content=generator.model_dump(), status_code=generator.error.code
        )
770
    elif isinstance(generator, CompletionResponse):
771
772
773
774
        return JSONResponse(
            content=generator.model_dump(),
            headers=metrics_header(metrics_header_format),
        )
Zhuohan Li's avatar
Zhuohan Li committed
775

776
777
    return StreamingResponse(content=generator, media_type="text/event-stream")

Zhuohan Li's avatar
Zhuohan Li committed
778

779
780
781
782
783
784
785
786
787
@router.post(
    "/v1/audio/transcriptions",
    responses={
        HTTPStatus.OK.value: {"content": {"text/event-stream": {}}},
        HTTPStatus.BAD_REQUEST.value: {"model": ErrorResponse},
        HTTPStatus.UNPROCESSABLE_ENTITY.value: {"model": ErrorResponse},
        HTTPStatus.INTERNAL_SERVER_ERROR.value: {"model": ErrorResponse},
    },
)
788
@with_cancellation
789
@load_aware_call
790
791
792
async def create_transcriptions(
    raw_request: Request, request: Annotated[TranscriptionRequest, Form()]
):
793
794
795
    handler = transcription(raw_request)
    if handler is None:
        return base(raw_request).create_error_response(
796
797
            message="The model does not support Transcriptions API"
        )
798
799

    audio_data = await request.file.read()
800
    try:
801
        generator = await handler.create_transcription(audio_data, request, raw_request)
802
    except Exception as e:
803
804
805
        raise HTTPException(
            status_code=HTTPStatus.INTERNAL_SERVER_ERROR.value, detail=str(e)
        ) from e
806
807

    if isinstance(generator, ErrorResponse):
808
809
810
        return JSONResponse(
            content=generator.model_dump(), status_code=generator.error.code
        )
811
812
813
814
815
816
817

    elif isinstance(generator, TranscriptionResponse):
        return JSONResponse(content=generator.model_dump())

    return StreamingResponse(content=generator, media_type="text/event-stream")


818
819
820
821
822
823
824
825
826
@router.post(
    "/v1/audio/translations",
    responses={
        HTTPStatus.OK.value: {"content": {"text/event-stream": {}}},
        HTTPStatus.BAD_REQUEST.value: {"model": ErrorResponse},
        HTTPStatus.UNPROCESSABLE_ENTITY.value: {"model": ErrorResponse},
        HTTPStatus.INTERNAL_SERVER_ERROR.value: {"model": ErrorResponse},
    },
)
827
828
@with_cancellation
@load_aware_call
829
830
831
async def create_translations(
    request: Annotated[TranslationRequest, Form()], raw_request: Request
):
832
833
834
    handler = translation(raw_request)
    if handler is None:
        return base(raw_request).create_error_response(
835
836
            message="The model does not support Translations API"
        )
837
838

    audio_data = await request.file.read()
839
    try:
840
        generator = await handler.create_translation(audio_data, request, raw_request)
841
    except Exception as e:
842
843
844
        raise HTTPException(
            status_code=HTTPStatus.INTERNAL_SERVER_ERROR.value, detail=str(e)
        ) from e
845
846

    if isinstance(generator, ErrorResponse):
847
848
849
        return JSONResponse(
            content=generator.model_dump(), status_code=generator.error.code
        )
850
851
852
853
854
855
856

    elif isinstance(generator, TranslationResponse):
        return JSONResponse(content=generator.model_dump())

    return StreamingResponse(content=generator, media_type="text/event-stream")


857
if envs.VLLM_SERVER_DEV_MODE:
858
859
860
861
    logger.warning(
        "SECURITY WARNING: Development endpoints are enabled! "
        "This should NOT be used in production!"
    )
862

863
864
    PydanticVllmConfig = pydantic.TypeAdapter(VllmConfig)

865
    @router.get("/server_info")
866
867
    async def show_server_info(
        raw_request: Request,
868
        config_format: Annotated[Literal["text", "json"], Query()] = "text",
869
870
871
    ):
        vllm_config: VllmConfig = raw_request.app.state.vllm_config
        server_info = {
872
873
874
            "vllm_config": str(vllm_config)
            if config_format == "text"
            else PydanticVllmConfig.dump_python(vllm_config, mode="json", fallback=str)
875
876
            # fallback=str is needed to handle e.g. torch.dtype
        }
877
878
        return JSONResponse(content=server_info)

879
880
881
882
883
884
    @router.post("/reset_prefix_cache")
    async def reset_prefix_cache(raw_request: Request):
        """
        Reset the prefix cache. Note that we currently do not check if the
        prefix cache is successfully reset in the API server.
        """
885
886
        logger.info("Resetting prefix cache...")
        await engine_client(raw_request).reset_prefix_cache()
887
888
        return Response(status_code=200)

889
890
891
892
893
894
895
896
897
898
    @router.post("/reset_mm_cache")
    async def reset_mm_cache(raw_request: Request):
        """
        Reset the multi-modal cache. Note that we currently do not check if the
        multi-modal cache is successfully reset in the API server.
        """
        logger.info("Resetting multi-modal cache...")
        await engine_client(raw_request).reset_mm_cache()
        return Response(status_code=200)

899
900
901
902
903
904
905
906
907
908
909
    @router.post("/sleep")
    async def sleep(raw_request: Request):
        # get POST params
        level = raw_request.query_params.get("level", "1")
        await engine_client(raw_request).sleep(int(level))
        # FIXME: in v0 with frontend multiprocessing, the sleep command
        # is sent but does not finish yet when we return a response.
        return Response(status_code=200)

    @router.post("/wake_up")
    async def wake_up(raw_request: Request):
910
911
912
913
914
915
        tags = raw_request.query_params.getlist("tags")
        if tags == []:
            # set to None to wake up all tags if no tags are provided
            tags = None
        logger.info("wake up the engine with tags: %s", tags)
        await engine_client(raw_request).wake_up(tags)
916
917
918
919
        # FIXME: in v0 with frontend multiprocessing, the wake-up command
        # is sent but does not finish yet when we return a response.
        return Response(status_code=200)

920
921
922
923
924
925
    @router.get("/is_sleeping")
    async def is_sleeping(raw_request: Request):
        logger.info("check whether the engine is sleeping")
        is_sleeping = await engine_client(raw_request).is_sleeping()
        return JSONResponse(content={"is_sleeping": is_sleeping})

926
927
928
929
930
    @router.post("/collective_rpc")
    async def collective_rpc(raw_request: Request):
        try:
            body = await raw_request.json()
        except json.JSONDecodeError as e:
931
932
933
934
            raise HTTPException(
                status_code=HTTPStatus.BAD_REQUEST.value,
                detail=f"JSON decode error: {e}",
            ) from e
935
936
        method = body.get("method")
        if method is None:
937
938
939
940
            raise HTTPException(
                status_code=HTTPStatus.BAD_REQUEST.value,
                detail="Missing 'method' in request body",
            )
941
        # For security reason, only serialized string args/kwargs are passed.
942
        # User-defined `method` is responsible for deserialization if needed.
943
944
        args: list[str] = body.get("args", [])
        kwargs: dict[str, str] = body.get("kwargs", {})
945
        timeout: float | None = body.get("timeout")
946
        results = await engine_client(raw_request).collective_rpc(
947
948
            method=method, timeout=timeout, args=tuple(args), kwargs=kwargs
        )
949
950
951
952
953
954
955
956
957
958
        if results is None:
            return Response(status_code=200)
        response: list[Any] = []
        for result in results:
            if result is None or isinstance(result, (dict, list)):
                response.append(result)
            else:
                response.append(str(result))
        return JSONResponse(content={"results": response})

959

960
961
962
963
964
965
966
967
968
969
@router.post(
    "/scale_elastic_ep",
    dependencies=[Depends(validate_json_request)],
    responses={
        HTTPStatus.OK.value: {"model": dict},
        HTTPStatus.BAD_REQUEST.value: {"model": ErrorResponse},
        HTTPStatus.REQUEST_TIMEOUT.value: {"model": ErrorResponse},
        HTTPStatus.INTERNAL_SERVER_ERROR.value: {"model": ErrorResponse},
    },
)
970
971
972
973
async def scale_elastic_ep(raw_request: Request):
    try:
        body = await raw_request.json()
    except json.JSONDecodeError as e:
974
        raise HTTPException(status_code=400, detail="Invalid JSON format") from e  # noqa: B904
975
976
977
978
979

    new_data_parallel_size = body.get("new_data_parallel_size")
    drain_timeout = body.get("drain_timeout", 120)  # Default 2 minutes

    if new_data_parallel_size is None:
980
981
982
        raise HTTPException(
            status_code=400, detail="new_data_parallel_size is required"
        )
983

984
    if not isinstance(new_data_parallel_size, int) or new_data_parallel_size <= 0:
985
        raise HTTPException(
986
987
            status_code=400, detail="new_data_parallel_size must be a positive integer"
        )
988
989

    if not isinstance(drain_timeout, int) or drain_timeout <= 0:
990
991
992
        raise HTTPException(
            status_code=400, detail="drain_timeout must be a positive integer"
        )
993
994
995
996
997
998
999

    # Set scaling flag to prevent new requests
    global _scaling_elastic_ep
    _scaling_elastic_ep = True
    client = engine_client(raw_request)
    try:
        await client.scale_elastic_ep(new_data_parallel_size, drain_timeout)
1000
1001
1002
1003
1004
        return JSONResponse(
            {
                "message": f"Scaled to {new_data_parallel_size} data parallel engines",
            }
        )
1005
    except TimeoutError as e:
1006
1007
1008
1009
1010
        raise HTTPException(
            status_code=408,
            detail="Scale failed due to request drain timeout "
            f"after {drain_timeout} seconds",
        ) from e
1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021
1022
    except Exception as e:
        logger.error("Scale failed: %s", e)
        raise HTTPException(status_code=500, detail="Scale failed") from e
    finally:
        _scaling_elastic_ep = False


@router.post("/is_scaling_elastic_ep")
async def is_scaling_elastic_ep(raw_request: Request):
    return JSONResponse({"is_scaling_elastic_ep": _scaling_elastic_ep})


1023
1024
1025
1026
1027
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
1040
1041
1042
1043
1044
1045
1046
1047
1048
1049
1050
1051
1052
1053
1054
1055
1056
1057
@router.post(
    "/inference/v1/generate",
    dependencies=[Depends(validate_json_request)],
    responses={
        HTTPStatus.OK.value: {"content": {"text/event-stream": {}}},
        HTTPStatus.BAD_REQUEST.value: {"model": ErrorResponse},
        HTTPStatus.NOT_FOUND.value: {"model": ErrorResponse},
        HTTPStatus.INTERNAL_SERVER_ERROR.value: {"model": ErrorResponse},
    },
)
@with_cancellation
@load_aware_call
async def generate(request: GenerateRequest, raw_request: Request):
    handler = generate_tokens(raw_request)
    if handler is None:
        return base(raw_request).create_error_response(
            message="The model does not support generate tokens API"
        )
    try:
        generator = await handler.serve_tokens(request, raw_request)
    except Exception as e:
        raise HTTPException(
            status_code=HTTPStatus.INTERNAL_SERVER_ERROR.value, detail=str(e)
        ) from e
    if isinstance(generator, ErrorResponse):
        return JSONResponse(
            content=generator.model_dump(), status_code=generator.error.code
        )

    elif isinstance(generator, GenerateResponse):
        return JSONResponse(content=generator.model_dump())

    return StreamingResponse(content=generator, media_type="text/event-stream")


1058
if envs.VLLM_TORCH_PROFILER_DIR:
1059
    logger.warning_once(
1060
        "Torch Profiler is enabled in the API server. This should ONLY be "
1061
1062
        "used for local development!"
    )
1063
1064
1065
1066
1067
1068
elif envs.VLLM_TORCH_CUDA_PROFILE:
    logger.warning_once(
        "CUDA Profiler is enabled in the API server. This should ONLY be "
        "used for local development!"
    )
if envs.VLLM_TORCH_PROFILER_DIR or envs.VLLM_TORCH_CUDA_PROFILE:
1069
1070

    @router.post("/start_profile")
1071
    async def start_profile(raw_request: Request):
1072
        logger.info("Starting profiler...")
1073
        await engine_client(raw_request).start_profile()
1074
1075
1076
1077
        logger.info("Profiler started.")
        return Response(status_code=200)

    @router.post("/stop_profile")
1078
    async def stop_profile(raw_request: Request):
1079
        logger.info("Stopping profiler...")
1080
        await engine_client(raw_request).stop_profile()
1081
1082
1083
1084
        logger.info("Profiler stopped.")
        return Response(status_code=200)


1085
def load_log_config(log_config_file: str | None) -> dict | None:
1086
1087
1088
1089
1090
1091
    if not log_config_file:
        return None
    try:
        with open(log_config_file) as f:
            return json.load(f)
    except Exception as e:
1092
1093
1094
        logger.warning(
            "Failed to load log config from file %s: error %s", log_config_file, e
        )
1095
1096
1097
        return None


1098
1099
1100
class AuthenticationMiddleware:
    """
    Pure ASGI middleware that authenticates each request by checking
1101
    if the Authorization Bearer token exists and equals anyof "{api_key}".
1102
1103
1104
1105
1106
1107
1108
1109

    Notes
    -----
    There are two cases in which authentication is skipped:
        1. The HTTP method is OPTIONS.
        2. The request path doesn't start with /v1 (e.g. /health).
    """

1110
    def __init__(self, app: ASGIApp, tokens: list[str]) -> None:
1111
        self.app = app
1112
        self.api_tokens = [hashlib.sha256(t.encode("utf-8")).digest() for t in tokens]
1113
1114
1115
1116
1117
1118
1119
1120
1121
1122
1123
1124
1125
1126
1127
1128
1129

    def verify_token(self, headers: Headers) -> bool:
        authorization_header_value = headers.get("Authorization")
        if not authorization_header_value:
            return False

        scheme, _, param = authorization_header_value.partition(" ")
        if scheme.lower() != "bearer":
            return False

        param_hash = hashlib.sha256(param.encode("utf-8")).digest()

        token_match = False
        for token_hash in self.api_tokens:
            token_match |= secrets.compare_digest(param_hash, token_hash)

        return token_match
1130

1131
1132
    def __call__(self, scope: Scope, receive: Receive, send: Send) -> Awaitable[None]:
        if scope["type"] not in ("http", "websocket") or scope["method"] == "OPTIONS":
1133
1134
1135
1136
1137
1138
1139
            # scope["type"] can be "lifespan" or "startup" for example,
            # in which case we don't need to do anything
            return self.app(scope, receive, send)
        root_path = scope.get("root_path", "")
        url_path = URL(scope=scope).path.removeprefix(root_path)
        headers = Headers(scope=scope)
        # Type narrow to satisfy mypy.
1140
        if url_path.startswith("/v1") and not self.verify_token(headers):
1141
            response = JSONResponse(content={"error": "Unauthorized"}, status_code=401)
1142
1143
1144
1145
1146
1147
1148
1149
1150
1151
1152
1153
1154
1155
            return response(scope, receive, send)
        return self.app(scope, receive, send)


class XRequestIdMiddleware:
    """
    Middleware the set's the X-Request-Id header for each response
    to a random uuid4 (hex) value if the header isn't already
    present in the request, otherwise use the provided request id.
    """

    def __init__(self, app: ASGIApp) -> None:
        self.app = app

1156
    def __call__(self, scope: Scope, receive: Receive, send: Send) -> Awaitable[None]:
1157
1158
1159
1160
1161
1162
1163
1164
1165
1166
1167
1168
1169
        if scope["type"] not in ("http", "websocket"):
            return self.app(scope, receive, send)

        # Extract the request headers.
        request_headers = Headers(scope=scope)

        async def send_with_request_id(message: Message) -> None:
            """
            Custom send function to mutate the response headers
            and append X-Request-Id to it.
            """
            if message["type"] == "http.response.start":
                response_headers = MutableHeaders(raw=message["headers"])
1170
                request_id = request_headers.get("X-Request-Id", uuid.uuid4().hex)
1171
1172
1173
1174
1175
1176
                response_headers.append("X-Request-Id", request_id)
            await send(message)

        return self.app(scope, receive, send_with_request_id)


1177
1178
1179
1180
1181
1182
1183
1184
# Global variable to track scaling state
_scaling_elastic_ep = False


class ScalingMiddleware:
    """
    Middleware that checks if the model is currently scaling and
    returns a 503 Service Unavailable response if it is.
1185

1186
1187
1188
1189
1190
1191
1192
    This middleware applies to all HTTP requests and prevents
    processing when the model is in a scaling state.
    """

    def __init__(self, app: ASGIApp) -> None:
        self.app = app

1193
    def __call__(self, scope: Scope, receive: Receive, send: Send) -> Awaitable[None]:
1194
1195
1196
1197
1198
1199
1200
        if scope["type"] != "http":
            return self.app(scope, receive, send)

        # Check global scaling state
        global _scaling_elastic_ep
        if _scaling_elastic_ep:
            # Return 503 Service Unavailable response
1201
1202
1203
1204
1205
1206
            response = JSONResponse(
                content={
                    "error": "The model is currently scaling. Please try again later."
                },
                status_code=503,
            )
1207
1208
1209
1210
1211
            return response(scope, receive, send)

        return self.app(scope, receive, send)


1212
1213
1214
1215
def _extract_content_from_chunk(chunk_data: dict) -> str:
    """Extract content from a streaming response chunk."""
    try:
        from vllm.entrypoints.openai.protocol import (
1216
1217
1218
            ChatCompletionStreamResponse,
            CompletionStreamResponse,
        )
1219
1220

        # Try using Completion types for type-safe parsing
1221
1222
        if chunk_data.get("object") == "chat.completion.chunk":
            chat_response = ChatCompletionStreamResponse.model_validate(chunk_data)
1223
1224
            if chat_response.choices and chat_response.choices[0].delta.content:
                return chat_response.choices[0].delta.content
1225
1226
1227
        elif chunk_data.get("object") == "text_completion":
            completion_response = CompletionStreamResponse.model_validate(chunk_data)
            if completion_response.choices and completion_response.choices[0].text:
1228
1229
1230
                return completion_response.choices[0].text
    except pydantic.ValidationError:
        # Fallback to manual parsing
1231
1232
1233
1234
1235
1236
        if "choices" in chunk_data and chunk_data["choices"]:
            choice = chunk_data["choices"][0]
            if "delta" in choice and choice["delta"].get("content"):
                return choice["delta"]["content"]
            elif choice.get("text"):
                return choice["text"]
1237
1238
1239
1240
1241
1242
1243
1244
1245
1246
1247
1248
1249
1250
1251
    return ""


class SSEDecoder:
    """Robust Server-Sent Events decoder for streaming responses."""

    def __init__(self):
        self.buffer = ""
        self.content_buffer = []

    def decode_chunk(self, chunk: bytes) -> list[dict]:
        """Decode a chunk of SSE data and return parsed events."""
        import json

        try:
1252
            chunk_str = chunk.decode("utf-8")
1253
1254
1255
1256
1257
1258
1259
1260
        except UnicodeDecodeError:
            # Skip malformed chunks
            return []

        self.buffer += chunk_str
        events = []

        # Process complete lines
1261
1262
1263
        while "\n" in self.buffer:
            line, self.buffer = self.buffer.split("\n", 1)
            line = line.rstrip("\r")  # Handle CRLF
1264

1265
            if line.startswith("data: "):
1266
                data_str = line[6:].strip()
1267
1268
                if data_str == "[DONE]":
                    events.append({"type": "done"})
1269
1270
1271
                elif data_str:
                    try:
                        event_data = json.loads(data_str)
1272
                        events.append({"type": "data", "data": event_data})
1273
1274
1275
1276
1277
1278
1279
1280
1281
1282
1283
1284
1285
1286
1287
1288
1289
                    except json.JSONDecodeError:
                        # Skip malformed JSON
                        continue

        return events

    def extract_content(self, event_data: dict) -> str:
        """Extract content from event data."""
        return _extract_content_from_chunk(event_data)

    def add_content(self, content: str) -> None:
        """Add content to the buffer."""
        if content:
            self.content_buffer.append(content)

    def get_complete_content(self) -> str:
        """Get the complete buffered content."""
1290
        return "".join(self.content_buffer)
1291
1292
1293
1294
1295
1296
1297
1298
1299
1300
1301
1302
1303
1304
1305
1306
1307
1308
1309
1310


def _log_streaming_response(response, response_body: list) -> None:
    """Log streaming response with robust SSE parsing."""
    from starlette.concurrency import iterate_in_threadpool

    sse_decoder = SSEDecoder()
    chunk_count = 0

    def buffered_iterator():
        nonlocal chunk_count

        for chunk in response_body:
            chunk_count += 1
            yield chunk

            # Parse SSE events from chunk
            events = sse_decoder.decode_chunk(chunk)

            for event in events:
1311
1312
                if event["type"] == "data":
                    content = sse_decoder.extract_content(event["data"])
1313
                    sse_decoder.add_content(content)
1314
                elif event["type"] == "done":
1315
1316
1317
1318
1319
1320
1321
1322
                    # Log complete content when done
                    full_content = sse_decoder.get_complete_content()
                    if full_content:
                        # Truncate if too long
                        if len(full_content) > 2048:
                            full_content = full_content[:2048] + ""
                            "...[truncated]"
                        logger.info(
1323
                            "response_body={streaming_complete: content=%r, chunks=%d}",
1324
1325
1326
                            full_content,
                            chunk_count,
                        )
1327
1328
                    else:
                        logger.info(
1329
1330
1331
                            "response_body={streaming_complete: no_content, chunks=%d}",
                            chunk_count,
                        )
1332
1333
1334
                    return

    response.body_iterator = iterate_in_threadpool(buffered_iterator())
1335
    logger.info("response_body={streaming_started: chunks=%d}", len(response_body))
1336
1337
1338
1339
1340
1341
1342
1343
1344
1345
1346


def _log_non_streaming_response(response_body: list) -> None:
    """Log non-streaming response."""
    try:
        decoded_body = response_body[0].decode()
        logger.info("response_body={%s}", decoded_body)
    except UnicodeDecodeError:
        logger.info("response_body={<binary_data>}")


1347
def build_app(args: Namespace) -> FastAPI:
1348
    if args.disable_fastapi_docs:
1349
1350
1351
        app = FastAPI(
            openapi_url=None, docs_url=None, redoc_url=None, lifespan=lifespan
        )
1352
1353
    else:
        app = FastAPI(lifespan=lifespan)
1354
1355
1356
1357
1358
1359
1360
1361
1362
1363
1364
1365
1366

    if envs.VLLM_ALLOW_RUNTIME_LORA_UPDATING:
        logger.warning(
            "LoRA dynamic loading & unloading is enabled in the API server. "
            "This should ONLY be used for local development!"
        )
        from vllm.entrypoints.dynamic_lora import register_dynamic_lora_routes

        register_dynamic_lora_routes(router)

    from vllm.entrypoints.sagemaker.routes import register_sagemaker_routes

    register_sagemaker_routes(router)
Ethan Xu's avatar
Ethan Xu committed
1367
    app.include_router(router)
1368

Ethan Xu's avatar
Ethan Xu committed
1369
    app.root_path = args.root_path
Zhuohan Li's avatar
Zhuohan Li committed
1370

1371
1372
    mount_metrics(app)

1373
1374
1375
1376
    from vllm.entrypoints.pooling import register_pooling_api_routers

    register_pooling_api_routers(app)

Zhuohan Li's avatar
Zhuohan Li committed
1377
1378
1379
1380
1381
1382
1383
1384
    app.add_middleware(
        CORSMiddleware,
        allow_origins=args.allowed_origins,
        allow_credentials=args.allow_credentials,
        allow_methods=args.allowed_methods,
        allow_headers=args.allowed_headers,
    )

1385
1386
    @app.exception_handler(HTTPException)
    async def http_exception_handler(_: Request, exc: HTTPException):
1387
        err = ErrorResponse(
1388
1389
1390
1391
1392
1393
            error=ErrorInfo(
                message=exc.detail,
                type=HTTPStatus(exc.status_code).phrase,
                code=exc.status_code,
            )
        )
1394
1395
        return JSONResponse(err.model_dump(), status_code=exc.status_code)

Ethan Xu's avatar
Ethan Xu committed
1396
    @app.exception_handler(RequestValidationError)
1397
    async def validation_exception_handler(_: Request, exc: RequestValidationError):
1398
1399
1400
1401
1402
1403
1404
1405
        exc_str = str(exc)
        errors_str = str(exc.errors())

        if exc.errors() and errors_str and errors_str != exc_str:
            message = f"{exc_str} {errors_str}"
        else:
            message = exc_str

1406
1407
1408
1409
1410
1411
1412
1413
        err = ErrorResponse(
            error=ErrorInfo(
                message=message,
                type=HTTPStatus.BAD_REQUEST.phrase,
                code=HTTPStatus.BAD_REQUEST,
            )
        )
        return JSONResponse(err.model_dump(), status_code=HTTPStatus.BAD_REQUEST)
Ethan Xu's avatar
Ethan Xu committed
1414

1415
    # Ensure --api-key option from CLI takes precedence over VLLM_API_KEY
1416
1417
    if tokens := [key for key in (args.api_key or [envs.VLLM_API_KEY]) if key]:
        app.add_middleware(AuthenticationMiddleware, tokens=tokens)
1418

1419
    if args.enable_request_id_headers:
1420
        app.add_middleware(XRequestIdMiddleware)
1421

1422
1423
1424
    # Add scaling middleware to check for scaling state
    app.add_middleware(ScalingMiddleware)

1425
    if envs.VLLM_DEBUG_LOG_API_SERVER_RESPONSE:
1426
1427
1428
1429
1430
        logger.warning(
            "CAUTION: Enabling log response in the API Server. "
            "This can include sensitive information and should be "
            "avoided in production."
        )
1431
1432
1433
1434

        @app.middleware("http")
        async def log_response(request: Request, call_next):
            response = await call_next(request)
1435
            response_body = [section async for section in response.body_iterator]
1436
            response.body_iterator = iterate_in_threadpool(iter(response_body))
1437
1438
1439
1440
1441
1442
1443
1444
1445
1446
1447
            # Check if this is a streaming response by looking at content-type
            content_type = response.headers.get("content-type", "")
            is_streaming = content_type == "text/event-stream; charset=utf-8"

            # Log response body based on type
            if not response_body:
                logger.info("response_body={<empty>}")
            elif is_streaming:
                _log_streaming_response(response, response_body)
            else:
                _log_non_streaming_response(response_body)
1448
            return response
1449

1450
1451
1452
1453
    for middleware in args.middleware:
        module_path, object_name = middleware.rsplit(".", 1)
        imported = getattr(importlib.import_module(module_path), object_name)
        if inspect.isclass(imported):
1454
            app.add_middleware(imported)  # type: ignore[arg-type]
1455
1456
1457
        elif inspect.iscoroutinefunction(imported):
            app.middleware("http")(imported)
        else:
1458
1459
1460
            raise ValueError(
                f"Invalid middleware {middleware}. Must be a function or a class."
            )
1461

1462
    app = sagemaker_standards.bootstrap(app)
1463
1464
1465
1466
1467
1468
1469
1470
1471
1472
1473
1474
1475
1476
1477
1478
1479
1480
1481
1482
1483
1484
1485
1486
1487
    # Optional endpoints
    if args.tokens_only:

        @app.post("/abort_requests")
        async def abort_requests(raw_request: Request):
            """
            Abort one or more requests. To be used in a
            Disaggregated Everything setup.
            """
            try:
                body = await raw_request.json()
            except json.JSONDecodeError as e:
                raise HTTPException(
                    status_code=HTTPStatus.BAD_REQUEST.value,
                    detail=f"JSON decode error: {e}",
                ) from e
            request_ids = body.get("request_ids")
            if request_ids is None:
                raise HTTPException(
                    status_code=HTTPStatus.BAD_REQUEST.value,
                    detail="Missing 'request_ids' in request body",
                )
            # Abort requests in background
            asyncio.create_task(engine_client(raw_request).abort(request_ids))
            return Response(status_code=200)
1488

Ethan Xu's avatar
Ethan Xu committed
1489
1490
1491
    return app


1492
async def init_app_state(
1493
    engine_client: EngineClient,
1494
    state: State,
1495
    args: Namespace,
1496
) -> None:
1497
1498
    vllm_config = engine_client.vllm_config

1499
    if args.served_model_name is not None:
1500
        served_model_names = args.served_model_name
1501
    else:
1502
        served_model_names = [args.model]
1503

1504
    if args.enable_log_requests:
1505
        request_logger = RequestLogger(max_log_len=args.max_log_len)
1506
1507
    else:
        request_logger = None
1508

1509
    base_model_paths = [
1510
        BaseModelPath(name=name, model_path=args.model) for name in served_model_names
1511
1512
    ]

1513
    state.engine_client = engine_client
1514
    state.log_stats = not args.disable_log_stats
1515
    state.vllm_config = vllm_config
Ethan Xu's avatar
Ethan Xu committed
1516

1517
    supported_tasks = await engine_client.get_supported_tasks()
1518
    logger.info("Supported tasks: %s", supported_tasks)
1519

1520
1521
1522
    resolved_chat_template = await process_chat_template(
        args.chat_template, engine_client, vllm_config.model_config
    )
1523

1524
    if args.tool_server == "demo":
1525
        tool_server: ToolServer | None = DemoToolServer()
1526
1527
        assert isinstance(tool_server, DemoToolServer)
        await tool_server.init_and_validate()
1528
1529
1530
    elif args.tool_server:
        tool_server = MCPToolServer()
        await tool_server.add_tool_server(args.tool_server)
1531
1532
1533
    else:
        tool_server = None

1534
    # Merge default_mm_loras into the static lora_modules
1535
1536
1537
1538
1539
    default_mm_loras = (
        vllm_config.lora_config.default_mm_loras
        if vllm_config.lora_config is not None
        else {}
    )
1540

1541
1542
1543
1544
1545
1546
    default_mm_loras = (
        vllm_config.lora_config.default_mm_loras
        if vllm_config.lora_config is not None
        else {}
    )
    lora_modules = process_lora_modules(args.lora_modules, default_mm_loras)
1547

1548
    state.openai_serving_models = OpenAIServingModels(
1549
        engine_client=engine_client,
1550
        base_model_paths=base_model_paths,
1551
        lora_modules=lora_modules,
1552
    )
1553
    await state.openai_serving_models.init_static_loras()
1554
1555
1556
1557
1558
1559
1560
1561
1562
1563
1564
1565
1566
1567
1568
1569
1570
1571
1572
1573
1574
1575
1576
1577
1578
1579
1580
1581
1582
1583
1584
1585
1586
1587
1588
1589
1590
1591
1592
1593
1594
1595
1596
1597
1598
1599
1600
1601
1602
1603
1604
1605
1606
1607
1608
1609
    state.openai_serving_responses = (
        OpenAIServingResponses(
            engine_client,
            state.openai_serving_models,
            request_logger=request_logger,
            chat_template=resolved_chat_template,
            chat_template_content_format=args.chat_template_content_format,
            return_tokens_as_token_ids=args.return_tokens_as_token_ids,
            enable_auto_tools=args.enable_auto_tool_choice,
            tool_parser=args.tool_call_parser,
            tool_server=tool_server,
            reasoning_parser=args.structured_outputs_config.reasoning_parser,
            enable_prompt_tokens_details=args.enable_prompt_tokens_details,
            enable_force_include_usage=args.enable_force_include_usage,
            enable_log_outputs=args.enable_log_outputs,
            log_error_stack=args.log_error_stack,
        )
        if "generate" in supported_tasks
        else None
    )
    state.openai_serving_chat = (
        OpenAIServingChat(
            engine_client,
            state.openai_serving_models,
            args.response_role,
            request_logger=request_logger,
            chat_template=resolved_chat_template,
            chat_template_content_format=args.chat_template_content_format,
            trust_request_chat_template=args.trust_request_chat_template,
            return_tokens_as_token_ids=args.return_tokens_as_token_ids,
            enable_auto_tools=args.enable_auto_tool_choice,
            exclude_tools_when_tool_choice_none=args.exclude_tools_when_tool_choice_none,
            tool_parser=args.tool_call_parser,
            reasoning_parser=args.structured_outputs_config.reasoning_parser,
            enable_prompt_tokens_details=args.enable_prompt_tokens_details,
            enable_force_include_usage=args.enable_force_include_usage,
            enable_log_outputs=args.enable_log_outputs,
            log_error_stack=args.log_error_stack,
        )
        if "generate" in supported_tasks
        else None
    )
    state.openai_serving_completion = (
        OpenAIServingCompletion(
            engine_client,
            state.openai_serving_models,
            request_logger=request_logger,
            return_tokens_as_token_ids=args.return_tokens_as_token_ids,
            enable_prompt_tokens_details=args.enable_prompt_tokens_details,
            enable_force_include_usage=args.enable_force_include_usage,
            log_error_stack=args.log_error_stack,
        )
        if "generate" in supported_tasks
        else None
    )
    state.openai_serving_pooling = (
1610
1611
1612
1613
1614
1615
1616
1617
1618
1619
1620
        (
            OpenAIServingPooling(
                engine_client,
                state.openai_serving_models,
                supported_tasks=supported_tasks,
                request_logger=request_logger,
                chat_template=resolved_chat_template,
                chat_template_content_format=args.chat_template_content_format,
                trust_request_chat_template=args.trust_request_chat_template,
                log_error_stack=args.log_error_stack,
            )
1621
        )
1622
        if any(task in POOLING_TASKS for task in supported_tasks)
1623
1624
1625
1626
1627
1628
1629
1630
1631
1632
1633
1634
1635
1636
1637
1638
1639
1640
1641
1642
        else None
    )
    state.openai_serving_embedding = (
        OpenAIServingEmbedding(
            engine_client,
            state.openai_serving_models,
            request_logger=request_logger,
            chat_template=resolved_chat_template,
            chat_template_content_format=args.chat_template_content_format,
            trust_request_chat_template=args.trust_request_chat_template,
            log_error_stack=args.log_error_stack,
        )
        if "embed" in supported_tasks
        else None
    )
    state.openai_serving_classification = (
        ServingClassification(
            engine_client,
            state.openai_serving_models,
            request_logger=request_logger,
1643
1644
1645
            chat_template=resolved_chat_template,
            chat_template_content_format=args.chat_template_content_format,
            trust_request_chat_template=args.trust_request_chat_template,
1646
1647
1648
1649
1650
1651
1652
1653
1654
1655
1656
1657
1658
1659
1660
            log_error_stack=args.log_error_stack,
        )
        if "classify" in supported_tasks
        else None
    )
    state.openai_serving_scores = (
        ServingScores(
            engine_client,
            state.openai_serving_models,
            request_logger=request_logger,
            log_error_stack=args.log_error_stack,
        )
        if ("embed" in supported_tasks or "score" in supported_tasks)
        else None
    )
1661
    state.openai_serving_tokenization = OpenAIServingTokenization(
1662
        engine_client,
1663
        state.openai_serving_models,
1664
        request_logger=request_logger,
1665
1666
        chat_template=resolved_chat_template,
        chat_template_content_format=args.chat_template_content_format,
1667
        trust_request_chat_template=args.trust_request_chat_template,
1668
        log_error_stack=args.log_error_stack,
1669
    )
1670
1671
1672
1673
1674
1675
    state.openai_serving_transcription = (
        OpenAIServingTranscription(
            engine_client,
            state.openai_serving_models,
            request_logger=request_logger,
            log_error_stack=args.log_error_stack,
1676
            enable_force_include_usage=args.enable_force_include_usage,
1677
1678
1679
1680
1681
1682
1683
1684
1685
1686
        )
        if "transcription" in supported_tasks
        else None
    )
    state.openai_serving_translation = (
        OpenAIServingTranslation(
            engine_client,
            state.openai_serving_models,
            request_logger=request_logger,
            log_error_stack=args.log_error_stack,
1687
            enable_force_include_usage=args.enable_force_include_usage,
1688
1689
1690
1691
        )
        if "transcription" in supported_tasks
        else None
    )
1692
1693
1694
1695
1696
1697
1698
1699
1700
1701
1702
1703
1704
1705
1706
1707
1708
1709
    state.anthropic_serving_messages = (
        AnthropicServingMessages(
            engine_client,
            state.openai_serving_models,
            args.response_role,
            request_logger=request_logger,
            chat_template=resolved_chat_template,
            chat_template_content_format=args.chat_template_content_format,
            return_tokens_as_token_ids=args.return_tokens_as_token_ids,
            enable_auto_tools=args.enable_auto_tool_choice,
            tool_parser=args.tool_call_parser,
            reasoning_parser=args.structured_outputs_config.reasoning_parser,
            enable_prompt_tokens_details=args.enable_prompt_tokens_details,
            enable_force_include_usage=args.enable_force_include_usage,
        )
        if "generate" in supported_tasks
        else None
    )
1710
1711
1712
1713
1714
1715
1716
1717
1718
1719
1720
1721
1722
1723
    state.serving_tokens = (
        ServingTokens(
            engine_client,
            state.openai_serving_models,
            request_logger=request_logger,
            return_tokens_as_token_ids=args.return_tokens_as_token_ids,
            log_error_stack=args.log_error_stack,
            enable_prompt_tokens_details=args.enable_prompt_tokens_details,
            enable_log_outputs=args.enable_log_outputs,
            force_no_detokenize=args.tokens_only,
        )
        if "generate" in supported_tasks
        else None
    )
1724

1725
1726
1727
    state.enable_server_load_tracking = args.enable_server_load_tracking
    state.server_load_metrics = 0

1728

1729
def create_server_socket(addr: tuple[str, int]) -> socket.socket:
1730
1731
1732
1733
1734
1735
    family = socket.AF_INET
    if is_valid_ipv6_address(addr[0]):
        family = socket.AF_INET6

    sock = socket.socket(family=family, type=socket.SOCK_STREAM)
    sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
1736
    sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, 1)
1737
1738
1739
1740
1741
    sock.bind(addr)

    return sock


1742
1743
1744
1745
1746
1747
def create_server_unix_socket(path: str) -> socket.socket:
    sock = socket.socket(family=socket.AF_UNIX, type=socket.SOCK_STREAM)
    sock.bind(path)
    return sock


1748
def validate_api_server_args(args):
1749
    valid_tool_parses = ToolParserManager.list_registered()
1750
1751
1752
1753
1754
    if args.enable_auto_tool_choice and args.tool_call_parser not in valid_tool_parses:
        raise KeyError(
            f"invalid tool call parser: {args.tool_call_parser} "
            f"(chose from {{ {','.join(valid_tool_parses)} }})"
        )
1755

1756
    valid_reasoning_parsers = ReasoningParserManager.list_registered()
1757
1758
    if (
        reasoning_parser := args.structured_outputs_config.reasoning_parser
1759
    ) and reasoning_parser not in valid_reasoning_parsers:
1760
        raise KeyError(
1761
            f"invalid reasoning parser: {reasoning_parser} "
1762
            f"(chose from {{ {','.join(valid_reasoning_parsers)} }})"
1763
        )
1764

1765
1766
1767
1768
1769
1770
1771
1772
1773
1774
1775

def setup_server(args):
    """Validate API server args, set up signal handler, create socket
    ready to serve."""

    logger.info("vLLM API server version %s", VLLM_VERSION)
    log_non_default_args(args)

    if args.tool_parser_plugin and len(args.tool_parser_plugin) > 3:
        ToolParserManager.import_tool_parser(args.tool_parser_plugin)

1776
1777
1778
    if args.reasoning_parser_plugin and len(args.reasoning_parser_plugin) > 3:
        ReasoningParserManager.import_reasoning_parser(args.reasoning_parser_plugin)

1779
1780
    validate_api_server_args(args)

1781
1782
1783
    # workaround to make sure that we bind the port before the engine is set up.
    # This avoids race conditions with ray.
    # see https://github.com/vllm-project/vllm/issues/8204
1784
1785
1786
1787
1788
    if args.uds:
        sock = create_server_unix_socket(args.uds)
    else:
        sock_addr = (args.host or "", args.port)
        sock = create_server_socket(sock_addr)
1789

1790
1791
1792
1793
    # workaround to avoid footguns where uvicorn drops requests with too
    # many concurrent requests active
    set_ulimit()

1794
1795
1796
1797
1798
1799
    def signal_handler(*_) -> None:
        # Interrupt server on sigterm while initializing
        raise KeyboardInterrupt("terminated")

    signal.signal(signal.SIGTERM, signal_handler)

1800
1801
1802
1803
1804
    if args.uds:
        listen_address = f"unix:{args.uds}"
    else:
        addr, port = sock_addr
        is_ssl = args.ssl_keyfile and args.ssl_certfile
1805
        host_part = f"[{addr}]" if is_valid_ipv6_address(addr) else addr or "0.0.0.0"
1806
        listen_address = f"http{'s' if is_ssl else ''}://{host_part}:{port}"
1807
1808
1809
1810
1811
    return listen_address, sock


async def run_server(args, **uvicorn_kwargs) -> None:
    """Run a single-worker API server."""
1812
1813

    # Add process-specific prefix to stdout and stderr.
1814
    decorate_logs("APIServer")
1815

1816
1817
1818
1819
    listen_address, sock = setup_server(args)
    await run_server_worker(listen_address, sock, args, **uvicorn_kwargs)


1820
1821
1822
async def run_server_worker(
    listen_address, sock, args, client_config=None, **uvicorn_kwargs
) -> None:
1823
1824
1825
1826
1827
    """Run a single API server worker."""

    if args.tool_parser_plugin and len(args.tool_parser_plugin) > 3:
        ToolParserManager.import_tool_parser(args.tool_parser_plugin)

1828
1829
1830
    if args.reasoning_parser_plugin and len(args.reasoning_parser_plugin) > 3:
        ReasoningParserManager.import_reasoning_parser(args.reasoning_parser_plugin)

1831
1832
1833
    # Load logging config for uvicorn if specified
    log_config = load_log_config(args.log_config_file)
    if log_config is not None:
1834
        uvicorn_kwargs["log_config"] = log_config
1835

1836
    async with build_async_engine_client(
1837
1838
        args,
        client_config=client_config,
1839
    ) as engine_client:
1840
        maybe_register_tokenizer_info_endpoint(args)
1841
1842
        app = build_app(args)

1843
        await init_app_state(engine_client, app.state, args)
1844

1845
1846
        logger.info(
            "Starting vLLM API server %d on %s",
1847
            engine_client.vllm_config.parallel_config._api_process_rank,
1848
1849
            listen_address,
        )
1850
1851
        shutdown_task = await serve_http(
            app,
1852
            sock=sock,
1853
            enable_ssl_refresh=args.enable_ssl_refresh,
1854
1855
1856
            host=args.host,
            port=args.port,
            log_level=args.uvicorn_log_level,
1857
1858
1859
            # NOTE: When the 'disable_uvicorn_access_log' value is True,
            # no access log will be output.
            access_log=not args.disable_uvicorn_access_log,
1860
            timeout_keep_alive=envs.VLLM_HTTP_TIMEOUT_KEEP_ALIVE,
1861
1862
1863
1864
            ssl_keyfile=args.ssl_keyfile,
            ssl_certfile=args.ssl_certfile,
            ssl_ca_certs=args.ssl_ca_certs,
            ssl_cert_reqs=args.ssl_cert_reqs,
1865
1866
            h11_max_incomplete_event_size=args.h11_max_incomplete_event_size,
            h11_max_header_count=args.h11_max_header_count,
1867
1868
1869
            **uvicorn_kwargs,
        )

1870
    # NB: Await server shutdown only after the backend context is exited
1871
1872
1873
1874
    try:
        await shutdown_task
    finally:
        sock.close()
1875

Ethan Xu's avatar
Ethan Xu committed
1876
1877
1878

if __name__ == "__main__":
    # NOTE(simon):
1879
1880
    # This section should be in sync with vllm/entrypoints/cli/main.py for CLI
    # entrypoints.
1881
    cli_env_setup()
Ethan Xu's avatar
Ethan Xu committed
1882
    parser = FlexibleArgumentParser(
1883
1884
        description="vLLM OpenAI-Compatible RESTful API server."
    )
Ethan Xu's avatar
Ethan Xu committed
1885
1886
    parser = make_arg_parser(parser)
    args = parser.parse_args()
1887
    validate_parsed_serve_args(args)
1888

1889
    uvloop.run(run_server(args))