api_server.py 51.3 KB
Newer Older
1
2
# SPDX-License-Identifier: Apache-2.0

3
import asyncio
4
import atexit
5
import gc
6
7
import importlib
import inspect
8
import json
9
import multiprocessing
10
import os
11
import signal
12
import socket
13
import tempfile
14
import uuid
15
from argparse import Namespace
16
from collections.abc import AsyncIterator
17
from contextlib import asynccontextmanager
18
from functools import partial
19
from http import HTTPStatus
20
from typing import Annotated, Any, Optional
21

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

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

105
TIMEOUT_KEEP_ALIVE = 5  # seconds
Zhuohan Li's avatar
Zhuohan Li committed
106

107
prometheus_multiproc_dir: tempfile.TemporaryDirectory
108

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

112
_running_tasks: set[asyncio.Task] = set()
113

114

115
@asynccontextmanager
116
async def lifespan(app: FastAPI):
117
118
    try:
        if app.state.log_stats:
119
            engine_client: EngineClient = app.state.engine_client
120
121
122

            async def _force_log():
                while True:
123
124
                    await asyncio.sleep(10.)
                    await engine_client.do_log_stats()
125
126
127
128
129
130

            task = asyncio.create_task(_force_log())
            _running_tasks.add(task)
            task.add_done_callback(_running_tasks.remove)
        else:
            task = None
131
132
133
134
135

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


146
@asynccontextmanager
147
async def build_async_engine_client(
148
149
150
    args: Namespace,
    client_config: Optional[dict[str, Any]] = None,
) -> AsyncIterator[EngineClient]:
151

152
    # Context manager to handle engine_client lifecycle
153
154
155
    # Ensures everything is shutdown and cleaned up on error/exit
    engine_args = AsyncEngineArgs.from_cli_args(args)

156
    async with build_async_engine_client_from_engine_args(
157
158
            engine_args, args.disable_frontend_multiprocessing,
            client_config) as engine:
159
160
161
162
163
164
165
        yield engine


@asynccontextmanager
async def build_async_engine_client_from_engine_args(
    engine_args: AsyncEngineArgs,
    disable_frontend_multiprocessing: bool = False,
166
    client_config: Optional[dict[str, Any]] = None,
167
) -> AsyncIterator[EngineClient]:
168
    """
169
    Create EngineClient, either:
170
171
172
173
174
175
        - in-process using the AsyncLLMEngine Directly
        - multiprocess using AsyncLLMEngine RPC

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

176
177
178
179
180
181
182
183
184
185
186
187
188
    # Create the EngineConfig (determines if we can use V1).
    usage_context = UsageContext.OPENAI_API_SERVER
    vllm_config = engine_args.create_engine_config(usage_context=usage_context)

    # V1 AsyncLLM.
    if envs.VLLM_USE_V1:
        if disable_frontend_multiprocessing:
            logger.warning(
                "V1 is enabled, but got --disable-frontend-multiprocessing. "
                "To disable frontend multiprocessing, set VLLM_USE_V1=0.")

        from vllm.v1.engine.async_llm import AsyncLLM
        async_llm: Optional[AsyncLLM] = None
189
190
        client_index = client_config.pop(
            "client_index") if client_config else 0
191
192
193
194
195
        try:
            async_llm = AsyncLLM.from_vllm_config(
                vllm_config=vllm_config,
                usage_context=usage_context,
                disable_log_requests=engine_args.disable_log_requests,
196
197
198
                disable_log_stats=engine_args.disable_log_stats,
                client_addresses=client_config,
                client_index=client_index)
199
200
201
202

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

203
204
205
206
207
208
209
210
            yield async_llm
        finally:
            if async_llm:
                async_llm.shutdown()

    # V0 AsyncLLM.
    elif (MQLLMEngineClient.is_unsupported_config(vllm_config)
          or disable_frontend_multiprocessing):
211

212
213
        engine_client: Optional[EngineClient] = None
        try:
214
215
216
217
218
            engine_client = AsyncLLMEngine.from_vllm_config(
                vllm_config=vllm_config,
                usage_context=usage_context,
                disable_log_requests=engine_args.disable_log_requests,
                disable_log_stats=engine_args.disable_log_stats)
219
220
221
222
            yield engine_client
        finally:
            if engine_client and hasattr(engine_client, "shutdown"):
                engine_client.shutdown()
223

224
    # V0MQLLMEngine.
225
    else:
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
        if "PROMETHEUS_MULTIPROC_DIR" not in os.environ:
            # Make TemporaryDirectory for prometheus multiprocessing
            # Note: global TemporaryDirectory will be automatically
            #   cleaned up upon exit.
            global prometheus_multiproc_dir
            prometheus_multiproc_dir = tempfile.TemporaryDirectory()
            os.environ[
                "PROMETHEUS_MULTIPROC_DIR"] = prometheus_multiproc_dir.name
        else:
            logger.warning(
                "Found PROMETHEUS_MULTIPROC_DIR was set by user. "
                "This directory must be wiped between vLLM runs or "
                "you will find inaccurate metrics. Unset the variable "
                "and vLLM will properly handle cleanup.")

241
        # Select random path for IPC.
242
        ipc_path = get_open_zmq_ipc_path()
243
244
        logger.debug("Multiprocessing frontend to use %s for IPC Path.",
                     ipc_path)
245

246
        # Start RPCServer in separate process (holds the LLMEngine).
247
248
        # the current process might have CUDA context,
        # so we need to spawn a new process
249
250
        context = multiprocessing.get_context("spawn")

251
252
253
        # Ensure we can serialize transformer config before spawning
        maybe_register_config_serialize_by_value()

254
255
256
257
        # The Process can raise an exception during startup, which may
        # not actually result in an exitcode being reported. As a result
        # we use a shared variable to communicate the information.
        engine_alive = multiprocessing.Value('b', True, lock=False)
258
259
260
261
262
        engine_process = context.Process(
            target=run_mp_engine,
            args=(vllm_config, UsageContext.OPENAI_API_SERVER, ipc_path,
                  engine_args.disable_log_stats,
                  engine_args.disable_log_requests, engine_alive))
263
        engine_process.start()
264
        engine_pid = engine_process.pid
265
        assert engine_pid is not None, "Engine process failed to start."
266
        logger.info("Started engine process with PID %d", engine_pid)
267

268
269
270
271
272
273
274
275
        def _cleanup_ipc_path():
            socket_path = ipc_path.replace("ipc://", "")
            if os.path.exists(socket_path):
                os.remove(socket_path)

        # Ensure we clean up the local IPC socket file on exit.
        atexit.register(_cleanup_ipc_path)

276
        # Build RPCClient, which conforms to EngineClient Protocol.
277
        build_client = partial(MQLLMEngineClient, ipc_path, vllm_config,
278
279
280
                               engine_pid)
        mq_engine_client = await asyncio.get_running_loop().run_in_executor(
            None, build_client)
281
        try:
282
283
            while True:
                try:
284
                    await mq_engine_client.setup()
285
                    break
286
                except TimeoutError:
287
288
                    if (not engine_process.is_alive()
                            or not engine_alive.value):
289
                        raise RuntimeError(
290
291
                            "Engine process failed to start. See stack "
                            "trace for the root cause.") from None
292

293
            yield mq_engine_client  # type: ignore[misc]
294
295
        finally:
            # Ensure rpc server process was terminated
296
            engine_process.terminate()
297
298

            # Close all open connections to the backend
299
            mq_engine_client.close()
300

301
302
303
304
305
            # Wait for engine process to join
            engine_process.join(4)
            if engine_process.exitcode is None:
                # Kill if taking longer than 5 seconds to stop
                engine_process.kill()
306

307
308
309
310
311
            # Lazy import for prometheus multiprocessing.
            # We need to set PROMETHEUS_MULTIPROC_DIR environment variable
            # before prometheus_client is imported.
            # See https://prometheus.github.io/client_python/multiprocess/
            from prometheus_client import multiprocess
312
            multiprocess.mark_process_dead(engine_process.pid)
313

314

315
316
async def validate_json_request(raw_request: Request):
    content_type = raw_request.headers.get("content-type", "").lower()
317
318
    media_type = content_type.split(";", maxsplit=1)[0]
    if media_type != "application/json":
319
320
321
        raise RequestValidationError(errors=[
            "Unsupported Media Type: Only 'application/json' is allowed"
        ])
322
323


Ethan Xu's avatar
Ethan Xu committed
324
router = APIRouter()
Zhuohan Li's avatar
Zhuohan Li committed
325

326

327
328
329
330
class PrometheusResponse(Response):
    media_type = prometheus_client.CONTENT_TYPE_LATEST


331
def mount_metrics(app: FastAPI):
332
333
334
    """Mount prometheus metrics to a FastAPI app."""

    registry = get_prometheus_registry()
335

336
337
338
339
    # `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
340
341
342
343
344
345
346
347
348
349
    Instrumentator(
        excluded_handlers=[
            "/metrics",
            "/health",
            "/load",
            "/ping",
            "/version",
            "/server_info",
        ],
        registry=registry,
350
    ).add().instrument(app).expose(app, response_class=PrometheusResponse)
351
352
353

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

355
    # Workaround for 307 Redirect for /metrics
356
    metrics_route.path_regex = re.compile("^/metrics(?P<path>.*)$")
357
    app.routes.append(metrics_route)
358
359


360
361
362
363
364
def base(request: Request) -> OpenAIServing:
    # Reuse the existing instance
    return tokenization(request)


365
366
367
368
def models(request: Request) -> OpenAIServingModels:
    return request.app.state.openai_serving_models


369
def chat(request: Request) -> Optional[OpenAIServingChat]:
370
371
372
    return request.app.state.openai_serving_chat


373
def completion(request: Request) -> Optional[OpenAIServingCompletion]:
374
375
376
    return request.app.state.openai_serving_completion


377
378
379
380
def pooling(request: Request) -> Optional[OpenAIServingPooling]:
    return request.app.state.openai_serving_pooling


381
382
def embedding(request: Request) -> Optional[OpenAIServingEmbedding]:
    return request.app.state.openai_serving_embedding
383
384


385
def score(request: Request) -> Optional[ServingScores]:
386
387
388
    return request.app.state.openai_serving_scores


389
390
391
392
def classify(request: Request) -> Optional[ServingClassification]:
    return request.app.state.openai_serving_classification


393
394
def rerank(request: Request) -> Optional[ServingScores]:
    return request.app.state.openai_serving_scores
395
396


397
398
def tokenization(request: Request) -> OpenAIServingTokenization:
    return request.app.state.openai_serving_tokenization
399
400


401
402
403
404
def transcription(request: Request) -> OpenAIServingTranscription:
    return request.app.state.openai_serving_transcription


405
def engine_client(request: Request) -> EngineClient:
406
407
408
    return request.app.state.engine_client


409
410
@router.get("/health", response_class=Response)
async def health(raw_request: Request) -> Response:
411
    """Health check."""
412
    await engine_client(raw_request).check_health()
413
    return Response(status_code=200)
414
415


416
417
418
419
420
421
422
423
424
@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
    # - /v1/embeddings
    # - /pooling
425
    # - /classify
426
427
428
429
430
431
432
433
434
    # - /score
    # - /v1/score
    # - /rerank
    # - /v1/rerank
    # - /v2/rerank
    return JSONResponse(
        content={'server_load': request.app.state.server_load_metrics})


435
436
437
@router.get("/ping", response_class=Response)
@router.post("/ping", response_class=Response)
async def ping(raw_request: Request) -> Response:
438
439
440
441
    """Ping check. Endpoint required for SageMaker"""
    return await health(raw_request)


442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
@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
                 },
             })
458
@with_cancellation
459
async def tokenize(request: TokenizeRequest, raw_request: Request):
460
461
    handler = tokenization(raw_request)

462
463
464
465
466
467
468
469
470
    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

471
472
473
    if isinstance(generator, ErrorResponse):
        return JSONResponse(content=generator.model_dump(),
                            status_code=generator.code)
474
    elif isinstance(generator, TokenizeResponse):
475
476
        return JSONResponse(content=generator.model_dump())

477
478
    assert_never(generator)

479

480
481
482
483
484
485
486
487
488
489
490
491
492
@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
                 },
             })
493
@with_cancellation
494
async def detokenize(request: DetokenizeRequest, raw_request: Request):
495
496
    handler = tokenization(raw_request)

497
498
499
500
501
502
503
504
    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

505
506
507
    if isinstance(generator, ErrorResponse):
        return JSONResponse(content=generator.model_dump(),
                            status_code=generator.code)
508
    elif isinstance(generator, DetokenizeResponse):
509
510
        return JSONResponse(content=generator.model_dump())

511
512
    assert_never(generator)

513

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

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


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


528
@router.post("/v1/chat/completions",
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
             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
                 }
             })
546
@with_cancellation
547
@load_aware_call
548
549
async def create_chat_completion(request: ChatCompletionRequest,
                                 raw_request: Request):
550
551
552
553
    handler = chat(raw_request)
    if handler is None:
        return base(raw_request).create_error_response(
            message="The model does not support Chat Completions API")
554

555
    generator = await handler.create_chat_completion(request, raw_request)
556

557
558
559
    if isinstance(generator, ErrorResponse):
        return JSONResponse(content=generator.model_dump(),
                            status_code=generator.code)
560

561
    elif isinstance(generator, ChatCompletionResponse):
562
        return JSONResponse(content=generator.model_dump())
563

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

566

567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
@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
                 },
             })
585
@with_cancellation
586
@load_aware_call
587
async def create_completion(request: CompletionRequest, raw_request: Request):
588
589
590
591
592
    handler = completion(raw_request)
    if handler is None:
        return base(raw_request).create_error_response(
            message="The model does not support Completions API")

593
594
595
596
597
598
599
600
601
    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

602
603
604
    if isinstance(generator, ErrorResponse):
        return JSONResponse(content=generator.model_dump(),
                            status_code=generator.code)
605
    elif isinstance(generator, CompletionResponse):
606
        return JSONResponse(content=generator.model_dump())
Zhuohan Li's avatar
Zhuohan Li committed
607

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

Zhuohan Li's avatar
Zhuohan Li committed
610

611
612
613
614
615
616
617
618
619
620
@router.post("/v1/embeddings",
             dependencies=[Depends(validate_json_request)],
             responses={
                 HTTPStatus.BAD_REQUEST.value: {
                     "model": ErrorResponse
                 },
                 HTTPStatus.INTERNAL_SERVER_ERROR.value: {
                     "model": ErrorResponse
                 },
             })
621
@with_cancellation
622
@load_aware_call
623
async def create_embedding(request: EmbeddingRequest, raw_request: Request):
624
625
    handler = embedding(raw_request)
    if handler is None:
626
627
628
629
        return base(raw_request).create_error_response(
            message="The model does not support Embeddings API")

    generator = await handler.create_embedding(request, raw_request)
630

631
632
633
    if isinstance(generator, ErrorResponse):
        return JSONResponse(content=generator.model_dump(),
                            status_code=generator.code)
634
    elif isinstance(generator, EmbeddingResponse):
635
636
        return JSONResponse(content=generator.model_dump())

637
638
    assert_never(generator)

639

640
641
642
643
644
645
646
647
648
649
@router.post("/pooling",
             dependencies=[Depends(validate_json_request)],
             responses={
                 HTTPStatus.BAD_REQUEST.value: {
                     "model": ErrorResponse
                 },
                 HTTPStatus.INTERNAL_SERVER_ERROR.value: {
                     "model": ErrorResponse
                 },
             })
650
@with_cancellation
651
@load_aware_call
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
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")

    generator = await handler.create_pooling(request, raw_request)
    if isinstance(generator, ErrorResponse):
        return JSONResponse(content=generator.model_dump(),
                            status_code=generator.code)
    elif isinstance(generator, PoolingResponse):
        return JSONResponse(content=generator.model_dump())

    assert_never(generator)


668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
@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")

    generator = await handler.create_classify(request, raw_request)
    if isinstance(generator, ErrorResponse):
        return JSONResponse(content=generator.model_dump(),
                            status_code=generator.code)

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

    assert_never(generator)


689
690
691
692
693
694
695
696
697
698
@router.post("/score",
             dependencies=[Depends(validate_json_request)],
             responses={
                 HTTPStatus.BAD_REQUEST.value: {
                     "model": ErrorResponse
                 },
                 HTTPStatus.INTERNAL_SERVER_ERROR.value: {
                     "model": ErrorResponse
                 },
             })
699
@with_cancellation
700
@load_aware_call
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
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")

    generator = await handler.create_score(request, raw_request)
    if isinstance(generator, ErrorResponse):
        return JSONResponse(content=generator.model_dump(),
                            status_code=generator.code)
    elif isinstance(generator, ScoreResponse):
        return JSONResponse(content=generator.model_dump())

    assert_never(generator)


717
718
719
720
721
722
723
724
725
726
@router.post("/v1/score",
             dependencies=[Depends(validate_json_request)],
             responses={
                 HTTPStatus.BAD_REQUEST.value: {
                     "model": ErrorResponse
                 },
                 HTTPStatus.INTERNAL_SERVER_ERROR.value: {
                     "model": ErrorResponse
                 },
             })
727
@with_cancellation
728
@load_aware_call
729
730
731
732
733
734
735
736
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)


737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
@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
                 },
             })
754
@with_cancellation
755
@load_aware_call
756
757
758
async def create_transcriptions(raw_request: Request,
                                request: Annotated[TranscriptionRequest,
                                                   Form()]):
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
    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()
    generator = await handler.create_transcription(audio_data, request,
                                                   raw_request)

    if isinstance(generator, ErrorResponse):
        return JSONResponse(content=generator.model_dump(),
                            status_code=generator.code)

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

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


778
779
780
781
782
783
784
785
786
787
@router.post("/rerank",
             dependencies=[Depends(validate_json_request)],
             responses={
                 HTTPStatus.BAD_REQUEST.value: {
                     "model": ErrorResponse
                 },
                 HTTPStatus.INTERNAL_SERVER_ERROR.value: {
                     "model": ErrorResponse
                 },
             })
788
@with_cancellation
789
@load_aware_call
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
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")
    generator = await handler.do_rerank(request, raw_request)
    if isinstance(generator, ErrorResponse):
        return JSONResponse(content=generator.model_dump(),
                            status_code=generator.code)
    elif isinstance(generator, RerankResponse):
        return JSONResponse(content=generator.model_dump())

    assert_never(generator)


805
806
807
808
809
810
811
812
813
814
@router.post("/v1/rerank",
             dependencies=[Depends(validate_json_request)],
             responses={
                 HTTPStatus.BAD_REQUEST.value: {
                     "model": ErrorResponse
                 },
                 HTTPStatus.INTERNAL_SERVER_ERROR.value: {
                     "model": ErrorResponse
                 },
             })
815
816
@with_cancellation
async def do_rerank_v1(request: RerankRequest, raw_request: Request):
817
    logger.warning_once(
818
        "To indicate that the rerank API is not part of the standard OpenAI"
819
        " API, we have located it at `/rerank`. Please update your client "
820
821
822
823
824
        "accordingly. (Note: Conforms to JinaAI rerank API)")

    return await do_rerank(request, raw_request)


825
826
827
828
829
830
831
832
833
834
@router.post("/v2/rerank",
             dependencies=[Depends(validate_json_request)],
             responses={
                 HTTPStatus.BAD_REQUEST.value: {
                     "model": ErrorResponse
                 },
                 HTTPStatus.INTERNAL_SERVER_ERROR.value: {
                     "model": ErrorResponse
                 },
             })
835
836
837
838
839
@with_cancellation
async def do_rerank_v2(request: RerankRequest, raw_request: Request):
    return await do_rerank(request, raw_request)


840
TASK_HANDLERS: dict[str, dict[str, tuple]] = {
841
842
843
844
845
846
847
848
849
    "generate": {
        "messages": (ChatCompletionRequest, create_chat_completion),
        "default": (CompletionRequest, create_completion),
    },
    "embed": {
        "messages": (EmbeddingChatRequest, create_embedding),
        "default": (EmbeddingCompletionRequest, create_embedding),
    },
    "score": {
850
851
852
853
        "default": (RerankRequest, do_rerank)
    },
    "rerank": {
        "default": (RerankRequest, do_rerank)
854
855
856
857
858
859
860
861
862
863
864
    },
    "reward": {
        "messages": (PoolingChatRequest, create_pooling),
        "default": (PoolingCompletionRequest, create_pooling),
    },
    "classify": {
        "messages": (PoolingChatRequest, create_pooling),
        "default": (PoolingCompletionRequest, create_pooling),
    },
}

865
866
if envs.VLLM_SERVER_DEV_MODE:

867
868
869
870
871
    @router.get("/server_info")
    async def show_server_info(raw_request: Request):
        server_info = {"vllm_config": str(raw_request.app.state.vllm_config)}
        return JSONResponse(content=server_info)

872
873
874
875
876
877
    @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.
        """
878
879
880
881
882
883
        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)
884
885
        return Response(status_code=200)

886
887
888
889
890
891
892
893
894
895
896
    @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):
897
898
899
900
901
902
        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)
903
904
905
906
        # 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)

907
908
909
910
911
912
    @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})

913

914
915
916
917
918
919
920
921
922
923
924
925
926
@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
                 },
             })
927
928
929
930
async def invocations(raw_request: Request):
    """
    For SageMaker, routes requests to other handlers based on model `task`.
    """
931
932
    try:
        body = await raw_request.json()
933
    except json.JSONDecodeError as e:
934
935
936
        raise HTTPException(status_code=HTTPStatus.BAD_REQUEST.value,
                            detail=f"JSON decode error: {e}") from e

937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
    task = raw_request.app.state.task

    if task not in TASK_HANDLERS:
        raise HTTPException(
            status_code=400,
            detail=f"Unsupported task: '{task}' for '/invocations'. "
            f"Expected one of {set(TASK_HANDLERS.keys())}")

    handler_config = TASK_HANDLERS[task]
    if "messages" in body:
        request_model, handler = handler_config["messages"]
    else:
        request_model, handler = handler_config["default"]

    # this is required since we lose the FastAPI automatic casting
    request = request_model.model_validate(body)
    return await handler(request, raw_request)


956
957
958
959
960
961
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")
962
    async def start_profile(raw_request: Request):
963
        logger.info("Starting profiler...")
964
        await engine_client(raw_request).start_profile()
965
966
967
968
        logger.info("Profiler started.")
        return Response(status_code=200)

    @router.post("/stop_profile")
969
    async def stop_profile(raw_request: Request):
970
        logger.info("Stopping profiler...")
971
        await engine_client(raw_request).stop_profile()
972
973
974
975
        logger.info("Profiler stopped.")
        return Response(status_code=200)


976
977
if envs.VLLM_ALLOW_RUNTIME_LORA_UPDATING:
    logger.warning(
978
        "LoRA dynamic loading & unloading is enabled in the API server. "
979
980
        "This should ONLY be used for local development!")

981
982
    @router.post("/v1/load_lora_adapter",
                 dependencies=[Depends(validate_json_request)])
983
    async def load_lora_adapter(request: LoadLoRAAdapterRequest,
984
                                raw_request: Request):
985
986
987
988
989
        handler = models(raw_request)
        response = await handler.load_lora_adapter(request)
        if isinstance(response, ErrorResponse):
            return JSONResponse(content=response.model_dump(),
                                status_code=response.code)
990
991
992

        return Response(status_code=200, content=response)

993
994
    @router.post("/v1/unload_lora_adapter",
                 dependencies=[Depends(validate_json_request)])
995
    async def unload_lora_adapter(request: UnloadLoRAAdapterRequest,
996
                                  raw_request: Request):
997
998
999
1000
1001
        handler = models(raw_request)
        response = await handler.unload_lora_adapter(request)
        if isinstance(response, ErrorResponse):
            return JSONResponse(content=response.model_dump(),
                                status_code=response.code)
1002
1003
1004
1005

        return Response(status_code=200, content=response)


1006
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
1017
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


1018
def build_app(args: Namespace) -> FastAPI:
1019
1020
1021
1022
1023
1024
1025
    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
1026
1027
    app.include_router(router)
    app.root_path = args.root_path
Zhuohan Li's avatar
Zhuohan Li committed
1028

1029
1030
    mount_metrics(app)

Zhuohan Li's avatar
Zhuohan Li committed
1031
1032
1033
1034
1035
1036
1037
1038
    app.add_middleware(
        CORSMiddleware,
        allow_origins=args.allowed_origins,
        allow_credentials=args.allow_credentials,
        allow_methods=args.allowed_methods,
        allow_headers=args.allowed_headers,
    )

1039
1040
1041
1042
1043
1044
1045
    @app.exception_handler(HTTPException)
    async def http_exception_handler(_: Request, exc: HTTPException):
        err = ErrorResponse(message=exc.detail,
                            type=HTTPStatus(exc.status_code).phrase,
                            code=exc.status_code)
        return JSONResponse(err.model_dump(), status_code=exc.status_code)

Ethan Xu's avatar
Ethan Xu committed
1046
    @app.exception_handler(RequestValidationError)
1047
1048
1049
1050
1051
1052
1053
1054
1055
1056
1057
1058
    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

        err = ErrorResponse(message=message,
                            type=HTTPStatus.BAD_REQUEST.phrase,
1059
                            code=HTTPStatus.BAD_REQUEST)
Ethan Xu's avatar
Ethan Xu committed
1060
1061
1062
        return JSONResponse(err.model_dump(),
                            status_code=HTTPStatus.BAD_REQUEST)

1063
1064
    # Ensure --api-key option from CLI takes precedence over VLLM_API_KEY
    if token := args.api_key or envs.VLLM_API_KEY:
1065
1066
1067

        @app.middleware("http")
        async def authentication(request: Request, call_next):
1068
1069
            if request.method == "OPTIONS":
                return await call_next(request)
1070
1071
1072
1073
            url_path = request.url.path
            if app.root_path and url_path.startswith(app.root_path):
                url_path = url_path[len(app.root_path):]
            if not url_path.startswith("/v1"):
1074
1075
1076
1077
1078
1079
                return await call_next(request)
            if request.headers.get("Authorization") != "Bearer " + token:
                return JSONResponse(content={"error": "Unauthorized"},
                                    status_code=401)
            return await call_next(request)

1080
1081
1082
1083
1084
1085
1086
1087
1088
1089
1090
1091
    if args.enable_request_id_headers:
        logger.warning(
            "CAUTION: Enabling X-Request-Id headers in the API Server. "
            "This can harm performance at high QPS.")

        @app.middleware("http")
        async def add_request_id(request: Request, call_next):
            request_id = request.headers.get(
                "X-Request-Id") or uuid.uuid4().hex
            response = await call_next(request)
            response.headers["X-Request-Id"] = request_id
            return response
1092
1093
1094
1095
1096
1097
1098
1099
1100
1101
1102
1103
1104

    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))
1105
1106
            logger.info("response_body={%s}",
                        response_body[0].decode() if response_body else None)
1107
            return response
1108

1109
1110
1111
1112
    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):
1113
            app.add_middleware(imported)  # type: ignore[arg-type]
1114
1115
1116
        elif inspect.iscoroutinefunction(imported):
            app.middleware("http")(imported)
        else:
1117
1118
            raise ValueError(f"Invalid middleware {middleware}. "
                             f"Must be a function or a class.")
1119

Ethan Xu's avatar
Ethan Xu committed
1120
1121
1122
    return app


1123
async def init_app_state(
1124
    engine_client: EngineClient,
1125
    vllm_config: VllmConfig,
1126
    state: State,
1127
    args: Namespace,
1128
) -> None:
1129
    if args.served_model_name is not None:
1130
        served_model_names = args.served_model_name
1131
    else:
1132
        served_model_names = [args.model]
1133

1134
1135
1136
1137
1138
    if args.disable_log_requests:
        request_logger = None
    else:
        request_logger = RequestLogger(max_log_len=args.max_log_len)

1139
1140
1141
1142
1143
    base_model_paths = [
        BaseModelPath(name=name, model_path=args.model)
        for name in served_model_names
    ]

1144
    state.engine_client = engine_client
1145
    state.log_stats = not args.disable_log_stats
1146
1147
    state.vllm_config = vllm_config
    model_config = vllm_config.model_config
Ethan Xu's avatar
Ethan Xu committed
1148

1149
    resolved_chat_template = load_chat_template(args.chat_template)
1150
    if resolved_chat_template is not None:
1151
1152
1153
1154
1155
1156
1157
1158
1159
        # 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(
1160
                tokenizer=tokenizer,
1161
1162
                chat_template=None,
                tools=None,
1163
                model_config=vllm_config.model_config,
1164
            )
1165
1166
1167
1168
1169
1170
1171

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

1173
    state.openai_serving_models = OpenAIServingModels(
1174
        engine_client=engine_client,
1175
1176
1177
1178
1179
        model_config=model_config,
        base_model_paths=base_model_paths,
        lora_modules=args.lora_modules,
        prompt_adapters=args.prompt_adapters,
    )
1180
    await state.openai_serving_models.init_static_loras()
1181
    state.openai_serving_chat = OpenAIServingChat(
1182
        engine_client,
1183
        model_config,
1184
        state.openai_serving_models,
1185
1186
        args.response_role,
        request_logger=request_logger,
1187
1188
        chat_template=resolved_chat_template,
        chat_template_content_format=args.chat_template_content_format,
1189
        return_tokens_as_token_ids=args.return_tokens_as_token_ids,
1190
        enable_auto_tools=args.enable_auto_tool_choice,
1191
        tool_parser=args.tool_call_parser,
1192
        reasoning_parser=args.reasoning_parser,
1193
        enable_prompt_tokens_details=args.enable_prompt_tokens_details,
1194
    ) if model_config.runner_type == "generate" else None
1195
    state.openai_serving_completion = OpenAIServingCompletion(
1196
        engine_client,
1197
        model_config,
1198
        state.openai_serving_models,
1199
        request_logger=request_logger,
1200
        return_tokens_as_token_ids=args.return_tokens_as_token_ids,
1201
    ) if model_config.runner_type == "generate" else None
1202
    state.openai_serving_pooling = OpenAIServingPooling(
1203
        engine_client,
1204
        model_config,
1205
        state.openai_serving_models,
1206
        request_logger=request_logger,
1207
1208
        chat_template=resolved_chat_template,
        chat_template_content_format=args.chat_template_content_format,
1209
    ) if model_config.runner_type == "pooling" else None
1210
1211
1212
    state.openai_serving_embedding = OpenAIServingEmbedding(
        engine_client,
        model_config,
1213
        state.openai_serving_models,
1214
1215
1216
1217
        request_logger=request_logger,
        chat_template=resolved_chat_template,
        chat_template_content_format=args.chat_template_content_format,
    ) if model_config.task == "embed" else None
1218
    state.openai_serving_scores = ServingScores(
1219
1220
        engine_client,
        model_config,
1221
        state.openai_serving_models,
1222
1223
        request_logger=request_logger) if model_config.task in (
            "score", "embed", "pooling") else None
1224
1225
1226
1227
1228
1229
    state.openai_serving_classification = ServingClassification(
        engine_client,
        model_config,
        state.openai_serving_models,
        request_logger=request_logger,
    ) if model_config.task == "classify" else None
1230
    state.jinaai_serving_reranking = ServingScores(
1231
1232
1233
1234
1235
        engine_client,
        model_config,
        state.openai_serving_models,
        request_logger=request_logger
    ) if model_config.task == "score" else None
1236
    state.openai_serving_tokenization = OpenAIServingTokenization(
1237
        engine_client,
1238
        model_config,
1239
        state.openai_serving_models,
1240
        request_logger=request_logger,
1241
1242
        chat_template=resolved_chat_template,
        chat_template_content_format=args.chat_template_content_format,
1243
    )
1244
1245
1246
1247
1248
1249
    state.openai_serving_transcription = OpenAIServingTranscription(
        engine_client,
        model_config,
        state.openai_serving_models,
        request_logger=request_logger,
    ) if model_config.runner_type == "transcription" else None
1250
    state.task = model_config.task
1251

1252
1253
1254
    state.enable_server_load_tracking = args.enable_server_load_tracking
    state.server_load_metrics = 0

1255

1256
def create_server_socket(addr: tuple[str, int]) -> socket.socket:
1257
1258
1259
1260
1261
1262
    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)
1263
    sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, 1)
1264
1265
1266
1267
1268
    sock.bind(addr)

    return sock


1269
def validate_api_server_args(args):
1270
    valid_tool_parses = ToolParserManager.tool_parsers.keys()
1271
    if args.enable_auto_tool_choice \
1272
            and args.tool_call_parser not in valid_tool_parses:
1273
        raise KeyError(f"invalid tool call parser: {args.tool_call_parser} "
1274
                       f"(chose from {{ {','.join(valid_tool_parses)} }})")
1275

1276
    valid_reasoning_parses = ReasoningParserManager.reasoning_parsers.keys()
1277
    if args.reasoning_parser \
1278
1279
1280
1281
1282
        and args.reasoning_parser not in valid_reasoning_parses:
        raise KeyError(
            f"invalid reasoning parser: {args.reasoning_parser} "
            f"(chose from {{ {','.join(valid_reasoning_parses)} }})")

1283
1284
1285
1286
1287
1288
1289
1290
1291
1292
1293
1294
1295

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)

1296
1297
1298
    # 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
1299
1300
    sock_addr = (args.host or "", args.port)
    sock = create_server_socket(sock_addr)
1301

1302
1303
1304
1305
    # workaround to avoid footguns where uvicorn drops requests with too
    # many concurrent requests active
    set_ulimit()

1306
1307
1308
1309
1310
1311
    def signal_handler(*_) -> None:
        # Interrupt server on sigterm while initializing
        raise KeyboardInterrupt("terminated")

    signal.signal(signal.SIGTERM, signal_handler)

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

    return listen_address, sock


async def run_server(args, **uvicorn_kwargs) -> None:
    """Run a single-worker API server."""
    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)

    server_index = client_config.get("client_index", 0) if client_config else 0

1339
1340
1341
1342
1343
    # 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

1344
    async with build_async_engine_client(args, client_config) as engine_client:
1345
1346
        app = build_app(args)

1347
1348
        vllm_config = await engine_client.get_vllm_config()
        await init_app_state(engine_client, vllm_config, app.state, args)
1349

1350
1351
        logger.info("Starting vLLM API server %d on %s", server_index,
                    listen_address)
1352
1353
        shutdown_task = await serve_http(
            app,
1354
            sock=sock,
1355
            enable_ssl_refresh=args.enable_ssl_refresh,
1356
1357
1358
            host=args.host,
            port=args.port,
            log_level=args.uvicorn_log_level,
1359
1360
1361
            # NOTE: When the 'disable_uvicorn_access_log' value is True,
            # no access log will be output.
            access_log=not args.disable_uvicorn_access_log,
1362
1363
1364
1365
1366
            timeout_keep_alive=TIMEOUT_KEEP_ALIVE,
            ssl_keyfile=args.ssl_keyfile,
            ssl_certfile=args.ssl_certfile,
            ssl_ca_certs=args.ssl_ca_certs,
            ssl_cert_reqs=args.ssl_cert_reqs,
1367
1368
1369
            **uvicorn_kwargs,
        )

1370
    # NB: Await server shutdown only after the backend context is exited
1371
1372
1373
1374
    try:
        await shutdown_task
    finally:
        sock.close()
1375

Ethan Xu's avatar
Ethan Xu committed
1376
1377
1378

if __name__ == "__main__":
    # NOTE(simon):
1379
1380
    # This section should be in sync with vllm/entrypoints/cli/main.py for CLI
    # entrypoints.
1381
    cli_env_setup()
Ethan Xu's avatar
Ethan Xu committed
1382
1383
1384
1385
    parser = FlexibleArgumentParser(
        description="vLLM OpenAI-Compatible RESTful API server.")
    parser = make_arg_parser(parser)
    args = parser.parse_args()
1386
    validate_parsed_serve_args(args)
1387

1388
    uvloop.run(run_server(args))