深度解析 SGLang：大模型编程新范式——从 Prompt Engineering 到 Structured Generation 的系统性跃迁

# Traditional: Fragile & Unverifiableprompt = f"""You are a helpful assistant. Extract entities from the text.Text: "{text}"Output format: JSON with keys: person, organization, location."""response = llm.generate(prompt)try:    data = json.loads(response)except:    # Retry? Fallback? Give up?    data = retry_with_stronger_hints(prompt)

# SGLang: Structured & Deterministic@sgl.functiondef extract_entities(s, text):    s += "Text: " + text + "\n"    s += "Entities:\n"    with s.fork():        s += "person: " + s.gen("person", stop=",") + ", "        s += "organization: " + s.gen("org", regex=r"[A-Z][a-z]+( [A-Z][a-z]+)*") + ", "        s += "location: " + s.gen("loc", choices=["Paris", "Tokyo", "New York"])    return s["person"], s["org"], s["loc"]

[User Program]        ↓ (Parse + Semantic Analysis)[Constraint IR] → (CFG / Regex / Choices → Finite-State Machine)       ↓ (Lowering)[Execution Plan] → (Token-wise Constraints + KV Cache Management)       ↓[Runtime Engine] → (vLLM Integration + Constrained Sampling Kernel)

# SGLang: s.gen("date", regex=r"\d{4}-\d{2}-\d{2}")

class ConstraintIR:    def __init__(self, dfa: DFA, vocab: List[str]):        self.dfa = dfa        self.state = dfa.start_state        self.vocab_mask = self._build_vocab_mask(vocab)  # [V] bool tensor    def update(self, token_id: int) -> bool:        """Consume token, update state, return if still valid"""        token = self.vocab[token_id]        next_state = self.dfa.transition(self.state, token)        if next_state is None:             return False  # Invalid token        self.state = next_state        self._update_vocab_mask()  # Recompute allowed tokens        return True    def get_allowed_tokens(self) -> torch.Tensor:        return self.vocab_mask  # [V] bool tensor for sampling

// Extended Block Table Entrystruct BlockTableEntry {  int64_t physical_block_id;  std::optional<ConstraintState> constraint_state; // DFA state + metadata};// During fork():// 1. Share physical blocks for common prefix// 2. Copy constraint_state for diverging part// 3. New tokens get new blocks with independent constraint_state

__global__ void constrained_sampling_kernel(    float* logits,          // [V] unnormalized scores    bool* allowed_mask,     // [V] from ConstraintIR    int* output_token       // [1]) {  int tid = threadIdx.x;  float max_logit = -1e9;  int selected_id = -1;  // Warp-level reduction: find max among allowed tokens  for (int i = tid; i < VOCAB_SIZE; i += blockDim.x) {    if (allowed_mask[i] && logits[i] > max_logit) {      max_logit = logits[i];      selected_id = i;    }  }  // Softmax over allowed tokens only (numerically stable)  __shared__ float s_max, s_sum;  if (tid == 0) { s_max = max_logit; }  __syncthreads();  float exp_val = (selected_id != -1) ? expf(logits[selected_id] - s_max) : 0.0f;  float sum = warp_reduce_sum(exp_val); // Custom reduction  if (tid == 0) {    s_sum = sum;    float prob = exp_val / sum;    if (curand_uniform(...) < prob) {      output_token[0] = selected_id;    }  }}

@sgl.functiondef generate_user(s):    s += "Generate a user profile in JSON:\n"    with s.json_object():        s += '"name": "' + s.gen("name", regex=r"[A-Z][a-z]+") + '",\n'        s += '"age": ' + s.gen("age", regex=r"\d{1,3}") + ',\n'        s += '"email": "' + s.gen("email", regex=r"[a-z]+@[a-z]+\.[a-z]+") + '"'    return s["name"], int(s["age"]), s["email"]

@sgl.functiondef answer_math_question(s, question):    s += f"Question: {question}\n"    s += "Let's solve step by step:\n"        steps = []    for i in range(5):        # Generate next step with tool hint        step = s.gen(f"step_{i}",                     choices=["CALC", "SEARCH", "FINISH"],                    stop="\n")        steps.append(step)                if step == "CALC":            expr = s.gen("expr", regex=r"[\d+\-*/(). ]+")            result = calculator.eval(expr)  # ← External tool call            s += f" = {result}\n"          # ← Backfill result        elif step == "SEARCH":            query = s.gen("query", max_tokens=20)            docs = search_engine(query)            s += f"Found: {docs[0][:100]}...\n"        else:  # FINISH            break        s += "Answer: " + s.gen("answer", stop=".")    return s["answer"]

@sgl.functiondef describe_image(s, image):    s += s.image(image)  # Inject image embedding    s += "Describe this image with:\n"        # Enforce structured output    s += "- Main object: " + s.gen("obj", choices=["cat", "dog", "car"]) + "\n"    s += "- Color: " + s.gen("color", regex=r"(red|blue|green|black)") + "\n"    s += "- Action: " + s.gen("action",                            choices=["sitting", "running", "driving"]) + "\n"        # Cross-field constraint: if obj=="car", action must be "driving"    if s["obj"] == "car" and s["action"] != "driving":        s.rollback_to("action")  # ← Re-generate action        s += "- Action: driving\n"        return s["obj"], s["color"], s["action"]

# Before fusion: regex ∧ choicess.gen("city", regex=r"[A-Z][a-z]+", choices=["Paris", "Tokyo"])# Compiler fuses to: DFA accepting ONLY {"Paris", "Tokyo"}# → States reduced from 12 (regex) + 2 (choices) → 5 (minimal DFA)

@sgl.functiondef demo(s):    x = s.gen("x", choices=[1,2,3])    if x == "4":  # ← Impossible! choices=[1,2,3]        s.gen("y", regex=r"a+")  # ← Dead branch

s.gen("zip", regex=r"94\d{3}")  # After "94", only "0"-"9" allowed# If prompt already has "943", next must be digit → no sampling needed

[User Program] → SGLang Compiler → Constraint IR                          ↓[vLLM Engine] ← SGLang Runtime Adapter  ├─ PagedAttention (Block Table + Constraint State)  ├─ Custom Sampler (Constrained Sampling Kernel)  └─ Speculative Decoding (with Constraint Propagation)

class ConstrainedSampler(Sampler):    def __init__(self, constraint_engine: ConstraintEngine):        self.constraint_engine = constraint_engine    def forward(self, logits: torch.Tensor,                 request_states: List[RequestState]) -> torch.Tensor:        # 1. Get allowed tokens for each request        masks = self.constraint_engine.get_masks(request_states)                # 2. Apply mask: set disallowed tokens to -inf        logits = logits.masked_fill(~masks, float('-inf'))                # 3. Call original sampling (top-p, temperature, etc.)        return super().forward(logits)

# Draft model also uses constrained sampling!draft_tokens = draft_model.generate_constrained(prompt, constraint_ir)# Verification: Check both logits AND constraint statefor i, token in enumerate(draft_tokens):    if not constraint_ir.update(token):  # ← Constraint violated!        accepted = draft_tokens[:i]        break    if not acceptance_sampling(logits[i], token):        accepted = draft_tokens[:i]        break

with sgl.debug():    result = extract_entities.run("John works at Google in Mountain View.")

[Step 0] Prompt: "Text: John works at Google in Mountain View.\nEntities:\n"[Step 1] → gen("person", stop=",")          Allowed: [A-Za-z]+ → Tokens: ['J','o','h','n'] → "John"[Step 2] → gen("org", regex=r"[A-Z][a-z]+( [A-Z][a-z]+)*")         DFA State: S0 → 'G' → S1 → 'o' → ... → "Google"[Step 3] → gen("loc", choices=[...])         Allowed: ["Mountain View"] → "Mountain View"✅ Success: ('John', 'Google', 'Mountain View')

@sgl.functiondef safe_chat(s, user_input):    # Auto-escape dangerous tokens    clean_input = s.sanitize(user_input, policy="llm-safe")    s += f"User: {clean_input}\n"    s += "Assistant: " + s.gen("response",                             max_tokens=100,                             prevent=["<script>", "http://"])

@sgl.functiondef label_news_article(s, text):    with s.json_object():        s += '"headline": "' + s.gen("headline", max_tokens=20) + '",\n'        s += '"entities": ['        for i in range(5):            if i > 0: s += ", "            with s.json_object():                s += '"type": "' + s.gen("type", choices=["PERSON","ORG"]) + '", '                s += '"name": "' + s.gen("name", regex=r"[A-Z][a-z]+( [A-Z][a-z]+)*") + '"'        s += ']'    return s.value  # ← Guaranteed valid JSON

# "Always cite source before stating fact"@sgl.functiondef fact_checking_agent(s, query):    s += f"Query: {query}\n"    with s.ltl_constraint("[] (fact → ◇source)"):  # Always fact implies eventually source        s += "Answer: "        while not s.finished():            token = s.gen()            if is_fact_statement(token):                 s.require_future("source")  # ← Sets constraint flag            if token == "Source:":                s.fulfill("source")         # ← Clears flag