api_server.py 73 KB
Newer Older
1
# SPDX-License-Identifier: Apache-2.0
2
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
3

4
import asyncio
5
import gc
6
import hashlib
7
8
import importlib
import inspect
9
import json
10
import multiprocessing
11
import multiprocessing.forkserver as forkserver
12
import os
13
import secrets
14
import signal
15
import socket
16
import tempfile
17
import uuid
18
from argparse import Namespace
19
from collections.abc import AsyncGenerator, AsyncIterator, Awaitable
20
from contextlib import asynccontextmanager
21
from http import HTTPStatus
22
from typing import Annotated, Any, Callable, Literal, Optional
23

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

41
import vllm.envs as envs
42
from vllm.config import VllmConfig
Woosuk Kwon's avatar
Woosuk Kwon committed
43
from vllm.engine.arg_utils import AsyncEngineArgs
44
from vllm.engine.protocol import EngineClient
45
46
47
from vllm.entrypoints.chat_utils import (load_chat_template,
                                         resolve_hf_chat_template,
                                         resolve_mistral_chat_template)
48
from vllm.entrypoints.launcher import serve_http
49
from vllm.entrypoints.logger import RequestLogger
50
from vllm.entrypoints.openai.cli_args import (make_arg_parser,
51
                                              validate_parsed_serve_args)
52
53
# yapf conflicts with isort for this block
# yapf: disable
54
from vllm.entrypoints.openai.protocol import (ChatCompletionRequest,
55
                                              ChatCompletionResponse,
56
57
                                              ClassificationRequest,
                                              ClassificationResponse,
58
                                              CompletionRequest,
59
                                              CompletionResponse,
60
61
                                              DetokenizeRequest,
                                              DetokenizeResponse,
62
                                              EmbeddingRequest,
63
64
                                              EmbeddingResponse, ErrorInfo,
                                              ErrorResponse,
65
                                              IOProcessorResponse,
66
                                              LoadLoRAAdapterRequest,
67
                                              PoolingRequest, PoolingResponse,
68
                                              RerankRequest, RerankResponse,
69
70
                                              ResponsesRequest,
                                              ResponsesResponse, ScoreRequest,
71
72
73
                                              ScoreResponse,
                                              StreamingResponsesResponse,
                                              TokenizeRequest,
74
                                              TokenizeResponse,
75
76
                                              TranscriptionRequest,
                                              TranscriptionResponse,
77
78
                                              TranslationRequest,
                                              TranslationResponse,
79
                                              UnloadLoRAAdapterRequest)
80
# yapf: enable
81
from vllm.entrypoints.openai.serving_chat import OpenAIServingChat
82
83
from vllm.entrypoints.openai.serving_classification import (
    ServingClassification)
84
from vllm.entrypoints.openai.serving_completion import OpenAIServingCompletion
85
from vllm.entrypoints.openai.serving_embedding import OpenAIServingEmbedding
86
87
from vllm.entrypoints.openai.serving_engine import OpenAIServing
from vllm.entrypoints.openai.serving_models import (BaseModelPath,
88
                                                    LoRAModulePath,
89
                                                    OpenAIServingModels)
90
from vllm.entrypoints.openai.serving_pooling import OpenAIServingPooling
91
from vllm.entrypoints.openai.serving_responses import OpenAIServingResponses
92
from vllm.entrypoints.openai.serving_score import ServingScores
93
94
from vllm.entrypoints.openai.serving_tokenization import (
    OpenAIServingTokenization)
95
from vllm.entrypoints.openai.serving_transcription import (
96
    OpenAIServingTranscription, OpenAIServingTranslation)
97
from vllm.entrypoints.openai.tool_parsers import ToolParserManager
98
99
from vllm.entrypoints.tool_server import (DemoToolServer, MCPToolServer,
                                          ToolServer)
100
from vllm.entrypoints.utils import (cli_env_setup, load_aware_call,
101
                                    log_non_default_args, with_cancellation)
102
from vllm.logger import init_logger
103
from vllm.reasoning import ReasoningParserManager
104
from vllm.transformers_utils.tokenizer import MistralTokenizer
yhu422's avatar
yhu422 committed
105
from vllm.usage.usage_lib import UsageContext
106
from vllm.utils import (Device, FlexibleArgumentParser, decorate_logs,
107
                        is_valid_ipv6_address, set_ulimit)
108
from vllm.v1.engine.exceptions import EngineDeadError
109
from vllm.v1.metrics.prometheus import get_prometheus_registry
110
from vllm.version import __version__ as VLLM_VERSION
Zhuohan Li's avatar
Zhuohan Li committed
111

112
prometheus_multiproc_dir: tempfile.TemporaryDirectory
113

114
# Cannot use __name__ (https://github.com/vllm-project/vllm/pull/4765)
115
logger = init_logger('vllm.entrypoints.openai.api_server')
116

117
_running_tasks: set[asyncio.Task] = set()
118

119

120
@asynccontextmanager
121
async def lifespan(app: FastAPI):
122
123
    try:
        if app.state.log_stats:
124
            engine_client: EngineClient = app.state.engine_client
125
126
127

            async def _force_log():
                while True:
128
                    await asyncio.sleep(envs.VLLM_LOG_STATS_INTERVAL)
129
                    await engine_client.do_log_stats()
130
131
132
133
134
135

            task = asyncio.create_task(_force_log())
            _running_tasks.add(task)
            task.add_done_callback(_running_tasks.remove)
        else:
            task = None
136
137
138
139
140

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


151
@asynccontextmanager
152
async def build_async_engine_client(
153
    args: Namespace,
154
155
156
    *,
    usage_context: UsageContext = UsageContext.OPENAI_API_SERVER,
    disable_frontend_multiprocessing: Optional[bool] = None,
157
158
    client_config: Optional[dict[str, Any]] = None,
) -> AsyncIterator[EngineClient]:
159

160
161
162
163
164
165
166
167
168
    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")
        multiprocessing.set_start_method('forkserver')
        multiprocessing.set_forkserver_preload(["vllm.v1.engine.async_llm"])
        forkserver.ensure_running()
        logger.debug("Forkserver setup complete!")

169
    # Context manager to handle engine_client lifecycle
170
171
    # Ensures everything is shutdown and cleaned up on error/exit
    engine_args = AsyncEngineArgs.from_cli_args(args)
172
173
174
    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)
175

176
177
178
179
    if disable_frontend_multiprocessing is None:
        disable_frontend_multiprocessing = bool(
            args.disable_frontend_multiprocessing)

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


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

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

205
206
207
208
    # Create the EngineConfig (determines if we can use V1).
    vllm_config = engine_args.create_engine_config(usage_context=usage_context)

    # V1 AsyncLLM.
209
    assert envs.VLLM_USE_V1
210

211
212
213
214
    if disable_frontend_multiprocessing:
        logger.warning(
            "V1 is enabled, but got --disable-frontend-multiprocessing. "
            "To disable frontend multiprocessing, set VLLM_USE_V1=0.")
215

216
217
    from vllm.v1.engine.async_llm import AsyncLLM
    async_llm: Optional[AsyncLLM] = None
218
219
220
221
222
223

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

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

        # Don't keep the dummy data in memory
        await async_llm.reset_mm_cache()

        yield async_llm
    finally:
        if async_llm:
            async_llm.shutdown()
241
242


243
244
async def validate_json_request(raw_request: Request):
    content_type = raw_request.headers.get("content-type", "").lower()
245
246
    media_type = content_type.split(";", maxsplit=1)[0]
    if media_type != "application/json":
247
248
249
        raise RequestValidationError(errors=[
            "Unsupported Media Type: Only 'application/json' is allowed"
        ])
250
251


Ethan Xu's avatar
Ethan Xu committed
252
router = APIRouter()
Zhuohan Li's avatar
Zhuohan Li committed
253

254

255
256
257
258
class PrometheusResponse(Response):
    media_type = prometheus_client.CONTENT_TYPE_LATEST


259
def mount_metrics(app: FastAPI):
260
261
262
    """Mount prometheus metrics to a FastAPI app."""

    registry = get_prometheus_registry()
263

264
265
266
267
    # `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
268
269
270
271
272
273
274
275
276
277
    Instrumentator(
        excluded_handlers=[
            "/metrics",
            "/health",
            "/load",
            "/ping",
            "/version",
            "/server_info",
        ],
        registry=registry,
278
    ).add().instrument(app).expose(app, response_class=PrometheusResponse)
279
280
281

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

283
    # Workaround for 307 Redirect for /metrics
284
    metrics_route.path_regex = re.compile("^/metrics(?P<path>.*)$")
285
    app.routes.append(metrics_route)
286
287


288
289
290
291
292
def base(request: Request) -> OpenAIServing:
    # Reuse the existing instance
    return tokenization(request)


293
294
295
296
def models(request: Request) -> OpenAIServingModels:
    return request.app.state.openai_serving_models


297
298
299
300
def responses(request: Request) -> Optional[OpenAIServingResponses]:
    return request.app.state.openai_serving_responses


301
def chat(request: Request) -> Optional[OpenAIServingChat]:
302
303
304
    return request.app.state.openai_serving_chat


305
def completion(request: Request) -> Optional[OpenAIServingCompletion]:
306
307
308
    return request.app.state.openai_serving_completion


309
310
311
312
def pooling(request: Request) -> Optional[OpenAIServingPooling]:
    return request.app.state.openai_serving_pooling


313
314
def embedding(request: Request) -> Optional[OpenAIServingEmbedding]:
    return request.app.state.openai_serving_embedding
315
316


317
def score(request: Request) -> Optional[ServingScores]:
318
319
320
    return request.app.state.openai_serving_scores


321
322
323
324
def classify(request: Request) -> Optional[ServingClassification]:
    return request.app.state.openai_serving_classification


325
326
def rerank(request: Request) -> Optional[ServingScores]:
    return request.app.state.openai_serving_scores
327
328


329
330
def tokenization(request: Request) -> OpenAIServingTokenization:
    return request.app.state.openai_serving_tokenization
331
332


333
334
335
336
def transcription(request: Request) -> OpenAIServingTranscription:
    return request.app.state.openai_serving_transcription


337
338
339
340
def translation(request: Request) -> OpenAIServingTranslation:
    return request.app.state.openai_serving_translation


341
def engine_client(request: Request) -> EngineClient:
342
343
344
    return request.app.state.engine_client


345
346
@router.get("/health", response_class=Response)
async def health(raw_request: Request) -> Response:
347
    """Health check."""
348
349
350
351
352
    try:
        await engine_client(raw_request).check_health()
        return Response(status_code=200)
    except EngineDeadError:
        return Response(status_code=503)
353
354


355
356
357
358
359
360
361
@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
362
    # - /v1/audio/translations
363
364
    # - /v1/embeddings
    # - /pooling
365
    # - /classify
366
367
368
369
370
371
372
373
374
    # - /score
    # - /v1/score
    # - /rerank
    # - /v1/rerank
    # - /v2/rerank
    return JSONResponse(
        content={'server_load': request.app.state.server_load_metrics})


375
376
377
@router.get("/ping", response_class=Response)
@router.post("/ping", response_class=Response)
async def ping(raw_request: Request) -> Response:
378
379
380
381
    """Ping check. Endpoint required for SageMaker"""
    return await health(raw_request)


382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
@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
                 },
             })
398
@with_cancellation
399
async def tokenize(request: TokenizeRequest, raw_request: Request):
400
401
    handler = tokenization(raw_request)

402
403
404
405
406
407
408
409
410
    try:
        generator = await handler.create_tokenize(request, raw_request)
    except NotImplementedError as e:
        raise HTTPException(status_code=HTTPStatus.NOT_IMPLEMENTED.value,
                            detail=str(e)) from e
    except Exception as e:
        raise HTTPException(status_code=HTTPStatus.INTERNAL_SERVER_ERROR.value,
                            detail=str(e)) from e

411
412
    if isinstance(generator, ErrorResponse):
        return JSONResponse(content=generator.model_dump(),
413
                            status_code=generator.error.code)
414
    elif isinstance(generator, TokenizeResponse):
415
416
        return JSONResponse(content=generator.model_dump())

417
418
    assert_never(generator)

419

420
421
422
423
424
425
426
427
428
429
430
431
432
@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
                 },
             })
433
@with_cancellation
434
async def detokenize(request: DetokenizeRequest, raw_request: Request):
435
436
    handler = tokenization(raw_request)

437
438
439
440
441
442
443
444
    try:
        generator = await handler.create_detokenize(request, raw_request)
    except OverflowError as e:
        raise RequestValidationError(errors=[str(e)]) from e
    except Exception as e:
        raise HTTPException(status_code=HTTPStatus.INTERNAL_SERVER_ERROR.value,
                            detail=str(e)) from e

445
446
    if isinstance(generator, ErrorResponse):
        return JSONResponse(content=generator.model_dump(),
447
                            status_code=generator.error.code)
448
    elif isinstance(generator, DetokenizeResponse):
449
450
        return JSONResponse(content=generator.model_dump())

451
452
    assert_never(generator)

453

454
455
456
457
458
459
460
461
462
def maybe_register_tokenizer_info_endpoint(args):
    """Conditionally register the tokenizer info endpoint if enabled."""
    if getattr(args, 'enable_tokenizer_info_endpoint', False):

        @router.get("/tokenizer_info")
        async def get_tokenizer_info(raw_request: Request):
            """Get comprehensive tokenizer information."""
            result = await tokenization(raw_request).get_tokenizer_info()
            return JSONResponse(content=result.model_dump(),
463
                                status_code=result.error.code if isinstance(
464
465
466
                                    result, ErrorResponse) else 200)


Ethan Xu's avatar
Ethan Xu committed
467
@router.get("/v1/models")
468
async def show_available_models(raw_request: Request):
469
    handler = models(raw_request)
470

471
472
    models_ = await handler.show_available_models()
    return JSONResponse(content=models_.model_dump())
Zhuohan Li's avatar
Zhuohan Li committed
473
474


Ethan Xu's avatar
Ethan Xu committed
475
@router.get("/version")
476
async def show_version():
477
    ver = {"version": VLLM_VERSION}
478
479
480
    return JSONResponse(content=ver)


481
async def _convert_stream_to_sse_events(
482
483
    generator: AsyncGenerator[StreamingResponsesResponse, None]
) -> AsyncGenerator[str, None]:
484
485
486
487
488
489
490
491
492
    """Convert the generator to a stream of events in SSE format"""
    async for event in generator:
        event_type = getattr(event, 'type', 'unknown')
        # https://developer.mozilla.org/en-US/docs/Web/API/Server-sent_events/Using_server-sent_events#event_stream_format
        event_data = (f"event: {event_type}\n"
                      f"data: {event.model_dump_json(indent=None)}\n\n")
        yield event_data


493
494
495
496
497
498
499
500
501
502
503
504
505
506
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
                 },
             })
@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(
            message="The model does not support Responses API")
517
518
519
520
521
    try:
        generator = await handler.create_responses(request, raw_request)
    except Exception as e:
        raise HTTPException(status_code=HTTPStatus.INTERNAL_SERVER_ERROR.value,
                            detail=str(e)) from e
522
523
524

    if isinstance(generator, ErrorResponse):
        return JSONResponse(content=generator.model_dump(),
525
                            status_code=generator.error.code)
526
527
    elif isinstance(generator, ResponsesResponse):
        return JSONResponse(content=generator.model_dump())
528
529
530

    return StreamingResponse(content=_convert_stream_to_sse_events(generator),
                             media_type="text/event-stream")
531
532
533


@router.get("/v1/responses/{response_id}")
534
535
536
537
538
539
async def retrieve_responses(
    response_id: str,
    raw_request: Request,
    starting_after: Optional[int] = None,
    stream: Optional[bool] = False,
):
540
541
542
543
544
    handler = responses(raw_request)
    if handler is None:
        return base(raw_request).create_error_response(
            message="The model does not support Responses API")

545
    try:
546
547
548
549
550
        response = await handler.retrieve_responses(
            response_id,
            starting_after=starting_after,
            stream=stream,
        )
551
552
553
    except Exception as e:
        raise HTTPException(status_code=HTTPStatus.INTERNAL_SERVER_ERROR.value,
                            detail=str(e)) from e
554
555
556

    if isinstance(response, ErrorResponse):
        return JSONResponse(content=response.model_dump(),
557
                            status_code=response.error.code)
558
559
560
561
    elif isinstance(response, ResponsesResponse):
        return JSONResponse(content=response.model_dump())
    return StreamingResponse(content=_convert_stream_to_sse_events(response),
                             media_type="text/event-stream")
562
563
564
565
566
567
568
569
570


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

571
572
573
574
575
    try:
        response = await handler.cancel_responses(response_id)
    except Exception as e:
        raise HTTPException(status_code=HTTPStatus.INTERNAL_SERVER_ERROR.value,
                            detail=str(e)) from e
576
577
578

    if isinstance(response, ErrorResponse):
        return JSONResponse(content=response.model_dump(),
579
                            status_code=response.error.code)
580
581
582
    return JSONResponse(content=response.model_dump())


583
@router.post("/v1/chat/completions",
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
             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
                 }
             })
601
@with_cancellation
602
@load_aware_call
603
604
async def create_chat_completion(request: ChatCompletionRequest,
                                 raw_request: Request):
605
606
607
608
    handler = chat(raw_request)
    if handler is None:
        return base(raw_request).create_error_response(
            message="The model does not support Chat Completions API")
609
610
611
612
613
    try:
        generator = await handler.create_chat_completion(request, raw_request)
    except Exception as e:
        raise HTTPException(status_code=HTTPStatus.INTERNAL_SERVER_ERROR.value,
                            detail=str(e)) from e
614
615
    if isinstance(generator, ErrorResponse):
        return JSONResponse(content=generator.model_dump(),
616
                            status_code=generator.error.code)
617

618
    elif isinstance(generator, ChatCompletionResponse):
619
        return JSONResponse(content=generator.model_dump())
620

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

623

624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
@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
                 },
             })
642
@with_cancellation
643
@load_aware_call
644
async def create_completion(request: CompletionRequest, raw_request: Request):
645
646
647
648
649
    handler = completion(raw_request)
    if handler is None:
        return base(raw_request).create_error_response(
            message="The model does not support Completions API")

650
651
652
653
654
655
656
657
658
    try:
        generator = await handler.create_completion(request, raw_request)
    except OverflowError as e:
        raise HTTPException(status_code=HTTPStatus.BAD_REQUEST.value,
                            detail=str(e)) from e
    except Exception as e:
        raise HTTPException(status_code=HTTPStatus.INTERNAL_SERVER_ERROR.value,
                            detail=str(e)) from e

659
660
    if isinstance(generator, ErrorResponse):
        return JSONResponse(content=generator.model_dump(),
661
                            status_code=generator.error.code)
662
    elif isinstance(generator, CompletionResponse):
663
        return JSONResponse(content=generator.model_dump())
Zhuohan Li's avatar
Zhuohan Li committed
664

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

Zhuohan Li's avatar
Zhuohan Li committed
667

668
669
670
671
672
673
674
675
676
677
@router.post("/v1/embeddings",
             dependencies=[Depends(validate_json_request)],
             responses={
                 HTTPStatus.BAD_REQUEST.value: {
                     "model": ErrorResponse
                 },
                 HTTPStatus.INTERNAL_SERVER_ERROR.value: {
                     "model": ErrorResponse
                 },
             })
678
@with_cancellation
679
@load_aware_call
680
async def create_embedding(request: EmbeddingRequest, raw_request: Request):
681
682
    handler = embedding(raw_request)
    if handler is None:
683
684
685
        return base(raw_request).create_error_response(
            message="The model does not support Embeddings API")

686
687
688
689
690
    try:
        generator = await handler.create_embedding(request, raw_request)
    except Exception as e:
        raise HTTPException(status_code=HTTPStatus.INTERNAL_SERVER_ERROR.value,
                            detail=str(e)) from e
691

692
693
    if isinstance(generator, ErrorResponse):
        return JSONResponse(content=generator.model_dump(),
694
                            status_code=generator.error.code)
695
    elif isinstance(generator, EmbeddingResponse):
696
697
        return JSONResponse(content=generator.model_dump())

698
699
    assert_never(generator)

700

701
702
703
704
705
706
707
708
709
710
@router.post("/pooling",
             dependencies=[Depends(validate_json_request)],
             responses={
                 HTTPStatus.BAD_REQUEST.value: {
                     "model": ErrorResponse
                 },
                 HTTPStatus.INTERNAL_SERVER_ERROR.value: {
                     "model": ErrorResponse
                 },
             })
711
@with_cancellation
712
@load_aware_call
713
714
715
716
717
async def create_pooling(request: PoolingRequest, raw_request: Request):
    handler = pooling(raw_request)
    if handler is None:
        return base(raw_request).create_error_response(
            message="The model does not support Pooling API")
718
719
720
721
722
    try:
        generator = await handler.create_pooling(request, raw_request)
    except Exception as e:
        raise HTTPException(status_code=HTTPStatus.INTERNAL_SERVER_ERROR.value,
                            detail=str(e)) from e
723
724
    if isinstance(generator, ErrorResponse):
        return JSONResponse(content=generator.model_dump(),
725
                            status_code=generator.error.code)
726
    elif isinstance(generator, (PoolingResponse, IOProcessorResponse)):
727
728
729
730
731
        return JSONResponse(content=generator.model_dump())

    assert_never(generator)


732
733
734
735
736
737
738
739
740
741
@router.post("/classify", dependencies=[Depends(validate_json_request)])
@with_cancellation
@load_aware_call
async def create_classify(request: ClassificationRequest,
                          raw_request: Request):
    handler = classify(raw_request)
    if handler is None:
        return base(raw_request).create_error_response(
            message="The model does not support Classification API")

742
743
744
745
746
    try:
        generator = await handler.create_classify(request, raw_request)
    except Exception as e:
        raise HTTPException(status_code=HTTPStatus.INTERNAL_SERVER_ERROR.value,
                            detail=str(e)) from e
747
748
    if isinstance(generator, ErrorResponse):
        return JSONResponse(content=generator.model_dump(),
749
                            status_code=generator.error.code)
750
751
752
753
754
755
756

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

    assert_never(generator)


757
758
759
760
761
762
763
764
765
766
@router.post("/score",
             dependencies=[Depends(validate_json_request)],
             responses={
                 HTTPStatus.BAD_REQUEST.value: {
                     "model": ErrorResponse
                 },
                 HTTPStatus.INTERNAL_SERVER_ERROR.value: {
                     "model": ErrorResponse
                 },
             })
767
@with_cancellation
768
@load_aware_call
769
770
771
772
773
774
async def create_score(request: ScoreRequest, raw_request: Request):
    handler = score(raw_request)
    if handler is None:
        return base(raw_request).create_error_response(
            message="The model does not support Score API")

775
776
777
778
779
    try:
        generator = await handler.create_score(request, raw_request)
    except Exception as e:
        raise HTTPException(status_code=HTTPStatus.INTERNAL_SERVER_ERROR.value,
                            detail=str(e)) from e
780
781
    if isinstance(generator, ErrorResponse):
        return JSONResponse(content=generator.model_dump(),
782
                            status_code=generator.error.code)
783
784
785
786
787
788
    elif isinstance(generator, ScoreResponse):
        return JSONResponse(content=generator.model_dump())

    assert_never(generator)


789
790
791
792
793
794
795
796
797
798
@router.post("/v1/score",
             dependencies=[Depends(validate_json_request)],
             responses={
                 HTTPStatus.BAD_REQUEST.value: {
                     "model": ErrorResponse
                 },
                 HTTPStatus.INTERNAL_SERVER_ERROR.value: {
                     "model": ErrorResponse
                 },
             })
799
@with_cancellation
800
@load_aware_call
801
802
803
804
805
806
807
808
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 "
        "have moved it to `/score`. Please update your client accordingly.")

    return await create_score(request, raw_request)


809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
@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
                 },
             })
826
@with_cancellation
827
@load_aware_call
828
829
830
async def create_transcriptions(raw_request: Request,
                                request: Annotated[TranscriptionRequest,
                                                   Form()]):
831
832
833
834
835
836
    handler = transcription(raw_request)
    if handler is None:
        return base(raw_request).create_error_response(
            message="The model does not support Transcriptions API")

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

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

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

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


854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
@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
                 },
             })
@with_cancellation
@load_aware_call
async def create_translations(request: Annotated[TranslationRequest,
                                                 Form()],
                              raw_request: Request):
    handler = translation(raw_request)
    if handler is None:
        return base(raw_request).create_error_response(
            message="The model does not support Translations API")

    audio_data = await request.file.read()
882
883
884
885
886
887
    try:
        generator = await handler.create_translation(audio_data, request,
                                                     raw_request)
    except Exception as e:
        raise HTTPException(status_code=HTTPStatus.INTERNAL_SERVER_ERROR.value,
                            detail=str(e)) from e
888
889
890

    if isinstance(generator, ErrorResponse):
        return JSONResponse(content=generator.model_dump(),
891
                            status_code=generator.error.code)
892
893
894
895
896
897
898

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

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


899
900
901
902
903
904
905
906
907
908
@router.post("/rerank",
             dependencies=[Depends(validate_json_request)],
             responses={
                 HTTPStatus.BAD_REQUEST.value: {
                     "model": ErrorResponse
                 },
                 HTTPStatus.INTERNAL_SERVER_ERROR.value: {
                     "model": ErrorResponse
                 },
             })
909
@with_cancellation
910
@load_aware_call
911
912
913
914
915
async def do_rerank(request: RerankRequest, raw_request: Request):
    handler = rerank(raw_request)
    if handler is None:
        return base(raw_request).create_error_response(
            message="The model does not support Rerank (Score) API")
916
917
918
919
920
    try:
        generator = await handler.do_rerank(request, raw_request)
    except Exception as e:
        raise HTTPException(status_code=HTTPStatus.INTERNAL_SERVER_ERROR.value,
                            detail=str(e)) from e
921
922
    if isinstance(generator, ErrorResponse):
        return JSONResponse(content=generator.model_dump(),
923
                            status_code=generator.error.code)
924
925
926
927
928
929
    elif isinstance(generator, RerankResponse):
        return JSONResponse(content=generator.model_dump())

    assert_never(generator)


930
931
932
933
934
935
936
937
938
939
@router.post("/v1/rerank",
             dependencies=[Depends(validate_json_request)],
             responses={
                 HTTPStatus.BAD_REQUEST.value: {
                     "model": ErrorResponse
                 },
                 HTTPStatus.INTERNAL_SERVER_ERROR.value: {
                     "model": ErrorResponse
                 },
             })
940
941
@with_cancellation
async def do_rerank_v1(request: RerankRequest, raw_request: Request):
942
    logger.warning_once(
943
        "To indicate that the rerank API is not part of the standard OpenAI"
944
        " API, we have located it at `/rerank`. Please update your client "
945
946
947
948
949
        "accordingly. (Note: Conforms to JinaAI rerank API)")

    return await do_rerank(request, raw_request)


950
951
952
953
954
955
956
957
958
959
@router.post("/v2/rerank",
             dependencies=[Depends(validate_json_request)],
             responses={
                 HTTPStatus.BAD_REQUEST.value: {
                     "model": ErrorResponse
                 },
                 HTTPStatus.INTERNAL_SERVER_ERROR.value: {
                     "model": ErrorResponse
                 },
             })
960
961
962
963
964
@with_cancellation
async def do_rerank_v2(request: RerankRequest, raw_request: Request):
    return await do_rerank(request, raw_request)


965
if envs.VLLM_SERVER_DEV_MODE:
966
967
    logger.warning("SECURITY WARNING: Development endpoints are enabled! "
                   "This should NOT be used in production!")
968

969
970
    PydanticVllmConfig = pydantic.TypeAdapter(VllmConfig)

971
    @router.get("/server_info")
972
973
974
975
976
977
978
979
980
981
982
983
984
    async def show_server_info(
        raw_request: Request,
        config_format: Annotated[Literal["text", "json"],
                                 Query()] = "text",
    ):
        vllm_config: VllmConfig = raw_request.app.state.vllm_config
        server_info = {
            "vllm_config":
            str(vllm_config)
            if config_format == "text" else PydanticVllmConfig.dump_python(
                vllm_config, mode="json", fallback=str)
            # fallback=str is needed to handle e.g. torch.dtype
        }
985
986
        return JSONResponse(content=server_info)

987
988
989
990
991
992
    @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.
        """
993
994
995
996
997
998
        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)
999
1000
        return Response(status_code=200)

1001
1002
1003
1004
1005
1006
1007
1008
1009
1010
1011
    @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):
1012
1013
1014
1015
1016
1017
        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)
1018
1019
1020
1021
        # 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)

1022
1023
1024
1025
1026
1027
    @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})

1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
    @router.post("/collective_rpc")
    async def collective_rpc(raw_request: Request):
        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
        method = body.get("method")
        if method is None:
            raise HTTPException(status_code=HTTPStatus.BAD_REQUEST.value,
                                detail="Missing 'method' in request body")
        # For security reason, only serialized string args/kwargs are passed.
1040
        # User-defined `method` is responsible for deserialization if needed.
1041
1042
1043
1044
1045
1046
1047
1048
1049
1050
1051
1052
1053
1054
1055
        args: list[str] = body.get("args", [])
        kwargs: dict[str, str] = body.get("kwargs", {})
        timeout: Optional[float] = body.get("timeout")
        results = await engine_client(raw_request).collective_rpc(
            method=method, timeout=timeout, args=tuple(args), kwargs=kwargs)
        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})

1056

1057
1058
1059
1060
1061
1062
1063
1064
1065
1066
1067
1068
1069
1070
1071
1072
1073
1074
1075
1076
1077
1078
1079
1080
1081
1082
1083
1084
1085
1086
1087
1088
1089
1090
1091
1092
1093
1094
1095
1096
1097
1098
1099
1100
1101
1102
1103
1104
1105
1106
1107
1108
1109
1110
1111
1112
1113
1114
1115
1116
1117
1118
1119
1120
1121
1122
1123
@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
                 },
             })
async def scale_elastic_ep(raw_request: Request):
    try:
        body = await raw_request.json()
    except json.JSONDecodeError as e:
        raise HTTPException(status_code=400,
                            detail="Invalid JSON format") from e  # noqa: B904

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

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

    if not isinstance(drain_timeout, int) or drain_timeout <= 0:
        raise HTTPException(status_code=400,
                            detail="drain_timeout must be a positive integer")

    # 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)
        return JSONResponse({
            "message":
            f"Scaled to {new_data_parallel_size} "
            "data parallel engines",
        })
    except TimeoutError as e:
        raise HTTPException(status_code=408,
                            detail="Scale failed due to request drain timeout "
                            f"after {drain_timeout} seconds") from e
    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})


1124
1125
1126
1127
1128
1129
1130
1131
1132
1133
1134
1135
1136
1137
1138
1139
1140
1141
1142
1143
1144
1145
1146
1147
# TODO: RequestType = TypeForm[BaseModel] when recognized by type checkers
# (requires typing_extensions >= 4.13)
RequestType = Any
GetHandlerFn = Callable[[Request], Optional[OpenAIServing]]
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
]


1148
1149
1150
1151
1152
1153
1154
1155
1156
1157
1158
1159
1160
@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
                 },
             })
1161
async def invocations(raw_request: Request):
1162
    """For SageMaker, routes requests based on the request type."""
1163
1164
    try:
        body = await raw_request.json()
1165
    except json.JSONDecodeError as e:
1166
1167
1168
        raise HTTPException(status_code=HTTPStatus.BAD_REQUEST.value,
                            detail=f"JSON decode error: {e}") from e

1169
1170
1171
1172
    valid_endpoints = [(validator, endpoint)
                       for validator, (get_handler,
                                       endpoint) in INVOCATION_VALIDATORS
                       if get_handler(raw_request) is not None]
1173

1174
1175
1176
1177
1178
1179
1180
1181
1182
1183
1184
1185
1186
1187
1188
    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
    ]
    msg = ("Cannot find suitable handler for request. "
           f"Expected one of: {type_names}")
    res = base(raw_request).create_error_response(message=msg)
1189
    return JSONResponse(content=res.model_dump(), status_code=res.error.code)
1190
1191


1192
1193
1194
1195
1196
1197
if envs.VLLM_TORCH_PROFILER_DIR:
    logger.warning(
        "Torch Profiler is enabled in the API server. This should ONLY be "
        "used for local development!")

    @router.post("/start_profile")
1198
    async def start_profile(raw_request: Request):
1199
        logger.info("Starting profiler...")
1200
        await engine_client(raw_request).start_profile()
1201
1202
1203
1204
        logger.info("Profiler started.")
        return Response(status_code=200)

    @router.post("/stop_profile")
1205
    async def stop_profile(raw_request: Request):
1206
        logger.info("Stopping profiler...")
1207
        await engine_client(raw_request).stop_profile()
1208
1209
1210
1211
        logger.info("Profiler stopped.")
        return Response(status_code=200)


1212
1213
if envs.VLLM_ALLOW_RUNTIME_LORA_UPDATING:
    logger.warning(
1214
        "LoRA dynamic loading & unloading is enabled in the API server. "
1215
1216
        "This should ONLY be used for local development!")

1217
1218
    @router.post("/v1/load_lora_adapter",
                 dependencies=[Depends(validate_json_request)])
1219
    async def load_lora_adapter(request: LoadLoRAAdapterRequest,
1220
                                raw_request: Request):
1221
1222
1223
1224
        handler = models(raw_request)
        response = await handler.load_lora_adapter(request)
        if isinstance(response, ErrorResponse):
            return JSONResponse(content=response.model_dump(),
1225
                                status_code=response.error.code)
1226
1227
1228

        return Response(status_code=200, content=response)

1229
1230
    @router.post("/v1/unload_lora_adapter",
                 dependencies=[Depends(validate_json_request)])
1231
    async def unload_lora_adapter(request: UnloadLoRAAdapterRequest,
1232
                                  raw_request: Request):
1233
1234
1235
1236
        handler = models(raw_request)
        response = await handler.unload_lora_adapter(request)
        if isinstance(response, ErrorResponse):
            return JSONResponse(content=response.model_dump(),
1237
                                status_code=response.error.code)
1238
1239
1240
1241

        return Response(status_code=200, content=response)


1242
1243
1244
1245
1246
1247
1248
1249
1250
1251
1252
1253
def load_log_config(log_config_file: Optional[str]) -> Optional[dict]:
    if not log_config_file:
        return None
    try:
        with open(log_config_file) as f:
            return json.load(f)
    except Exception as e:
        logger.warning("Failed to load log config from file %s: error %s",
                       log_config_file, e)
        return None


1254
1255
1256
class AuthenticationMiddleware:
    """
    Pure ASGI middleware that authenticates each request by checking
1257
    if the Authorization Bearer token exists and equals anyof "{api_key}".
1258
1259
1260
1261
1262
1263
1264
1265

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

1266
    def __init__(self, app: ASGIApp, tokens: list[str]) -> None:
1267
        self.app = app
1268
1269
1270
1271
1272
1273
1274
1275
1276
1277
1278
1279
1280
1281
1282
1283
1284
1285
1286
1287
        self.api_tokens = [
            hashlib.sha256(t.encode("utf-8")).digest() for t in tokens
        ]

    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
1288
1289
1290
1291
1292
1293
1294
1295
1296
1297
1298
1299

    def __call__(self, scope: Scope, receive: Receive,
                 send: Send) -> Awaitable[None]:
        if scope["type"] not in ("http",
                                 "websocket") or scope["method"] == "OPTIONS":
            # 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.
1300
        if url_path.startswith("/v1") and not self.verify_token(headers):
1301
1302
1303
1304
1305
1306
1307
1308
1309
1310
1311
1312
1313
1314
1315
1316
1317
1318
1319
1320
1321
1322
1323
1324
1325
1326
1327
1328
1329
1330
1331
1332
1333
1334
1335
1336
1337
1338
1339
            response = JSONResponse(content={"error": "Unauthorized"},
                                    status_code=401)
            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

    def __call__(self, scope: Scope, receive: Receive,
                 send: Send) -> Awaitable[None]:
        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"])
                request_id = request_headers.get("X-Request-Id",
                                                 uuid.uuid4().hex)
                response_headers.append("X-Request-Id", request_id)
            await send(message)

        return self.app(scope, receive, send_with_request_id)


1340
1341
1342
1343
1344
1345
1346
1347
# 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.
1348

1349
1350
1351
1352
1353
1354
1355
1356
1357
1358
1359
1360
1361
1362
1363
1364
1365
1366
1367
1368
1369
1370
1371
1372
1373
1374
    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

    def __call__(self, scope: Scope, receive: Receive,
                 send: Send) -> Awaitable[None]:
        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
            response = JSONResponse(content={
                "error":
                "The model is currently scaling. Please try again later."
            },
                                    status_code=503)
            return response(scope, receive, send)

        return self.app(scope, receive, send)


1375
1376
1377
1378
1379
1380
1381
1382
1383
1384
1385
1386
1387
1388
1389
1390
1391
1392
1393
1394
1395
1396
1397
1398
1399
1400
1401
1402
1403
1404
1405
1406
1407
1408
1409
1410
1411
1412
1413
1414
1415
1416
1417
1418
1419
1420
1421
1422
1423
1424
1425
1426
1427
1428
1429
1430
1431
1432
1433
1434
1435
1436
1437
1438
1439
1440
1441
1442
1443
1444
1445
1446
1447
1448
1449
1450
1451
1452
1453
1454
1455
1456
1457
1458
1459
1460
1461
1462
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
1488
1489
1490
1491
1492
1493
1494
1495
1496
1497
1498
1499
1500
1501
1502
1503
1504
1505
1506
1507
1508
1509
1510
def _extract_content_from_chunk(chunk_data: dict) -> str:
    """Extract content from a streaming response chunk."""
    try:
        from vllm.entrypoints.openai.protocol import (
            ChatCompletionStreamResponse, CompletionStreamResponse)

        # Try using Completion types for type-safe parsing
        if chunk_data.get('object') == 'chat.completion.chunk':
            chat_response = ChatCompletionStreamResponse.model_validate(
                chunk_data)
            if chat_response.choices and chat_response.choices[0].delta.content:
                return chat_response.choices[0].delta.content
        elif chunk_data.get('object') == 'text_completion':
            completion_response = CompletionStreamResponse.model_validate(
                chunk_data)
            if completion_response.choices and completion_response.choices[
                    0].text:
                return completion_response.choices[0].text
    except pydantic.ValidationError:
        # Fallback to manual parsing
        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']
    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:
            chunk_str = chunk.decode('utf-8')
        except UnicodeDecodeError:
            # Skip malformed chunks
            return []

        self.buffer += chunk_str
        events = []

        # Process complete lines
        while '\n' in self.buffer:
            line, self.buffer = self.buffer.split('\n', 1)
            line = line.rstrip('\r')  # Handle CRLF

            if line.startswith('data: '):
                data_str = line[6:].strip()
                if data_str == '[DONE]':
                    events.append({'type': 'done'})
                elif data_str:
                    try:
                        event_data = json.loads(data_str)
                        events.append({'type': 'data', 'data': event_data})
                    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."""
        return ''.join(self.content_buffer)


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:
                if event['type'] == 'data':
                    content = sse_decoder.extract_content(event['data'])
                    sse_decoder.add_content(content)
                elif event['type'] == 'done':
                    # 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(
                            "response_body={streaming_complete: " \
                            "content='%s', chunks=%d}",
                            full_content, chunk_count)
                    else:
                        logger.info(
                            "response_body={streaming_complete: " \
                            "no_content, chunks=%d}",
                            chunk_count)
                    return

    response.body_iterator = iterate_in_threadpool(buffered_iterator())
    logger.info("response_body={streaming_started: chunks=%d}",
                len(response_body))


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


1511
def build_app(args: Namespace) -> FastAPI:
1512
1513
1514
1515
1516
1517
1518
    if args.disable_fastapi_docs:
        app = FastAPI(openapi_url=None,
                      docs_url=None,
                      redoc_url=None,
                      lifespan=lifespan)
    else:
        app = FastAPI(lifespan=lifespan)
Ethan Xu's avatar
Ethan Xu committed
1519
1520
    app.include_router(router)
    app.root_path = args.root_path
Zhuohan Li's avatar
Zhuohan Li committed
1521

1522
1523
    mount_metrics(app)

Zhuohan Li's avatar
Zhuohan Li committed
1524
1525
1526
1527
1528
1529
1530
1531
    app.add_middleware(
        CORSMiddleware,
        allow_origins=args.allowed_origins,
        allow_credentials=args.allow_credentials,
        allow_methods=args.allowed_methods,
        allow_headers=args.allowed_headers,
    )

1532
1533
    @app.exception_handler(HTTPException)
    async def http_exception_handler(_: Request, exc: HTTPException):
1534
1535
        err = ErrorResponse(
            error=ErrorInfo(message=exc.detail,
1536
                            type=HTTPStatus(exc.status_code).phrase,
1537
                            code=exc.status_code))
1538
1539
        return JSONResponse(err.model_dump(), status_code=exc.status_code)

Ethan Xu's avatar
Ethan Xu committed
1540
    @app.exception_handler(RequestValidationError)
1541
1542
1543
1544
1545
1546
1547
1548
1549
1550
    async def validation_exception_handler(_: Request,
                                           exc: RequestValidationError):
        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

1551
1552
1553
        err = ErrorResponse(error=ErrorInfo(message=message,
                                            type=HTTPStatus.BAD_REQUEST.phrase,
                                            code=HTTPStatus.BAD_REQUEST))
Ethan Xu's avatar
Ethan Xu committed
1554
1555
1556
        return JSONResponse(err.model_dump(),
                            status_code=HTTPStatus.BAD_REQUEST)

1557
    # Ensure --api-key option from CLI takes precedence over VLLM_API_KEY
1558
1559
    if tokens := [key for key in (args.api_key or [envs.VLLM_API_KEY]) if key]:
        app.add_middleware(AuthenticationMiddleware, tokens=tokens)
1560

1561
    if args.enable_request_id_headers:
1562
        app.add_middleware(XRequestIdMiddleware)
1563

1564
1565
1566
    # Add scaling middleware to check for scaling state
    app.add_middleware(ScalingMiddleware)

1567
1568
1569
1570
1571
1572
1573
1574
1575
1576
1577
1578
    if envs.VLLM_DEBUG_LOG_API_SERVER_RESPONSE:
        logger.warning("CAUTION: Enabling log response in the API Server. "
                       "This can include sensitive information and should be "
                       "avoided in production.")

        @app.middleware("http")
        async def log_response(request: Request, call_next):
            response = await call_next(request)
            response_body = [
                section async for section in response.body_iterator
            ]
            response.body_iterator = iterate_in_threadpool(iter(response_body))
1579
1580
1581
1582
1583
1584
1585
1586
1587
1588
1589
            # 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)
1590
            return response
1591

1592
1593
1594
1595
    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):
1596
            app.add_middleware(imported)  # type: ignore[arg-type]
1597
1598
1599
        elif inspect.iscoroutinefunction(imported):
            app.middleware("http")(imported)
        else:
1600
1601
            raise ValueError(f"Invalid middleware {middleware}. "
                             f"Must be a function or a class.")
1602

Ethan Xu's avatar
Ethan Xu committed
1603
1604
1605
    return app


1606
async def init_app_state(
1607
    engine_client: EngineClient,
1608
    vllm_config: VllmConfig,
1609
    state: State,
1610
    args: Namespace,
1611
) -> None:
1612
    if args.served_model_name is not None:
1613
        served_model_names = args.served_model_name
1614
    else:
1615
        served_model_names = [args.model]
1616

1617
    if args.enable_log_requests:
1618
        request_logger = RequestLogger(max_log_len=args.max_log_len)
1619
1620
    else:
        request_logger = None
1621

1622
1623
1624
1625
1626
    base_model_paths = [
        BaseModelPath(name=name, model_path=args.model)
        for name in served_model_names
    ]

1627
    state.engine_client = engine_client
1628
    state.log_stats = not args.disable_log_stats
1629
1630
    state.vllm_config = vllm_config
    model_config = vllm_config.model_config
Ethan Xu's avatar
Ethan Xu committed
1631

1632
    supported_tasks = await engine_client.get_supported_tasks()
1633
1634
1635

    logger.info("Supported_tasks: %s", supported_tasks)

1636
    resolved_chat_template = load_chat_template(args.chat_template)
1637
    if resolved_chat_template is not None:
1638
1639
1640
1641
1642
1643
1644
1645
1646
        # Get the tokenizer to check official template
        tokenizer = await engine_client.get_tokenizer()

        if isinstance(tokenizer, MistralTokenizer):
            # The warning is logged in resolve_mistral_chat_template.
            resolved_chat_template = resolve_mistral_chat_template(
                chat_template=resolved_chat_template)
        else:
            hf_chat_template = resolve_hf_chat_template(
1647
                tokenizer=tokenizer,
1648
1649
                chat_template=None,
                tools=None,
1650
                model_config=vllm_config.model_config,
1651
            )
1652
1653
1654
1655
1656
1657
1658

            if hf_chat_template != resolved_chat_template:
                logger.warning(
                    "Using supplied chat template: %s\n"
                    "It is different from official chat template '%s'. "
                    "This discrepancy may lead to performance degradation.",
                    resolved_chat_template, args.model)
1659

1660
1661
    if args.tool_server == "demo":
        tool_server: Optional[ToolServer] = DemoToolServer()
1662
1663
        assert isinstance(tool_server, DemoToolServer)
        await tool_server.init_and_validate()
1664
1665
1666
    elif args.tool_server:
        tool_server = MCPToolServer()
        await tool_server.add_tool_server(args.tool_server)
1667
1668
1669
    else:
        tool_server = None

1670
1671
1672
1673
1674
1675
1676
1677
1678
1679
1680
1681
1682
1683
1684
1685
1686
    # Merge default_mm_loras into the static lora_modules
    default_mm_loras = (vllm_config.lora_config.default_mm_loras
                        if vllm_config.lora_config is not None else {})

    lora_modules = args.lora_modules
    if default_mm_loras:
        default_mm_lora_paths = [
            LoRAModulePath(
                name=modality,
                path=lora_path,
            ) for modality, lora_path in default_mm_loras.items()
        ]
        if args.lora_modules is None:
            lora_modules = default_mm_lora_paths
        else:
            lora_modules += default_mm_lora_paths

1687
    state.openai_serving_models = OpenAIServingModels(
1688
        engine_client=engine_client,
1689
1690
        model_config=model_config,
        base_model_paths=base_model_paths,
1691
        lora_modules=lora_modules,
1692
    )
1693
    await state.openai_serving_models.init_static_loras()
1694
1695
1696
1697
1698
1699
1700
1701
1702
1703
    state.openai_serving_responses = OpenAIServingResponses(
        engine_client,
        model_config,
        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,
1704
        tool_server=tool_server,
1705
        reasoning_parser=args.structured_outputs_config.reasoning_parser,
1706
1707
        enable_prompt_tokens_details=args.enable_prompt_tokens_details,
        enable_force_include_usage=args.enable_force_include_usage,
1708
        enable_log_outputs=args.enable_log_outputs,
1709
        log_error_stack=args.log_error_stack,
1710
    ) if "generate" in supported_tasks else None
1711
    state.openai_serving_chat = OpenAIServingChat(
1712
        engine_client,
1713
        model_config,
1714
        state.openai_serving_models,
1715
1716
        args.response_role,
        request_logger=request_logger,
1717
1718
        chat_template=resolved_chat_template,
        chat_template_content_format=args.chat_template_content_format,
1719
        return_tokens_as_token_ids=args.return_tokens_as_token_ids,
1720
        enable_auto_tools=args.enable_auto_tool_choice,
1721
1722
        exclude_tools_when_tool_choice_none=args.
        exclude_tools_when_tool_choice_none,
1723
        tool_parser=args.tool_call_parser,
1724
        reasoning_parser=args.structured_outputs_config.reasoning_parser,
1725
        enable_prompt_tokens_details=args.enable_prompt_tokens_details,
1726
        enable_force_include_usage=args.enable_force_include_usage,
1727
        enable_log_outputs=args.enable_log_outputs,
1728
        log_error_stack=args.log_error_stack,
1729
    ) if "generate" in supported_tasks else None
1730
    state.openai_serving_completion = OpenAIServingCompletion(
1731
        engine_client,
1732
        model_config,
1733
        state.openai_serving_models,
1734
        request_logger=request_logger,
1735
        return_tokens_as_token_ids=args.return_tokens_as_token_ids,
1736
        enable_prompt_tokens_details=args.enable_prompt_tokens_details,
1737
        enable_force_include_usage=args.enable_force_include_usage,
1738
        log_error_stack=args.log_error_stack,
1739
    ) if "generate" in supported_tasks else None
1740
    state.openai_serving_pooling = OpenAIServingPooling(
1741
        engine_client,
1742
        vllm_config,
1743
        state.openai_serving_models,
1744
        request_logger=request_logger,
1745
1746
        chat_template=resolved_chat_template,
        chat_template_content_format=args.chat_template_content_format,
1747
        log_error_stack=args.log_error_stack,
1748
    ) if "encode" in supported_tasks else None
1749
1750
1751
    state.openai_serving_embedding = OpenAIServingEmbedding(
        engine_client,
        model_config,
1752
        state.openai_serving_models,
1753
1754
1755
        request_logger=request_logger,
        chat_template=resolved_chat_template,
        chat_template_content_format=args.chat_template_content_format,
1756
        log_error_stack=args.log_error_stack,
1757
    ) if "embed" in supported_tasks else None
1758
1759
1760
1761
1762
    state.openai_serving_classification = ServingClassification(
        engine_client,
        model_config,
        state.openai_serving_models,
        request_logger=request_logger,
1763
        log_error_stack=args.log_error_stack,
1764
    ) if "classify" in supported_tasks else None
1765
1766
1767
1768
    state.openai_serving_scores = ServingScores(
        engine_client,
        model_config,
        state.openai_serving_models,
1769
        request_logger=request_logger,
1770
        log_error_stack=args.log_error_stack,
1771
    ) if ("embed" in supported_tasks or "score" in supported_tasks) else None
1772
    state.openai_serving_tokenization = OpenAIServingTokenization(
1773
        engine_client,
1774
        model_config,
1775
        state.openai_serving_models,
1776
        request_logger=request_logger,
1777
1778
        chat_template=resolved_chat_template,
        chat_template_content_format=args.chat_template_content_format,
1779
        log_error_stack=args.log_error_stack,
1780
    )
1781
1782
1783
    state.openai_serving_transcription = OpenAIServingTranscription(
        engine_client,
        model_config,
1784
1785
        state.openai_serving_models,
        request_logger=request_logger,
1786
        log_error_stack=args.log_error_stack,
1787
    ) if "transcription" in supported_tasks else None
1788
1789
1790
    state.openai_serving_translation = OpenAIServingTranslation(
        engine_client,
        model_config,
1791
1792
        state.openai_serving_models,
        request_logger=request_logger,
1793
        log_error_stack=args.log_error_stack,
1794
    ) if "transcription" in supported_tasks else None
1795

1796
1797
1798
    state.enable_server_load_tracking = args.enable_server_load_tracking
    state.server_load_metrics = 0

1799

1800
def create_server_socket(addr: tuple[str, int]) -> socket.socket:
1801
1802
1803
1804
1805
1806
    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)
1807
    sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, 1)
1808
1809
1810
1811
1812
    sock.bind(addr)

    return sock


1813
1814
1815
1816
1817
1818
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


1819
def validate_api_server_args(args):
1820
    valid_tool_parses = ToolParserManager.tool_parsers.keys()
1821
    if args.enable_auto_tool_choice \
1822
            and args.tool_call_parser not in valid_tool_parses:
1823
        raise KeyError(f"invalid tool call parser: {args.tool_call_parser} "
1824
                       f"(chose from {{ {','.join(valid_tool_parses)} }})")
1825

1826
    valid_reasoning_parses = ReasoningParserManager.reasoning_parsers.keys()
1827
1828
    if ((reasoning_parser := args.structured_outputs_config.reasoning_parser)
            and reasoning_parser not in valid_reasoning_parses):
1829
        raise KeyError(
1830
            f"invalid reasoning parser: {reasoning_parser} "
1831
1832
            f"(chose from {{ {','.join(valid_reasoning_parses)} }})")

1833
1834
1835
1836
1837
1838
1839
1840
1841
1842
1843
1844
1845

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)

    validate_api_server_args(args)

1846
1847
1848
    # 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
1849
1850
1851
1852
1853
    if args.uds:
        sock = create_server_unix_socket(args.uds)
    else:
        sock_addr = (args.host or "", args.port)
        sock = create_server_socket(sock_addr)
1854

1855
1856
1857
1858
    # workaround to avoid footguns where uvicorn drops requests with too
    # many concurrent requests active
    set_ulimit()

1859
1860
1861
1862
1863
1864
    def signal_handler(*_) -> None:
        # Interrupt server on sigterm while initializing
        raise KeyboardInterrupt("terminated")

    signal.signal(signal.SIGTERM, signal_handler)

1865
1866
1867
1868
1869
1870
1871
1872
    if args.uds:
        listen_address = f"unix:{args.uds}"
    else:
        addr, port = sock_addr
        is_ssl = args.ssl_keyfile and args.ssl_certfile
        host_part = f"[{addr}]" if is_valid_ipv6_address(
            addr) else addr or "0.0.0.0"
        listen_address = f"http{'s' if is_ssl else ''}://{host_part}:{port}"
1873
1874
1875
1876
1877
    return listen_address, sock


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

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

1882
1883
1884
1885
1886
1887
1888
1889
1890
1891
1892
1893
1894
1895
    listen_address, sock = setup_server(args)
    await run_server_worker(listen_address, sock, args, **uvicorn_kwargs)


async def run_server_worker(listen_address,
                            sock,
                            args,
                            client_config=None,
                            **uvicorn_kwargs) -> None:
    """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)

1896
1897
1898
1899
1900
    # Load logging config for uvicorn if specified
    log_config = load_log_config(args.log_config_file)
    if log_config is not None:
        uvicorn_kwargs['log_config'] = log_config

1901
1902
1903
1904
    async with build_async_engine_client(
            args,
            client_config=client_config,
    ) as engine_client:
1905
        maybe_register_tokenizer_info_endpoint(args)
1906
1907
        app = build_app(args)

1908
1909
        vllm_config = await engine_client.get_vllm_config()
        await init_app_state(engine_client, vllm_config, app.state, args)
1910

1911
1912
        logger.info("Starting vLLM API server %d on %s",
                    vllm_config.parallel_config._api_process_rank,
1913
                    listen_address)
1914
1915
        shutdown_task = await serve_http(
            app,
1916
            sock=sock,
1917
            enable_ssl_refresh=args.enable_ssl_refresh,
1918
1919
1920
            host=args.host,
            port=args.port,
            log_level=args.uvicorn_log_level,
1921
1922
1923
            # NOTE: When the 'disable_uvicorn_access_log' value is True,
            # no access log will be output.
            access_log=not args.disable_uvicorn_access_log,
1924
            timeout_keep_alive=envs.VLLM_HTTP_TIMEOUT_KEEP_ALIVE,
1925
1926
1927
1928
            ssl_keyfile=args.ssl_keyfile,
            ssl_certfile=args.ssl_certfile,
            ssl_ca_certs=args.ssl_ca_certs,
            ssl_cert_reqs=args.ssl_cert_reqs,
1929
1930
            h11_max_incomplete_event_size=args.h11_max_incomplete_event_size,
            h11_max_header_count=args.h11_max_header_count,
1931
1932
1933
            **uvicorn_kwargs,
        )

1934
    # NB: Await server shutdown only after the backend context is exited
1935
1936
1937
1938
    try:
        await shutdown_task
    finally:
        sock.close()
1939

Ethan Xu's avatar
Ethan Xu committed
1940
1941
1942

if __name__ == "__main__":
    # NOTE(simon):
1943
1944
    # This section should be in sync with vllm/entrypoints/cli/main.py for CLI
    # entrypoints.
1945
    cli_env_setup()
Ethan Xu's avatar
Ethan Xu committed
1946
1947
1948
1949
    parser = FlexibleArgumentParser(
        description="vLLM OpenAI-Compatible RESTful API server.")
    parser = make_arg_parser(parser)
    args = parser.parse_args()
1950
    validate_parsed_serve_args(args)
1951

1952
    uvloop.run(run_server(args))