sglang/docs/backend/native_api.ipynb

SGLang Native APIs

Apart from the OpenAI compatible APIs, the SGLang Runtime also provides its native server APIs. We introduce these following APIs:

• /generate (text generation model)
• /get_model_info
• /get_server_info
• /health
• /health_generate
• /flush_cache
• /update_weights
• /encode(embedding model)
• /classify(reward model)
• /start_expert_distribution_record
• /stop_expert_distribution_record
• /dump_expert_distribution_record

We mainly use requests to test these APIs in the following examples. You can also use curl.

Launch A Server

In [ ]:

import requests
from sglang.test.test_utils import is_in_ci

if is_in_ci():
    from patch import launch_server_cmd
else:
    from sglang.utils import launch_server_cmd

from sglang.utils import wait_for_server, print_highlight, terminate_process


server_process, port = launch_server_cmd(
    "python -m sglang.launch_server --model-path meta-llama/Llama-3.2-1B-Instruct --host 0.0.0.0"
)

wait_for_server(f"http://localhost:{port}")

Generate (text generation model)

Generate completions. This is similar to the /v1/completions in OpenAI API. Detailed parameters can be found in the sampling parameters.

In [ ]:

url = f"http://localhost:{port}/generate"
data = {"text": "What is the capital of France?"}

response = requests.post(url, json=data)
print_highlight(response.json())

Get Model Info

Get the information of the model.

• model_path: The path/name of the model.
• is_generation: Whether the model is used as generation model or embedding model.
• tokenizer_path: The path/name of the tokenizer.

In [ ]:

url = f"http://localhost:{port}/get_model_info"

response = requests.get(url)
response_json = response.json()
print_highlight(response_json)
assert response_json["model_path"] == "meta-llama/Llama-3.2-1B-Instruct"
assert response_json["is_generation"] is True
assert response_json["tokenizer_path"] == "meta-llama/Llama-3.2-1B-Instruct"
assert response_json.keys() == {"model_path", "is_generation", "tokenizer_path"}

Get Server Info

Gets the server information including CLI arguments, token limits, and memory pool sizes.

• Note: get_server_info merges the following deprecated endpoints:
  • get_server_args
  • get_memory_pool_size
  • get_max_total_num_tokens

In [ ]:

# get_server_info

url = f"http://localhost:{port}/get_server_info"

response = requests.get(url)
print_highlight(response.text)

Health Check

• /health: Check the health of the server.
• /health_generate: Check the health of the server by generating one token.

In [ ]:

url = f"http://localhost:{port}/health_generate"

response = requests.get(url)
print_highlight(response.text)

In [ ]:

url = f"http://localhost:{port}/health"

response = requests.get(url)
print_highlight(response.text)

Flush Cache

Flush the radix cache. It will be automatically triggered when the model weights are updated by the /update_weights API.

In [ ]:

# flush cache

url = f"http://localhost:{port}/flush_cache"

response = requests.post(url)
print_highlight(response.text)

Update Weights From Disk

Update model weights from disk without restarting the server. Only applicable for models with the same architecture and parameter size.

SGLang support update_weights_from_disk API for continuous evaluation during training (save checkpoint to disk and update weights from disk).

In [ ]:

# successful update with same architecture and size

url = f"http://localhost:{port}/update_weights_from_disk"
data = {"model_path": "meta-llama/Llama-3.2-1B"}

response = requests.post(url, json=data)
print_highlight(response.text)
assert response.json()["success"] is True
assert response.json()["message"] == "Succeeded to update model weights."

In [ ]:

# failed update with different parameter size or wrong name

url = f"http://localhost:{port}/update_weights_from_disk"
data = {"model_path": "meta-llama/Llama-3.2-1B-wrong"}

response = requests.post(url, json=data)
response_json = response.json()
print_highlight(response_json)
assert response_json["success"] is False
assert response_json["message"] == (
    "Failed to get weights iterator: "
    "meta-llama/Llama-3.2-1B-wrong"
    " (repository not found)."
)

Encode (embedding model)

Encode text into embeddings. Note that this API is only available for embedding models and will raise an error for generation models. Therefore, we launch a new server to server an embedding model.

In [ ]:

terminate_process(server_process)

embedding_process, port = launch_server_cmd(
    """
python -m sglang.launch_server --model-path Alibaba-NLP/gte-Qwen2-7B-instruct \
    --host 0.0.0.0 --is-embedding
"""
)

wait_for_server(f"http://localhost:{port}")

In [ ]:

# successful encode for embedding model

url = f"http://localhost:{port}/encode"
data = {"model": "Alibaba-NLP/gte-Qwen2-7B-instruct", "text": "Once upon a time"}

response = requests.post(url, json=data)
response_json = response.json()
print_highlight(f"Text embedding (first 10): {response_json['embedding'][:10]}")

In [ ]:

terminate_process(embedding_process)

Classify (reward model)

SGLang Runtime also supports reward models. Here we use a reward model to classify the quality of pairwise generations.

In [ ]:

terminate_process(embedding_process)

# Note that SGLang now treats embedding models and reward models as the same type of models.
# This will be updated in the future.

reward_process, port = launch_server_cmd(
    """
python -m sglang.launch_server --model-path Skywork/Skywork-Reward-Llama-3.1-8B-v0.2 --host 0.0.0.0 --is-embedding
"""
)

wait_for_server(f"http://localhost:{port}")

In [ ]:

from transformers import AutoTokenizer

PROMPT = (
    "What is the range of the numeric output of a sigmoid node in a neural network?"
)

RESPONSE1 = "The output of a sigmoid node is bounded between -1 and 1."
RESPONSE2 = "The output of a sigmoid node is bounded between 0 and 1."

CONVS = [
    [{"role": "user", "content": PROMPT}, {"role": "assistant", "content": RESPONSE1}],
    [{"role": "user", "content": PROMPT}, {"role": "assistant", "content": RESPONSE2}],
]

tokenizer = AutoTokenizer.from_pretrained("Skywork/Skywork-Reward-Llama-3.1-8B-v0.2")
prompts = tokenizer.apply_chat_template(CONVS, tokenize=False)

url = f"http://localhost:{port}/classify"
data = {"model": "Skywork/Skywork-Reward-Llama-3.1-8B-v0.2", "text": prompts}

responses = requests.post(url, json=data).json()
for response in responses:
    print_highlight(f"reward: {response['embedding'][0]}")

In [ ]:

terminate_process(reward_process)

Capture expert selection distribution in MoE models

SGLang Runtime supports recording the number of times an expert is selected in a MoE model run for each expert in the model. This is useful when analyzing the throughput of the model and plan for optimization.

Note: We only print out the first 10 lines of the csv below for better readability. Please adjust accordingly if you want to analyze the results more deeply.

In [ ]:

expert_record_server_process, port = launch_server_cmd(
    "python -m sglang.launch_server --model-path Qwen/Qwen1.5-MoE-A2.7B --host 0.0.0.0"
)

wait_for_server(f"http://localhost:{port}")

In [ ]:

response = requests.post(f"http://localhost:{port}/start_expert_distribution_record")
print_highlight(response)

url = f"http://localhost:{port}/generate"
data = {"text": "What is the capital of France?"}

response = requests.post(url, json=data)
print_highlight(response.json())

response = requests.post(f"http://localhost:{port}/stop_expert_distribution_record")
print_highlight(response)

response = requests.post(f"http://localhost:{port}/dump_expert_distribution_record")
print_highlight(response)

import glob

output_file = glob.glob("expert_distribution_*.csv")[0]
with open(output_file, "r") as f:
    print_highlight("\n| Layer ID | Expert ID | Count |")
    print_highlight("|----------|-----------|--------|")
    next(f)
    for i, line in enumerate(f):
        if i < 9:
            layer_id, expert_id, count = line.strip().split(",")
            print_highlight(f"| {layer_id:8} | {expert_id:9} | {count:6} |")

In [ ]:

terminate_process(expert_record_server_process)

Skip Tokenizer and Detokenizer

SGLang Runtime also supports skip tokenizer and detokenizer. This is useful in cases like integrating with RLHF workflow.

In [ ]:

tokenizer_free_server_process, port = launch_server_cmd(
    """
python3 -m sglang.launch_server --model-path meta-llama/Llama-3.2-1B-Instruct --skip-tokenizer-init
"""
)

wait_for_server(f"http://localhost:{port}")

In [ ]:

from transformers import AutoTokenizer

tokenizer = AutoTokenizer.from_pretrained("meta-llama/Llama-3.2-1B-Instruct")

input_text = "What is the capital of France?"

input_tokens = tokenizer.encode(input_text)
print_highlight(f"Input Text: {input_text}")
print_highlight(f"Tokenized Input: {input_tokens}")

response = requests.post(
    f"http://localhost:{port}/generate",
    json={
        "input_ids": input_tokens,
        "sampling_params": {
            "temperature": 0,
            "max_new_tokens": 256,
            "stop_token_ids": [tokenizer.eos_token_id],
        },
        "stream": False,
    },
)
output = response.json()
output_tokens = output["output_ids"]

output_text = tokenizer.decode(output_tokens, skip_special_tokens=False)
print_highlight(f"Tokenized Output: {output_tokens}")
print_highlight(f"Decoded Output: {output_text}")
print_highlight(f"Output Text: {output['meta_info']['finish_reason']}")

In [ ]:

terminate_process(tokenizer_free_server_process)