进阶系列 (`advanced/`) Implementation Plan

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: 在 advanced/ 下新增 16 步进阶教程(adv01–adv16),覆盖主系列缺失的推理优化技术,风格与主系列一致、独立可运行。

Architecture: 独立 advanced/ 子目录,内部从 adv01 起编号,自带 advanced/README.md 学习路线。每步复用主系列 step07 的 TinyTransformerWithKVCache(importlib shim,路径 ../../step07_kvcache_for_single_request)。能在 CPU 上真实现的写可运行代码;纯架构类(分布式分离)用 Python 模拟器讲清原理。

Tech Stack: Python 3 + PyTorch(CPU 即可),标准库。无新增依赖。

Model API(来自 step07,全系列复用):

python

# TinyTransformerWithKVCache(vocab_size=256, d_model=4, num_heads=1, num_layers=1)
# forward(token_ids: Tensor, past_key_values=None) -> (logits, new_past_key_values)
# KVCache = (K, V) tuple, 各形状 [seq, num_heads, d_head]

复用 model shim(每个 advXX/model.py 都用这个模板,改 module 名即可):

python

"""advXX: 复用 step07 的 TinyTransformerWithKVCache"""
import sys, os, importlib
_path = os.path.join(os.path.dirname(__file__), '..', '..', 'step07_kvcache_for_single_request')
sys.path.insert(0, os.path.abspath(_path))
_mod = importlib.import_module('model')
TinyTransformerWithKVCache = _mod.TinyTransformerWithKVCache

每步 README 固定章节(执行时由子代理填写完整正文):

教学目标(一句话)
问题:主系列为什么没解决这个
原理(ASCII 图)
实现细节(核心代码讲解)
教学版 vs 真实框架(vLLM/SGLang)对比
运行(python run.py)
下一步(指向下一 adv 步)

通用验证(每步的"测试"): cd advanced/advXX_<name> && python run.py 退出码 0 且打印 ✅ advXX_<name> 通过。

通用提交规范: 每步一个 commit,信息形如 docs(advXX): <step name> 进阶教程。

File Structure

advanced/
  README.md                 ← 学习路线 + Phase 分组 + 与主系列衔接
  adv01_quantization/        ← model.py(shim) + quant.py + run.py + README.md
  adv02_sampling_advanced/   ← sampler.py + run.py + README.md
  adv03_speculative_decoding/← draft.py + target.py(均 shim) + spec_decode.py + run.py + README.md
  adv04_flash_decoding/      ← flash_decode.py + run.py + README.md
  adv05_radix_prefix_cache/  ← radix_tree.py + engine.py + run.py + README.md
  adv06_pipeline_parallel/   ← pp_sim.py + run.py + README.md
  adv07_sequence_parallel/   ← sp_sim.py + run.py + README.md
  adv08_data_parallel_dplb/  ← dp_sim.py + run.py + README.md
  adv09_tbo_dbo_overlap/     ← overlap_sim.py + run.py + README.md
  adv10_pd_disaggregation/   ← pd_sim.py + run.py + README.md
  adv11_afd_attention_ffn/   ← afd_sim.py + run.py + README.md
  adv12_moe_eplb/            ← moe_layer.py + run.py + README.md
  adv13_linear_attention/    ← linear_attn.py + run.py + README.md
  adv14_multi_lora/          ← lora.py + multi_lora_engine.py + run.py + README.md
  adv15_guided_decoder/      ← guided.py + run.py + README.md
  adv16_function_call/       ← tool_loop.py + run.py + README.md

顶层 README.md 末尾追加进阶系列指针。

Task 0: 脚手架 — `advanced/README.md` + 顶层指针

Files:

Create: advanced/README.md
Modify: README.md(末尾追加一行指针)
[ ] Step 1: 写 advanced/README.md

内容:标题 # mini-vllm-tutorial 进阶系列,一段引言("主系列 step01–20 覆盖了…仍有 16 项推理优化手段未讲,本进阶系列补齐"),然后 7 个 Phase 的学习路线(把下面这段直接写入):

Phase A — 精度与采样进阶
  adv01_quantization        ← W4A16/W8A16 权重量化
  adv02_sampling_advanced   ← MinP / 惩罚项 / Beam Search

Phase B — 解码加速
  adv03_speculative_decoding ← 投机解码(草稿+验证)
  adv04_flash_decoding       ← 长序列 decode 切分 (+FlashInfer 简介)

Phase C — 缓存结构进阶
  adv05_radix_prefix_cache   ← Radix Attention + Copy-on-Write

Phase D — 并行进阶(承接主系列 step17 TP)
  adv06_pipeline_parallel    ← Pipeline Parallel (PP, 微批次/1F1B)
  adv07_sequence_parallel    ← Sequence Parallel (SP)
  adv08_data_parallel_dplb   ← Data Parallel + DPLB 负载均衡
  adv09_tbo_dbo_overlap      ← TBO/DBO 计算-通信重叠

Phase E — 分离式架构
  adv10_pd_disaggregation    ← Prefill/Decode 分离
  adv11_afd_attention_ffn    ← Attention/FFN 分离

Phase F — 模型架构变体
  adv12_moe_eplb             ← MoE Top-k 路由 + EPLB
  adv13_linear_attention     ← 线性注意力 / SSM

Phase G — 服务与输出控制
  adv14_multi_lora           ← Multi-LoRA 动态切换
  adv15_guided_decoder       ← JSON/regex 结构化输出
  adv16_function_call        ← Function Call / Tool Call

再加"学习方式"段(沿用主系列:先读 README→运行 run.py→读代码→diff 相邻步)和"前置:建议先学完主系列 step01–20"。

[ ] Step 2: 顶层 README 追加指针

在 README.md 文件末尾(## 与 nano-vllm 的关系 段之后,或文末)追加:

markdown


## 进阶系列

学完主系列 15 步后,继续 16 步进阶优化(量化/投机解码/PD 分离/MoE/结构化输出…):
→ 见 [advanced/README.md](advanced/README.md)

[ ] Step 3: 验证

Run: ls advanced/README.md && grep -q "进阶系列" README.md && echo OK Expected: OK

[ ] Step 4: Commit

bash

git add advanced/README.md README.md
git commit -m "docs(adv): 新增进阶系列学习路线与顶层指针"

Task 1: adv01 — 量化 Quantization (W4A16 / W8A16) 🟢

Files:

Create: advanced/adv01_quantization/model.py(shim)
Create: advanced/adv01_quantization/quant.py
Create: advanced/adv01_quantization/run.py
Create: advanced/adv01_quantization/README.md

核心实现 quant.py(weight-only,对称量化):

python

import torch

def quantize_weight(w: torch.Tensor, bits: int = 8):
    """对称 per-tensor 量化。返回 (qweight_int, scale)。bits=4 或 8。"""
    qmax = (1 << (bits - 1)) - 1          # 7 (int4) 或 127 (int8)
    scale = w.abs().max() / qmax
    q = torch.round(w / scale).clamp(-qmax, qmax).to(torch.int8)  # 用 int8 容器存
    return q, scale

def dequantize_weight(q: torch.Tensor, scale: torch.Tensor, bits: int = 8):
    return q.to(torch.float32) * scale

class QuantizedLinear:
    """W4A16/W8A16: 权重低比特存,激活 FP16,matmul 前反量化。"""
    def __init__(self, linear: torch.nn.Linear, bits: int = 8):
        self.bits = bits
        self.q, self.scale = quantize_weight(linear.weight.data, bits)
        self.bias = linear.bias

    def __call__(self, x: torch.Tensor):
        w = dequantize_weight(self.q, self.scale, self.bits).to(x.dtype)
        out = x @ w.t()
        if self.bias is not None:
            out = out + self.bias
        return out

def quantize_model(model, bits: int = 8):
    """把 model 的所有 nn.Linear 换成 QuantizedLinear,原地替换。"""
    import torch.nn as nn
    for name, mod in model.named_children():
        if isinstance(mod, nn.Linear):
            setattr(model, name, QuantizedLinear(mod, bits))  # 简化:仅顶层属性
        else:
            quantize_model(mod, bits)  # 递归子模块
    return model

注:named_children+setattr 只覆盖直接属性层;对 TinyTransformer(单层)够用。README 说明真实框架用 per-channel/groupwise + 打包存储。

[ ] Step 1: 写 quant.py(上面代码)
[ ] Step 2: 写 run.py — 构建 TinyTransformer,对比 FP32 vs INT8 vs INT4:① 输出 logits 数值接近(相对误差 < 1e-2);② 统计权重大小(INT8≈1/4, INT4≈1/8);③ 打印 ✅ adv01_quantization 通过。
[ ] Step 3: 写 README.md(7 章节固定结构,ASCII 图:FP32 权重 vs 量化后存储布局;对比 vLLM 的 AWQ/GPTQ/FP8)
[ ] Step 4: 验证 cd advanced/adv01_quantization && python run.py → 退出 0 且打印通过标记
[ ] Step 5: Commit docs(adv01): 量化 Quantization 进阶教程

Task 2: adv02 — 采样进阶 (MinP / 惩罚项 / Beam Search) 🟢

Files:

Create: advanced/adv02_sampling_advanced/sampler.py
Create: advanced/adv02_sampling_advanced/run.py
Create: advanced/adv02_sampling_advanced/README.md

核心 sampler.py(新增函数,基于 step06 风格):

python

import torch

def min_p_sample(logits: torch.Tensor, min_p: float, temperature: float = 1.0) -> torch.Tensor:
    """保留概率 >= max_prob * min_p 的候选,再从中采样。"""
    probs = torch.softmax(logits / temperature, dim=-1)
    max_p = probs.max()
    mask = probs >= max_p * min_p
    filtered = torch.where(mask, probs, torch.zeros_like(probs))
    return torch.multinomial(filtered, num_samples=1).squeeze(-1)

def apply_frequency_penalty(logits, token_ids, penalty):
    """频率惩罚:出现次数越多,logit 衰减越多。"""
    freq = torch.bincount(token_ids, minlength=logits.size(-1)).float()
    return logits - penalty * freq

def apply_presence_penalty(logits, token_ids, penalty):
    """存在惩罚:出现过就减固定值(至多一次)。"""
    appeared = torch.bincount(token_ids, minlength=logits.size(-1)).clamp(0, 1)
    return logits - penalty * appeared

def apply_repetition_penalty(logits, token_ids, penalty):
    """重复惩罚:出现过的 token,logit 除以 penalty(>1 则降低)。"""
    appeared = torch.bincount(token_ids, minlength=logits.size(-1)).clamp(0, 1).bool()
    logits = logits.clone()
    logits[appeared] = logits[appeared] / penalty
    return logits

def beam_search(model, prompt_ids, beam_width=3, max_new=10):
    """束搜索:每步保留 top-beam 个候选序列。返回最优序列。"""
    beams = [(prompt_ids, 0.0)]  # (token_ids, 累积 logprob)
    for _ in range(max_new):
        all_cands = []
        for ids, score in beams:
            logits, _ = model(ids[-1:], past_key_values=None)  # 简化:每步全量重算
            logp = torch.log_softmax(logits[-1], dim=-1)
            topk = torch.topk(logp, beam_width)
            for v, idx in zip(topk.values, topk.indices):
                all_cands.append((torch.cat([ids, idx.view(1)]), score + v.item()))
        beams = sorted(all_cands, key=lambda c: c[1])[-beam_width:]
    return beams[-1][0]

[ ] Step 1: 写 sampler.py
[ ] Step 2: 写 run.py — 构造假 logits 张量,断言:MinP 过滤后概率和≈1;三种 penalty 对已出现 token 的 logit 降低;Beam Search 输出长度 == prompt_len + max_new。打印通过标记。
[ ] Step 3: 写 README.md(ASCII:TopK vs TopP vs MinP 的候选集差异;Beam Search 宽度=2 的搜索树)
[ ] Step 4: 验证 cd advanced/adv02_sampling_advanced && python run.py
[ ] Step 5: Commit docs(adv02): 采样进阶(MinP/惩罚/Beam)教程

Task 3: adv03 — 投机解码 Speculative Decoding 🟢

Files:

Create: advanced/adv03_speculative_decoding/model.py(shim)
Create: advanced/adv03_speculative_decoding/spec_decode.py
Create: advanced/adv03_speculative_decoding/run.py
Create: advanced/adv03_speculative_decoding/README.md

核心 spec_decode.py(草稿模型 = 更小 num_layers/d_model;同词表):

python

import torch

def draft_speculate(draft_model, token_ids, past_kv_draft, k=4):
    """草稿模型一次生成 k 个候选 token。返回 (draft_tokens, draft_probs, new_past_kv)。"""
    tokens, probs = [], []
    cur = token_ids
    for _ in range(k):
        logits, past_kv_draft = draft_model(cur[-1:], past_key_values=past_kv_draft)
        p = torch.softmax(logits[-1], dim=-1)
        nxt = p.argmax(dim=-1)
        tokens.append(nxt.item()); probs.append(p)
        cur = torch.cat([cur, nxt.view(1)])
    return tokens, probs, past_kv_draft

def target_verify(target_model, draft_tokens, draft_probs, token_ids, past_kv_target):
    """目标模型一次前向验证所有草稿 token,按概率比接受/拒绝(经典 speculative sampling)。
    返回接受的 token 列表 + 是否到达末尾。教学简化:用 argmax 比对。"""
    logits, past_kv_target = target_model(
        torch.tensor(draft_tokens, dtype=torch.long), past_key_values=past_kv_target
    )
    target_p = torch.softmax(logits, dim=-1)
    accepted = []
    for i, dt in enumerate(draft_tokens):
        tgt = target_p[i].argmax().item()
        if tgt == dt:
            accepted.append(dt)          # 接受
        else:
            accepted.append(tgt)         # 拒绝,用目标模型的 token,后续丢弃
            break                        # 第一个不匹配即停(教学版)
    return accepted, past_kv_target

[ ] Step 1: 写 spec_decode.py
[ ] Step 2: 写 run.py — 用 step07 模型当 target,更小配置(d_model=2,num_layers=1)当 draft。对比:① 普通自回归生成 N token 的 forward 次数;② 投机解码生成同样 N token 的 forward 次数(应更少)。断言投机解码结果与纯自回归在贪婪模式下完全一致(因都 argmax)。打印通过标记 + 两次 forward 次数对比。
[ ] Step 3: 写 README.md(ASCII:草稿串行生成→目标并行验证→接受/拒绝;对比 vLLM EAGLE/Medusa、SGLang}
[ ] Step 4: 验证 cd advanced/adv03_speculative_decoding && python run.py
[ ] Step 5: Commit docs(adv03): 投机解码 Speculative Decoding 教程

Task 4: adv04 — Flash-Decoding (+ FlashInfer 简介) 🟡

Files:

Create: advanced/adv04_flash_decoding/flash_decode.py
Create: advanced/adv04_flash_decoding/run.py
Create: advanced/adv04_flash_decoding/README.md

核心 flash_decode.py(演示 split-K 思想,不依赖 CUDA):

python

import torch, math

def naive_decode_attention(q, K, V):
    """单 token decode:q [heads,d_head],K/V [seq,heads,d_head]。"""
    scores = torch.einsum('hd,shd->sh', q, K) / math.sqrt(q.size(-1))  # [seq, heads]
    # 简化:逐 head softmax
    attn = torch.softmax(scores, dim=0)  # [seq, heads]
    return torch.einsum('sh,shd->hd', attn, V)  # [heads, d_head]

def flash_decode_splitk(q, K, V, num_splits=4):
    """把长 KV 切成 num_splits 段并行算,再合并(split-K)。
    教学版用串行循环模拟"各段并行",展示分段→局部 max/sum→全局归约。"""
    seq = K.size(0)
    chunk = (seq + num_splits - 1) // num_splits
    local_outs, local_max, local_sum = [], [], []
    for s in range(num_splits):
        lo, hi = s*chunk, min((s+1)*chunk, seq)
        if lo >= hi: continue
        Kc, Vc = K[lo:hi], V[lo:hi]
        scores = torch.einsum('hd,shd->sh', q, Kc) / math.sqrt(q.size(-1))
        m = scores.max(dim=0).values
        p = torch.exp(scores - m.unsqueeze(0))
        s_ = p.sum(dim=0)
        o = torch.einsum('sh,shd->hd', p, Vc)
        local_outs.append(o); local_max.append(m); local_sum.append(s_)
    # 全局归约(online softmax 合并)
    Gm = torch.stack(local_max).max(dim=0).values
    Go = torch.zeros_like(q)
    Gs = torch.zeros_like(q)
    for o, m, s in zip(local_outs, local_max, local_sum):
        w = torch.exp(m - Gm)
        Go += o * w; Gs += s * w
    return Go / Gs

[ ] Step 1: 写 flash_decode.py
[ ] Step 2: 写 run.py — 构造 q/K/V,断言 flash_decode_splitk 与 naive_decode_attention 输出 allclose(atol=1e-5)。打印通过标记。README 注明:单机串行无真实加速,图解展示长 KV 切多 SM 并行。
[ ] Step 3: 写 README.md(ASCII:长 KV 按段切到多 SM,每段局部 softmax,全局 online 合并;简介 FlashInfer 库对各阶段算子优化)
[ ] Step 4: 验证 cd advanced/adv04_flash_decoding && python run.py
[ ] Step 5: Commit docs(adv04): Flash-Decoding 长序列 decode 切分教程

Task 5: adv05 — Radix Attention + Copy-on-Write 🟢

Files:

Create: advanced/adv05_radix_prefix_cache/radix_tree.py
Create: advanced/adv05_radix_prefix_cache/engine.py
Create: advanced/adv05_radix_prefix_cache/run.py
Create: advanced/adv05_radix_prefix_cache/README.md

核心 radix_tree.py(基数树前缀复用 + CoW):

python

class RadixNode:
    def __init__(self, key, value=None):
        self.key = list(key)          # 该节点代表的 token 序列片段
        self.value = value            # 叶子:对应的 KV cache 引用
        self.children = {}            # token -> RadixNode
        self.ref = 0                  # 引用计数(被多少请求共享)

class RadixTree:
    def __init__(self):
        self.root = RadixNode([])

    def _split(self, node, key, idx):
        """把 node 在 key[idx] 处分裂:公共前缀留原节点,后缀变新子节点。"""
        child = RadixNode(node.key[idx:], node.value)
        child.children = node.children
        node.key = node.key[:idx]
        node.value = None
        node.children = {node.key[idx]: child}  # 简化:idx 处分叉
        node.children[child.key[0]] = child

    def insert(self, tokens, value):
        node, i = self.root, 0
        while i < len(tokens):
            tok = tokens[i]
            if tok not in node.children:
                node.children[tok] = RadixNode(tokens[i:], value)
                node.children[tok].ref = 1
                return
            child = node.children[tok]
            # 找公共前缀长度
            j = 0
            while j < len(child.key) and i+j < len(tokens) and child.key[j] == tokens[i+j]:
                j += 1
            if j < len(child.key):
                self._split(child, child.key, j)  # CoW 触发点:前缀部分共享,后续分叉
                child = node.children[tok]
            i += j
            if i >= len(tokens):
                break
            node = child
        node.ref += 1
        node.value = value

    def match_prefix(self, tokens):
        """返回最长已缓存前缀的长度 + 对应 value(供复用 KV)。"""
        node, i, hit_len, hit_val = self.root, 0, 0, None
        while i < len(tokens) and tokens[i] in node.children:
            child = node.children[tokens[i]]
            j = 0
            while j < len(child.key) and i+j < len(tokens) and child.key[j] == tokens[i+j]:
                j += 1
            i += j
            if child.value is not None:
                hit_len, hit_val = i, child.value
            node = child
        return hit_len, hit_val

[ ] Step 1: 写 radix_tree.py
[ ] Step 2: 写 engine.py — 用 RadixTree 做 prefix cache:新请求来,match_prefix 查命中长度,命中的 KV 复用(跳过 prefill 那段),不命中则 prefill 后 insert。CoW:两个请求共享前缀节点(ref=2),某请求后续 token 不同时 _split 分叉,各自拥有独立后缀 KV。
[ ] Step 3: 写 run.py — 插入 ["sys","prompt","A"] 与 ["sys","prompt","B"],断言:match_prefix(["sys","prompt","A"]) 命中长度=3;两个请求共享 "sys","prompt" 前缀节点(ref=2);分叉后各自独立。打印通过标记。
[ ] Step 4: 写 README.md(ASCII:基数树多叉结构 + CoW 分叉;对比主系列 step13 hash 方案 vs SGLang RadixAttention)
[ ] Step 5: 验证 cd advanced/adv05_radix_prefix_cache && python run.py
[ ] Step 5: Commit docs(adv05): Radix Attention + CoW 前缀缓存教程

Task 6: adv06 — Pipeline Parallel (PP) 🟡

Files:

Create: advanced/adv06_pipeline_parallel/pp_sim.py
Create: advanced/adv06_pipeline_parallel/run.py
Create: advanced/adv06_pipeline_parallel/README.md

核心 pp_sim.py(模拟 GPipe 微批次 + 1F1B):

python

import time

class Device:
    def __init__(self, name, layers, comm_latency=0.01, fwd_time=0.02, bwd_time=0.04):
        self.name, self.layers = name, layers
        self.comm_latency, self.fwd_time, self.bwd_time = comm_latency, fwd_time, bwd_time

def gpipe_schedule(devices, num_microbatches):
    """GPipe:所有 microbatch 正向跑完,再反向。返回总时间 + 时间轴事件列表。"""
    events = []
    t = 0.0
    # 前向:每个设备依次处理所有 microbatch
    for mb in range(num_microbatches):
        for d in devices:
            t += d.comm_latency
            events.append((t, d.name, 'F', mb))
            t += d.fwd_time
    # 反向:倒序
    for mb in range(num_microbatches):
        for d in reversed(devices):
            t += d.comm_latency
            events.append((t, d.name, 'B', mb))
            t += d.bwd_time
    return t, events

def onef_oneb_schedule(devices, num_microbatches):
    """1F1B:稳定后每设备一前一向一反向,显存占用更小。返回总时间 + 事件。"""
    events = []
    t = 0.0
    # 简化:交错排列前向/反向,演示 bubble 缩小
    n = num_microbatches
    for i in range(2*n):
        for d in devices:
            t += d.comm_latency
            phase = 'F' if i < n else 'B'
            mb = i if i < n else (2*n-1-i)
            events.append((t, d.name, phase, mb))
            t += d.fwd_time if phase=='F' else d.bwd_time
    return t, events

[ ] Step 1: 写 pp_sim.py
[ ] Step 2: 写 run.py — 4 设备 × 4 microbatch,跑两种调度,断言 1F1B 的"最大同时驻留 microbatch 数" < GPipe(打印每设备显存占用峰值对比)。打印通过标记。注意:总时间可能 1F1B 不更短(单机串行模拟),README 诚实说明模拟只演示"显存气泡"差异非真实加速。
[ ] Step 3: 写 README.md(ASCII:GPipe 全前向后反向 vs 1F1B 交错的 pipeline 时间轴 + bubble;对比 vLLM/DeepSpeed PP)
[ ] Step 4: 验证 cd advanced/adv06_pipeline_parallel && python run.py
[ ] Step 5: Commit docs(adv06): Pipeline Parallel 模拟器教程

Task 7: adv07 — Sequence Parallel (SP) 🟡

Files:

Create: advanced/adv07_sequence_parallel/sp_sim.py
Create: advanced/adv07_sequence_parallel/run.py
Create: advanced/adv07_sequence_parallel/README.md

核心 sp_sim.py(序列维切分 + all-gather / reduce-scatter 模拟):

python

import torch

def sp_attention(q, k, v, seq_splits=2):
    """把序列维切成 seq_splits 段(模拟分到多卡),各段算 attention 后拼回。
    教学版串行循环模拟"各卡并行"。演示 LayerNorm/Attention 的序列切分。"""
    seq = q.size(0)
    chunk = (seq + seq_splits - 1) // seq_splits
    outs = []
    for s in range(seq_splits):
        lo, hi = s*chunk, min((s+1)*chunk, seq)
        if lo >= hi: continue
        qc, kc, vc = q[lo:hi], k[lo:hi], v[lo:hi]
        scores = qc @ kc.transpose(-2,-1) / (qc.size(-1) ** 0.5)
        attn = torch.softmax(scores, dim=-1)
        outs.append(attn @ vc)
    return torch.cat(outs, dim=0)

def all_gather(local_shard, dim=0):
    """模拟 all-gather:各卡分片拼成完整张量。教学版单进程直接 cat。"""
    return torch.cat([local_shard, local_shard.clone()], dim=dim)  # 2 卡模拟

def reduce_scatter(full, dim=0):
    """模拟 reduce-scatter:规约后按卡切分。教学版:sum 后切两半各取一。"""
    summed = full + full  # 模拟两卡各有一份相加
    half = full.size(dim) // 2
    return summed.narrow(dim, 0, half)

[ ] Step 1: 写 sp_sim.py
[ ] Step 2: 写 run.py — 构造 q/k/v,断言 sp_attention 与朴素 attention allclose;演示 all_gather/reduce_scatter 形状正确。打印通过标记。
[ ] Step 3: 写 README.md(ASCII:序列维切分到多卡 + LayerNorm all-gather / Attention reduce-scatter 通信;对比 Megatron-LM SP)
[ ] Step 4: 验证 cd advanced/adv07_sequence_parallel && python run.py
[ ] Step 5: Commit docs(adv07): Sequence Parallel 模拟器教程

Task 8: adv08 — Data Parallel + DPLB 🟢

Files:

Create: advanced/adv08_data_parallel_dplb/dp_sim.py
Create: advanced/adv08_data_parallel_dplb/run.py
Create: advanced/adv08_data_parallel_dplb/README.md

核心 dp_sim.py(多副本 + 负载均衡路由):

python

import torch, time

class Replica:
    def __init__(self, name, speed=1.0):
        self.name, self.speed = name, speed
        self.queue = []   # 待处理 token 数
        self.in_flight = 0
    def load(self):
        return self.in_flight + sum(self.queue)
    def step(self, dt):
        done = min(self.in_flight, int(self.speed * dt * 100))
        self.in_flight -= done
        while self.queue and self.in_flight < 100:
            self.in_flight += self.queue.pop(0)

class RoundRobinLB:
    def route(self, replicas, req_size):  # 轮询
        r = replicas[len([1]) % len(replicas)]  # 简化:始终选第一个(占位,实际用计数器)
        return min(replicas, key=lambda r: r.name)  # 占位,run.py 中用真 RR

class LeastLoadLB:  # = DPLB 的核心:按负载路由
    def route(self, replicas, req_size):
        return min(replicas, key=lambda r: r.load())

def simulate(replicas, lb, arrivals, total_time=50):
    """arrivals: [(time, req_size), ...]。返回各副本完成数 + 是否有副本过载。"""
    log = []
    for t in range(total_time):
        # 注入到达
        while arrivals and arrivals[0][0] == t:
            _, size = arrivals.pop(0)
            r = lb.route(replicas, size)
            r.queue.append(size)
        for r in replicas:
            r.step(1)
        log.append({r.name: r.load() for r in replicas})
    return log

[ ] Step 1: 写 dp_sim.py
[ ] Step 2: 写 run.py — 3 副本(速度不同模拟异构),注入突发负载。对比 RoundRobin vs LeastLoad(DPLB),断言 LeastLoad 下"各副本负载方差"更小。打印通过标记 + 负载分布对比。
[ ] Step 3: 写 README.md(ASCII:多副本 + LB 路由;对比 DPLB 数据并行负载均衡)
[ ] Step 4: 验证 cd advanced/adv08_data_parallel_dplb && python run.py
[ ] Step 5: Commit docs(adv08): Data Parallel + DPLB 负载均衡教程

Task 9: adv09 — TBO/DBO 计算-通信重叠 🟡

Files:

Create: advanced/adv09_tbo_dbo_overlap/overlap_sim.py
Create: advanced/adv09_tbo_dbo_overlap/run.py
Create: advanced/adv09_tbo_dbo_overlap/README.md

核心 overlap_sim.py(微批次交错,计算流与通信流重叠):

python

import time
from concurrent.futures import ThreadPoolExecutor

def compute(t):  # 模拟 attention 计算
    time.sleep(t)
    return 'compute_done'

def comm(t):     # 模拟 dispatch/combine 通信
    time.sleep(t)
    return 'comm_done'

def no_overlap(microbatches, ct=0.05, mt=0.03):
    """朴素:每 microbatch 先算再通信,串行。"""
    t0 = time.time()
    for mb in microbatches:
        compute(ct); comm(mt)
    return time.time() - t0

def tbo_overlap(microbatches, ct=0.05, mt=0.03):
    """TBO:微批次 i 的通信 与 微批次 i+1 的计算 重叠(用线程模拟并行)。"""
    t0 = time.time()
    with ThreadPoolExecutor(max_workers=2) as ex:
        prev_comm = None
        for mb in microbatches:
            if prev_comm:
                prev_comm.result()       # 等 i-1 的通信完(与 i 的计算重叠过)
            comp = ex.submit(compute, ct)
            prev_comm = ex.submit(comm, mt)
            comp.result()
        if prev_comm: prev_comm.result()
    return time.time() - t0

[ ] Step 1: 写 overlap_sim.py
[ ] Step 2: 写 run.py — 8 microbatch,断言 tbo_overlap < no_overlap(应接近 (ct+mt) vs max(ct,mt)*n 级别改善)。打印通过标记 + 两种耗时对比。
[ ] Step 3: 写 README.md(ASCII:朴素串行 vs TBO 计算通信交错时间轴;对比 DeepSeek TBO/DBO)
[ ] Step 4: 验证 cd advanced/adv09_tbo_dbo_overlap && python run.py
[ ] Step 5: Commit docs(adv09): TBO/DBO 计算通信重叠教程

Task 10: adv10 — PD Disaggregation (Prefill/Decode 分离) 🟡

Files:

Create: advanced/adv10_pd_disaggregation/pd_sim.py
Create: advanced/adv10_pd_disaggregation/run.py
Create: advanced/adv10_pd_disaggregation/README.md

核心 pd_sim.py(分离的 prefill 引擎 + decode 引擎 + KV 迁移):

python

import time

class PrefillEngine:
    def __init__(self, speed=1.0): self.speed = speed; self.busy = 0
    def prefill(self, prompt_len):
        """prefill 耗时 ∝ prompt_len²(算力密集)。"""
        t = (prompt_len ** 2) / (self.speed * 1e6)
        self.busy += t
        return {'kv_size': prompt_len}, t

class DecodeEngine:
    def __init__(self, speed=1.0): self.speed = speed; self.busy = 0
    def decode(self, kv_state, steps):
        """decode 每步 ∝ kv_size(存储密集)。"""
        t = kv_state['kv_size'] * steps / (self.speed * 1e6)
        self.busy += t
        return t

def transfer_kv(kv_state, latency=0.01):
    """模拟 KV 从 prefill 节点迁移到 decode 节点。"""
    time.sleep(latency)
    return latency

def colocated(reqs, p_speed=1.0, d_speed=1.0):
    """合并部署:同一引擎既 prefill 又 decode,互相阻塞。"""
    pe, de = PrefillEngine(p_speed), DecodeEngine(d_speed)
    total = 0
    for prompt_len, steps in reqs:
        kv, t1 = pe.prefill(prompt_len)
        t2 = de.decode(kv, steps)
        total += t1 + t2
    return total

def disaggregated(reqs, p_speed=1.0, d_speed=1.0, kv_latency=0.01):
    """分离部署:prefill 与 decode 各自独立配比,KV 迁移。"""
    pe, de = PrefillEngine(p_speed), DecodeEngine(d_speed)
    total = 0
    for prompt_len, steps in reqs:
        kv, t1 = pe.prefill(prompt_len)
        total += transfer_kv(kv, kv_latency)
        t2 = de.decode(kv, steps)
        total += t1 + t2
    return total

[ ] Step 1: 写 pd_sim.py
[ ] Step 2: 写 run.py — 多请求,对比 colocated vs disaggregated,断言:分离后 prefill/decode 可独立扩容(改 p_speed/d_speed 配比),长 prompt 场景下分离吞吐更高(打印两组耗时 + 各引擎利用率)。打印通过标记。
[ ] Step 3: 写 README.md(ASCII:合并部署互相阻塞 vs PD 分离 + KV 迁移;对比 DeepSeek/VLLM PD disaggregation、Mooncake KVStore)
[ ] Step 4: 验证 cd advanced/adv10_pd_disaggregation && python run.py
[ ] Step 5: Commit docs(adv10): PD Disaggregation 分离部署教程

Task 11: adv11 — AFD (Attention-FFN Disaggregation) 🟡

Files:

Create: advanced/adv11_afd_attention_ffn/afd_sim.py
Create: advanced/adv11_afd_attention_ffn/run.py
Create: advanced/adv11_afd_attention_ffn/README.md

核心 afd_sim.py(Attention 与 FFN 分到不同"设备",按 A/F 配比):

python

import torch, time

class AttentionDevice:
    def __init__(self, n=1, t=0.02): self.n, self.t = n, t  # n 个并行单元
    def forward(self, x):
        time.sleep(self.t / self.n)
        return x  # 占位:真实应做 attention

class FFNDevice:
    def __init__(self, n=1, t=0.03): self.n, self.t = n, t
    def forward(self, x):
        time.sleep(self.t / self.n)
        return x

def run_layer(seq_len, attn_dev, ffn_dev, a_f_ratio):
    """一层 = Attention + FFN。a_f_ratio 决定 A/F 设备配比。
    教学版:固定吞吐,演示调 A/F 配比让两端利用率均衡。"""
    x = torch.zeros(seq_len)
    x = attn_dev.forward(x)
    # 模拟 dispatch(把激活送 FFN 设备)
    x = ffn_dev.forward(x)
    return x

def balanced_config(attn_time, ffn_time, target_util=1.0):
    """计算 A/F 配比使两端耗时接近(利用率均衡)。"""
    # A 单元数 : F 单元数 = attn_time : ffn_time 的反比
    a_units = max(1, int(round(ffn_time / attn_time * 2)) // 1)
    f_units = max(1, int(round(attn_time / ffn_time * 2)) // 1)
    return a_units, f_units

[ ] Step 1: 写 afd_sim.py
[ ] Step 2: 写 run.py — 设 attn/ffn 耗时不等,算出均衡 A/F 配比,断言均衡后两端利用率差 < 10%。打印通过标记 + 配比对比。
[ ] Step 3: 写 README.md(ASCII:Attention 与 FFN 分设备,A/F 配比调优;对比原论文 AFD)
[ ] Step 4: 验证 cd advanced/adv11_afd_attention_ffn && python run.py
[ ] Step 5: Commit docs(adv11): AFD Attention-FFN 分离教程

Task 12: adv12 — MoE + EPLB 🟢

Files:

Create: advanced/adv12_moe_eplb/moe_layer.py
Create: advanced/adv12_moe_eplb/run.py
Create: advanced/adv12_moe_eplb/README.md

核心 moe_layer.py(Top-k 路由 + 专家负载统计 + 重均衡):

python

import torch, torch.nn as nn, torch.nn.functional as F

class MoELayer(nn.Module):
    def __init__(self, d_model, num_experts=4, top_k=2):
        super().__init__()
        self.num_experts, self.top_k = num_experts, top_k
        self.gate = nn.Linear(d_model, num_experts, bias=False)
        self.experts = nn.ModuleList([nn.Linear(d_model, d_model) for _ in range(num_experts)])
    def forward(self, x):  # x: [seq, d_model]
        scores = F.softmax(self.gate(x), dim=-1)          # [seq, num_experts]
        topk_val, topk_idx = torch.topk(scores, self.top_k, dim=-1)
        topk_val = topk_val / topk_val.sum(dim=-1, keepdim=True)
        out = torch.zeros_like(x)
        for i in range(self.top_k):
            for e in range(self.num_experts):
                mask = (topk_idx[:, i] == e)
                if mask.any():
                    out[mask] += topk_val[mask, i:i+1] * self.experts[e](x[mask])
        # 记录每个专家被选中的 token 数(负载)
        load = torch.bincount(topk_idx.view(-1), minlength=self.num_experts)
        return out, load

def expert_imbalance(load):
    """负载不均衡度:最大/最小专家负载比。"""
    return load.max().float() / (load.min().float() + 1e-6)

def eplb_rebalance(load, num_devices):
    """EPLB:把专家重新分配到设备,使每设备负载尽量均。
    教学版:按负载排序后贪心装箱到 num_devices 个设备。"""
    experts_sorted = torch.argsort(load, descending=True)
    device_load = [0]*num_devices
    assignment = {}
    for e in experts_sorted.tolist():
        d = device_load.index(min(device_load))
        assignment[e] = d
        device_load[d] += load[e].item()
    return assignment, device_load

[ ] Step 1: 写 moe_layer.py
[ ] Step 2: 写 run.py — 构造 MoE,跑一批 token,断言 expert_imbalance > 1(天然不均),eplb_rebalance 后各设备负载方差更小。打印通过标记 + 路由分布 + 重均衡结果。
[ ] Step 3: 写 README.md(ASCII:Top-k 路由 + 专家负载柱状图 + EPLB 重分布;对比 DeepSeek-MoE、vLLM MoE)
[ ] Step 4: 验证 cd advanced/adv12_moe_eplb && python run.py
[ ] Step 5: Commit docs(adv12): MoE + EPLB 专家负载均衡教程

Task 13: adv13 — Linear Attention / SSM 🟢

Files:

Create: advanced/adv13_linear_attention/linear_attn.py
Create: advanced/adv13_linear_attention/run.py
Create: advanced/adv13_linear_attention/README.md

核心 linear_attn.py(线性注意力 + 与标准 attention 对比):

python

import torch, torch.nn as nn, math

def standard_attention(q, k, v):
    """q/k/v: [seq, d]. 标准 softmax attention,O(seq² d)。"""
    scores = q @ k.transpose(-2, -1) / math.sqrt(q.size(-1))  # [seq, seq]
    return torch.softmax(scores, dim=-1) @ v

def linear_attention(q, k, v, feature_map=None):
    """线性 attention:用 φ(q)φ(k)^T 替代 softmax,O(seq d²)。
    feature_map 默认 elu+1。可递推:第 t 步输出 = φ(q_t) @ (Σ φ(k_i)v_i)。"""
    if feature_map is None:
        feature_map = lambda x: torch.nn.functional.elu(x) + 1
    qf, kf = feature_map(q), feature_map(k)        # [seq, d]
    # 递推形式(展示 decode 时 O(1) 更新)
    S = torch.zeros(q.size(-1), q.size(-1))        # 状态矩阵
    out = []
    for t in range(q.size(0)):
        S = S + torch.outer(kf[t], v[t])           # 状态更新
        o = qf[t] @ S                               # 输出
        out.append(o)
    return torch.stack(out)

class LinearAttentionLayer(nn.Module):
    def __init__(self, d_model):
        super().__init__()
        self.wq = nn.Linear(d_model, d_model, bias=False)
        self.wk = nn.Linear(d_model, d_model, bias=False)
        self.wv = nn.Linear(d_model, d_model, bias=False)
    def forward(self, x):
        return linear_attention(self.wq(x), self.wk(x), self.wv(x))

[ ] Step 1: 写 linear_attn.py
[ ] Step 2: 写 run.py — 断言:linear 与 standard 输出形状一致(数值不同,因近似);linear 的递推形式与矩阵形式结果一致(allclose)。打印通过标记 + 两种复杂度对比说明。
[ ] Step 3: 写 README.md(ASCII:softmax 注意力 O(seq²) vs 线性 φ(q)φ(k) 状态递推 O(seq);提及 SSM/Mamba 状态空间模型同源思想;对比 Mamba/Linear Attention 路线)
[ ] Step 4: 验证 cd advanced/adv13_linear_attention && python run.py
[ ] Step 5: Commit docs(adv13): Linear Attention / SSM 线性注意力教程

Task 14: adv14 — Multi-LoRA 动态切换 🟢

Files:

Create: advanced/adv14_multi_lora/lora.py
Create: advanced/adv14_multi_lora/multi_lora_engine.py
Create: advanced/adv14_multi_lora/run.py
Create: advanced/adv14_multi_lora/README.md

核心 lora.py(LoRA 层 + 多适配器切换):

python

import torch, torch.nn as nn

class LoRALinear(nn.Module):
    """包裹一个 nn.Linear,加可切换的低秩适配器 A/B。base 冻结。"""
    def __init__(self, base: nn.Linear, r=2, num_adapters=2):
        super().__init__()
        self.base = base
        for p in self.base.parameters(): p.requires_grad = False
        self.r = r
        self.adapters = nn.ModuleDict({
            str(i): nn.ModuleList([
                nn.Linear(base.out_features, r, bias=False),   # 注意:形状按 base
                nn.Linear(r, base.in_features, bias=False),    # 简化教学版,见 README
            ]) for i in range(num_adapters)
        })
        self.active = '0'
    def set_adapter(self, idx): self.active = str(idx)
    def forward(self, x):
        out = self.base(x)
        A, B = self.adapters[self.active]
        return out + B(A(x))  # 教学:演示旁路增量

class MultiLoRAEngine:
    """按请求动态选 adapter。"""
    def __init__(self, model, num_adapters):
        self.model = model
        self.num_adapters = num_adapters
    def generate(self, token_ids, adapter_idx, steps=4):
        # 把模型里所有 LoRALinear 切到 adapter_idx
        for m in self.model.modules():
            if isinstance(m, LoRALinear): m.set_adapter(adapter_idx)
        # 简化:返回固定输出证明切换生效(实际应 forward)
        return adapter_idx

[ ] Step 1: 写 lora.py
[ ] Step 2: 写 run.py — 包一个 nn.Linear 成 LoRALinear(2 适配器),断言:set_adapter(0) 与 set_adapter(1) 输出不同(因 A/B 随机初始化不同);切换后立即生效。打印通过标记。
[ ] Step 3: 写 README.md(ASCII:base 模型 + 多个 LoRA 旁路按请求切换;对比 vLLM/PEFT Multi-LoRA serving)
[ ] Step 4: 验证 cd advanced/adv14_multi_lora && python run.py
[ ] Step 5: Commit docs(adv14): Multi-LoRA 多适配器切换教程

Task 15: adv15 — Guided Decoder (JSON/regex 结构化输出) 🟢

Files:

Create: advanced/adv15_guided_decoder/guided.py
Create: advanced/adv15_guided_decoder/run.py
Create: advanced/adv15_guided_decoder/README.md

核心 guided.py(基于 regex 的 token 掩码约束,简化版):

python

import re

def build_allowed_chars(pattern):
    """从 regex 提取允许的字符集(简化:只支持字符类 [..] 和字面量)。"""
    # 教学版:直接返回 regex 编译后的字符集近似——用暴力法生成小字符集
    return set(chr(i) for i in range(256) if re.match(pattern, chr(i), re.PARTIAL_MATCH))

def mask_logits(logits, allowed_token_ids):
    """把不在 allowed 集合的 token logit 置 -inf。"""
    mask = torch.full_like(logits, float('-inf'))
    mask[allowed_token_ids] = 0
    return logits + mask

class RegexGuide:
    """维护当前已生成文本,根据 regex 前缀匹配决定下一步允许的 token。"""
    def __init__(self, pattern, tokenizer_vocab):
        self.pattern, self.vocab = pattern, tokenizer_vocab
        self.generated = ''
    def next_allowed(self, logits):
        # 暴力:对每个候选 token,试拼到 generated 后,若仍能 partial match regex 则允许
        import torch
        allowed = []
        for tid, tok in self.vocab.items():
            trial = self.generated + tok
            try:
                if re.match(self.pattern, trial, re.PARTIAL_MATCH):
                    allowed.append(tid)
            except re.error:
                continue
        return mask_logits(logits, torch.tensor(allowed))
    def consume(self, token_str):
        self.generated += token_str

教学简化:re.PARTIAL_MATCH 在 Python 3.11+ 支持;低版本用 regex 库或放宽为字符级。README 注明。

[ ] Step 1: 写 guided.py
[ ] Step 2: 写 run.py — 用 regex r'-?\d+(\.\d+)?'(数字),构造假 logits+小词表,贪婪采样多步,断言最终生成串完全匹配 regex。打印通过标记 + 生成结果。
[ ] Step 3: 写 README.md(ASCII:每步用 regex 前缀匹配裁剪候选 token;对比 vLLM/Outlines/lm-format-enforcer 的 CFG/指导解码)
[ ] Step 4: 验证 cd advanced/adv15_guided_decoder && python run.py
[ ] Step 5: Commit docs(adv15): Guided Decoder 结构化输出教程

Task 16: adv16 — Function Call / Tool Call 🟢

Files:

Create: advanced/adv16_function_call/tool_loop.py
Create: advanced/adv16_function_call/run.py
Create: advanced/adv16_function_call/README.md

核心 tool_loop.py(基于 adv15 guided,约束模型输出工具调用 JSON,执行后回填):

python

import json, re

TOOL_SCHEMA = {
    "get_weather": {"args": ["city"], "returns": "str"},
    "calculator": {"args": ["expr"], "returns": "str"},
}

def fake_model_output(prompt, forced_json):
    """教学版:不跑真模型,直接返回被 guided decoder 约束出的工具调用 JSON。"""
    return forced_json

def parse_tool_call(text):
    """从模型输出解析 {name, args}。失败返回 None。"""
    m = re.search(r'\{.*\}', text, re.S)
    if not m: return None
    try:
        obj = json.loads(m.group(0))
        if 'name' in obj and 'args' in obj: return obj
    except json.JSONDecodeError: pass
    return None

def execute_tool(call):
    """模拟执行工具,返回结果字符串。"""
    name, args = call['name'], call['args']
    if name == 'get_weather':
        return f"{args.get('city','?')}: 晴 25°C"
    if name == 'calculator':
        try: return str(eval(args.get('expr','0')))
        except Exception as e: return f"err: {e}"
    return "unknown tool"

def tool_loop(user_query, max_iters=3):
    """ReAct 风格循环:模型输出工具调用 → 执行 → 把结果拼回 prompt → 直到给出最终答案。"""
    prompt = user_query
    for _ in range(max_iters):
        # guided decoder 约束输出为合法工具调用 JSON(教学版硬编码)
        out = fake_model_output(prompt, '{"name":"get_weather","args":{"city":"北京"}}')
        call = parse_tool_call(out)
        if call is None:
            return prompt + " → 最终答案"   # 不再调用工具
        result = execute_tool(call)
        prompt += f"\n[tool:{call['name']}]->{result}"
    return prompt

[ ] Step 1: 写 tool_loop.py
[ ] Step 2: 写 run.py — 跑 tool_loop,断言:解析出工具调用、execute_tool 返回非空、最终 prompt 含 [tool: 标记。打印通过标记 + 循环轨迹。
[ ] Step 3: 写 README.md(ASCII:ReAct 循环 model↔tool,guided decoder 保证输出合法 JSON;对比 OpenAI function calling、vLLM guided tool)
[ ] Step 4: 验证 cd advanced/adv16_function_call && python run.py
[ ] Step 5: Commit docs(adv16): Function Call / Tool Call 教程

Task 17: 收尾 — 全系列联跑 + SUMMARY 索引

Files:

Modify: SUMMARY.md(追加进阶系列索引小节)
[ ] Step 1: 全系列联跑验证

Run(对每个 adv 目录):

bash

for d in advanced/adv*/; do echo "=== $d ==="; (cd "$d" && python run.py >/dev/null 2>&1 && echo "PASS: $d" || echo "FAIL: $d"); done

Expected: 全部 PASS。任何 FAIL 回到对应 Task 修复。

[ ] Step 2: SUMMARY.md 追加进阶系列索引

在 SUMMARY.md 末尾追加:

markdown


## 进阶系列索引(adv01–adv16)

| 优化手段 | 对应步骤 | 主要收益 | 代价 |
|---------|---------|---------|------|
| 量化 W4A16/W8A16 | adv01 | 显存↓4–8×,速度↑ | 精度损失 |
| 采样进阶 MinP/Penalty/Beam | adv02 | 输出可控 | — |
| 投机解码 | adv03 | decode 2–3× | 草稿模型开销 |
| Flash-Decoding | adv04 | 长序列 decode↑ | 需要分块 |
| Radix + CoW | adv05 | 前缀复用↑ | 树管理 |
| Pipeline Parallel | adv06 | 支持更大模型 | bubble |
| Sequence Parallel | adv07 | 序列维切分 | 通信 |
| Data Parallel + DPLB | adv08 | 多副本均衡 | 副本开销 |
| TBO/DBO | adv09 | 计算-通信重叠 | 复杂调度 |
| PD Disaggregation | adv10 | 吞吐↑ | KV 迁移 |
| AFD | adv11 | A/F 配比均衡 | 跨设备通信 |
| MoE + EPLB | adv12 | 稀疏激活省算力 | 路由+均衡 |
| Linear Attention/SSM | adv13 | 长序列 O(n) | 精度近似 |
| Multi-LoRA | adv14 | 多任务共享 base | 适配器管理 |
| Guided Decoder | adv15 | 结构化输出 | 约束开销 |
| Function Call | adv16 | 工具调用 | 循环延迟 |

[ ] Step 3: Commit docs(adv): 全系列联跑通过 + SUMMARY 进阶索引

Self-Review (执行前自查)

Spec 覆盖: spec 16 项 → Task 1–16 一一对应 + Task 0 脚手架 + Task 17 收尾。✓ Placeholder 扫描: 每步含核心代码;run.py 断言标准已写明;README 章节固定 7 段(正文执行时填)。无 "TODO/TBD"。✓ 类型一致性: 复用 TinyTransformerWithKVCache(token_ids, past_key_values) 调用一致;shim 路径 ../../step07_kvcache_for_single_request 全系列统一。✓ 风险标注: 🟡 模拟器步骤的"加速"为原理演示,README 须诚实标注(已在各 Task 注明)。✓

执行方式选择

Plan complete and saved to docs/superpowers/plans/2026-07-15-advanced-series.md. Two execution options:

1. Subagent-Driven (recommended) - 每个 Task 派一个 fresh 子代理写完整内容,Task 间我做 review,快速迭代。适合 16 个独立 step。

2. Inline Execution - 在本会话用 executing-plans 批量执行,带 checkpoint 回顾。

Which approach?

进阶系列 (advanced/) Implementation Plan ​

File Structure ​

Task 0: 脚手架 — advanced/README.md + 顶层指针 ​

Task 1: adv01 — 量化 Quantization (W4A16 / W8A16) 🟢 ​

Task 2: adv02 — 采样进阶 (MinP / 惩罚项 / Beam Search) 🟢 ​

Task 3: adv03 — 投机解码 Speculative Decoding 🟢 ​

Task 4: adv04 — Flash-Decoding (+ FlashInfer 简介) 🟡 ​

Task 5: adv05 — Radix Attention + Copy-on-Write 🟢 ​

Task 6: adv06 — Pipeline Parallel (PP) 🟡 ​

Task 7: adv07 — Sequence Parallel (SP) 🟡 ​

Task 8: adv08 — Data Parallel + DPLB 🟢 ​

Task 9: adv09 — TBO/DBO 计算-通信重叠 🟡 ​

Task 10: adv10 — PD Disaggregation (Prefill/Decode 分离) 🟡 ​

Task 11: adv11 — AFD (Attention-FFN Disaggregation) 🟡 ​

Task 12: adv12 — MoE + EPLB 🟢 ​

Task 13: adv13 — Linear Attention / SSM 🟢 ​

Task 14: adv14 — Multi-LoRA 动态切换 🟢 ​

Task 15: adv15 — Guided Decoder (JSON/regex 结构化输出) 🟢 ​

Task 16: adv16 — Function Call / Tool Call 🟢 ​

Task 17: 收尾 — 全系列联跑 + SUMMARY 索引 ​

Self-Review (执行前自查) ​

执行方式选择 ​

进阶系列 (`advanced/`) Implementation Plan

File Structure

Task 0: 脚手架 — `advanced/README.md` + 顶层指针

Task 1: adv01 — 量化 Quantization (W4A16 / W8A16) 🟢

Task 2: adv02 — 采样进阶 (MinP / 惩罚项 / Beam Search) 🟢

Task 3: adv03 — 投机解码 Speculative Decoding 🟢

Task 4: adv04 — Flash-Decoding (+ FlashInfer 简介) 🟡

Task 5: adv05 — Radix Attention + Copy-on-Write 🟢

Task 6: adv06 — Pipeline Parallel (PP) 🟡

Task 7: adv07 — Sequence Parallel (SP) 🟡

Task 8: adv08 — Data Parallel + DPLB 🟢

Task 9: adv09 — TBO/DBO 计算-通信重叠 🟡

Task 10: adv10 — PD Disaggregation (Prefill/Decode 分离) 🟡

Task 11: adv11 — AFD (Attention-FFN Disaggregation) 🟡

Task 12: adv12 — MoE + EPLB 🟢

Task 13: adv13 — Linear Attention / SSM 🟢

Task 14: adv14 — Multi-LoRA 动态切换 🟢

Task 15: adv15 — Guided Decoder (JSON/regex 结构化输出) 🟢

Task 16: adv16 — Function Call / Tool Call 🟢

Task 17: 收尾 — 全系列联跑 + SUMMARY 索引

Self-Review (执行前自查)

执行方式选择