DSL 程序走读：从原语到执行计划

谁该读这一篇？ 想用 SGLang DSL 写真实业务程序（agent / RAG / 多分支）的开发者。 前置阅读： 02-core-concepts/02-frontend-dsl.md。 耗时： 40 分钟 学完能： 1. 写一个完整的 ReAct agent； 2. 写一个"k 路推理 + 投票"的程序； 3. 解释客户端 StreamExecutor 怎么把 DSL 翻成 HTTP 请求； 4. 调试 DSL 程序（看 prompt、看 cache、看 fork 是否真的共享）； 5. 在 DSL 和 OpenAI API 之间做出正确的取舍。

1. 案例 1：简单 QA

import sglang as sgl

sgl.set_default_backend(sgl.RuntimeEndpoint("http://localhost:30000"))

@sgl.function
def qa(s, question):
    s += sgl.system("You are concise and accurate.")
    s += sgl.user(question)
    s += sgl.assistant_begin()
    s += sgl.gen("answer", max_tokens=200, temperature=0.1)
    s += sgl.assistant_end()

print(qa.run(question="什么是 RadixAttention？")["answer"])

要点：

sgl.system / user / assistant_begin/end 自动按模型 chat template 渲染。
s += sgl.gen(name, ...) 把生成结果存进 s[name]。
默认非流式；stream=True 触发流式。

2. 案例 2：受限选择（select）

@sgl.function
def classify(s, text):
    s += sgl.system("You classify text into emotions.")
    s += sgl.user(f"Text: {text}\nWhat emotion?")
    s += sgl.assistant_begin()
    s += sgl.select("emotion", choices=["happy", "sad", "angry", "neutral"])
    s += sgl.assistant_end()

print(classify.run(text="It's a wonderful day!")["emotion"])
# → "happy"

select 在 runtime 算每个 choice 的概率，挑分数最高的。比 OpenAI 的 tool_choice 严格得多——这里物理上不可能生成 choices 之外的字。

3. 案例 3：并行多分支（fork + 投票）

@sgl.function
def reason_and_vote(s, question, num_branches=5):
    s += sgl.system("Think step by step.")
    s += sgl.user(question)
    forks = s.fork(num_branches)
    for i, f in enumerate(forks):
        f += sgl.assistant_begin()
        f += sgl.gen(f"reasoning_{i}", max_tokens=300, temperature=0.7)
        f += sgl.assistant_end()
    # 收集答案
    answers = [forks[i][f"reasoning_{i}"] for i in range(num_branches)]
    s += "Candidate answers:\n"
    for a in answers:
        s += f"- {a}\n"
    s += "Final majority answer: "
    s += sgl.gen("final", max_tokens=100, temperature=0.0)

state = reason_and_vote.run(question="2+3+5=?")
print(state["final"])

为什么这比客户端循环 5 次快：

5 个 fork 共享 system + user prefix（约 200 token）。
Runtime 用 Cascade Attention 一次性算 5 个分支的 attention。
客户端循环 5 次需要 5 次完整 prefill。

5 个分支接近线性并发；不开 fork 时延约 5×。

4. 案例 4：受约束 JSON 输出

schema = {
    "type": "object",
    "properties": {
        "name": {"type": "string"},
        "age": {"type": "integer"},
        "email": {"type": "string"}
    },
    "required": ["name", "age"]
}

@sgl.function
def extract(s, text):
    s += sgl.user(f"Extract person info from: {text}")
    s += sgl.assistant_begin()
    s += sgl.gen("person", max_tokens=200, json_schema=json.dumps(schema))
    s += sgl.assistant_end()

result = json.loads(extract.run(text="Alice is 30, alice@x.com")["person"])
# {"name": "Alice", "age": 30, "email": "alice@x.com"}

runtime 用压缩 FSM 强制输出符合 schema。

5. 案例 5：完整 ReAct Agent

TOOLS = {
    "weather": lambda q: f"{q}: sunny 22°C",
    "calculator": lambda q: str(eval(q)),
    "search": lambda q: f"search results for: {q}",
}

@sgl.function
def react_agent(s, question, max_steps=4):
    s += sgl.system(
        "You are a tool-using assistant. Use:\n"
        "Thought: <think>\nAction: <tool_name>\n"
        "Action Input: <input>\n"
        "Observation: <result>\n"
        "Repeat until you have the answer, then output:\n"
        "Final Answer: <answer>"
    )
    s += sgl.user(question)
    s += sgl.assistant_begin()

    for step in range(max_steps):
        s += "Thought: "
        s += sgl.gen(f"thought_{step}", max_tokens=200, stop="\n")
        s += "\nAction: "
        s += sgl.select(f"action_{step}", choices=list(TOOLS.keys()) + ["FINISH"])
        if s[f"action_{step}"] == "FINISH":
            break
        s += "\nAction Input: "
        s += sgl.gen(f"input_{step}", max_tokens=100, stop="\n")
        obs = TOOLS[s[f"action_{step}"]](s[f"input_{step}"])
        s += f"\nObservation: {obs}\n"

    s += "\nFinal Answer: "
    s += sgl.gen("final", max_tokens=300)
    s += sgl.assistant_end()

print(react_agent.run(question="今天上海天气如何，温度乘以 2 等于多少？")["final"])

和纯 OpenAI 实现的差距：

system prompt 长（200 token），每次 react 步骤会复用，命中率 100%。
select 严格限定 action 选项，不会出现幻觉 tool name。
多步迭代共享前面所有 step 的 KV。

实测在 SGLang 上比"客户端循环 + OpenAI API"快 3-5×。

6. StreamExecutor 内部

源码：lang/interpreter.py:274 (class StreamExecutor)。

调用栈：

qa.run(question="...")
    └── _run_program (interpreter.py:57)
            └── StreamExecutor.__init__
            └── StreamExecutor.run
                    └── for expr in program.expressions:
                            └── execute(expr)
                                    └── 触发 backend HTTP 请求

核心数据：

prompt: str 当前累积的完整 prompt 字符串。
variables: Dict[str, str] 命名结果。
stream_var_event: Dict[str, Event] 流式 sync 用。

执行逻辑：每个 sgl.gen(...) 在 executor 里变成一个 HTTP /generate 请求；带上当前 prompt，runtime 命中 cache 后只 prefill 末尾差异。

fork 拷贝当前 executor state，多个子 executor 并发。

7. 调试技巧

7.1 看完整 prompt

state = qa.run(question="...")
print(state.text())                       # 拼接后的完整 prompt + 生成

7.2 看 metadata

print(state.get_meta_info("answer"))
# {'tokens': 42, 'completion_time': 0.123, ...}

7.3 看 cache 命中

curl http://localhost:30000/metrics | grep cache_hit_rate

跑两次同样的程序，第二次命中率应该接近 100%。

7.4 看 fork 是否真的共享

启动 server 时加 --log-level debug，看 scheduler 日志里有没有 cascade 字样。或者跑 benchmark 对比 fork=1 和 fork=5 的总时间，应该接近线性而非 5×。

8. 性能 cheatsheet

写法	runtime 收益
多次 `qa.run(...)` 用相同 system	100% cache 命中 system
`s.fork(N)` 并发分支	shared prefix attention 一次算
长字符串拼一次给 LLM	比多次小请求快
`select` 替代 free-form gen + parse	省 forward + 0 失败率
短 max_tokens	显存压力小、可调度更多并发

9. DSL vs OpenAI API：什么时候选 DSL

场景	推荐
单轮独立 chat	OpenAI API（简单）
多轮 chat（保留历史）	OpenAI API + 把 history 当 messages
Agent + 多 tool 切换	DSL（select 限定 action）
K 路并行推理	DSL（fork）
严格 JSON 输出	两者都行，DSL 略快
多步流水线（链式 gen）	DSL（cache 命中）
团队全员 Python，无 LLM 经验	OpenAI API（学习曲线低）

DSL 不是必选项，但复杂程序场景它的"压榨 runtime 复用能力"无可替代。

10. 错误处理

try:
    state = qa.run(question="...")
except sgl.SglEndpointError as e:
    print("HTTP 错误:", e)
except sgl.SglRuntimeError as e:
    print("DSL 执行错误:", e)

常见错误：

EndpointError：HTTP 调用失败（server 挂、超时）。
RuntimeError：DSL 程序结构问题（fork 之后变量名冲突）。
输出不符合 schema：受约束解码兜底，但极端情况 grammar 报错。

11. 小结

DSL 用 gen / select / fork / role 原语描述程序结构。
StreamExecutor 客户端按表达式顺序执行，每次 gen 走一次 HTTP。
Fork 是 DSL 的杀手特性，配合 RadixAttention 收益 3-5×。
调试用 state.text() 和 /metrics。
复杂 agent / 多分支 / 严格输出场景优先考虑 DSL。

12. 自检

s.fork(5) 在 runtime 里看到的是几个独立请求？

答案

**5 个独立请求**——每个 fork 分支对应一次独立 HTTP `/generate` 调用，从 runtime 视角看就是 5 个 rid。但它们都带了相同的"父 ProgramState rid"作为 cache hint，runtime 用这个 hint 让 `match_prefix` 直接命中父节点 KV，不需 hash 算前缀。实际效果：5 个请求虽然是独立 rid，但共享父节点 KV 的 lock_ref += 5；attention kernel 用 cascade wrapper 把 5 个 Q 共算在同段 K/V 上。

select(choices=[...]) 为什么不可能生成 choices 外的内容？

答案

`select` 不是"生成"——它根本不调采样器。实现：把每个 choice 各自 tokenize，跑一次 forward 拿到 prompt 末尾 logits + 计算每个 choice 的 token 序列的 logprob 累积；选 logprob 最大的 choice，**把对应 token 序列直接写回 ProgramState**。既然不调采样、不生成新 token，输出物理上不可能是 choices 之外的字符串。对比 `gen + regex`：那是生成路径，靠 vocab mask 强制约束，理论上还有数值边角 case；`select` 是分类路径，0 失败率。

ReAct agent 用 DSL 比纯 OpenAI 实现快的核心原因？

答案

三层叠加： (1) **多步 prefix 累积命中**：每个 step 的 prompt = 前面所有 step 拼接，DSL 让 runtime 知道这是同一 ProgramState 续写，`match_prefix` 100% 命中前面；OpenAI 每次客户端发完整 messages，靠后端 hash 盲匹配，命中也只到 block 级。 (2) **`select` 强制 action 限定**：模型不会乱编 tool name，避免 parse 失败重试；OpenAI function calling 偶有 hallucination。 (3) **共享 system prompt 跨调用**：所有 `react_agent.run(...)` 共享同 system，cache 沉淀后 system 部分 0 算力。实测 3-5×。

state.text() 看到的内容和 client 发送的 HTTP 请求一致吗？为什么？

答案

**大体一致但不完全**。`state.text()` 是客户端 ProgramState 累积的拼接字符串（含 system + user + assistant + 生成结果），是发给 server 的 `text` 字段或者推断出 messages 后的等价物。不一致的点：(a) HTTP 请求里还带元数据（sampling_params、rid、cache hint、chat_template 渲染参数）；(b) 多模态请求里 `state.text()` 显示 placeholder（如 ``）而 HTTP 里 image_data 是 base64 / URL；(c) `state.text()` 显示的是模型已经吐出的内容，HTTP 请求里 prompt 截止到当前 gen 开始位置。调试时主要用 `state.text()` 看"模型实际看到的 prompt"对不对。

同一个 @sgl.function 100 次并发调用，runtime 看到的请求数？

答案

取决于函数里有几个 `sgl.gen` / `sgl.select` 调用。 - 函数里 1 个 gen：100 次调用 → 100 个请求。 - 函数里 3 个 gen：100 次调用 → 300 个请求（每个 ProgramState 串行 3 步）。 - 函数里有 `fork(5)`：每次调用产生 1 + 5 = 6 个请求；100 次 → 600 个。 100 次并发的客户端并行用 asyncio gather 提交，runtime 视角下短时间内涌入对应数量的请求；cache-aware 调度 + RadixAttention 让共享 prefix 的请求高效合批。监控时看 `sglang_num_requests_total` 涨速能验证。

13. 下一步

02-multimodal.md — 图像 / 视频。
03-json-schema.md — 受约束 JSON 进阶。
04-tool-use.md — Function calling。
源码：lang/。

上游源码：sglang/python/sglang/lang/interpreter.py。