SGLang 네이티브 API

SGLang 네이티브 API (Native APIs)

OpenAI 호환 API 외에도 SGLang Runtime은 자체 네이티브 서버 API를 제공합니다. 채팅 템플릿을 거치지 않고 런타임을 좀 더 직접 다루고 싶을 때 유용해요. 이 글에서는 주요 네이티브 엔드포인트를 하나씩 살펴봅니다.

출처: 공식 문서 - SGLang Native APIs

SGLang 런타임이 제공하는 네이티브 엔드포인트는 다음과 같아요.

/generate (텍스트 생성 모델)
/get_model_info
/server_info
/health
/health_generate
/flush_cache
/update_weights
/encode (임베딩 모델)
/v1/rerank (크로스 인코더 리랭크 모델)
/v1/score (디코더 전용 스코어링)
/classify (리워드 모델)
/start_expert_distribution_record
/stop_expert_distribution_record
/dump_expert_distribution_record
/tokenize
/detokenize
전체 목록은 http_server.py에서 확인할 수 있어요.

예시에서는 주로 requests로 테스트합니다. curl을 써도 됩니다.

서버 실행 (Launch A Server)

from sglang.test.doc_patch import launch_server_cmd
from sglang.utils import wait_for_server, print_highlight, terminate_process

server_process, port = launch_server_cmd(
    "python3 -m sglang.launch_server --model-path qwen/qwen2.5-0.5b-instruct --host 0.0.0.0 --log-level warning"
)

wait_for_server(f"http://localhost:{port}", process=server_process)

Generate (텍스트 생성 모델)

컴플리션을 생성합니다. OpenAI API의 /v1/completions와 비슷해요. 자세한 파라미터는 sampling parameters에서 확인할 수 있습니다.

import requests

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)

모델의 정보를 조회합니다.

model_path: 모델의 경로/이름
is_generation: 모델이 생성 모델인지 임베딩 모델인지 여부
tokenizer_path: 토크나이저의 경로/이름
preferred_sampling_params: --preferred-sampling-params로 지정한 기본 샘플링 파라미터. 이 예시에선 서버 인자로 명시하지 않아 None이 반환돼요.
weight_version: 모델 가중치의 버전. 모델의 학습된 파라미터 변경을 추적할 때 자주 씁니다.
has_image_understanding: 모델이 이미지 이해 능력을 가졌는지 여부
has_audio_understanding: 모델이 오디오 이해 능력을 가졌는지 여부
model_type: HuggingFace config의 모델 타입 (예: "qwen2", "llama")
architectures: HuggingFace config의 모델 아키텍처 (예: ["Qwen2ForCausalLM"])
embedding: 해석된 임베딩 서빙 계획. pooling, 정규화, 실행·어텐션 스타일, Matryoshka 차원, 캐시 정책, 유효한 BCG prefill 설정을 포함해요. 모델 설정이 임베딩 능력 계약을 노출할 때 이 필드를 사용할 수 있습니다.

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"] == "qwen/qwen2.5-0.5b-instruct"
assert response_json["is_generation"] is True
assert response_json["tokenizer_path"] == "qwen/qwen2.5-0.5b-instruct"
assert response_json["preferred_sampling_params"] is None
assert response_json.keys() == {
    "model_path",
    "is_generation",
    "tokenizer_path",
    "preferred_sampling_params",
    "weight_version",
    "has_image_understanding",
    "has_audio_understanding",
    "model_type",
    "architectures",
    "embedding",
}

서버 정보 가져오기 (Get Server Info)

CLI 인자, 토큰 한도, 메모리 풀 크기 등 서버 정보를 조회합니다.

참고: get_server_info는 다음의 폐기된 엔드포인트들을 하나로 합쳤어요.

get_server_args

get_memory_pool_size

get_max_total_num_tokens

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

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

헬스 체크 (Health Check)

/health: 서버가 살아있는지 확인
/health_generate: 토큰을 하나 생성해 서버가 정상 동작하는지 확인

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

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

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

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

캐시 비우기 (Flush Cache)

라딕스(radix) 캐시를 비웁니다. /update_weights API로 모델 가중치가 갱신되면 자동으로 트리거됩니다.

파라미터:

timeout (query, float, 기본 0, 단위: 초): 비우기 전에 idle 상태를 기다리는 시간. 0이면 idle이 아닐 때 바로 실패(fail fast)해요. HiCache 비동기 작업이 진행 중일 때 0이 아닌 timeout 값은 서버가 idle이 될 때까지 기다렸다가 비우게 해서 불필요한 400 오류를 피할 수 있게 합니다.

# With timeout (wait up to 30s for idle state)
curl -s -X POST "http://127.0.0.1:30000/flush_cache?timeout=30"

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

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

디스크에서 가중치 갱신 (Update Weights From Disk)

서버를 재시작하지 않고 디스크에서 모델 가중치를 갱신합니다. 같은 아키텍처와 파라미터 크기를 가진 모델에만 적용할 수 있어요.

SGLang은 학습 중 지속 평가를 위해 update_weights_from_disk API를 지원합니다 (체크포인트를 디스크에 저장하고 디스크에서 가중치를 갱신).

# successful update with same architecture and size

url = f"http://localhost:{port}/update_weights_from_disk"
data = {"model_path": "qwen/qwen2.5-0.5b-instruct"}

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

# failed update with different parameter size or wrong name

url = f"http://localhost:{port}/update_weights_from_disk"
data = {"model_path": "qwen/qwen2.5-0.5b-instruct-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: "
    "qwen/qwen2.5-0.5b-instruct-wrong"
    " (repository not found)."
)

terminate_process(server_process)

Encode (임베딩 모델)

텍스트를 임베딩으로 인코딩합니다. 이 API는 임베딩 모델에서만 사용 가능하고, 생성 모델에서는 오류가 발생해요. 그래서 임베딩 모델용으로 새 서버를 띄우는 방식으로 진행합니다.

embedding_process, port = launch_server_cmd("""
python3 -m sglang.launch_server --model-path Alibaba-NLP/gte-Qwen2-1.5B-instruct \
    --host 0.0.0.0 --is-embedding --log-level warning
""")

wait_for_server(f"http://localhost:{port}", process=embedding_process)

# successful encode for embedding model

url = f"http://localhost:{port}/encode"
data = {"model": "Alibaba-NLP/gte-Qwen2-1.5B-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]}")

terminate_process(embedding_process)

v1/rerank (크로스 인코더 리랭크 모델)

쿼리가 주어졌을 때 크로스 인코더 모델로 문서 목록을 리랭크합니다. 이 API는 BAAI/bge-reranker-v2-m3 같은 크로스 인코더 모델에서만 사용 가능하며 attention-backend가 triton·torch_native일 때 지원됩니다.

reranker_process, port = launch_server_cmd("""
python3 -m sglang.launch_server --model-path BAAI/bge-reranker-v2-m3 \
    --host 0.0.0.0 --disable-radix-cache --chunked-prefill-size -1 --attention-backend triton --is-embedding --log-level warning
""")

wait_for_server(f"http://localhost:{port}", process=reranker_process)

# compute rerank scores for query and documents

url = f"http://localhost:{port}/v1/rerank"
data = {
    "model": "BAAI/bge-reranker-v2-m3",
    "query": "what is panda?",
    "documents": [
        "hi",
        "The giant panda (Ailuropoda melanoleuca), sometimes called a panda bear or simply panda, is a bear species endemic to China.",
    ],
}

response = requests.post(url, json=data)
response_json = response.json()
for item in response_json:
    print_highlight(f"Score: {item['score']:.2f} - Document: '{item['document']}'")

terminate_process(reranker_process)

v1/score (디코더 전용 스코어링)

쿼리와 아이템이 주어졌을 때 특정 토큰의 토큰 확률을 계산합니다. 분류 작업, 응답 스코어링, 로그 확률 계산에 유용해요.

파라미터:

query: 쿼리 텍스트
items: 스코어링할 아이템 텍스트
label_token_ids: 확률을 계산할 토큰 ID
apply_softmax: 정규화된 확률을 얻기 위해 softmax를 적용할지 (기본: False)
item_first: 연결 순서에서 아이템이 먼저 오는지 (기본: False)
model: 모델 이름

응답에는 scores가 포함되는데, label_token_ids 순서대로 아이템마다 확률 리스트 하나씩 들어갑니다.

score_process, port = launch_server_cmd("""
python3 -m sglang.launch_server --model-path qwen/qwen2.5-0.5b-instruct \
    --host 0.0.0.0 --log-level warning
""")

wait_for_server(f"http://localhost:{port}", process=score_process)

# Score the probability of different completions given a query
query = "The capital of France is"
items = ["Paris", "London", "Berlin"]

url = f"http://localhost:{port}/v1/score"
data = {
    "model": "qwen/qwen2.5-0.5b-instruct",
    "query": query,
    "items": items,
    "label_token_ids": [9454, 2753],  # e.g. "Yes" and "No" token ids
    "apply_softmax": True,  # Normalize probabilities to sum to 1
}

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

# Display scores for each item
for item, scores in zip(items, response_json["scores"]):
    print_highlight(f"Item '{item}': probabilities = {[f'{s:.4f}' for s in scores]}")

terminate_process(score_process)

Classify (리워드 모델)

SGLang Runtime은 리워드 모델도 지원합니다. 여기서는 리워드 모델로 쌍(pairwise) 생성의 품질을 분류합니다.

# 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("""
python3 -m sglang.launch_server --model-path Skywork/Skywork-Reward-Llama-3.1-8B-v0.2 --host 0.0.0.0 --is-embedding --log-level warning
""")

wait_for_server(f"http://localhost:{port}", process=reward_process)

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, return_dict=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]}")

terminate_process(reward_process)

MoE 모델의 전문가 선택 분포 캡처 (Capture expert selection distribution)

SGLang Runtime은 MoE 모델 실행에서 각 전문가가 선택된 횟수를 기록할 수 있어요. 모델의 처리량을 분석하고 최적화를 계획할 때 유용합니다.

참고: 아래에서는 가독성을 위해 csv의 처음 10줄만 출력합니다. 더 깊이 분석하고 싶다면 그에 맞게 조정하세요.

expert_record_server_process, port = launch_server_cmd(
    "python3 -m sglang.launch_server --model-path Qwen/Qwen1.5-MoE-A2.7B --host 0.0.0.0 --expert-distribution-recorder-mode stat --log-level warning"
)

wait_for_server(f"http://localhost:{port}", process=expert_record_server_process)

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)

terminate_process(expert_record_server_process)

Tokenize/Detokenize 예시 (라운드 트립)

/tokenize와 /detokenize 엔드포인트를 함께 쓰는 예시입니다. 먼저 문자열을 토큰화한 뒤, 얻은 ID를 디토크나이즈해 원본 텍스트를 재구성해요. 토크나이제이션은 외부에서 처리하면서 디토크나이제이션은 서버에 맡겨야 할 때 유용한 워크플로우입니다.

tokenizer_free_server_process, port = launch_server_cmd("""
python3 -m sglang.launch_server --model-path qwen/qwen2.5-0.5b-instruct
""")

wait_for_server(f"http://localhost:{port}", process=tokenizer_free_server_process)

import requests
from sglang.utils import print_highlight

base_url = f"http://localhost:{port}"
tokenize_url = f"{base_url}/tokenize"
detokenize_url = f"{base_url}/detokenize"

model_name = "qwen/qwen2.5-0.5b-instruct"
input_text = "SGLang provides efficient tokenization endpoints."
print_highlight(f"Original Input Text:\n'{input_text}'")

# --- tokenize the input text ---
tokenize_payload = {
    "model": model_name,
    "prompt": input_text,
    "add_special_tokens": False,
}
try:
    tokenize_response = requests.post(tokenize_url, json=tokenize_payload)
    tokenize_response.raise_for_status()
    tokenization_result = tokenize_response.json()
    token_ids = tokenization_result.get("tokens")

    if not token_ids:
        raise ValueError("Tokenization returned empty tokens.")

    print_highlight(f"\nTokenized Output (IDs):\n{token_ids}")
    print_highlight(f"Token Count: {tokenization_result.get('count')}")
    print_highlight(f"Max Model Length: {tokenization_result.get('max_model_len')}")

    # --- detokenize the obtained token IDs ---
    detokenize_payload = {
        "model": model_name,
        "tokens": token_ids,
        "skip_special_tokens": True,
    }

    detokenize_response = requests.post(detokenize_url, json=detokenize_payload)
    detokenize_response.raise_for_status()
    detokenization_result = detokenize_response.json()
    reconstructed_text = detokenization_result.get("text")

    print_highlight(f"\nDetokenized Output (Text):\n'{reconstructed_text}'")

    if input_text == reconstructed_text:
        print_highlight(
            "\nRound Trip Successful: Original and reconstructed text match."
        )
    else:
        print_highlight(
            "\nRound Trip Mismatch: Original and reconstructed text differ."
        )

except requests.exceptions.RequestException as e:
    print_highlight(f"\nHTTP Request Error: {e}")
except Exception as e:
    print_highlight(f"\nAn error occurred: {e}")

terminate_process(tokenizer_free_server_process)

더 알아보기 (Learn more)

/generate가 받는 상세 파라미터는 Sampling Parameters를 참고해요.
OpenAI 호환 방식으로 서빙하고 싶다면 OpenAI APIs - Completions를 보세요.