api_server.py 72.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, Callable
18
from contextlib import asynccontextmanager
19
from http import HTTPStatus
20
from typing import Annotated, Any, Literal
21

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

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

126
prometheus_multiproc_dir: tempfile.TemporaryDirectory
127

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

131
132
ENDPOINT_LOAD_METRICS_FORMAT_HEADER_LABEL = "endpoint-load-metrics-format"

133
_running_tasks: set[asyncio.Task] = set()
134

135

136
@asynccontextmanager
137
async def lifespan(app: FastAPI):
138
139
    try:
        if app.state.log_stats:
140
            engine_client: EngineClient = app.state.engine_client
141
142
143

            async def _force_log():
                while True:
144
                    await asyncio.sleep(envs.VLLM_LOG_STATS_INTERVAL)
145
                    await engine_client.do_log_stats()
146
147
148
149
150
151

            task = asyncio.create_task(_force_log())
            _running_tasks.add(task)
            task.add_done_callback(_running_tasks.remove)
        else:
            task = None
152
153
154

        # Mark the startup heap as static so that it's ignored by GC.
        # Reduces pause times of oldest generation collections.
155
        freeze_gc_heap()
156
157
158
159
160
161
162
163
        try:
            yield
        finally:
            if task is not None:
                task.cancel()
    finally:
        # Ensure app state including engine ref is gc'd
        del app.state
164
165


166
@asynccontextmanager
167
async def build_async_engine_client(
168
    args: Namespace,
169
170
    *,
    usage_context: UsageContext = UsageContext.OPENAI_API_SERVER,
171
172
    disable_frontend_multiprocessing: bool | None = None,
    client_config: dict[str, Any] | None = None,
173
) -> AsyncIterator[EngineClient]:
174
175
176
177
    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")
178
        multiprocessing.set_start_method("forkserver")
179
180
181
182
        multiprocessing.set_forkserver_preload(["vllm.v1.engine.async_llm"])
        forkserver.ensure_running()
        logger.debug("Forkserver setup complete!")

183
    # Context manager to handle engine_client lifecycle
184
185
    # Ensures everything is shutdown and cleaned up on error/exit
    engine_args = AsyncEngineArgs.from_cli_args(args)
186
187
188
    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)
189

190
    if disable_frontend_multiprocessing is None:
191
        disable_frontend_multiprocessing = bool(args.disable_frontend_multiprocessing)
192

193
    async with build_async_engine_client_from_engine_args(
194
195
196
197
        engine_args,
        usage_context=usage_context,
        disable_frontend_multiprocessing=disable_frontend_multiprocessing,
        client_config=client_config,
198
    ) as engine:
199
200
201
202
203
204
        yield engine


@asynccontextmanager
async def build_async_engine_client_from_engine_args(
    engine_args: AsyncEngineArgs,
205
206
    *,
    usage_context: UsageContext = UsageContext.OPENAI_API_SERVER,
207
    disable_frontend_multiprocessing: bool = False,
208
    client_config: dict[str, Any] | None = None,
209
) -> AsyncIterator[EngineClient]:
210
    """
211
    Create EngineClient, either:
212
213
214
215
216
217
        - in-process using the AsyncLLMEngine Directly
        - multiprocess using AsyncLLMEngine RPC

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

218
219
220
    # Create the EngineConfig (determines if we can use V1).
    vllm_config = engine_args.create_engine_config(usage_context=usage_context)

221
    if disable_frontend_multiprocessing:
222
        logger.warning("V1 is enabled, but got --disable-frontend-multiprocessing.")
223

224
    from vllm.v1.engine.async_llm import AsyncLLM
225

226
    async_llm: AsyncLLM | None = None
227
228
229
230
231
232

    # 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)

233
234
235
236
237
    try:
        async_llm = AsyncLLM.from_vllm_config(
            vllm_config=vllm_config,
            usage_context=usage_context,
            enable_log_requests=engine_args.enable_log_requests,
238
            aggregate_engine_logging=engine_args.aggregate_engine_logging,
239
240
241
            disable_log_stats=engine_args.disable_log_stats,
            client_addresses=client_config,
            client_count=client_count,
242
243
            client_index=client_index,
        )
244
245

        # Don't keep the dummy data in memory
246
        assert async_llm is not None
247
248
249
250
251
252
        await async_llm.reset_mm_cache()

        yield async_llm
    finally:
        if async_llm:
            async_llm.shutdown()
253
254


255
256
async def validate_json_request(raw_request: Request):
    content_type = raw_request.headers.get("content-type", "").lower()
257
258
    media_type = content_type.split(";", maxsplit=1)[0]
    if media_type != "application/json":
259
260
261
        raise RequestValidationError(
            errors=["Unsupported Media Type: Only 'application/json' is allowed"]
        )
262
263


Ethan Xu's avatar
Ethan Xu committed
264
router = APIRouter()
Zhuohan Li's avatar
Zhuohan Li committed
265

266

267
268
269
270
class PrometheusResponse(Response):
    media_type = prometheus_client.CONTENT_TYPE_LATEST


271
def mount_metrics(app: FastAPI):
272
273
274
    """Mount prometheus metrics to a FastAPI app."""

    registry = get_prometheus_registry()
275

276
277
278
279
    # `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
280
281
282
283
284
285
286
287
288
289
    Instrumentator(
        excluded_handlers=[
            "/metrics",
            "/health",
            "/load",
            "/ping",
            "/version",
            "/server_info",
        ],
        registry=registry,
290
    ).add().instrument(app).expose(app, response_class=PrometheusResponse)
291
292
293

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

295
    # Workaround for 307 Redirect for /metrics
296
    metrics_route.path_regex = re.compile("^/metrics(?P<path>.*)$")
297
    app.routes.append(metrics_route)
298
299


300
301
302
303
304
def base(request: Request) -> OpenAIServing:
    # Reuse the existing instance
    return tokenization(request)


305
306
307
308
def models(request: Request) -> OpenAIServingModels:
    return request.app.state.openai_serving_models


309
def responses(request: Request) -> OpenAIServingResponses | None:
310
311
312
    return request.app.state.openai_serving_responses


313
314
315
316
def messages(request: Request) -> AnthropicServingMessages:
    return request.app.state.anthropic_serving_messages


317
def chat(request: Request) -> OpenAIServingChat | None:
318
319
320
    return request.app.state.openai_serving_chat


321
def completion(request: Request) -> OpenAIServingCompletion | None:
322
323
324
    return request.app.state.openai_serving_completion


325
def pooling(request: Request) -> OpenAIServingPooling | None:
326
327
328
    return request.app.state.openai_serving_pooling


329
def embedding(request: Request) -> OpenAIServingEmbedding | None:
330
    return request.app.state.openai_serving_embedding
331
332


333
def score(request: Request) -> ServingScores | None:
334
335
336
    return request.app.state.openai_serving_scores


337
def classify(request: Request) -> ServingClassification | None:
338
339
340
    return request.app.state.openai_serving_classification


341
def rerank(request: Request) -> ServingScores | None:
342
    return request.app.state.openai_serving_scores
343
344


345
346
def tokenization(request: Request) -> OpenAIServingTokenization:
    return request.app.state.openai_serving_tokenization
347
348


349
350
351
352
def transcription(request: Request) -> OpenAIServingTranscription:
    return request.app.state.openai_serving_transcription


353
354
355
356
def translation(request: Request) -> OpenAIServingTranslation:
    return request.app.state.openai_serving_translation


357
def engine_client(request: Request) -> EngineClient:
358
359
360
    return request.app.state.engine_client


361
362
@router.get("/health", response_class=Response)
async def health(raw_request: Request) -> Response:
363
    """Health check."""
364
365
366
367
368
    try:
        await engine_client(raw_request).check_health()
        return Response(status_code=200)
    except EngineDeadError:
        return Response(status_code=503)
369
370


371
372
373
374
375
376
377
@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
378
    # - /v1/audio/translations
379
380
    # - /v1/embeddings
    # - /pooling
381
    # - /classify
382
383
384
385
386
    # - /score
    # - /v1/score
    # - /rerank
    # - /v1/rerank
    # - /v2/rerank
387
    return JSONResponse(content={"server_load": request.app.state.server_load_metrics})
388
389


390
391
392
@router.get("/ping", response_class=Response)
@router.post("/ping", response_class=Response)
async def ping(raw_request: Request) -> Response:
393
394
395
396
    """Ping check. Endpoint required for SageMaker"""
    return await health(raw_request)


397
398
399
400
401
402
403
404
405
406
@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},
    },
)
407
@with_cancellation
408
async def tokenize(request: TokenizeRequest, raw_request: Request):
409
410
    handler = tokenization(raw_request)

411
412
413
    try:
        generator = await handler.create_tokenize(request, raw_request)
    except NotImplementedError as e:
414
415
416
        raise HTTPException(
            status_code=HTTPStatus.NOT_IMPLEMENTED.value, detail=str(e)
        ) from e
417
    except Exception as e:
418
419
420
        raise HTTPException(
            status_code=HTTPStatus.INTERNAL_SERVER_ERROR.value, detail=str(e)
        ) from e
421

422
    if isinstance(generator, ErrorResponse):
423
424
425
        return JSONResponse(
            content=generator.model_dump(), status_code=generator.error.code
        )
426
    elif isinstance(generator, TokenizeResponse):
427
428
        return JSONResponse(content=generator.model_dump())

429
430
    assert_never(generator)

431

432
433
434
435
436
437
438
439
440
@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},
    },
)
441
@with_cancellation
442
async def detokenize(request: DetokenizeRequest, raw_request: Request):
443
444
    handler = tokenization(raw_request)

445
446
447
448
449
    try:
        generator = await handler.create_detokenize(request, raw_request)
    except OverflowError as e:
        raise RequestValidationError(errors=[str(e)]) from e
    except Exception as e:
450
451
452
        raise HTTPException(
            status_code=HTTPStatus.INTERNAL_SERVER_ERROR.value, detail=str(e)
        ) from e
453

454
    if isinstance(generator, ErrorResponse):
455
456
457
        return JSONResponse(
            content=generator.model_dump(), status_code=generator.error.code
        )
458
    elif isinstance(generator, DetokenizeResponse):
459
460
        return JSONResponse(content=generator.model_dump())

461
462
    assert_never(generator)

463

464
465
def maybe_register_tokenizer_info_endpoint(args):
    """Conditionally register the tokenizer info endpoint if enabled."""
466
    if getattr(args, "enable_tokenizer_info_endpoint", False):
467
468
469
470
471

        @router.get("/tokenizer_info")
        async def get_tokenizer_info(raw_request: Request):
            """Get comprehensive tokenizer information."""
            result = await tokenization(raw_request).get_tokenizer_info()
472
473
474
475
476
477
            return JSONResponse(
                content=result.model_dump(),
                status_code=result.error.code
                if isinstance(result, ErrorResponse)
                else 200,
            )
478
479


Ethan Xu's avatar
Ethan Xu committed
480
@router.get("/v1/models")
481
async def show_available_models(raw_request: Request):
482
    handler = models(raw_request)
483

484
485
    models_ = await handler.show_available_models()
    return JSONResponse(content=models_.model_dump())
Zhuohan Li's avatar
Zhuohan Li committed
486
487


Ethan Xu's avatar
Ethan Xu committed
488
@router.get("/version")
489
async def show_version():
490
    ver = {"version": VLLM_VERSION}
491
492
493
    return JSONResponse(content=ver)


494
async def _convert_stream_to_sse_events(
495
    generator: AsyncGenerator[StreamingResponsesResponse, None],
496
) -> AsyncGenerator[str, None]:
497
498
    """Convert the generator to a stream of events in SSE format"""
    async for event in generator:
499
        event_type = getattr(event, "type", "unknown")
500
        # https://developer.mozilla.org/en-US/docs/Web/API/Server-sent_events/Using_server-sent_events#event_stream_format
501
502
503
        event_data = (
            f"event: {event_type}\ndata: {event.model_dump_json(indent=None)}\n\n"
        )
504
505
506
        yield event_data


507
508
509
510
511
512
513
514
515
516
@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},
    },
)
517
518
519
520
521
@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(
522
523
            message="The model does not support Responses API"
        )
524
525
526
    try:
        generator = await handler.create_responses(request, raw_request)
    except Exception as e:
527
528
529
        raise HTTPException(
            status_code=HTTPStatus.INTERNAL_SERVER_ERROR.value, detail=str(e)
        ) from e
530
531

    if isinstance(generator, ErrorResponse):
532
533
534
        return JSONResponse(
            content=generator.model_dump(), status_code=generator.error.code
        )
535
536
    elif isinstance(generator, ResponsesResponse):
        return JSONResponse(content=generator.model_dump())
537

538
539
540
    return StreamingResponse(
        content=_convert_stream_to_sse_events(generator), media_type="text/event-stream"
    )
541
542
543


@router.get("/v1/responses/{response_id}")
544
545
546
async def retrieve_responses(
    response_id: str,
    raw_request: Request,
547
548
    starting_after: int | None = None,
    stream: bool | None = False,
549
):
550
551
552
    handler = responses(raw_request)
    if handler is None:
        return base(raw_request).create_error_response(
553
554
            message="The model does not support Responses API"
        )
555

556
    try:
557
558
559
560
561
        response = await handler.retrieve_responses(
            response_id,
            starting_after=starting_after,
            stream=stream,
        )
562
    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(response, ErrorResponse):
568
569
570
        return JSONResponse(
            content=response.model_dump(), status_code=response.error.code
        )
571
572
    elif isinstance(response, ResponsesResponse):
        return JSONResponse(content=response.model_dump())
573
574
575
    return StreamingResponse(
        content=_convert_stream_to_sse_events(response), media_type="text/event-stream"
    )
576
577
578
579
580
581
582


@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(
583
584
            message="The model does not support Responses API"
        )
585

586
587
588
    try:
        response = await handler.cancel_responses(response_id)
    except Exception as e:
589
590
591
        raise HTTPException(
            status_code=HTTPStatus.INTERNAL_SERVER_ERROR.value, detail=str(e)
        ) from e
592
593

    if isinstance(response, ErrorResponse):
594
595
596
        return JSONResponse(
            content=response.model_dump(), status_code=response.error.code
        )
597
598
599
    return JSONResponse(content=response.model_dump())


600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
@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):
649
650
651
        resp = generator.model_dump(exclude_none=True)
        logger.debug("Anthropic Messages Response: %s", resp)
        return JSONResponse(content=resp)
652
653
654
655

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


656
657
658
659
660
661
662
663
664
665
@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},
    },
)
666
@with_cancellation
667
@load_aware_call
668
async def create_chat_completion(request: ChatCompletionRequest, raw_request: Request):
669
670
671
    metrics_header_format = raw_request.headers.get(
        ENDPOINT_LOAD_METRICS_FORMAT_HEADER_LABEL, ""
    )
672
673
674
    handler = chat(raw_request)
    if handler is None:
        return base(raw_request).create_error_response(
675
676
            message="The model does not support Chat Completions API"
        )
677
678
679
    try:
        generator = await handler.create_chat_completion(request, raw_request)
    except Exception as e:
680
681
682
        raise HTTPException(
            status_code=HTTPStatus.INTERNAL_SERVER_ERROR.value, detail=str(e)
        ) from e
683
    if isinstance(generator, ErrorResponse):
684
685
686
        return JSONResponse(
            content=generator.model_dump(), status_code=generator.error.code
        )
687

688
    elif isinstance(generator, ChatCompletionResponse):
689
690
691
692
        return JSONResponse(
            content=generator.model_dump(),
            headers=metrics_header(metrics_header_format),
        )
693

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

696

697
698
699
700
701
702
703
704
705
706
@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},
    },
)
707
@with_cancellation
708
@load_aware_call
709
async def create_completion(request: CompletionRequest, raw_request: Request):
710
711
712
    metrics_header_format = raw_request.headers.get(
        ENDPOINT_LOAD_METRICS_FORMAT_HEADER_LABEL, ""
    )
713
714
715
    handler = completion(raw_request)
    if handler is None:
        return base(raw_request).create_error_response(
716
717
            message="The model does not support Completions API"
        )
718

719
720
721
    try:
        generator = await handler.create_completion(request, raw_request)
    except OverflowError as e:
722
723
724
        raise HTTPException(
            status_code=HTTPStatus.BAD_REQUEST.value, detail=str(e)
        ) from e
725
    except Exception as e:
726
727
728
        raise HTTPException(
            status_code=HTTPStatus.INTERNAL_SERVER_ERROR.value, detail=str(e)
        ) from e
729

730
    if isinstance(generator, ErrorResponse):
731
732
733
        return JSONResponse(
            content=generator.model_dump(), status_code=generator.error.code
        )
734
    elif isinstance(generator, CompletionResponse):
735
736
737
738
        return JSONResponse(
            content=generator.model_dump(),
            headers=metrics_header(metrics_header_format),
        )
Zhuohan Li's avatar
Zhuohan Li committed
739

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

Zhuohan Li's avatar
Zhuohan Li committed
742

743
744
745
746
747
748
749
750
@router.post(
    "/v1/embeddings",
    dependencies=[Depends(validate_json_request)],
    responses={
        HTTPStatus.BAD_REQUEST.value: {"model": ErrorResponse},
        HTTPStatus.INTERNAL_SERVER_ERROR.value: {"model": ErrorResponse},
    },
)
751
@with_cancellation
752
@load_aware_call
753
754
755
756
async def create_embedding(
    request: EmbeddingRequest,
    raw_request: Request,
):
757
758
    handler = embedding(raw_request)
    if handler is None:
759
        return base(raw_request).create_error_response(
760
761
            message="The model does not support Embeddings API"
        )
762

763
764
765
    try:
        generator = await handler.create_embedding(request, raw_request)
    except Exception as e:
766
767
768
        raise HTTPException(
            status_code=HTTPStatus.INTERNAL_SERVER_ERROR.value, detail=str(e)
        ) from e
769

770
    if isinstance(generator, ErrorResponse):
771
772
773
        return JSONResponse(
            content=generator.model_dump(), status_code=generator.error.code
        )
774
    elif isinstance(generator, EmbeddingResponse):
775
        return JSONResponse(content=generator.model_dump())
776
777
778
779
780
781
    elif isinstance(generator, EmbeddingBytesResponse):
        return StreamingResponse(
            content=generator.body,
            headers={"metadata": generator.metadata},
            media_type=generator.media_type,
        )
782

783
784
    assert_never(generator)

785

786
787
788
789
790
791
792
793
@router.post(
    "/pooling",
    dependencies=[Depends(validate_json_request)],
    responses={
        HTTPStatus.BAD_REQUEST.value: {"model": ErrorResponse},
        HTTPStatus.INTERNAL_SERVER_ERROR.value: {"model": ErrorResponse},
    },
)
794
@with_cancellation
795
@load_aware_call
796
797
798
799
async def create_pooling(request: PoolingRequest, raw_request: Request):
    handler = pooling(raw_request)
    if handler is None:
        return base(raw_request).create_error_response(
800
801
            message="The model does not support Pooling API"
        )
802
803
804
    try:
        generator = await handler.create_pooling(request, raw_request)
    except Exception as e:
805
806
807
        raise HTTPException(
            status_code=HTTPStatus.INTERNAL_SERVER_ERROR.value, detail=str(e)
        ) from e
808
    if isinstance(generator, ErrorResponse):
809
810
811
        return JSONResponse(
            content=generator.model_dump(), status_code=generator.error.code
        )
812
    elif isinstance(generator, (PoolingResponse, IOProcessorResponse)):
813
        return JSONResponse(content=generator.model_dump())
814
815
816
817
818
819
    elif isinstance(generator, PoolingBytesResponse):
        return StreamingResponse(
            content=generator.body,
            headers={"metadata": generator.metadata},
            media_type=generator.media_type,
        )
820
821
822
823

    assert_never(generator)


824
825
826
@router.post("/classify", dependencies=[Depends(validate_json_request)])
@with_cancellation
@load_aware_call
827
async def create_classify(request: ClassificationRequest, raw_request: Request):
828
829
830
    handler = classify(raw_request)
    if handler is None:
        return base(raw_request).create_error_response(
831
832
            message="The model does not support Classification API"
        )
833

834
835
836
    try:
        generator = await handler.create_classify(request, raw_request)
    except Exception as e:
837
838
839
        raise HTTPException(
            status_code=HTTPStatus.INTERNAL_SERVER_ERROR.value, detail=str(e)
        ) from e
840
    if isinstance(generator, ErrorResponse):
841
842
843
        return JSONResponse(
            content=generator.model_dump(), status_code=generator.error.code
        )
844
845
846
847
848
849
850

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

    assert_never(generator)


851
852
853
854
855
856
857
858
@router.post(
    "/score",
    dependencies=[Depends(validate_json_request)],
    responses={
        HTTPStatus.BAD_REQUEST.value: {"model": ErrorResponse},
        HTTPStatus.INTERNAL_SERVER_ERROR.value: {"model": ErrorResponse},
    },
)
859
@with_cancellation
860
@load_aware_call
861
862
863
864
async def create_score(request: ScoreRequest, raw_request: Request):
    handler = score(raw_request)
    if handler is None:
        return base(raw_request).create_error_response(
865
866
            message="The model does not support Score API"
        )
867

868
869
870
    try:
        generator = await handler.create_score(request, raw_request)
    except Exception as e:
871
872
873
        raise HTTPException(
            status_code=HTTPStatus.INTERNAL_SERVER_ERROR.value, detail=str(e)
        ) from e
874
    if isinstance(generator, ErrorResponse):
875
876
877
        return JSONResponse(
            content=generator.model_dump(), status_code=generator.error.code
        )
878
879
880
881
882
883
    elif isinstance(generator, ScoreResponse):
        return JSONResponse(content=generator.model_dump())

    assert_never(generator)


884
885
886
887
888
889
890
891
@router.post(
    "/v1/score",
    dependencies=[Depends(validate_json_request)],
    responses={
        HTTPStatus.BAD_REQUEST.value: {"model": ErrorResponse},
        HTTPStatus.INTERNAL_SERVER_ERROR.value: {"model": ErrorResponse},
    },
)
892
@with_cancellation
893
@load_aware_call
894
895
896
async def create_score_v1(request: ScoreRequest, raw_request: Request):
    logger.warning(
        "To indicate that Score API is not part of standard OpenAI API, we "
897
898
        "have moved it to `/score`. Please update your client accordingly."
    )
899
900
901
902

    return await create_score(request, raw_request)


903
904
905
906
907
908
909
910
911
@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},
    },
)
912
@with_cancellation
913
@load_aware_call
914
915
916
async def create_transcriptions(
    raw_request: Request, request: Annotated[TranscriptionRequest, Form()]
):
917
918
919
    handler = transcription(raw_request)
    if handler is None:
        return base(raw_request).create_error_response(
920
921
            message="The model does not support Transcriptions API"
        )
922
923

    audio_data = await request.file.read()
924
    try:
925
        generator = await handler.create_transcription(audio_data, request, raw_request)
926
    except Exception as e:
927
928
929
        raise HTTPException(
            status_code=HTTPStatus.INTERNAL_SERVER_ERROR.value, detail=str(e)
        ) from e
930
931

    if isinstance(generator, ErrorResponse):
932
933
934
        return JSONResponse(
            content=generator.model_dump(), status_code=generator.error.code
        )
935
936
937
938
939
940
941

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

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


942
943
944
945
946
947
948
949
950
@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},
    },
)
951
952
@with_cancellation
@load_aware_call
953
954
955
async def create_translations(
    request: Annotated[TranslationRequest, Form()], raw_request: Request
):
956
957
958
    handler = translation(raw_request)
    if handler is None:
        return base(raw_request).create_error_response(
959
960
            message="The model does not support Translations API"
        )
961
962

    audio_data = await request.file.read()
963
    try:
964
        generator = await handler.create_translation(audio_data, request, raw_request)
965
    except Exception as e:
966
967
968
        raise HTTPException(
            status_code=HTTPStatus.INTERNAL_SERVER_ERROR.value, detail=str(e)
        ) from e
969
970

    if isinstance(generator, ErrorResponse):
971
972
973
        return JSONResponse(
            content=generator.model_dump(), status_code=generator.error.code
        )
974
975
976
977
978
979
980

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

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


981
982
983
984
985
986
987
988
@router.post(
    "/rerank",
    dependencies=[Depends(validate_json_request)],
    responses={
        HTTPStatus.BAD_REQUEST.value: {"model": ErrorResponse},
        HTTPStatus.INTERNAL_SERVER_ERROR.value: {"model": ErrorResponse},
    },
)
989
@with_cancellation
990
@load_aware_call
991
992
993
994
async def do_rerank(request: RerankRequest, raw_request: Request):
    handler = rerank(raw_request)
    if handler is None:
        return base(raw_request).create_error_response(
995
996
            message="The model does not support Rerank (Score) API"
        )
997
998
999
    try:
        generator = await handler.do_rerank(request, raw_request)
    except Exception as e:
1000
1001
1002
        raise HTTPException(
            status_code=HTTPStatus.INTERNAL_SERVER_ERROR.value, detail=str(e)
        ) from e
1003
    if isinstance(generator, ErrorResponse):
1004
1005
1006
        return JSONResponse(
            content=generator.model_dump(), status_code=generator.error.code
        )
1007
1008
1009
1010
1011
1012
    elif isinstance(generator, RerankResponse):
        return JSONResponse(content=generator.model_dump())

    assert_never(generator)


1013
1014
1015
1016
1017
1018
1019
1020
@router.post(
    "/v1/rerank",
    dependencies=[Depends(validate_json_request)],
    responses={
        HTTPStatus.BAD_REQUEST.value: {"model": ErrorResponse},
        HTTPStatus.INTERNAL_SERVER_ERROR.value: {"model": ErrorResponse},
    },
)
1021
1022
@with_cancellation
async def do_rerank_v1(request: RerankRequest, raw_request: Request):
1023
    logger.warning_once(
1024
        "To indicate that the rerank API is not part of the standard OpenAI"
1025
        " API, we have located it at `/rerank`. Please update your client "
1026
1027
        "accordingly. (Note: Conforms to JinaAI rerank API)"
    )
1028
1029
1030
1031

    return await do_rerank(request, raw_request)


1032
1033
1034
1035
1036
1037
1038
1039
@router.post(
    "/v2/rerank",
    dependencies=[Depends(validate_json_request)],
    responses={
        HTTPStatus.BAD_REQUEST.value: {"model": ErrorResponse},
        HTTPStatus.INTERNAL_SERVER_ERROR.value: {"model": ErrorResponse},
    },
)
1040
1041
1042
1043
1044
@with_cancellation
async def do_rerank_v2(request: RerankRequest, raw_request: Request):
    return await do_rerank(request, raw_request)


1045
if envs.VLLM_SERVER_DEV_MODE:
1046
1047
1048
1049
    logger.warning(
        "SECURITY WARNING: Development endpoints are enabled! "
        "This should NOT be used in production!"
    )
1050

1051
1052
    PydanticVllmConfig = pydantic.TypeAdapter(VllmConfig)

1053
    @router.get("/server_info")
1054
1055
    async def show_server_info(
        raw_request: Request,
1056
        config_format: Annotated[Literal["text", "json"], Query()] = "text",
1057
1058
1059
    ):
        vllm_config: VllmConfig = raw_request.app.state.vllm_config
        server_info = {
1060
1061
1062
            "vllm_config": str(vllm_config)
            if config_format == "text"
            else PydanticVllmConfig.dump_python(vllm_config, mode="json", fallback=str)
1063
1064
            # fallback=str is needed to handle e.g. torch.dtype
        }
1065
1066
        return JSONResponse(content=server_info)

1067
1068
1069
1070
1071
1072
    @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.
        """
1073
1074
1075
1076
1077
1078
        device = None
        device_str = raw_request.query_params.get("device")
        if device_str is not None:
            device = Device[device_str.upper()]
        logger.info("Resetting prefix cache with specific %s...", str(device))
        await engine_client(raw_request).reset_prefix_cache(device)
1079
1080
        return Response(status_code=200)

1081
1082
1083
1084
1085
1086
1087
1088
1089
1090
    @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)

1091
1092
1093
1094
1095
1096
1097
1098
1099
1100
1101
    @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):
1102
1103
1104
1105
1106
1107
        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)
1108
1109
1110
1111
        # 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)

1112
1113
1114
1115
1116
1117
    @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})

1118
1119
1120
1121
1122
    @router.post("/collective_rpc")
    async def collective_rpc(raw_request: Request):
        try:
            body = await raw_request.json()
        except json.JSONDecodeError as e:
1123
1124
1125
1126
            raise HTTPException(
                status_code=HTTPStatus.BAD_REQUEST.value,
                detail=f"JSON decode error: {e}",
            ) from e
1127
1128
        method = body.get("method")
        if method is None:
1129
1130
1131
1132
            raise HTTPException(
                status_code=HTTPStatus.BAD_REQUEST.value,
                detail="Missing 'method' in request body",
            )
1133
        # For security reason, only serialized string args/kwargs are passed.
1134
        # User-defined `method` is responsible for deserialization if needed.
1135
1136
        args: list[str] = body.get("args", [])
        kwargs: dict[str, str] = body.get("kwargs", {})
1137
        timeout: float | None = body.get("timeout")
1138
        results = await engine_client(raw_request).collective_rpc(
1139
1140
            method=method, timeout=timeout, args=tuple(args), kwargs=kwargs
        )
1141
1142
1143
1144
1145
1146
1147
1148
1149
1150
        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})

1151

1152
1153
1154
1155
1156
1157
1158
1159
1160
1161
@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},
    },
)
1162
1163
1164
1165
async def scale_elastic_ep(raw_request: Request):
    try:
        body = await raw_request.json()
    except json.JSONDecodeError as e:
1166
        raise HTTPException(status_code=400, detail="Invalid JSON format") from e  # noqa: B904
1167
1168
1169
1170
1171

    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:
1172
1173
1174
        raise HTTPException(
            status_code=400, detail="new_data_parallel_size is required"
        )
1175

1176
    if not isinstance(new_data_parallel_size, int) or new_data_parallel_size <= 0:
1177
        raise HTTPException(
1178
1179
            status_code=400, detail="new_data_parallel_size must be a positive integer"
        )
1180
1181

    if not isinstance(drain_timeout, int) or drain_timeout <= 0:
1182
1183
1184
        raise HTTPException(
            status_code=400, detail="drain_timeout must be a positive integer"
        )
1185
1186
1187
1188
1189
1190
1191

    # 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)
1192
1193
1194
1195
1196
        return JSONResponse(
            {
                "message": f"Scaled to {new_data_parallel_size} data parallel engines",
            }
        )
1197
    except TimeoutError as e:
1198
1199
1200
1201
1202
        raise HTTPException(
            status_code=408,
            detail="Scale failed due to request drain timeout "
            f"after {drain_timeout} seconds",
        ) from e
1203
1204
1205
1206
1207
1208
1209
1210
1211
1212
1213
1214
    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})


1215
1216
1217
# TODO: RequestType = TypeForm[BaseModel] when recognized by type checkers
# (requires typing_extensions >= 4.13)
RequestType = Any
1218
GetHandlerFn = Callable[[Request], OpenAIServing | None]
1219
1220
1221
1222
1223
1224
1225
1226
1227
1228
1229
1230
1231
1232
1233
1234
1235
1236
1237
1238
EndpointFn = Callable[[RequestType, Request], Awaitable[Any]]

# NOTE: Items defined earlier take higher priority
INVOCATION_TYPES: list[tuple[RequestType, tuple[GetHandlerFn, EndpointFn]]] = [
    (ChatCompletionRequest, (chat, create_chat_completion)),
    (CompletionRequest, (completion, create_completion)),
    (EmbeddingRequest, (embedding, create_embedding)),
    (ClassificationRequest, (classify, create_classify)),
    (ScoreRequest, (score, create_score)),
    (RerankRequest, (rerank, do_rerank)),
    (PoolingRequest, (pooling, create_pooling)),
]

# NOTE: Construct the TypeAdapters only once
INVOCATION_VALIDATORS = [
    (pydantic.TypeAdapter(request_type), (get_handler, endpoint))
    for request_type, (get_handler, endpoint) in INVOCATION_TYPES
]


1239
1240
1241
1242
1243
1244
1245
1246
1247
@router.post(
    "/invocations",
    dependencies=[Depends(validate_json_request)],
    responses={
        HTTPStatus.BAD_REQUEST.value: {"model": ErrorResponse},
        HTTPStatus.UNSUPPORTED_MEDIA_TYPE.value: {"model": ErrorResponse},
        HTTPStatus.INTERNAL_SERVER_ERROR.value: {"model": ErrorResponse},
    },
)
1248
async def invocations(raw_request: Request):
1249
    """For SageMaker, routes requests based on the request type."""
1250
1251
    try:
        body = await raw_request.json()
1252
    except json.JSONDecodeError as e:
1253
1254
1255
        raise HTTPException(
            status_code=HTTPStatus.BAD_REQUEST.value, detail=f"JSON decode error: {e}"
        ) from e
1256

1257
1258
1259
1260
1261
    valid_endpoints = [
        (validator, endpoint)
        for validator, (get_handler, endpoint) in INVOCATION_VALIDATORS
        if get_handler(raw_request) is not None
    ]
1262

1263
1264
1265
1266
1267
1268
1269
1270
1271
1272
1273
1274
    for request_validator, endpoint in valid_endpoints:
        try:
            request = request_validator.validate_python(body)
        except pydantic.ValidationError:
            continue

        return await endpoint(request, raw_request)

    type_names = [
        t.__name__ if isinstance(t := validator._type, type) else str(t)
        for validator, _ in valid_endpoints
    ]
1275
    msg = f"Cannot find suitable handler for request. Expected one of: {type_names}"
1276
    res = base(raw_request).create_error_response(message=msg)
1277
    return JSONResponse(content=res.model_dump(), status_code=res.error.code)
1278
1279


1280
if envs.VLLM_TORCH_PROFILER_DIR:
1281
    logger.warning_once(
1282
        "Torch Profiler is enabled in the API server. This should ONLY be "
1283
1284
        "used for local development!"
    )
1285
1286
1287
1288
1289
1290
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:
1291
1292

    @router.post("/start_profile")
1293
    async def start_profile(raw_request: Request):
1294
        logger.info("Starting profiler...")
1295
        await engine_client(raw_request).start_profile()
1296
1297
1298
1299
        logger.info("Profiler started.")
        return Response(status_code=200)

    @router.post("/stop_profile")
1300
    async def stop_profile(raw_request: Request):
1301
        logger.info("Stopping profiler...")
1302
        await engine_client(raw_request).stop_profile()
1303
1304
1305
1306
        logger.info("Profiler stopped.")
        return Response(status_code=200)


1307
1308
if envs.VLLM_ALLOW_RUNTIME_LORA_UPDATING:
    logger.warning(
1309
        "LoRA dynamic loading & unloading is enabled in the API server. "
1310
1311
        "This should ONLY be used for local development!"
    )
1312

1313
1314
    @router.post("/v1/load_lora_adapter", dependencies=[Depends(validate_json_request)])
    async def load_lora_adapter(request: LoadLoRAAdapterRequest, raw_request: Request):
1315
1316
1317
        handler = models(raw_request)
        response = await handler.load_lora_adapter(request)
        if isinstance(response, ErrorResponse):
1318
1319
1320
            return JSONResponse(
                content=response.model_dump(), status_code=response.error.code
            )
1321
1322
1323

        return Response(status_code=200, content=response)

1324
1325
1326
1327
1328
1329
    @router.post(
        "/v1/unload_lora_adapter", dependencies=[Depends(validate_json_request)]
    )
    async def unload_lora_adapter(
        request: UnloadLoRAAdapterRequest, raw_request: Request
    ):
1330
1331
1332
        handler = models(raw_request)
        response = await handler.unload_lora_adapter(request)
        if isinstance(response, ErrorResponse):
1333
1334
1335
            return JSONResponse(
                content=response.model_dump(), status_code=response.error.code
            )
1336
1337
1338
1339

        return Response(status_code=200, content=response)


1340
def load_log_config(log_config_file: str | None) -> dict | None:
1341
1342
1343
1344
1345
1346
    if not log_config_file:
        return None
    try:
        with open(log_config_file) as f:
            return json.load(f)
    except Exception as e:
1347
1348
1349
        logger.warning(
            "Failed to load log config from file %s: error %s", log_config_file, e
        )
1350
1351
1352
        return None


1353
1354
1355
class AuthenticationMiddleware:
    """
    Pure ASGI middleware that authenticates each request by checking
1356
    if the Authorization Bearer token exists and equals anyof "{api_key}".
1357
1358
1359
1360
1361
1362
1363
1364

    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).
    """

1365
    def __init__(self, app: ASGIApp, tokens: list[str]) -> None:
1366
        self.app = app
1367
        self.api_tokens = [hashlib.sha256(t.encode("utf-8")).digest() for t in tokens]
1368
1369
1370
1371
1372
1373
1374
1375
1376
1377
1378
1379
1380
1381
1382
1383
1384

    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
1385

1386
1387
    def __call__(self, scope: Scope, receive: Receive, send: Send) -> Awaitable[None]:
        if scope["type"] not in ("http", "websocket") or scope["method"] == "OPTIONS":
1388
1389
1390
1391
1392
1393
1394
            # 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.
1395
        if url_path.startswith("/v1") and not self.verify_token(headers):
1396
            response = JSONResponse(content={"error": "Unauthorized"}, status_code=401)
1397
1398
1399
1400
1401
1402
1403
1404
1405
1406
1407
1408
1409
1410
            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

1411
    def __call__(self, scope: Scope, receive: Receive, send: Send) -> Awaitable[None]:
1412
1413
1414
1415
1416
1417
1418
1419
1420
1421
1422
1423
1424
        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"])
1425
                request_id = request_headers.get("X-Request-Id", uuid.uuid4().hex)
1426
1427
1428
1429
1430
1431
                response_headers.append("X-Request-Id", request_id)
            await send(message)

        return self.app(scope, receive, send_with_request_id)


1432
1433
1434
1435
1436
1437
1438
1439
# 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.
1440

1441
1442
1443
1444
1445
1446
1447
    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

1448
    def __call__(self, scope: Scope, receive: Receive, send: Send) -> Awaitable[None]:
1449
1450
1451
1452
1453
1454
1455
        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
1456
1457
1458
1459
1460
1461
            response = JSONResponse(
                content={
                    "error": "The model is currently scaling. Please try again later."
                },
                status_code=503,
            )
1462
1463
1464
1465
1466
            return response(scope, receive, send)

        return self.app(scope, receive, send)


1467
1468
1469
1470
def _extract_content_from_chunk(chunk_data: dict) -> str:
    """Extract content from a streaming response chunk."""
    try:
        from vllm.entrypoints.openai.protocol import (
1471
1472
1473
            ChatCompletionStreamResponse,
            CompletionStreamResponse,
        )
1474
1475

        # Try using Completion types for type-safe parsing
1476
1477
        if chunk_data.get("object") == "chat.completion.chunk":
            chat_response = ChatCompletionStreamResponse.model_validate(chunk_data)
1478
1479
            if chat_response.choices and chat_response.choices[0].delta.content:
                return chat_response.choices[0].delta.content
1480
1481
1482
        elif chunk_data.get("object") == "text_completion":
            completion_response = CompletionStreamResponse.model_validate(chunk_data)
            if completion_response.choices and completion_response.choices[0].text:
1483
1484
1485
                return completion_response.choices[0].text
    except pydantic.ValidationError:
        # Fallback to manual parsing
1486
1487
1488
1489
1490
1491
        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"]
1492
1493
1494
1495
1496
1497
1498
1499
1500
1501
1502
1503
1504
1505
1506
    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:
1507
            chunk_str = chunk.decode("utf-8")
1508
1509
1510
1511
1512
1513
1514
1515
        except UnicodeDecodeError:
            # Skip malformed chunks
            return []

        self.buffer += chunk_str
        events = []

        # Process complete lines
1516
1517
1518
        while "\n" in self.buffer:
            line, self.buffer = self.buffer.split("\n", 1)
            line = line.rstrip("\r")  # Handle CRLF
1519

1520
            if line.startswith("data: "):
1521
                data_str = line[6:].strip()
1522
1523
                if data_str == "[DONE]":
                    events.append({"type": "done"})
1524
1525
1526
                elif data_str:
                    try:
                        event_data = json.loads(data_str)
1527
                        events.append({"type": "data", "data": event_data})
1528
1529
1530
1531
1532
1533
1534
1535
1536
1537
1538
1539
1540
1541
1542
1543
1544
                    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."""
1545
        return "".join(self.content_buffer)
1546
1547
1548
1549
1550
1551
1552
1553
1554
1555
1556
1557
1558
1559
1560
1561
1562
1563
1564
1565


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:
1566
1567
                if event["type"] == "data":
                    content = sse_decoder.extract_content(event["data"])
1568
                    sse_decoder.add_content(content)
1569
                elif event["type"] == "done":
1570
1571
1572
1573
1574
1575
1576
1577
                    # 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(
1578
                            "response_body={streaming_complete: content=%r, chunks=%d}",
1579
1580
1581
                            full_content,
                            chunk_count,
                        )
1582
1583
                    else:
                        logger.info(
1584
1585
1586
                            "response_body={streaming_complete: no_content, chunks=%d}",
                            chunk_count,
                        )
1587
1588
1589
                    return

    response.body_iterator = iterate_in_threadpool(buffered_iterator())
1590
    logger.info("response_body={streaming_started: chunks=%d}", len(response_body))
1591
1592
1593
1594
1595
1596
1597
1598
1599
1600
1601


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>}")


1602
def build_app(args: Namespace) -> FastAPI:
1603
    if args.disable_fastapi_docs:
1604
1605
1606
        app = FastAPI(
            openapi_url=None, docs_url=None, redoc_url=None, lifespan=lifespan
        )
1607
1608
    else:
        app = FastAPI(lifespan=lifespan)
Ethan Xu's avatar
Ethan Xu committed
1609
1610
    app.include_router(router)
    app.root_path = args.root_path
Zhuohan Li's avatar
Zhuohan Li committed
1611

1612
1613
    mount_metrics(app)

Zhuohan Li's avatar
Zhuohan Li committed
1614
1615
1616
1617
1618
1619
1620
1621
    app.add_middleware(
        CORSMiddleware,
        allow_origins=args.allowed_origins,
        allow_credentials=args.allow_credentials,
        allow_methods=args.allowed_methods,
        allow_headers=args.allowed_headers,
    )

1622
1623
    @app.exception_handler(HTTPException)
    async def http_exception_handler(_: Request, exc: HTTPException):
1624
        err = ErrorResponse(
1625
1626
1627
1628
1629
1630
            error=ErrorInfo(
                message=exc.detail,
                type=HTTPStatus(exc.status_code).phrase,
                code=exc.status_code,
            )
        )
1631
1632
        return JSONResponse(err.model_dump(), status_code=exc.status_code)

Ethan Xu's avatar
Ethan Xu committed
1633
    @app.exception_handler(RequestValidationError)
1634
    async def validation_exception_handler(_: Request, exc: RequestValidationError):
1635
1636
1637
1638
1639
1640
1641
1642
        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

1643
1644
1645
1646
1647
1648
1649
1650
        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
1651

1652
    # Ensure --api-key option from CLI takes precedence over VLLM_API_KEY
1653
1654
    if tokens := [key for key in (args.api_key or [envs.VLLM_API_KEY]) if key]:
        app.add_middleware(AuthenticationMiddleware, tokens=tokens)
1655

1656
    if args.enable_request_id_headers:
1657
        app.add_middleware(XRequestIdMiddleware)
1658

1659
1660
1661
    # Add scaling middleware to check for scaling state
    app.add_middleware(ScalingMiddleware)

1662
    if envs.VLLM_DEBUG_LOG_API_SERVER_RESPONSE:
1663
1664
1665
1666
1667
        logger.warning(
            "CAUTION: Enabling log response in the API Server. "
            "This can include sensitive information and should be "
            "avoided in production."
        )
1668
1669
1670
1671

        @app.middleware("http")
        async def log_response(request: Request, call_next):
            response = await call_next(request)
1672
            response_body = [section async for section in response.body_iterator]
1673
            response.body_iterator = iterate_in_threadpool(iter(response_body))
1674
1675
1676
1677
1678
1679
1680
1681
1682
1683
1684
            # 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)
1685
            return response
1686

1687
1688
1689
1690
    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):
1691
            app.add_middleware(imported)  # type: ignore[arg-type]
1692
1693
1694
        elif inspect.iscoroutinefunction(imported):
            app.middleware("http")(imported)
        else:
1695
1696
1697
            raise ValueError(
                f"Invalid middleware {middleware}. Must be a function or a class."
            )
1698

Ethan Xu's avatar
Ethan Xu committed
1699
1700
1701
    return app


1702
async def init_app_state(
1703
    engine_client: EngineClient,
1704
    state: State,
1705
    args: Namespace,
1706
) -> None:
1707
1708
    vllm_config = engine_client.vllm_config

1709
    if args.served_model_name is not None:
1710
        served_model_names = args.served_model_name
1711
    else:
1712
        served_model_names = [args.model]
1713

1714
    if args.enable_log_requests:
1715
        request_logger = RequestLogger(max_log_len=args.max_log_len)
1716
1717
    else:
        request_logger = None
1718

1719
    base_model_paths = [
1720
        BaseModelPath(name=name, model_path=args.model) for name in served_model_names
1721
1722
    ]

1723
    state.engine_client = engine_client
1724
    state.log_stats = not args.disable_log_stats
1725
    state.vllm_config = vllm_config
Ethan Xu's avatar
Ethan Xu committed
1726

1727
    supported_tasks = await engine_client.get_supported_tasks()
1728
    logger.info("Supported tasks: %s", supported_tasks)
1729

1730
1731
1732
    resolved_chat_template = await process_chat_template(
        args.chat_template, engine_client, vllm_config.model_config
    )
1733

1734
    if args.tool_server == "demo":
1735
        tool_server: ToolServer | None = DemoToolServer()
1736
1737
        assert isinstance(tool_server, DemoToolServer)
        await tool_server.init_and_validate()
1738
1739
1740
    elif args.tool_server:
        tool_server = MCPToolServer()
        await tool_server.add_tool_server(args.tool_server)
1741
1742
1743
    else:
        tool_server = None

1744
    # Merge default_mm_loras into the static lora_modules
1745
1746
1747
1748
1749
    default_mm_loras = (
        vllm_config.lora_config.default_mm_loras
        if vllm_config.lora_config is not None
        else {}
    )
1750

1751
1752
1753
1754
1755
1756
    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)
1757

1758
    state.openai_serving_models = OpenAIServingModels(
1759
        engine_client=engine_client,
1760
        base_model_paths=base_model_paths,
1761
        lora_modules=lora_modules,
1762
    )
1763
    await state.openai_serving_models.init_static_loras()
1764
1765
1766
1767
1768
1769
1770
1771
1772
1773
1774
1775
1776
1777
1778
1779
1780
1781
1782
1783
1784
1785
1786
1787
1788
1789
1790
1791
1792
1793
1794
1795
1796
1797
1798
1799
1800
1801
1802
1803
1804
1805
1806
1807
1808
1809
1810
1811
1812
1813
1814
1815
1816
1817
1818
1819
    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 = (
1820
1821
1822
1823
1824
1825
1826
1827
1828
1829
1830
        (
            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,
            )
1831
        )
1832
        if any(task in POOLING_TASKS for task in supported_tasks)
1833
1834
1835
1836
1837
1838
1839
1840
1841
1842
1843
1844
1845
1846
1847
1848
1849
1850
1851
1852
1853
1854
1855
1856
1857
1858
1859
1860
1861
1862
1863
1864
1865
1866
1867
        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,
            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
    )
1868
    state.openai_serving_tokenization = OpenAIServingTokenization(
1869
        engine_client,
1870
        state.openai_serving_models,
1871
        request_logger=request_logger,
1872
1873
        chat_template=resolved_chat_template,
        chat_template_content_format=args.chat_template_content_format,
1874
        trust_request_chat_template=args.trust_request_chat_template,
1875
        log_error_stack=args.log_error_stack,
1876
    )
1877
1878
1879
1880
1881
1882
    state.openai_serving_transcription = (
        OpenAIServingTranscription(
            engine_client,
            state.openai_serving_models,
            request_logger=request_logger,
            log_error_stack=args.log_error_stack,
1883
            enable_force_include_usage=args.enable_force_include_usage,
1884
1885
1886
1887
1888
1889
1890
1891
1892
1893
        )
        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,
1894
            enable_force_include_usage=args.enable_force_include_usage,
1895
1896
1897
1898
        )
        if "transcription" in supported_tasks
        else None
    )
1899
1900
1901
1902
1903
1904
1905
1906
1907
1908
1909
1910
1911
1912
1913
1914
1915
1916
    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
    )
1917

1918
1919
1920
    state.enable_server_load_tracking = args.enable_server_load_tracking
    state.server_load_metrics = 0

1921

1922
def create_server_socket(addr: tuple[str, int]) -> socket.socket:
1923
1924
1925
1926
1927
1928
    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)
1929
    sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, 1)
1930
1931
1932
1933
1934
    sock.bind(addr)

    return sock


1935
1936
1937
1938
1939
1940
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


1941
def validate_api_server_args(args):
1942
    valid_tool_parses = ToolParserManager.list_registered()
1943
1944
1945
1946
1947
    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)} }})"
        )
1948

1949
    valid_reasoning_parsers = ReasoningParserManager.list_registered()
1950
1951
    if (
        reasoning_parser := args.structured_outputs_config.reasoning_parser
1952
    ) and reasoning_parser not in valid_reasoning_parsers:
1953
        raise KeyError(
1954
            f"invalid reasoning parser: {reasoning_parser} "
1955
            f"(chose from {{ {','.join(valid_reasoning_parsers)} }})"
1956
        )
1957

1958
1959
1960
1961
1962
1963
1964
1965
1966
1967
1968

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)

1969
1970
1971
    if args.reasoning_parser_plugin and len(args.reasoning_parser_plugin) > 3:
        ReasoningParserManager.import_reasoning_parser(args.reasoning_parser_plugin)

1972
1973
    validate_api_server_args(args)

1974
1975
1976
    # 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
1977
1978
1979
1980
1981
    if args.uds:
        sock = create_server_unix_socket(args.uds)
    else:
        sock_addr = (args.host or "", args.port)
        sock = create_server_socket(sock_addr)
1982

1983
1984
1985
1986
    # workaround to avoid footguns where uvicorn drops requests with too
    # many concurrent requests active
    set_ulimit()

1987
1988
1989
1990
1991
1992
    def signal_handler(*_) -> None:
        # Interrupt server on sigterm while initializing
        raise KeyboardInterrupt("terminated")

    signal.signal(signal.SIGTERM, signal_handler)

1993
1994
1995
1996
1997
    if args.uds:
        listen_address = f"unix:{args.uds}"
    else:
        addr, port = sock_addr
        is_ssl = args.ssl_keyfile and args.ssl_certfile
1998
        host_part = f"[{addr}]" if is_valid_ipv6_address(addr) else addr or "0.0.0.0"
1999
        listen_address = f"http{'s' if is_ssl else ''}://{host_part}:{port}"
2000
2001
2002
2003
2004
    return listen_address, sock


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

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

2009
2010
2011
2012
    listen_address, sock = setup_server(args)
    await run_server_worker(listen_address, sock, args, **uvicorn_kwargs)


2013
2014
2015
async def run_server_worker(
    listen_address, sock, args, client_config=None, **uvicorn_kwargs
) -> None:
2016
2017
2018
2019
2020
    """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)

2021
2022
2023
    if args.reasoning_parser_plugin and len(args.reasoning_parser_plugin) > 3:
        ReasoningParserManager.import_reasoning_parser(args.reasoning_parser_plugin)

2024
2025
2026
    # Load logging config for uvicorn if specified
    log_config = load_log_config(args.log_config_file)
    if log_config is not None:
2027
        uvicorn_kwargs["log_config"] = log_config
2028

2029
    async with build_async_engine_client(
2030
2031
        args,
        client_config=client_config,
2032
    ) as engine_client:
2033
        maybe_register_tokenizer_info_endpoint(args)
2034
2035
        app = build_app(args)

2036
        await init_app_state(engine_client, app.state, args)
2037

2038
2039
        logger.info(
            "Starting vLLM API server %d on %s",
2040
            engine_client.vllm_config.parallel_config._api_process_rank,
2041
2042
            listen_address,
        )
2043
2044
        shutdown_task = await serve_http(
            app,
2045
            sock=sock,
2046
            enable_ssl_refresh=args.enable_ssl_refresh,
2047
2048
2049
            host=args.host,
            port=args.port,
            log_level=args.uvicorn_log_level,
2050
2051
2052
            # NOTE: When the 'disable_uvicorn_access_log' value is True,
            # no access log will be output.
            access_log=not args.disable_uvicorn_access_log,
2053
            timeout_keep_alive=envs.VLLM_HTTP_TIMEOUT_KEEP_ALIVE,
2054
2055
2056
2057
            ssl_keyfile=args.ssl_keyfile,
            ssl_certfile=args.ssl_certfile,
            ssl_ca_certs=args.ssl_ca_certs,
            ssl_cert_reqs=args.ssl_cert_reqs,
2058
2059
            h11_max_incomplete_event_size=args.h11_max_incomplete_event_size,
            h11_max_header_count=args.h11_max_header_count,
2060
2061
2062
            **uvicorn_kwargs,
        )

2063
    # NB: Await server shutdown only after the backend context is exited
2064
2065
2066
2067
    try:
        await shutdown_task
    finally:
        sock.close()
2068

Ethan Xu's avatar
Ethan Xu committed
2069
2070
2071

if __name__ == "__main__":
    # NOTE(simon):
2072
2073
    # This section should be in sync with vllm/entrypoints/cli/main.py for CLI
    # entrypoints.
2074
    cli_env_setup()
Ethan Xu's avatar
Ethan Xu committed
2075
    parser = FlexibleArgumentParser(
2076
2077
        description="vLLM OpenAI-Compatible RESTful API server."
    )
Ethan Xu's avatar
Ethan Xu committed
2078
2079
    parser = make_arg_parser(parser)
    args = parser.parse_args()
2080
    validate_parsed_serve_args(args)
2081

2082
    uvloop.run(run_server(args))