http_server.py 33.7 KB
Newer Older
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
# Copyright 2023-2024 SGLang Team
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# ==============================================================================
"""
The entry point of inference server. (SRT = SGLang Runtime)

Wang Ran (汪然)'s avatar
Wang Ran (汪然) committed
17
This file implements HTTP APIs for the inference engine via fastapi.
18
19
20
21
"""

import asyncio
import dataclasses
22
import json
23
24
25
26
27
28
import logging
import multiprocessing as multiprocessing
import os
import threading
import time
from http import HTTPStatus
Lianmin Zheng's avatar
Lianmin Zheng committed
29
from typing import AsyncIterator, Callable, Dict, Optional
30
31
32
33

# Fix a bug of Python threading
setattr(threading, "_register_atexit", lambda *args, **kwargs: None)

34
35
36
from contextlib import asynccontextmanager

import numpy as np
37
38
39
40
import orjson
import requests
import uvicorn
import uvloop
41
42
from fastapi import Depends, FastAPI, Request, UploadFile
from fastapi.exceptions import RequestValidationError
43
44
45
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import ORJSONResponse, Response, StreamingResponse

46
from sglang.srt.disaggregation.utils import (
Byron Hsu's avatar
Byron Hsu committed
47
    FAKE_BOOTSTRAP_HOST,
48
49
    register_disaggregation_server,
)
50
from sglang.srt.entrypoints.engine import _launch_subprocesses
51
52
53
54
55
56
57
58
59
60
61
62
63
64
from sglang.srt.entrypoints.openai.protocol import (
    ChatCompletionRequest,
    CompletionRequest,
    EmbeddingRequest,
    ModelCard,
    ModelList,
    ScoringRequest,
    V1RerankReqInput,
)
from sglang.srt.entrypoints.openai.serving_chat import OpenAIServingChat
from sglang.srt.entrypoints.openai.serving_completions import OpenAIServingCompletion
from sglang.srt.entrypoints.openai.serving_embedding import OpenAIServingEmbedding
from sglang.srt.entrypoints.openai.serving_rerank import OpenAIServingRerank
from sglang.srt.entrypoints.openai.serving_score import OpenAIServingScore
65
from sglang.srt.function_call.function_call_parser import FunctionCallParser
66
from sglang.srt.managers.io_struct import (
Lianmin Zheng's avatar
Lianmin Zheng committed
67
    AbortReq,
68
69
70
71
72
73
74
    CloseSessionReqInput,
    ConfigureLoggingReq,
    EmbeddingReqInput,
    GenerateReqInput,
    GetWeightsByNameReqInput,
    InitWeightsUpdateGroupReqInput,
    OpenSessionReqInput,
75
    ParseFunctionCallReq,
76
    ProfileReqInput,
77
78
    ReleaseMemoryOccupationReqInput,
    ResumeMemoryOccupationReqInput,
Xihuai Wang's avatar
Xihuai Wang committed
79
    SeparateReasoningReqInput,
80
    SetInternalStateReq,
81
    SlowDownReqInput,
82
83
    UpdateWeightFromDiskReqInput,
    UpdateWeightsFromDistributedReqInput,
84
    UpdateWeightsFromTensorReqInput,
85
    VertexGenerateReqInput,
86
)
87
from sglang.srt.managers.template_manager import TemplateManager
88
from sglang.srt.managers.tokenizer_manager import TokenizerManager
89
from sglang.srt.metrics.func_timer import enable_func_timer
Xihuai Wang's avatar
Xihuai Wang committed
90
from sglang.srt.reasoning_parser import ReasoningParser
91
92
93
94
95
from sglang.srt.server_args import ServerArgs
from sglang.srt.utils import (
    add_api_key_middleware,
    add_prometheus_middleware,
    delete_directory,
96
    get_bool_env_var,
97
98
99
    kill_process_tree,
    set_uvicorn_logging_configs,
)
100
from sglang.srt.warmup import execute_warmups
101
102
103
104
105
106
107
108
109
110
from sglang.utils import get_exception_traceback
from sglang.version import __version__

logger = logging.getLogger(__name__)
asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())


# Store global states
@dataclasses.dataclass
class _GlobalState:
111
    tokenizer_manager: TokenizerManager
112
    template_manager: TemplateManager
113
114
115
116
117
118
119
120
121
122
123
    scheduler_info: Dict


_global_state: Optional[_GlobalState] = None


def set_global_state(global_state: _GlobalState):
    global _global_state
    _global_state = global_state


124
125
126
@asynccontextmanager
async def lifespan(fast_api_app: FastAPI):
    server_args: ServerArgs = fast_api_app.server_args
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144

    # Initialize OpenAI serving handlers
    fast_api_app.state.openai_serving_completion = OpenAIServingCompletion(
        _global_state.tokenizer_manager, _global_state.template_manager
    )
    fast_api_app.state.openai_serving_chat = OpenAIServingChat(
        _global_state.tokenizer_manager, _global_state.template_manager
    )
    fast_api_app.state.openai_serving_embedding = OpenAIServingEmbedding(
        _global_state.tokenizer_manager, _global_state.template_manager
    )
    fast_api_app.state.openai_serving_score = OpenAIServingScore(
        _global_state.tokenizer_manager
    )
    fast_api_app.state.openai_serving_rerank = OpenAIServingRerank(
        _global_state.tokenizer_manager
    )

145
146
147
148
149
150
151
152
153
154
155
156
157
    if server_args.warmups is not None:
        await execute_warmups(
            server_args.warmups.split(","), _global_state.tokenizer_manager
        )
        logger.info("Warmup ended")

    warmup_thread = getattr(fast_api_app, "warmup_thread", None)
    if warmup_thread is not None:
        warmup_thread.start()
    yield


# Fast API
158
159
160
161
app = FastAPI(
    lifespan=lifespan,
    openapi_url=None if get_bool_env_var("DISABLE_OPENAPI_DOC") else "/openapi.json",
)
162
163
164
165
166
167
168
169
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199

# Custom exception handlers to change validation error status codes
@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    """Override FastAPI's default 422 validation error with 400"""
    return ORJSONResponse(
        status_code=400,
        content={
            "detail": exc.errors(),
            "body": exc.body,
        },
    )


async def validate_json_request(raw_request: Request):
    """Validate that the request content-type is application/json."""
    content_type = raw_request.headers.get("content-type", "").lower()
    media_type = content_type.split(";", maxsplit=1)[0]
    if media_type != "application/json":
        raise RequestValidationError(
            errors=[
                {
                    "loc": ["header", "content-type"],
                    "msg": "Unsupported Media Type: Only 'application/json' is allowed",
                    "type": "value_error",
                }
            ]
        )


200
201
202
HEALTH_CHECK_TIMEOUT = int(os.getenv("SGLANG_HEALTH_CHECK_TIMEOUT", 20))


203
204
205
206
207
208
209
210
211
212
213
214
215
##### Native API endpoints #####


@app.get("/health")
async def health() -> Response:
    """Check the health of the http server."""
    return Response(status_code=200)


@app.get("/health_generate")
async def health_generate(request: Request) -> Response:
    """Check the health of the inference server by generating one token."""

216
217
    sampling_params = {"max_new_tokens": 1, "temperature": 0.0}
    rid = f"HEALTH_CHECK_{time.time()}"
218

219
220
221
    if _global_state.tokenizer_manager.is_image_gen:
        raise NotImplementedError()
    elif _global_state.tokenizer_manager.is_generation:
222
        gri = GenerateReqInput(
223
224
225
226
            rid=rid,
            input_ids=[0],
            sampling_params=sampling_params,
            log_metrics=False,
227
228
229
        )
    else:
        gri = EmbeddingReqInput(
230
            rid=rid, input_ids=[0], sampling_params=sampling_params, log_metrics=False
231
232
        )

233
    async def gen():
234
        async for _ in _global_state.tokenizer_manager.generate_request(gri, request):
235
            break
236

237
    tic = time.perf_counter()
238
    task = asyncio.create_task(gen())
239
    while time.perf_counter() < tic + HEALTH_CHECK_TIMEOUT:
240
241
242
243
        await asyncio.sleep(1)
        if _global_state.tokenizer_manager.last_receive_tstamp > tic:
            task.cancel()
            _global_state.tokenizer_manager.rid_to_state.pop(rid, None)
244
            _global_state.tokenizer_manager.health_check_failed = False
245
246
247
248
249
250
251
252
253
254
255
256
257
            return Response(status_code=200)

    task.cancel()
    tic_time = time.strftime("%H:%M:%S", time.localtime(tic))
    last_receive_time = time.strftime(
        "%H:%M:%S", time.localtime(_global_state.tokenizer_manager.last_receive_tstamp)
    )
    logger.error(
        f"Health check failed. Server couldn't get a response from detokenizer for last "
        f"{HEALTH_CHECK_TIMEOUT} seconds. tic start time: {tic_time}. "
        f"last_heartbeat time: {last_receive_time}"
    )
    _global_state.tokenizer_manager.rid_to_state.pop(rid, None)
258
    _global_state.tokenizer_manager.health_check_failed = True
259
    return Response(status_code=503)
260
261
262
263
264
265


@app.get("/get_model_info")
async def get_model_info():
    """Get the model information."""
    result = {
266
267
268
        "model_path": _global_state.tokenizer_manager.model_path,
        "tokenizer_path": _global_state.tokenizer_manager.server_args.tokenizer_path,
        "is_generation": _global_state.tokenizer_manager.is_generation,
269
270
271
272
273
274
    }
    return result


@app.get("/get_server_info")
async def get_server_info():
275
    internal_states = await _global_state.tokenizer_manager.get_internal_state()
276
    return {
277
        **dataclasses.asdict(_global_state.tokenizer_manager.server_args),
278
        **_global_state.scheduler_info,
279
        "internal_states": internal_states,
280
281
282
283
        "version": __version__,
    }


Liangsheng Yin's avatar
Liangsheng Yin committed
284
285
286
287
288
@app.get("/get_load")
async def get_load():
    return await _global_state.tokenizer_manager.get_load()


289
290
291
292
293
294
@app.api_route("/set_internal_state", methods=["POST", "PUT"])
async def set_internal_state(obj: SetInternalStateReq, request: Request):
    res = await _global_state.tokenizer_manager.set_internal_state(obj)
    return res


295
296
297
298
299
300
301
302
# fastapi implicitly converts json in the request to obj (dataclass)
@app.api_route("/generate", methods=["POST", "PUT"])
async def generate_request(obj: GenerateReqInput, request: Request):
    """Handle a generate request."""
    if obj.stream:

        async def stream_results() -> AsyncIterator[bytes]:
            try:
303
                async for out in _global_state.tokenizer_manager.generate_request(
304
305
306
307
308
309
310
                    obj, request
                ):
                    yield b"data: " + orjson.dumps(
                        out, option=orjson.OPT_NON_STR_KEYS
                    ) + b"\n\n"
            except ValueError as e:
                out = {"error": {"message": str(e)}}
311
                logger.error(f"[http_server] Error: {e}")
312
313
314
315
316
317
318
319
                yield b"data: " + orjson.dumps(
                    out, option=orjson.OPT_NON_STR_KEYS
                ) + b"\n\n"
            yield b"data: [DONE]\n\n"

        return StreamingResponse(
            stream_results(),
            media_type="text/event-stream",
320
            background=_global_state.tokenizer_manager.create_abort_task(obj),
321
322
323
        )
    else:
        try:
324
            ret = await _global_state.tokenizer_manager.generate_request(
325
326
327
328
                obj, request
            ).__anext__()
            return ret
        except ValueError as e:
329
            logger.error(f"[http_server] Error: {e}")
330
331
332
            return _create_error_response(e)


333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
@app.api_route("/generate_from_file", methods=["POST"])
async def generate_from_file_request(file: UploadFile, request: Request):
    """Handle a generate request, this is purely to work with input_embeds."""
    content = await file.read()
    input_embeds = json.loads(content.decode("utf-8"))

    obj = GenerateReqInput(
        input_embeds=input_embeds,
        sampling_params={
            "repetition_penalty": 1.2,
            "temperature": 0.2,
            "max_new_tokens": 512,
        },
    )

    try:
349
350
351
        ret = await _global_state.tokenizer_manager.generate_request(
            obj, request
        ).__anext__()
352
353
354
355
356
357
        return ret
    except ValueError as e:
        logger.error(f"Error: {e}")
        return _create_error_response(e)


358
359
360
361
@app.api_route("/encode", methods=["POST", "PUT"])
async def encode_request(obj: EmbeddingReqInput, request: Request):
    """Handle an embedding request."""
    try:
362
        ret = await _global_state.tokenizer_manager.generate_request(
363
364
365
366
367
368
369
370
371
372
373
            obj, request
        ).__anext__()
        return ret
    except ValueError as e:
        return _create_error_response(e)


@app.api_route("/classify", methods=["POST", "PUT"])
async def classify_request(obj: EmbeddingReqInput, request: Request):
    """Handle a reward model request. Now the arguments and return values are the same as embedding models."""
    try:
374
        ret = await _global_state.tokenizer_manager.generate_request(
375
376
377
378
379
380
381
            obj, request
        ).__anext__()
        return ret
    except ValueError as e:
        return _create_error_response(e)


382
383
384
385
386
387
388
389
@app.api_route(
    "/v1/rerank", methods=["POST", "PUT"], dependencies=[Depends(validate_json_request)]
)
async def v1_rerank_request(request: V1RerankReqInput, raw_request: Request):
    """Endpoint for reranking documents based on query relevance."""
    return await raw_request.app.state.openai_serving_rerank.handle_request(
        request, raw_request
    )
woodx's avatar
woodx committed
390
391


392
@app.api_route("/flush_cache", methods=["GET", "POST"])
393
394
async def flush_cache():
    """Flush the radix cache."""
395
    ret = await _global_state.tokenizer_manager.flush_cache()
396
397
398
    return Response(
        content="Cache flushed.\nPlease check backend logs for more details. "
        "(When there are running or waiting requests, the operation will not be performed.)\n",
399
        status_code=200 if ret.success else HTTPStatus.BAD_REQUEST,
400
401
402
403
    )


@app.api_route("/start_profile", methods=["GET", "POST"])
404
async def start_profile_async(obj: Optional[ProfileReqInput] = None):
405
    """Start profiling."""
406
407
408
409
    if obj is None:
        obj = ProfileReqInput()

    await _global_state.tokenizer_manager.start_profile(
410
411
412
413
414
        output_dir=obj.output_dir,
        num_steps=obj.num_steps,
        activities=obj.activities,
        with_stack=obj.with_stack,
        record_shapes=obj.record_shapes,
415
        profile_by_stage=obj.profile_by_stage,
416
    )
417
418
419
420
421
422
423
424
425
    return Response(
        content="Start profiling.\n",
        status_code=200,
    )


@app.api_route("/stop_profile", methods=["GET", "POST"])
async def stop_profile_async():
    """Stop profiling."""
426
    await _global_state.tokenizer_manager.stop_profile()
427
428
429
430
431
432
    return Response(
        content="Stop profiling. This will take some time.\n",
        status_code=200,
    )


433
434
435
@app.api_route("/start_expert_distribution_record", methods=["GET", "POST"])
async def start_expert_distribution_record_async():
    """Start recording the expert distribution. Clear the previous record if any."""
436
    await _global_state.tokenizer_manager.start_expert_distribution_record()
437
438
439
440
441
442
443
444
445
    return Response(
        content="Start recording the expert distribution.\n",
        status_code=200,
    )


@app.api_route("/stop_expert_distribution_record", methods=["GET", "POST"])
async def stop_expert_distribution_record_async():
    """Stop recording the expert distribution."""
446
    await _global_state.tokenizer_manager.stop_expert_distribution_record()
447
448
449
450
451
452
453
454
455
    return Response(
        content="Stop recording the expert distribution.\n",
        status_code=200,
    )


@app.api_route("/dump_expert_distribution_record", methods=["GET", "POST"])
async def dump_expert_distribution_record_async():
    """Dump expert distribution record."""
456
    await _global_state.tokenizer_manager.dump_expert_distribution_record()
457
458
459
460
461
462
    return Response(
        content="Dump expert distribution record.\n",
        status_code=200,
    )


463
464
@app.post("/update_weights_from_disk")
async def update_weights_from_disk(obj: UpdateWeightFromDiskReqInput, request: Request):
465
466
467
    """Update the weights from disk inplace without re-launching the server."""
    success, message, num_paused_requests = (
        await _global_state.tokenizer_manager.update_weights_from_disk(obj, request)
468
    )
469
470
471
472
473
    content = {
        "success": success,
        "message": message,
        "num_paused_requests": num_paused_requests,
    }
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
    if success:
        return ORJSONResponse(
            content,
            status_code=HTTPStatus.OK,
        )
    else:
        return ORJSONResponse(
            content,
            status_code=HTTPStatus.BAD_REQUEST,
        )


@app.post("/init_weights_update_group")
async def init_weights_update_group(
    obj: InitWeightsUpdateGroupReqInput, request: Request
):
    """Initialize the parameter update group."""
491
    success, message = await _global_state.tokenizer_manager.init_weights_update_group(
492
493
494
495
496
497
498
499
500
        obj, request
    )
    content = {"success": success, "message": message}
    if success:
        return ORJSONResponse(content, status_code=200)
    else:
        return ORJSONResponse(content, status_code=HTTPStatus.BAD_REQUEST)


501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
@app.post("/update_weights_from_tensor")
async def update_weights_from_tensor(
    obj: UpdateWeightsFromTensorReqInput, request: Request
):
    """Update the weights from tensor inplace without re-launching the server.
    Notes:
    1. Ensure that the model is on the correct device (e.g., GPU) before calling this endpoint. If the model is moved to the CPU unexpectedly, it may cause performance issues or runtime errors.
    2. HTTP will transmit only the metadata of the tensor, while the tensor itself will be directly copied to the model.
    3. Any binary data in the named tensors should be base64 encoded.
    """

    success, message = await _global_state.tokenizer_manager.update_weights_from_tensor(
        obj, request
    )
    content = {"success": success, "message": message}
    return ORJSONResponse(
        content, status_code=200 if success else HTTPStatus.BAD_REQUEST
    )


521
522
523
524
525
@app.post("/update_weights_from_distributed")
async def update_weights_from_distributed(
    obj: UpdateWeightsFromDistributedReqInput, request: Request
):
    """Update model parameter from distributed online."""
526
527
528
529
    success, message = (
        await _global_state.tokenizer_manager.update_weights_from_distributed(
            obj, request
        )
530
531
532
533
534
535
536
537
538
539
540
541
    )
    content = {"success": success, "message": message}
    if success:
        return ORJSONResponse(content, status_code=200)
    else:
        return ORJSONResponse(content, status_code=HTTPStatus.BAD_REQUEST)


@app.api_route("/get_weights_by_name", methods=["GET", "POST"])
async def get_weights_by_name(obj: GetWeightsByNameReqInput, request: Request):
    """Get model parameter by name."""
    try:
542
        ret = await _global_state.tokenizer_manager.get_weights_by_name(obj, request)
543
544
545
546
547
548
549
550
551
552
553
554
        if ret is None:
            return _create_error_response("Get parameter by name failed")
        else:
            return ORJSONResponse(ret, status_code=200)
    except Exception as e:
        return _create_error_response(e)


@app.api_route("/release_memory_occupation", methods=["GET", "POST"])
async def release_memory_occupation(
    obj: ReleaseMemoryOccupationReqInput, request: Request
):
555
    """Release GPU memory occupation temporarily."""
556
    try:
557
        await _global_state.tokenizer_manager.release_memory_occupation(obj, request)
558
559
560
561
562
563
564
565
    except Exception as e:
        return _create_error_response(e)


@app.api_route("/resume_memory_occupation", methods=["GET", "POST"])
async def resume_memory_occupation(
    obj: ResumeMemoryOccupationReqInput, request: Request
):
566
    """Resume GPU memory occupation."""
567
    try:
568
        await _global_state.tokenizer_manager.resume_memory_occupation(obj, request)
569
570
571
572
    except Exception as e:
        return _create_error_response(e)


573
574
575
576
577
578
579
580
581
582
583
584
585
@app.api_route("/slow_down", methods=["GET", "POST"])
async def slow_down(obj: SlowDownReqInput, request: Request):
    """Slow down the system deliberately. Only for testing. Example scenario:
    when we want to test performance of D in large-scale PD disaggregation and have no enough nodes for P,
    we can use this to slow down D to let it have enough running sequences, and then disable slowdown
    to let it run in full batch size.
    """
    try:
        await _global_state.tokenizer_manager.slow_down(obj, request)
    except Exception as e:
        return _create_error_response(e)


586
587
588
589
@app.api_route("/open_session", methods=["GET", "POST"])
async def open_session(obj: OpenSessionReqInput, request: Request):
    """Open a session, and return its unique session id."""
    try:
590
        session_id = await _global_state.tokenizer_manager.open_session(obj, request)
591
592
593
594
595
596
597
598
599
600
601
        if session_id is None:
            raise Exception(
                "Failed to open the session. Check if a session with the same id is still open."
            )
        return session_id
    except Exception as e:
        return _create_error_response(e)


@app.api_route("/close_session", methods=["GET", "POST"])
async def close_session(obj: CloseSessionReqInput, request: Request):
602
    """Close the session."""
603
    try:
604
        await _global_state.tokenizer_manager.close_session(obj, request)
605
606
607
608
609
610
611
        return Response(status_code=200)
    except Exception as e:
        return _create_error_response(e)


@app.api_route("/configure_logging", methods=["GET", "POST"])
async def configure_logging(obj: ConfigureLoggingReq, request: Request):
612
    """Configure the request logging options."""
613
    _global_state.tokenizer_manager.configure_logging(obj)
614
615
616
    return Response(status_code=200)


Lianmin Zheng's avatar
Lianmin Zheng committed
617
618
619
620
621
622
623
624
625
626
@app.post("/abort_request")
async def abort_request(obj: AbortReq, request: Request):
    """Abort a request."""
    try:
        _global_state.tokenizer_manager.abort_request(rid=obj.rid)
        return Response(status_code=200)
    except Exception as e:
        return _create_error_response(e)


627
628
@app.post("/parse_function_call")
async def parse_function_call_request(obj: ParseFunctionCallReq, request: Request):
YAMY's avatar
YAMY committed
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
    """
    A native API endpoint to parse function calls from a text.
    """
    # 1) Initialize the parser based on the request body
    parser = FunctionCallParser(tools=obj.tools, tool_call_parser=obj.tool_call_parser)

    # 2) Call the non-stream parsing method (non-stream)
    normal_text, calls = parser.parse_non_stream(obj.text)

    # 3) Organize the response content
    response_data = {
        "normal_text": normal_text,
        "calls": [
            call.model_dump() for call in calls
        ],  # Convert pydantic objects to dictionaries
    }

    return ORJSONResponse(content=response_data, status_code=200)


Xihuai Wang's avatar
Xihuai Wang committed
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
@app.post("/separate_reasoning")
async def separate_reasoning_request(obj: SeparateReasoningReqInput, request: Request):
    """
    A native API endpoint to separate reasoning from a text.
    """
    # 1) Initialize the parser based on the request body
    parser = ReasoningParser(model_type=obj.reasoning_parser)

    # 2) Call the non-stream parsing method (non-stream)
    reasoning_text, normal_text = parser.parse_non_stream(obj.text)

    # 3) Organize the response content
    response_data = {
        "reasoning_text": reasoning_text,
        "text": normal_text,
    }

    return ORJSONResponse(content=response_data, status_code=200)


669
670
671
##### OpenAI-compatible API endpoints #####


672
673
674
675
676
677
@app.post("/v1/completions", dependencies=[Depends(validate_json_request)])
async def openai_v1_completions(request: CompletionRequest, raw_request: Request):
    """OpenAI-compatible text completion endpoint."""
    return await raw_request.app.state.openai_serving_completion.handle_request(
        request, raw_request
    )
678
679


680
681
682
683
684
685
686
687
@app.post("/v1/chat/completions", dependencies=[Depends(validate_json_request)])
async def openai_v1_chat_completions(
    request: ChatCompletionRequest, raw_request: Request
):
    """OpenAI-compatible chat completion endpoint."""
    return await raw_request.app.state.openai_serving_chat.handle_request(
        request, raw_request
    )
688
689


690
691
692
693
694
695
696
697
698
699
@app.post(
    "/v1/embeddings",
    response_class=ORJSONResponse,
    dependencies=[Depends(validate_json_request)],
)
async def openai_v1_embeddings(request: EmbeddingRequest, raw_request: Request):
    """OpenAI-compatible embeddings endpoint."""
    return await raw_request.app.state.openai_serving_embedding.handle_request(
        request, raw_request
    )
700
701
702


@app.get("/v1/models", response_class=ORJSONResponse)
703
704
async def available_models():
    """Show available models. OpenAI-compatible endpoint."""
705
    served_model_names = [_global_state.tokenizer_manager.served_model_name]
706
707
    model_cards = []
    for served_model_name in served_model_names:
708
709
710
711
712
713
714
        model_cards.append(
            ModelCard(
                id=served_model_name,
                root=served_model_name,
                max_model_len=_global_state.tokenizer_manager.model_config.context_len,
            )
        )
715
716
717
    return ModelList(data=model_cards)


718
719
720
721
@app.get("/v1/models/{model:path}", response_class=ORJSONResponse)
async def retrieve_model(model: str):
    """Retrieves a model instance, providing basic information about the model."""
    served_model_names = [_global_state.tokenizer_manager.served_model_name]
722

723
724
725
726
727
728
729
730
731
732
733
734
    if model not in served_model_names:
        return ORJSONResponse(
            status_code=404,
            content={
                "error": {
                    "message": f"The model '{model}' does not exist",
                    "type": "invalid_request_error",
                    "param": "model",
                    "code": "model_not_found",
                }
            },
        )
735

736
737
738
739
740
    return ModelCard(
        id=model,
        root=model,
        max_model_len=_global_state.tokenizer_manager.model_config.context_len,
    )
741
742


743
744
745
746
747
748
749
750
## SageMaker API
@app.get("/ping")
async def sagemaker_health() -> Response:
    """Check the health of the http server."""
    return Response(status_code=200)


@app.post("/invocations")
751
752
753
754
755
756
757
async def sagemaker_chat_completions(
    request: ChatCompletionRequest, raw_request: Request
):
    """OpenAI-compatible chat completion endpoint."""
    return await raw_request.app.state.openai_serving_chat.handle_request(
        request, raw_request
    )
758
759


760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
## Vertex AI API
@app.post(os.environ.get("AIP_PREDICT_ROUTE", "/vertex_generate"))
async def vertex_generate(vertex_req: VertexGenerateReqInput, raw_request: Request):
    if not vertex_req.instances:
        return []
    inputs = {}
    for input_key in ("text", "input_ids", "input_embeds"):
        if vertex_req.instances[0].get(input_key):
            inputs[input_key] = [
                instance.get(input_key) for instance in vertex_req.instances
            ]
            break
    image_data = [
        instance.get("image_data")
        for instance in vertex_req.instances
        if instance.get("image_data") is not None
    ] or None
    req = GenerateReqInput(
        **inputs,
        image_data=image_data,
        **(vertex_req.parameters or {}),
    )
    ret = await generate_request(req, raw_request)
783
784
    if isinstance(ret, Response):
        return ret
785
786
787
    return ORJSONResponse({"predictions": ret})


788
789
@app.post("/v1/score", dependencies=[Depends(validate_json_request)])
async def v1_score_request(request: ScoringRequest, raw_request: Request):
790
    """Endpoint for the decoder-only scoring API. See Engine.score() for detailed documentation."""
791
792
793
    return await raw_request.app.state.openai_serving_score.handle_request(
        request, raw_request
    )
794
795


796
797
798
799
800
801
802
803
804
def _create_error_response(e):
    return ORJSONResponse(
        {"error": {"message": str(e)}}, status_code=HTTPStatus.BAD_REQUEST
    )


def launch_server(
    server_args: ServerArgs,
    pipe_finish_writer: Optional[multiprocessing.connection.Connection] = None,
805
    launch_callback: Optional[Callable[[], None]] = None,
806
807
808
809
810
811
812
813
):
    """
    Launch SRT (SGLang Runtime) Server.

    The SRT server consists of an HTTP server and an SRT engine.

    - HTTP server: A FastAPI server that routes requests to the engine.
    - The engine consists of three components:
814
        1. TokenizerManager: Tokenizes the requests and sends them to the scheduler.
815
816
817
818
        2. Scheduler (subprocess): Receives requests from the Tokenizer Manager, schedules batches, forwards them, and sends the output tokens to the Detokenizer Manager.
        3. DetokenizerManager (subprocess): Detokenizes the output tokens and sends the result back to the Tokenizer Manager.

    Note:
819
    1. The HTTP server, Engine, and TokenizerManager both run in the main process.
820
    2. Inter-process communication is done through IPC (each process uses a different port) via the ZMQ library.
821
    """
822
823
824
    tokenizer_manager, template_manager, scheduler_info = _launch_subprocesses(
        server_args=server_args
    )
825
826
    set_global_state(
        _GlobalState(
827
            tokenizer_manager=tokenizer_manager,
828
            template_manager=template_manager,
829
830
831
832
833
834
835
836
837
838
839
840
841
            scheduler_info=scheduler_info,
        )
    )

    # Add api key authorization
    if server_args.api_key:
        add_api_key_middleware(app, server_args.api_key)

    # Add prometheus middleware
    if server_args.enable_metrics:
        add_prometheus_middleware(app)
        enable_func_timer()

842
843
844
    # Send a warmup request - we will create the thread launch it
    # in the lifespan after all other warmups have fired.
    warmup_thread = threading.Thread(
845
846
847
848
        target=_wait_and_warmup,
        args=(
            server_args,
            pipe_finish_writer,
849
            _global_state.tokenizer_manager.image_token_id,
850
            launch_callback,
851
852
        ),
    )
853
    app.warmup_thread = warmup_thread
854
855
856
857

    try:
        # Update logging configs
        set_uvicorn_logging_configs()
858
        app.server_args = server_args
859
860
861
862
863
864
865
866
867
868
        # Listen for HTTP requests
        uvicorn.run(
            app,
            host=server_args.host,
            port=server_args.port,
            log_level=server_args.log_level_http or server_args.log_level,
            timeout_keep_alive=5,
            loop="uvloop",
        )
    finally:
869
        warmup_thread.join()
870
871


872
873
874
875
876
877
def _wait_and_warmup(
    server_args: ServerArgs,
    pipe_finish_writer: Optional[multiprocessing.connection.Connection],
    image_token_text: str,
    launch_callback: Optional[Callable[[], None]] = None,
):
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
    headers = {}
    url = server_args.url()
    if server_args.api_key:
        headers["Authorization"] = f"Bearer {server_args.api_key}"

    # Wait until the server is launched
    success = False
    for _ in range(120):
        time.sleep(1)
        try:
            res = requests.get(url + "/get_model_info", timeout=5, headers=headers)
            assert res.status_code == 200, f"{res=}, {res.text=}"
            success = True
            break
        except (AssertionError, requests.exceptions.RequestException):
            last_traceback = get_exception_traceback()
            pass

    if not success:
        if pipe_finish_writer is not None:
            pipe_finish_writer.send(last_traceback)
        logger.error(f"Initialization failed. warmup error: {last_traceback}")
        kill_process_tree(os.getpid())
        return

    model_info = res.json()

    # Send a warmup request
    request_name = "/generate" if model_info["is_generation"] else "/encode"
    max_new_tokens = 8 if model_info["is_generation"] else 1
    json_data = {
        "sampling_params": {
            "temperature": 0,
            "max_new_tokens": max_new_tokens,
        },
    }
    if server_args.skip_tokenizer_init:
fzyzcjy's avatar
fzyzcjy committed
915
        json_data["input_ids"] = [[10, 11, 12] for _ in range(server_args.dp_size)]
fzyzcjy's avatar
fzyzcjy committed
916
917
918
        # TODO Workaround the bug that embedding errors for list of size 1
        if server_args.dp_size == 1:
            json_data["input_ids"] = json_data["input_ids"][0]
919
    else:
fzyzcjy's avatar
fzyzcjy committed
920
        json_data["text"] = ["The capital city of France is"] * server_args.dp_size
fzyzcjy's avatar
fzyzcjy committed
921
922
923
        # TODO Workaround the bug that embedding errors for list of size 1
        if server_args.dp_size == 1:
            json_data["text"] = json_data["text"][0]
924

925
926
927
928
929
930
931
932
    # Debug dumping
    if server_args.debug_tensor_dump_input_file:
        json_data.pop("text", None)
        json_data["input_ids"] = np.load(
            server_args.debug_tensor_dump_input_file
        ).tolist()
        json_data["sampling_params"]["max_new_tokens"] = 0

933
    try:
934
935
936
937
938
939
940
941
942
        if server_args.disaggregation_mode == "null":
            res = requests.post(
                url + request_name,
                json=json_data,
                headers=headers,
                timeout=600,
            )
            assert res.status_code == 200, f"{res}"
        else:
943
944
945
946
947
948
949
            logger.info(f"Start of prefill warmup ...")
            json_data = {
                "sampling_params": {
                    "temperature": 0.0,
                    "max_new_tokens": 8,
                    "ignore_eos": True,
                },
Byron Hsu's avatar
Byron Hsu committed
950
                "bootstrap_host": [FAKE_BOOTSTRAP_HOST] * server_args.dp_size,
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
                # This is a hack to ensure fake transfer is enabled during prefill warmup
                # ensure each dp rank has a unique bootstrap_room during prefill warmup
                "bootstrap_room": [
                    i * (2**63 // server_args.dp_size) + (i % server_args.tp_size)
                    for i in range(server_args.dp_size)
                ],
                "input_ids": [[0, 1, 2, 3]] * server_args.dp_size,
            }
            res = requests.post(
                url + request_name,
                json=json_data,
                headers=headers,
                timeout=1800,  # because of deep gemm precache is very long if not precache.
            )
            logger.info(
                f"End of prefill warmup with status {res.status_code}, resp: {res.json()}"
            )

969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
    except Exception:
        last_traceback = get_exception_traceback()
        if pipe_finish_writer is not None:
            pipe_finish_writer.send(last_traceback)
        logger.error(f"Initialization failed. warmup error: {last_traceback}")
        kill_process_tree(os.getpid())
        return

    # Debug print
    # logger.info(f"{res.json()=}")

    logger.info("The server is fired up and ready to roll!")
    if pipe_finish_writer is not None:
        pipe_finish_writer.send("ready")

    if server_args.delete_ckpt_after_loading:
        delete_directory(server_args.model_path)
986
987
988
989

    if server_args.debug_tensor_dump_input_file:
        kill_process_tree(os.getpid())

990
991
992
993
994
995
996
997
    if server_args.pdlb_url is not None:
        register_disaggregation_server(
            server_args.disaggregation_mode,
            server_args.port,
            server_args.disaggregation_bootstrap_port,
            server_args.pdlb_url,
        )

998
999
    if launch_callback is not None:
        launch_callback()