api_server.py 25.9 KB
Newer Older
1
import asyncio
2
import atexit
3
4
import importlib
import inspect
5
import multiprocessing
6
import os
7
import re
8
import signal
9
import socket
10
import tempfile
11
import uuid
12
from argparse import Namespace
13
from contextlib import asynccontextmanager
14
from functools import partial
15
from http import HTTPStatus
16
from typing import AsyncIterator, Optional, Set, Tuple
17

18
import uvloop
19
from fastapi import APIRouter, FastAPI, Request
Zhuohan Li's avatar
Zhuohan Li committed
20
21
from fastapi.exceptions import RequestValidationError
from fastapi.middleware.cors import CORSMiddleware
22
from fastapi.responses import JSONResponse, Response, StreamingResponse
23
from starlette.datastructures import State
24
from starlette.routing import Mount
25
from typing_extensions import assert_never
Zhuohan Li's avatar
Zhuohan Li committed
26

27
import vllm.envs as envs
28
from vllm.config import ModelConfig
Woosuk Kwon's avatar
Woosuk Kwon committed
29
from vllm.engine.arg_utils import AsyncEngineArgs
30
31
32
from vllm.engine.multiprocessing.client import MQLLMEngineClient
from vllm.engine.multiprocessing.engine import run_mp_engine
from vllm.engine.protocol import EngineClient
33
from vllm.entrypoints.chat_utils import load_chat_template
34
from vllm.entrypoints.launcher import serve_http
35
from vllm.entrypoints.logger import RequestLogger
36
37
from vllm.entrypoints.openai.cli_args import (make_arg_parser,
                                              validate_parsed_serve_args)
38
39
# yapf conflicts with isort for this block
# yapf: disable
40
from vllm.entrypoints.openai.protocol import (ChatCompletionRequest,
41
                                              ChatCompletionResponse,
42
                                              CompletionRequest,
43
                                              CompletionResponse,
44
45
                                              DetokenizeRequest,
                                              DetokenizeResponse,
46
47
                                              EmbeddingRequest,
                                              EmbeddingResponse, ErrorResponse,
48
                                              LoadLoraAdapterRequest,
49
                                              ScoreRequest, ScoreResponse,
50
                                              TokenizeRequest,
51
52
53
                                              TokenizeResponse,
                                              UnloadLoraAdapterRequest)
# yapf: enable
54
55
from vllm.entrypoints.openai.serving_chat import OpenAIServingChat
from vllm.entrypoints.openai.serving_completion import OpenAIServingCompletion
56
from vllm.entrypoints.openai.serving_embedding import OpenAIServingEmbedding
57
from vllm.entrypoints.openai.serving_engine import BaseModelPath, OpenAIServing
58
from vllm.entrypoints.openai.serving_score import OpenAIServingScores
59
60
from vllm.entrypoints.openai.serving_tokenization import (
    OpenAIServingTokenization)
61
from vllm.entrypoints.openai.tool_parsers import ToolParserManager
62
from vllm.logger import init_logger
yhu422's avatar
yhu422 committed
63
from vllm.usage.usage_lib import UsageContext
64
65
from vllm.utils import (FlexibleArgumentParser, get_open_zmq_ipc_path,
                        is_valid_ipv6_address)
66
from vllm.version import __version__ as VLLM_VERSION
Zhuohan Li's avatar
Zhuohan Li committed
67

68
69
70
71
72
if envs.VLLM_USE_V1:
    from vllm.v1.engine.async_llm import AsyncLLMEngine  # type: ignore
else:
    from vllm.engine.async_llm_engine import AsyncLLMEngine  # type: ignore

73
TIMEOUT_KEEP_ALIVE = 5  # seconds
Zhuohan Li's avatar
Zhuohan Li committed
74

75
prometheus_multiproc_dir: tempfile.TemporaryDirectory
76

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

80
_running_tasks: Set[asyncio.Task] = set()
81

82

83
@asynccontextmanager
84
async def lifespan(app: FastAPI):
85
86
    try:
        if app.state.log_stats:
87
            engine_client: EngineClient = app.state.engine_client
88
89
90

            async def _force_log():
                while True:
91
92
                    await asyncio.sleep(10.)
                    await engine_client.do_log_stats()
93
94
95
96
97
98
99
100
101
102
103
104
105
106

            task = asyncio.create_task(_force_log())
            _running_tasks.add(task)
            task.add_done_callback(_running_tasks.remove)
        else:
            task = None
        try:
            yield
        finally:
            if task is not None:
                task.cancel()
    finally:
        # Ensure app state including engine ref is gc'd
        del app.state
107
108


109
@asynccontextmanager
110
async def build_async_engine_client(
111
        args: Namespace) -> AsyncIterator[EngineClient]:
112

113
    # Context manager to handle engine_client lifecycle
114
115
116
    # Ensures everything is shutdown and cleaned up on error/exit
    engine_args = AsyncEngineArgs.from_cli_args(args)

117
118
119
120
121
122
123
124
125
    async with build_async_engine_client_from_engine_args(
            engine_args, args.disable_frontend_multiprocessing) as engine:
        yield engine


@asynccontextmanager
async def build_async_engine_client_from_engine_args(
    engine_args: AsyncEngineArgs,
    disable_frontend_multiprocessing: bool = False,
126
) -> AsyncIterator[EngineClient]:
127
    """
128
    Create EngineClient, either:
129
130
131
132
133
134
        - in-process using the AsyncLLMEngine Directly
        - multiprocess using AsyncLLMEngine RPC

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

135
136
137
    # Fall back
    # TODO: fill out feature matrix.
    if (MQLLMEngineClient.is_unsupported_config(engine_args)
138
            or envs.VLLM_USE_V1 or disable_frontend_multiprocessing):
139
140
        engine_config = engine_args.create_engine_config(
            UsageContext.OPENAI_API_SERVER)
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
        uses_ray = getattr(AsyncLLMEngine._get_executor_cls(engine_config),
                           "uses_ray", False)

        build_engine = partial(AsyncLLMEngine.from_engine_args,
                               engine_args=engine_args,
                               engine_config=engine_config,
                               usage_context=UsageContext.OPENAI_API_SERVER)
        if uses_ray:
            # Must run in main thread with ray for its signal handlers to work
            engine_client = build_engine()
        else:
            engine_client = await asyncio.get_running_loop().run_in_executor(
                None, build_engine)

        yield engine_client
156
157
        if hasattr(engine_client, "shutdown"):
            engine_client.shutdown()
158
159
160
161
        return

    # Otherwise, use the multiprocessing AsyncLLMEngine.
    else:
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
        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.")

177
        # Select random path for IPC.
178
        ipc_path = get_open_zmq_ipc_path()
179
180
        logger.debug("Multiprocessing frontend to use %s for IPC Path.",
                     ipc_path)
181

182
        # Start RPCServer in separate process (holds the LLMEngine).
183
184
        # the current process might have CUDA context,
        # so we need to spawn a new process
185
186
        context = multiprocessing.get_context("spawn")

187
188
189
190
        # 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)
191
192
193
        engine_process = context.Process(target=run_mp_engine,
                                         args=(engine_args,
                                               UsageContext.OPENAI_API_SERVER,
194
                                               ipc_path, engine_alive))
195
        engine_process.start()
196
        engine_pid = engine_process.pid
197
        assert engine_pid is not None, "Engine process failed to start."
198
        logger.info("Started engine process with PID %d", engine_pid)
199

200
201
202
203
204
205
206
207
        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)

208
209
        # Build RPCClient, which conforms to EngineClient Protocol.
        engine_config = engine_args.create_engine_config()
210
211
212
213
        build_client = partial(MQLLMEngineClient, ipc_path, engine_config,
                               engine_pid)
        mq_engine_client = await asyncio.get_running_loop().run_in_executor(
            None, build_client)
214
        try:
215
216
            while True:
                try:
217
                    await mq_engine_client.setup()
218
                    break
219
                except TimeoutError:
220
221
                    if (not engine_process.is_alive()
                            or not engine_alive.value):
222
                        raise RuntimeError(
223
224
                            "Engine process failed to start. See stack "
                            "trace for the root cause.") from None
225

226
            yield mq_engine_client  # type: ignore[misc]
227
228
        finally:
            # Ensure rpc server process was terminated
229
            engine_process.terminate()
230
231

            # Close all open connections to the backend
232
            mq_engine_client.close()
233

234
235
236
237
238
            # 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()
239

240
241
242
243
244
            # 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
245
            multiprocess.mark_process_dead(engine_process.pid)
246

247

Ethan Xu's avatar
Ethan Xu committed
248
router = APIRouter()
Zhuohan Li's avatar
Zhuohan Li committed
249

250

251
def mount_metrics(app: FastAPI):
252
253
254
255
256
257
258
259
260
    # 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 (CollectorRegistry, make_asgi_app,
                                   multiprocess)

    prometheus_multiproc_dir_path = os.getenv("PROMETHEUS_MULTIPROC_DIR", None)
    if prometheus_multiproc_dir_path is not None:
261
262
        logger.debug("vLLM to use %s as PROMETHEUS_MULTIPROC_DIR",
                     prometheus_multiproc_dir_path)
263
264
265
266
267
268
269
270
271
        registry = CollectorRegistry()
        multiprocess.MultiProcessCollector(registry)

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

272
    # Workaround for 307 Redirect for /metrics
273
    metrics_route.path_regex = re.compile("^/metrics(?P<path>.*)$")
274
    app.routes.append(metrics_route)
275
276


277
278
279
280
281
282
def base(request: Request) -> OpenAIServing:
    # Reuse the existing instance
    return tokenization(request)


def chat(request: Request) -> Optional[OpenAIServingChat]:
283
284
285
    return request.app.state.openai_serving_chat


286
def completion(request: Request) -> Optional[OpenAIServingCompletion]:
287
288
289
    return request.app.state.openai_serving_completion


290
291
def embedding(request: Request) -> Optional[OpenAIServingEmbedding]:
    return request.app.state.openai_serving_embedding
292
293


294
295
296
297
def score(request: Request) -> Optional[OpenAIServingScores]:
    return request.app.state.openai_serving_scores


298
299
def tokenization(request: Request) -> OpenAIServingTokenization:
    return request.app.state.openai_serving_tokenization
300
301


302
def engine_client(request: Request) -> EngineClient:
303
304
305
    return request.app.state.engine_client


Ethan Xu's avatar
Ethan Xu committed
306
@router.get("/health")
307
async def health(raw_request: Request) -> Response:
308
    """Health check."""
309
    await engine_client(raw_request).check_health()
310
311
312
    return Response(status_code=200)


Ethan Xu's avatar
Ethan Xu committed
313
@router.post("/tokenize")
314
async def tokenize(request: TokenizeRequest, raw_request: Request):
315
316
    handler = tokenization(raw_request)

317
    generator = await handler.create_tokenize(request, raw_request)
318
319
320
    if isinstance(generator, ErrorResponse):
        return JSONResponse(content=generator.model_dump(),
                            status_code=generator.code)
321
    elif isinstance(generator, TokenizeResponse):
322
323
        return JSONResponse(content=generator.model_dump())

324
325
    assert_never(generator)

326

Ethan Xu's avatar
Ethan Xu committed
327
@router.post("/detokenize")
328
async def detokenize(request: DetokenizeRequest, raw_request: Request):
329
330
    handler = tokenization(raw_request)

331
    generator = await handler.create_detokenize(request, raw_request)
332
333
334
    if isinstance(generator, ErrorResponse):
        return JSONResponse(content=generator.model_dump(),
                            status_code=generator.code)
335
    elif isinstance(generator, DetokenizeResponse):
336
337
        return JSONResponse(content=generator.model_dump())

338
339
    assert_never(generator)

340

Ethan Xu's avatar
Ethan Xu committed
341
@router.get("/v1/models")
342
async def show_available_models(raw_request: Request):
343
344
345
    handler = base(raw_request)

    models = await handler.show_available_models()
346
    return JSONResponse(content=models.model_dump())
Zhuohan Li's avatar
Zhuohan Li committed
347
348


Ethan Xu's avatar
Ethan Xu committed
349
@router.get("/version")
350
async def show_version():
351
    ver = {"version": VLLM_VERSION}
352
353
354
    return JSONResponse(content=ver)


Ethan Xu's avatar
Ethan Xu committed
355
@router.post("/v1/chat/completions")
356
357
async def create_chat_completion(request: ChatCompletionRequest,
                                 raw_request: Request):
358
359
360
361
    handler = chat(raw_request)
    if handler is None:
        return base(raw_request).create_error_response(
            message="The model does not support Chat Completions API")
362

363
    generator = await handler.create_chat_completion(request, raw_request)
364

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

369
    elif isinstance(generator, ChatCompletionResponse):
370
        return JSONResponse(content=generator.model_dump())
371

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

374

Ethan Xu's avatar
Ethan Xu committed
375
@router.post("/v1/completions")
376
async def create_completion(request: CompletionRequest, raw_request: Request):
377
378
379
380
381
382
    handler = completion(raw_request)
    if handler is None:
        return base(raw_request).create_error_response(
            message="The model does not support Completions API")

    generator = await handler.create_completion(request, raw_request)
383
384
385
    if isinstance(generator, ErrorResponse):
        return JSONResponse(content=generator.model_dump(),
                            status_code=generator.code)
386
    elif isinstance(generator, CompletionResponse):
387
        return JSONResponse(content=generator.model_dump())
Zhuohan Li's avatar
Zhuohan Li committed
388

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

Zhuohan Li's avatar
Zhuohan Li committed
391

Ethan Xu's avatar
Ethan Xu committed
392
@router.post("/v1/embeddings")
393
async def create_embedding(request: EmbeddingRequest, raw_request: Request):
394
395
396
397
398
399
    handler = embedding(raw_request)
    if handler is None:
        return base(raw_request).create_error_response(
            message="The model does not support Embeddings API")

    generator = await handler.create_embedding(request, raw_request)
400
401
402
    if isinstance(generator, ErrorResponse):
        return JSONResponse(content=generator.model_dump(),
                            status_code=generator.code)
403
    elif isinstance(generator, EmbeddingResponse):
404
405
        return JSONResponse(content=generator.model_dump())

406
407
    assert_never(generator)

408

409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
@router.post("/v1/score")
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)


426
427
428
429
430
431
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")
432
    async def start_profile(raw_request: Request):
433
        logger.info("Starting profiler...")
434
        await engine_client(raw_request).start_profile()
435
436
437
438
        logger.info("Profiler started.")
        return Response(status_code=200)

    @router.post("/stop_profile")
439
    async def stop_profile(raw_request: Request):
440
        logger.info("Stopping profiler...")
441
        await engine_client(raw_request).stop_profile()
442
443
444
445
        logger.info("Profiler stopped.")
        return Response(status_code=200)


446
447
448
449
450
451
if envs.VLLM_ALLOW_RUNTIME_LORA_UPDATING:
    logger.warning(
        "Lora dynamic loading & unloading is enabled in the API server. "
        "This should ONLY be used for local development!")

    @router.post("/v1/load_lora_adapter")
452
453
    async def load_lora_adapter(request: LoadLoraAdapterRequest,
                                raw_request: Request):
454
455
456
457
458
459
460
        for route in [chat, completion, embedding]:
            handler = route(raw_request)
            if handler is not None:
                response = await handler.load_lora_adapter(request)
                if isinstance(response, ErrorResponse):
                    return JSONResponse(content=response.model_dump(),
                                        status_code=response.code)
461
462
463
464

        return Response(status_code=200, content=response)

    @router.post("/v1/unload_lora_adapter")
465
466
    async def unload_lora_adapter(request: UnloadLoraAdapterRequest,
                                  raw_request: Request):
467
468
469
470
471
472
473
        for route in [chat, completion, embedding]:
            handler = route(raw_request)
            if handler is not None:
                response = await handler.unload_lora_adapter(request)
                if isinstance(response, ErrorResponse):
                    return JSONResponse(content=response.model_dump(),
                                        status_code=response.code)
474
475
476
477

        return Response(status_code=200, content=response)


478
def build_app(args: Namespace) -> FastAPI:
479
480
481
482
483
484
485
    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
486
487
    app.include_router(router)
    app.root_path = args.root_path
Zhuohan Li's avatar
Zhuohan Li committed
488

489
490
    mount_metrics(app)

Zhuohan Li's avatar
Zhuohan Li committed
491
492
493
494
495
496
497
498
    app.add_middleware(
        CORSMiddleware,
        allow_origins=args.allowed_origins,
        allow_credentials=args.allow_credentials,
        allow_methods=args.allowed_methods,
        allow_headers=args.allowed_headers,
    )

Ethan Xu's avatar
Ethan Xu committed
499
500
    @app.exception_handler(RequestValidationError)
    async def validation_exception_handler(_, exc):
501
502
503
        err = ErrorResponse(message=str(exc),
                            type="BadRequestError",
                            code=HTTPStatus.BAD_REQUEST)
Ethan Xu's avatar
Ethan Xu committed
504
505
506
        return JSONResponse(err.model_dump(),
                            status_code=HTTPStatus.BAD_REQUEST)

507
    if token := envs.VLLM_API_KEY or args.api_key:
508
509
510

        @app.middleware("http")
        async def authentication(request: Request, call_next):
511
512
            if request.method == "OPTIONS":
                return await call_next(request)
513
514
515
516
            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"):
517
518
519
520
521
522
                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)

523
524
525
526
527
528
529
    @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

530
531
532
533
534
535
536
537
    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):
            app.add_middleware(imported)
        elif inspect.iscoroutinefunction(imported):
            app.middleware("http")(imported)
        else:
538
539
            raise ValueError(f"Invalid middleware {middleware}. "
                             f"Must be a function or a class.")
540

Ethan Xu's avatar
Ethan Xu committed
541
542
543
    return app


544
def init_app_state(
545
    engine_client: EngineClient,
546
547
    model_config: ModelConfig,
    state: State,
548
    args: Namespace,
549
) -> None:
550
    if args.served_model_name is not None:
551
        served_model_names = args.served_model_name
552
    else:
553
        served_model_names = [args.model]
554

555
556
557
558
559
    if args.disable_log_requests:
        request_logger = None
    else:
        request_logger = RequestLogger(max_log_len=args.max_log_len)

560
561
562
563
564
    base_model_paths = [
        BaseModelPath(name=name, model_path=args.model)
        for name in served_model_names
    ]

565
    state.engine_client = engine_client
566
    state.log_stats = not args.disable_log_stats
Ethan Xu's avatar
Ethan Xu committed
567

568
569
570
    resolved_chat_template = load_chat_template(args.chat_template)
    logger.info("Using supplied chat template:\n%s", resolved_chat_template)

571
    state.openai_serving_chat = OpenAIServingChat(
572
        engine_client,
573
        model_config,
574
        base_model_paths,
575
576
577
578
        args.response_role,
        lora_modules=args.lora_modules,
        prompt_adapters=args.prompt_adapters,
        request_logger=request_logger,
579
580
        chat_template=resolved_chat_template,
        chat_template_content_format=args.chat_template_content_format,
581
        return_tokens_as_token_ids=args.return_tokens_as_token_ids,
582
        enable_auto_tools=args.enable_auto_tool_choice,
583
        tool_parser=args.tool_call_parser,
584
        enable_prompt_tokens_details=args.enable_prompt_tokens_details,
585
    ) if model_config.runner_type == "generate" else None
586
    state.openai_serving_completion = OpenAIServingCompletion(
587
        engine_client,
588
        model_config,
589
        base_model_paths,
590
591
592
        lora_modules=args.lora_modules,
        prompt_adapters=args.prompt_adapters,
        request_logger=request_logger,
593
        return_tokens_as_token_ids=args.return_tokens_as_token_ids,
594
    ) if model_config.runner_type == "generate" else None
595
    state.openai_serving_embedding = OpenAIServingEmbedding(
596
        engine_client,
597
        model_config,
598
        base_model_paths,
599
        request_logger=request_logger,
600
601
        chat_template=resolved_chat_template,
        chat_template_content_format=args.chat_template_content_format,
602
    ) if model_config.runner_type == "pooling" else None
603
604
605
606
607
    state.openai_serving_scores = OpenAIServingScores(
        engine_client,
        model_config,
        base_model_paths,
        request_logger=request_logger
608
    ) if (model_config.runner_type == "pooling" \
609
          and model_config.is_cross_encoder) else None
610
    state.openai_serving_tokenization = OpenAIServingTokenization(
611
        engine_client,
612
        model_config,
613
        base_model_paths,
614
615
        lora_modules=args.lora_modules,
        request_logger=request_logger,
616
617
        chat_template=resolved_chat_template,
        chat_template_content_format=args.chat_template_content_format,
618
    )
619
620


621
622
623
624
625
626
627
628
629
630
631
632
def create_server_socket(addr: Tuple[str, int]) -> socket.socket:
    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)
    sock.bind(addr)

    return sock


633
async def run_server(args, **uvicorn_kwargs) -> None:
634
635
636
    logger.info("vLLM API server version %s", VLLM_VERSION)
    logger.info("args: %s", args)

637
638
639
640
641
642
643
644
645
    if args.tool_parser_plugin and len(args.tool_parser_plugin) > 3:
        ToolParserManager.import_tool_parser(args.tool_parser_plugin)

    valide_tool_parses = ToolParserManager.tool_parsers.keys()
    if args.enable_auto_tool_choice \
        and args.tool_call_parser not in valide_tool_parses:
        raise KeyError(f"invalid tool call parser: {args.tool_call_parser} "
                       f"(chose from {{ {','.join(valide_tool_parses)} }})")

646
647
648
    # 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
649
650
    sock_addr = (args.host or "", args.port)
    sock = create_server_socket(sock_addr)
651

652
653
654
655
656
657
    def signal_handler(*_) -> None:
        # Interrupt server on sigterm while initializing
        raise KeyboardInterrupt("terminated")

    signal.signal(signal.SIGTERM, signal_handler)

658
    async with build_async_engine_client(args) as engine_client:
659
660
        app = build_app(args)

661
662
        model_config = await engine_client.get_model_config()
        init_app_state(engine_client, model_config, app.state, args)
663
664
665
666
667
668
669
670
671
672
673

        shutdown_task = await serve_http(
            app,
            host=args.host,
            port=args.port,
            log_level=args.uvicorn_log_level,
            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,
674
675
676
            **uvicorn_kwargs,
        )

677
678
    # NB: Await server shutdown only after the backend context is exited
    await shutdown_task
679

680
681
    sock.close()

Ethan Xu's avatar
Ethan Xu committed
682
683
684
685
686
687
688
689

if __name__ == "__main__":
    # NOTE(simon):
    # This section should be in sync with vllm/scripts.py for CLI entrypoints.
    parser = FlexibleArgumentParser(
        description="vLLM OpenAI-Compatible RESTful API server.")
    parser = make_arg_parser(parser)
    args = parser.parse_args()
690
    validate_parsed_serve_args(args)
691

692
    uvloop.run(run_server(args))