SGLang — 结构化生成和高速 LLM 推理引擎

# 安装 Python 库
pip install sglang

# 或使用 Docker 获取 GPU 加速
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

# 4x A100-80GB 用于 70B 模型
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))

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