SGLang — 구조화된 생성 및 고속 LLM 추론 엔진

# Python 라이브러리 설치
pip install sglang

# 또는 GPU 가속을 위해 Docker 사용
docker pull sglang/sglang:latest
docker run --gpus all -p 30000:30000 sglang/sglang:latest \
  --model-path meta-llama/Llama-3.2-8B-Instruct \
  --host 0.0.0.0 --port 30000

# 하나의 GPU에서 모델 서빙
python -m sglang.launch_server \
  --model-path meta-llama/Llama-3.2-8B-Instruct \
  --port 30000

# 멀티 GPU 텐서 병렬성
python -m sglang.launch_server \
  --model-path meta-llama/Llama-3.2-70B-Instruct \
  --tensor-parallel-size 4 \
  --port 30000

# 양자화로 비용 절감
python -m sglang.launch_server \
  --model-path Qwen/Qwen2.5-72B-Instruct-AWQ \
  --quantization awq \
  --port 30000

curl http://localhost:30000/generate \
  -H "Content-Type: application/json" \
  -d '{
    "text": "프랑스의 수도는 어디인가요?",
    "sampling_params": {
      "max_new_tokens": 64,
      "temperature": 0
    }
  }'

{
  "text": "프랑스의 수도는 파리입니다.",
  "meta": {"prompt_tokens": 12, "completion_tokens": 8}
}

import sglang as sgl
from pydantic import BaseModel, Field
from typing import List, Optional

class ProductReview(BaseModel):
    product_name: str = Field(description="제품 이름")
    rating: int = Field(ge=1, le=5, description="1~5 점수")
    pros: List[str] = Field(max_length=5, description="주요 장점")
    cons: List[str] = Field(max_length=5, description="주요 단점")
    would_recommend: bool = Field(description="이 제품을 추천합니까")
    summary: str = Field(description="한 문장 요약")

backend = sgl.Runtime(host="localhost", port=30000)

@sgl.program
def review_analyzer(state, review_text: str):
    state += sgl.user("이 제품 리뷰를 분석하여 구조화된 데이터 추출:")
    state += sgl.assistant(sgl.gen("json_output", max_tokens=512))

program = review_analyzer()
result = program.run(
    review_text="좋은 노트북이지만 배터리 수명이 개선되면 좋겠습니다. 디스플레이가 아름답고 개발 성능이 훌륭합니다.",
    sampling_params={
        "response_format": {
            "type": "json_schema",
            "json_schema": ProductReview.model_json_schema()
        }
    }
)

review = ProductReview.model_validate_json(result["json_output"])
print(f"제품: {review.product_name}, 점수: {review.rating}/5")

@sgl.program
def email_extractor(state, text: str):
    state += sgl.user("이 텍스트에서 모든 이메일 주소를 추출하세요:")
    state += sgl.assistant(
        sgl.gen(
            "emails",
            regex=r"([a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,})(\s*,\s*|$)+",
            max_tokens=256
        )
    )

program = email_extractor()
result = program.run(
    text="support@example.com 또는 sales@example.com으로 연락하세요. 청구 문의는 billing@company.org로."
)
print(result["emails"])
# 출력: "support@example.com, sales@example.com, billing@company.org."

from pydantic import BaseModel

class SQLQuery(BaseModel):
    query: str = Field(description="유효한 SQL SELECT 문")
    explanation: str = Field(description="이 쿼리의 역할")
    estimated_rows: Optional[int] = Field(description="예상 행 수")

import sglang as sgl

@sgl.program
def chatbot(state, user_message: str):
    state += sgl.system("도움이 되는 어시스턴트입니다.")  # 이 프리픽스가 캐시됨!
    state += sgl.conversation(
        [{"role": "user", "content": "안녕"}, {"role": "assistant", "content": "안녕하세요!"}],
    )  # 캐시됨!
    state += sgl.user(user_message)
    state += sgl.assistant(sgl.gen("response", max_tokens=256))

# 첫 번째 요청: 전체 계산
r1 = chatbot().run("오늘 날씨가 어때?")

# 두 번째 요청이 동일한 시스템 프롬프트 + 대화 기록 사용:
# 새 user message에 대한 어텐션만 계산
r2 = chatbot().run("더 알려줘")

python -m sglang.launch_server \
  --model-path meta-llama/Llama-3.2-8B \
  --mem-fraction-static 0.85 \
  --context-length 8192

# 70B 모델용 4x A100-80GB
python -m sglang.launch_server \
  --model-path meta-llama/Llama-3.2-70B-Instruct \
  --tensor-parallel-size 4 \
  --mem-fraction-static 0.9 \
  --host 0.0.0.0 --port 30000

# GPU 사용량 확인
nvidia-smi
# 활성 추론 중 각 GPU가 ~95% 사용률 표시

@sgl.program
def react_agent(state, question: str):
    state += sgl.user(f"도구를 사용하여 단계별로 이 질문에 답하세요:\n{question}")
    
    for i in range(5):  # 최대 5단계 추론
        state += sgl.assistant(
            f"생각 {i+1}: " + sgl.gen("thought", stop="\n작업:", max_tokens=200)
        )
        
        action = sgl.gen("action", stop="\n관찰:", max_tokens=200)
        state += sgl.user(f"\n작업: {action}")
        
        obs = execute_tool(action)
        state += sgl.user(f"\n관찰: {obs}")
    
    state += sgl.assistant(sgl.gen("final_answer", max_tokens=500))

@sgl.program
def document_summarizer(state, doc: str):
    state += sgl.user(f"다음 문서를 3개 불릿 포인트로 요약:\n{doc}")
    state += sgl.assistant(sgl.gen("summary", max_tokens=256))

documents = load_documents("path/to/docs/")
results = sgl.compile(
    [document_summarizer(doc) for doc in documents[:100]],
    scheduler_policy="lookahead"
)

from sglang import RuntimeClient

client = RuntimeClient("http://localhost:30000")

stream = client.generate({
    "text": "이 보고서에서 핵심 지표를 추출하세요.",
    "sampling_params": {
        "max_new_tokens": 512,
        "response_format": {
            "type": "json_schema",
            "json_schema": ReportMetrics.model_json_schema()
        },
        "stream": True  # 스트리밍 활성화
    }
})

for chunk in stream:
    if chunk["event_type"] == "text":
        print(chunk["text"], end="", flush=True)

from pydantic import BaseModel
from typing import Literal

class WeatherRequest(BaseModel):
    city: str
    units: Literal["celsius", "fahrenheit"] = "celsius"

@sgl.program
def function_caller(state, user_input: str):
    state += sgl.user(user_input)
    state += sgl.assistant(sgl.gen("function_call", max_tokens=256))

from pydantic import BaseModel
from typing import Literal

class WeatherRequest(BaseModel):
    city: str
    units: Literal["celsius", "fahrenheit"] = "celsius"

@sgl.program
def function_caller(state, user_input: str):
    state += sgl.user(user_input)
    state += sgl.assistant(sgl.gen("function_call", max_tokens=256))

curl http://localhost:30000/health
# 반환: {"status": "ok", "gpu_memory_usage": "72%", "active_requests": 15}

python -m sglang.launch_server \
  --model-path meta-llama/Llama-3.2-8B \
  --mem-fraction-static 0.75 \
  --max-running-requests 32

curl -X POST http://localhost:30000/generate \
  -H "Content-Type: application/json" \
  -d '{"text": "warmup", "sampling_params": {"max_new_tokens": 1}}'

python -m sglang.launch_server \
  --model-path Qwen/Qwen2.5-72B-Instruct-AWQ \
  --quantization awq

async for event in program.run_async(stream=True):
    print(event.delta, end="", flush=True)

python -m sglang.launch_server \
  --model-path meta-llama/Llama-3.2-8B \
  --rate-limit-requests 100 \
  --rate-limit-tokens 50000 \
  --scheduler-policy lookahead