SGLang 探理引擎：结构化生成的高性能方案

传统 KV Cache（每请求独立）:
  请求 A "系统提示 + 问题1" → KV Cache A（2048 tokens）
  请求 B "系统提示 + 问题2" → KV Cache B（2048 tokens）
  → 同一系统提示被重复计算

RadixAttention（自动复用）:
  Radix Tree:
  ┌──────────────────────────┐
  │ [系统提示共享前缀]        │  ← 复用，不重复计算
  │  ├── [问题1] → KV Cache  │
  │  └── [问题2] → KV Cache  │
  └──────────────────────────┘
  → 系统提示只计算一次，后续请求直接复用

import sglang as sgl

# 定义结构化生成函数
@sgl.function
def extract_info(s, text):
    s += "从以下文本中提取信息，输出 JSON 格式：\n"
    s += text
    s += "\n\n请输出："
    # 使用正则约束输出格式
    s += sgl.gen(
        "json_output",
        regex=r'\{[^}]+\}',
        temperature=0.1,
        max_tokens=256,
    )

# 执行
state = extract_info.run(
    "张三，男，28岁，毕业于清华大学计算机系，现就职于字节跳动。"
)
print(state["json_output"])
# 输出确保是合法 JSON：{"name": "张三", "gender": "男", "age": 28, ...}

import sglang as sgl

@sgl.function
def multi_step_reasoning(s, question):
    # 步骤1：生成多个候选答案（并行）
    s += "问题：" + question + "\n"
    
    for i in range(3):
        s += f"\n方案{i+1}："
        s += sgl.gen(f"answer_{i}", temperature=0.8, max_tokens=200)
    
    # 步骤2：选择最佳答案
    s += "\n\n从以上方案中选择最佳的一个："
    s += sgl.gen("best_choice", temperature=0.1, max_tokens=100)

# 执行——SGLang 自动优化并行度
state = multi_step_reasoning.run("如何设计一个高可用系统？")
print(state["best_choice"])

# 安装 SGLang
pip install sglang[all]

# 安装 OpenAI 兼容服务器
pip install sglang[openai]

# 启动 OpenAI 兼容 API 服务器
python -m sglang.launch_server \
  --model-path meta-llama/Llama-3.3-70B-Instruct \
  --tp 4 \
  --port 30000 \
  --enable-radix-attention \
  --enable-structured-output

# 使用本地模型
python -m sglang.launch_server \
  --model-path /data/models/Llama-3.3-70B-Instruct \
  --tp 2 \
  --port 30000

import sglang as sgl

# 初始化运行时
runtime = sgl.Runtime(
    model_path="meta-llama/Llama-3.3-70B-Instruct",
    tp_size=4,
)

# 设置默认运行时
sgl.set_default_runtime(runtime)

# 基础生成
@sgl.function
def hello(s):
    s += "Hello! "
    s += sgl.gen("response", max_tokens=100)

state = hello.run()
print(state["response"])

# 批量生成
states = hello.run_batch([
    {"question": "你好"},
    {"question": "Hello"},
    {"question": "Bonjour"},
])
for s in states:
    print(s["response"])

import sglang as sgl
from pydantic import BaseModel

# 定义输出结构
class PersonInfo(BaseModel):
    name: str
    age: int
    skills: list[str]
    experience: int

@sgl.function
def extract_person(s, text: str):
    s += "从文本中提取人员信息：\n"
    s += text
    s += "\n\n输出 JSON："
    s += sgl.gen(
        "result",
        json_schema=PersonInfo.model_json_schema(),
        temperature=0.0,
        max_tokens=256,
    )

state = extract_person.run(
    "李四，30岁，精通 Python 和 Go，有5年后端开发经验。"
)

import json
person = json.loads(state["result"])
print(f"姓名：{person['name']}")
print(f"技能：{', '.join(person['skills'])}")

@sgl.function
def generate_sql(s, schema: str, question: str):
    s += f"数据库 Schema：\n{schema}\n\n"
    s += f"问题：{question}\n\n"
    s += "SQL 查询："
    s += sgl.gen(
        "sql",
        regex=r'SELECT\s+.*?\s+FROM\s+\w+(?:\s+WHERE\s+.*?\s*)?(?:\s+(?:ORDER BY|GROUP BY|LIMIT)\s+.*?\s*)?;',
        temperature=0.0,
        max_tokens=200,
    )

state = generate_sql.run(
    schema="users(id, name, age, department, salary)",
    question="找出薪水最高的前10个员工"
)
print(state["sql"])
# 输出保证是合法 SQL：SELECT * FROM users ORDER BY salary DESC LIMIT 10;

@sgl.function
def chat_with_memory(s, messages: list[dict]):
    """多轮对话自动复用 KV Cache"""
    for msg in messages:
        if msg["role"] == "user":
            s += f"User: {msg['content']}\n"
        elif msg["role"] == "assistant":
            s += f"Assistant: {msg['content']}\n"
    
    s += "Assistant: "
    s += sgl.gen("response", max_tokens=512, temperature=0.7)

# 多轮对话——RadixAttention 自动复用历史对话的 KV Cache
messages = [
    {"role": "user", "content": "你好，我叫小明"},
    {"role": "assistant", "content": "你好小明！很高兴认识你。"},
    {"role": "user", "content": "你还记得我的名字吗？"},
]

state = chat_with_memory.run(messages=messages)
print(state["response"])

# docker-compose.yml
version: '3.8'

services:
  sglang:
    image: lmsys/sglang:latest
    container_name: sglang-server
    runtime: nvidia
    environment:
      - NVIDIA_VISIBLE_DEVICES=0,1,2,3
      - HUGGING_FACE_HUB_TOKEN=${HF_TOKEN}
    ports:
      - "30000:30000"
    volumes:
      - model_cache:/root/.cache/huggingface
    command:
      - python
      - -m
      - sglang.launch_server
      - --model-path=meta-llama/Llama-3.3-70B-Instruct
      - --tp=4
      - --port=30000
      - --enable-radix-attention
      - --enable-structured-output
      - --mem-fraction-static=0.9
    deploy:
      resources:
        reservations:
          devices:
            - driver: nvidia
              count: 4
              capabilities: [gpu]

volumes:
  model_cache:

from openai import OpenAI

client = OpenAI(
    base_url="http://localhost:30000/v1",
    api_key="sglang",
)

# 标准对话
response = client.chat.completions.create(
    model="default",
    messages=[
        {"role": "user", "content": "解释 RadixAttention 原理"}
    ],
    max_tokens=1024,
    temperature=0.7,
)

# JSON 模式
response = client.chat.completions.create(
    model="default",
    messages=[
        {"role": "user", "content": "列出3种编程语言及其特点，用JSON格式"}
    ],
    response_format={"type": "json_object"},
    max_tokens=512,
    temperature=0.0,
)

print(response.choices[0].message.content)

import sglang as sgl

@sgl.function
def few_shot_classification(s, text: str):
    """Few-shot 分类模板"""
    # 系统提示
    s += "请对文本进行情感分类。\n\n"
    
    # Few-shot 示例
    examples = [
        ("这个产品太棒了！", "正面"),
        ("服务态度很差，不推荐。", "负面"),
        ("质量一般，价格还行。", "中性"),
    ]
    
    for ex_text, ex_label in examples:
        s += f"文本：{ex_text}\n分类：{ex_label}\n\n"
    
    # 待分类文本
    s += f"文本：{text}\n分类："
    s += sgl.gen(
        "label",
        choices=["正面", "负面", "中性"],
        temperature=0.0,
    )

# 批量分类
texts = [
    "体验非常好，超出预期！",
    "太失望了，完全不值这个价。",
    "还行吧，中规中矩。",
]

states = few_shot_classification.run_batch(
    [{"text": t} for t in texts]
)

for text, state in zip(texts, states):
    print(f"文本：{text} → 分类：{state['label']}")

# 健康检查
curl http://localhost:30000/health

# 获取服务器统计信息
curl http://localhost:30000/get_server_info

# 获取 RadixAttention 缓存统计
curl http://localhost:30000/get_radix_cache_info