38. Prefill Decode Disaggregation | Prefill / Decode 分离

难度： Medium | 环境： CPU-first | 标签： 推理优化, 推理服务, PD Disaggregation | 目标人群： 推理优化学习者

🚀 云端运行环境

本章节的实战代码可以点击以下链接在免费 GPU 算力平台上直接运行：

(国内推荐：魔搭社区免费实例)

---

本节导读

一次请求中的 prefill 更集中地消耗输入处理算力和带宽，decode 则持续占用 KV Cache 并受到排队影响。当一批请求的两类压力明显分化时，可以进一步观察独立资源池是否改善服务表现。

本节用最小教学模拟建立一条判断链：记录请求流量，形成 prefill / decode 分池计划，再比较吞吐、延迟和交接代价。

关键词： prefill, decode, queue, kv residency

---

前置阅读

导语： 进入本节前，先能比较 prefill 与 decode 的计算、带宽和 KV Cache 压力，再判断它们是否值得使用不同的资源池。

• 34. Prefix Caching and Chunked Prefill | 前缀缓存与分块预填充
• 36. Decode Scheduling | Decode 调度
• 37. KV Cache Scheduling | KV Cache 调度

---

Step 1: 先分清 Prefill 与 Decode 的压力来源

一次请求先处理输入，再逐 token 生成输出。Prefill 更集中地消耗输入处理算力和带宽，Decode 则持续占用 KV Cache 并受到排队影响。先按请求记录这两段的规模，才能判断拆分是否有依据。

本节的输入是请求的输入长度、生成长度和当前状态，输出是流量分层、资源池分配以及拆分收益与交接代价的比较。

请求形态	主要压力	先观察什么
长输入、短生成	prefill 的 attention 计算与输入访存	输入处理时间、prefill 吞吐、队列长度
短输入、长生成	decode 迭代、KV Cache 驻留与排队	TPOT、KV Cache 占用、decode 队列
输入和生成都长	两段压力叠加	端到端延迟、吞吐和两类队列

Step 2: 把请求流量写成显式账本

先把每个请求的输入长度、生成长度和当前状态记录下来。这样可以从同一份账本识别流量构成，再比较共享池和拆分方案的资源需求。一条记录可以写成“请求 a：输入 4000 token、生成 64 token、命中前缀、位于 prefill 队列”。

字段类别	记录内容	用途
核心：请求标识	请求名称或 ID	在分类、分池和结果表之间追踪同一个请求
核心：输入规模	prompt_tokens	估计 prefill 工作量和输入处理压力
核心：生成规模	decode_tokens	估计 decode 迭代次数和 KV Cache 驻留时间
状态：前缀复用	cache_hit	说明输入处理是否可能复用已有前缀
状态：所在队列	queue	观察请求当前处于 prefill、decode 或共享队列

Step 3: 比较拆分收益与交接代价

PD 分离需要同时比较收益和代价：独立资源池可能改善某类请求的等待，但也会增加交接、缓存迁移和空闲容量成本。先从吞吐、延迟和资源使用观察收益，再把交接时间纳入判断。CPU 机制实验可以比较方向；真实交接时间、KV Cache 迁移和 worker 空闲容量需要在 backend 中测量。

观察对象	需要比较的指标	读数代表什么
吞吐收益	throughput_gain	拆分后单位时间完成的请求或 token 是否增加
交互延迟	TTFT、TPOT、P95、latency_delta_ms	输入等待、生成速度和长尾请求是否改善
交接成本	handoff_time、KV Cache 迁移时间、通信量	请求从 prefill 池进入 decode 池需要付出的代价
资源代价	两类 worker 的显存、算力和空闲容量	分池后是否出现一侧拥塞、另一侧闲置
最终判断	accept / tune / reject	收益是否足以覆盖交接与资源组织成本

Step 4: 实现请求分类、分池与拆分决策

题目区把前面的账本和判断框架落成三个函数：先统计请求构成，再生成分池计划，最后根据吞吐、延迟和交接代价输出拆分结论。实现时保持输入字段与上面表格一致，便于把模拟结果替换成真实 backend 的记录。 完成 CPU 测试后，再把同样的请求账本和指标字段带入 Step 5；真实 backend / 多 GPU 证据在 Step 5 中记录。

实现位置	输入	输出	测试时观察
summarize_request_mix	请求的 prompt_tokens、decode_tokens 和分类阈值	三类请求数量	三类数量之和等于请求总数，边界请求分类稳定
plan_pd_split	请求名称、两类长度阈值	prefill_pool、decode_pool、shared_pool	每个请求只进入一个池，分池结果保留输入顺序
evaluate_pd_decision	baseline 与 split 的吞吐、P95	增益、延迟变化和 keep_split	吞吐增加且 P95 不恶化时才保留拆分

from typing import Dict, List

def summarize_request_mix(requests: List[Dict[str, int]], long_prompt_threshold: int) -> Dict[str, int]:
    """按输入长度和生成长度统计三类请求。
    
    返回 prefill-heavy、decode-heavy 和 mixed 的请求数量；这里只做流量分类，
    不代表真实 backend 已经完成了资源隔离。

    你需要返回：
    - prefill_heavy: 长 prompt、短 decode 的请求数
    - decode_heavy: 短 prompt、长 decode 的请求数
    - mixed: 其他请求数
    """
    # TODO 1：统计 prefill-heavy、decode-heavy 和 mixed 请求数量。
    # 提示：先创建结果字典，再逐个请求判断属于哪一类。
    # result = ???
    # if ???:
    #     result['prefill_heavy'] += 1
    # elif ???:
    #     result['decode_heavy'] += 1
    # else:
    #     result['mixed'] += 1
    raise NotImplementedError


def plan_pd_split(requests: List[Dict[str, int]], long_prompt_threshold: int, long_decode_threshold: int) -> Dict[str, object]:
    """根据请求规模生成三个逻辑池的分配计划。
    
    返回请求名称列表；它是调度账本，不会启动 worker 或搬运 KV Cache。
    """
    # TODO 2：根据请求规模生成三个逻辑池的分配计划。
    # 提示：先创建三个列表，再根据 prompt/decode 长度把 request name 放进去。
    # prefill_pool = ???
    # decode_pool = ???
    # shared_pool = ???
    raise NotImplementedError


def evaluate_pd_decision(baseline: Dict[str, float], split_run: Dict[str, float]) -> Dict[str, object]:
    """比较共享池与拆分方案的吞吐、P95 延迟，并输出保留建议。
    
    throughput_gain 是绝对差值，latency_delta_ms 是拆分减基线。
    """
    # TODO 3：比较共享池和拆分方案，并输出 keep_split。
    # 提示：先算 throughput_gain 和 latency_delta_ms，再决定 keep_split。
    # throughput_gain = ???
    # latency_delta_ms = ???
    # keep_split = ???
    raise NotImplementedError

def test_pd_disaggregation_template():
    try:
        requests = [
            {'name': 'a', 'prompt_tokens': 4000, 'decode_tokens': 64},
            {'name': 'b', 'prompt_tokens': 256, 'decode_tokens': 512},
            {'name': 'c', 'prompt_tokens': 1500, 'decode_tokens': 128},
        ]
        summary = summarize_request_mix(requests, long_prompt_threshold=2048)
        assert summary == {'prefill_heavy': 1, 'decode_heavy': 1, 'mixed': 1}
        assert sum(summary.values()) == len(requests), '请求分类数量必须守恒！'
        boundary = summarize_request_mix(
            [{'name': 'boundary', 'prompt_tokens': 2048, 'decode_tokens': 256}],
            long_prompt_threshold=2048,
        )
        assert boundary == {'prefill_heavy': 0, 'decode_heavy': 0, 'mixed': 1}, '阈值边界分类不明确！'
        plan = plan_pd_split(requests, long_prompt_threshold=2048, long_decode_threshold=256)
        assert plan['prefill_pool'] == ['a']
        assert plan['decode_pool'] == ['b']
        assert plan['shared_pool'] == ['c']
        planned = plan['prefill_pool'] + plan['decode_pool'] + plan['shared_pool']
        assert sorted(planned) == sorted(item['name'] for item in requests), '分池结果必须覆盖且只覆盖每个请求一次！'
        decision = evaluate_pd_decision(
            {'throughput': 100, 'p95_latency_ms': 180},
            {'throughput': 126, 'p95_latency_ms': 150},
        )
        assert decision['throughput_gain'] == 26
        assert decision['latency_delta_ms'] == -30
        assert decision['keep_split'] is True
        reject = evaluate_pd_decision(
            {'throughput': 100, 'p95_latency_ms': 180},
            {'throughput': 110, 'p95_latency_ms': 210},
        )
        assert reject['throughput_gain'] == 10
        assert reject['latency_delta_ms'] == 30
        assert reject['keep_split'] is False, '延迟恶化时不能只看吞吐收益！'
        print('测试通过：PD 分离模板可以工作。')
    except NotImplementedError:
        raise
    except (NameError, AttributeError) as e:
        raise NotImplementedError('请先完成 TODO 代码或检查字段名！') from e


test_pd_disaggregation_template()

---

🛑 STOP HERE 🛑

请先尝试自己完成代码并跑通测试。
如果你正在 Colab 中运行，并且遇到困难没有思路，可以向下滚动查看参考答案。

---

参考代码与解析

代码

def summarize_request_mix(requests: List[Dict[str, int]], long_prompt_threshold: int) -> Dict[str, int]:
    """按输入长度和生成长度统计三类请求。
    
    返回 prefill-heavy、decode-heavy 和 mixed 的请求数量；这里只做流量分类，
    不代表真实 backend 已经完成了资源隔离。

    你需要返回：
    - prefill_heavy: 长 prompt、短 decode 的请求数
    - decode_heavy: 短 prompt、长 decode 的请求数
    - mixed: 其他请求数
    """
    # 提示：先创建结果字典，再逐个请求判断属于哪一类。
    # result = ???
    # if ???:
    #     result['prefill_heavy'] += 1
    # elif ???:
    #     result['decode_heavy'] += 1
    # else:
    #     result['mixed'] += 1
    # TODO 1：统计 prefill-heavy、decode-heavy 和 mixed 请求数量。
    result = {'prefill_heavy': 0, 'decode_heavy': 0, 'mixed': 0}
    for request in requests:
        prompt_tokens = request.get('prompt_tokens', 0)
        decode_tokens = request.get('decode_tokens', 0)
        if prompt_tokens > long_prompt_threshold and decode_tokens <= long_prompt_threshold // 8:
            result['prefill_heavy'] += 1
        elif decode_tokens > long_prompt_threshold // 8 and prompt_tokens <= long_prompt_threshold:
            result['decode_heavy'] += 1
        else:
            result['mixed'] += 1
    return result


def plan_pd_split(requests: List[Dict[str, int]], long_prompt_threshold: int, long_decode_threshold: int) -> Dict[str, object]:
    """根据请求规模生成三个逻辑池的分配计划。
    
    返回请求名称列表；它是调度账本，不会启动 worker 或搬运 KV Cache。
    """
    # 提示：先创建三个列表，再根据 prompt/decode 长度把 request name 放进去。
    # prefill_pool = ???
    # decode_pool = ???
    # shared_pool = ???
    # TODO 2：根据请求规模生成三个逻辑池的分配计划。
    prefill_pool = []
    decode_pool = []
    shared_pool = []
    for request in requests:
        name = request.get('name', 'request')
        if request.get('prompt_tokens', 0) > long_prompt_threshold and request.get('decode_tokens', 0) <= long_decode_threshold:
            prefill_pool.append(name)
        elif request.get('decode_tokens', 0) > long_decode_threshold and request.get('prompt_tokens', 0) <= long_prompt_threshold:
            decode_pool.append(name)
        else:
            shared_pool.append(name)
    return {'prefill_pool': prefill_pool, 'decode_pool': decode_pool, 'shared_pool': shared_pool}


def evaluate_pd_decision(baseline: Dict[str, float], split_run: Dict[str, float]) -> Dict[str, object]:
    """比较共享池与拆分方案的吞吐、P95 延迟，并输出保留建议。
    
    throughput_gain 是绝对差值，latency_delta_ms 是拆分减基线。
    """
    # 提示：先算 throughput_gain 和 latency_delta_ms，再决定 keep_split。
    # throughput_gain = ???
    # latency_delta_ms = ???
    # keep_split = ???
    # TODO 3：比较共享池和拆分方案，并输出 keep_split。
    throughput_gain = split_run.get('throughput', 0.0) - baseline.get('throughput', 0.0)
    latency_delta_ms = split_run.get('p95_latency_ms', 0.0) - baseline.get('p95_latency_ms', 0.0)
    return {'throughput_gain': throughput_gain, 'latency_delta_ms': latency_delta_ms, 'keep_split': throughput_gain > 0 and latency_delta_ms <= 0}

解析

题目区和答案区保留同样的 3 个 TODO；答案区只补全实现，不改变函数接口。测试函数分别检查分类守恒、分池覆盖和收益/代价方向。

TODO	题目区要完成的机制	测试函数中的对应检查
1	根据输入长度和生成长度统计请求构成	三类数量守恒与阈值边界
2	将请求分配到 prefill、decode 或 shared 池	每个请求只进入一个池且顺序可追踪
3	比较共享池和拆分方案的吞吐、P95 与决策	收益明确时保留，延迟恶化时拒绝

1. TODO 1：统计请求流量构成

• 先读取每个请求的 prompt_tokens 和 decode_tokens，再把请求分成 prefill_heavy / decode_heavy / mixed。
• 这一步的意义不是精确建模，而是先判断流量是不是已经明显异构，是否存在拆池动机。

2. TODO 2：把请求分配到 prefill / decode / shared 三个池

• prefill_pool 放长 prompt、短 decode 的请求；decode_pool 放短 prompt、长 decode 的请求；其余进入 shared_pool。
• 这里先做最小分配账本，不涉及真实 worker 调度，只固定“谁更适合被单独隔离”。

3. TODO 3：输出 PD 分离决策

• 先计算 throughput_gain 和 latency_delta_ms，再判断拆分后的收益是否同时覆盖吞吐和延迟两侧。
• 这一步回答的是“拆池值不值得保留”，而不是“系统能不能拆”。

4. 这页的定位

• PD 分离首先回答“流量是否真的异构”，再回答“拆池是否值得”。
• 请求分类、资源分配和收益评估，构成了最小判断链路。

Step 5: 可选 GPU / backend：验证拆分收益与交接代价

CPU 题目区先验证流量分类、分池计划和决策逻辑；真实 PD 分离需要两个可运行的 worker 池、状态交接和匹配 workload。单 GPU 上的合成计算不能代替 prefill / decode worker 之间的真实交接，因此本步默认只做环境预检和实验配置，具备匹配 backend 后再执行真实对照。

实验路径	使用资产	学习者操作	可以验证 / 不能直接推出
CPU 机制验证	题目区三个函数和测试单元	先完成请求分类、分池计划和 baseline / split 决策	可以验证账本与决策逻辑；不能推出真实 worker 性能
环境预检	tools/environment_preflight.py、当前 Notebook runtime	检查 CUDA、GPU、显存、PyTorch 和可选 backend；需要时执行 python tools/environment_preflight.py --gpu --optional-packages vllm sglang	判断环境是否可执行；不产生 PD 性能结论
单 GPU backend	vLLM / SGLang 服务与统一请求集	可做共享池 baseline 或单 worker smoke test	可以测请求指标；不能证明跨设备交接成本
多 GPU / serving backend	两类 worker、状态交接配置和统一 workload	对比 shared pool 与 PD split，记录 TTFT、TPOT、吞吐、P95、交接时间和失败状态	可以验证当前模型与 workload；不能直接外推到其他 backend
结果登记	本节最后的 PD 实验记录表	每组新配置新增一行，不覆盖已有结果	形成可复核的 PD 对照记录

真实 backend 验证优先参考 70 Serving Scheduler Benchmark 和 81 Distributed Inference Project。本节 GPU 结果必须记录 evidence_level，CPU 模拟结果不能写成真实 PD 分离收益。

"""GPU/backend 配置单元：默认只做配置检查，不启动 backend。"""
RUN_PD_GPU_PREFLIGHT = False  # 改为 True 后检查当前 GPU 和可选 backend。
PD_RESULT_PATH = 'benchmarks/results/38_pd_disaggregation.json'  # 相对项目根目录。
PD_BACKEND = 'vllm'  # vllm / sglang / multi_backend；这里只记录计划，不自动启动服务。
PD_MODEL_ID = 'Qwen/Qwen2.5-0.5B-Instruct'
PD_DTYPE = 'float16'
PD_PROMPT_TOKENS = 2048
PD_DECODE_TOKENS = 256
PD_CONCURRENCY = 2
PD_WORKER_COUNT = 2  # 真实 PD 对照至少需要 prefill / decode 两类 worker。

"""GPU/backend 执行单元：记录环境和实验计划，不伪造 backend 结果。"""
if RUN_PD_GPU_PREFLIGHT:
    import importlib.util
    import json
    from pathlib import Path
    import torch

    if not torch.cuda.is_available():
        raise RuntimeError('PD 实验预检需要 CUDA；请先切换到 GPU runtime。')
    if PD_BACKEND not in {'vllm', 'sglang', 'multi_backend'}:
        raise ValueError('PD_BACKEND 必须是 vllm、sglang 或 multi_backend。')
    if any(value < 1 for value in (PD_PROMPT_TOKENS, PD_DECODE_TOKENS, PD_CONCURRENCY, PD_WORKER_COUNT)):
        raise ValueError('token 数、并发度和 worker 数必须为正数。')

    project_root = next((p for p in [Path.cwd(), *Path.cwd().parents] if (p / 'benchmarks').is_dir()), Path.cwd())
    report = {
        'task': 'pd_disaggregation_preflight',
        'evidence_level': 'gpu_environment_preflight_only',
        'runtime': {
            'device': torch.cuda.get_device_name(0),
            'gpu_memory_gb': round(torch.cuda.get_device_properties(0).total_memory / (1024 ** 3), 2),
            'torch': torch.__version__,
            'torch_cuda': torch.version.cuda,
            'backend_installed': importlib.util.find_spec(PD_BACKEND) is not None if PD_BACKEND != 'multi_backend' else False,
        },
        'config': {'model': PD_MODEL_ID, 'dtype': PD_DTYPE, 'prompt_tokens': PD_PROMPT_TOKENS, 'decode_tokens': PD_DECODE_TOKENS, 'concurrency': PD_CONCURRENCY, 'worker_count': PD_WORKER_COUNT, 'backend': PD_BACKEND},
        'decision': {'decision': 'measure', 'reason': '环境预检完成；真实 PD split 仍需两个 worker 池和统一 workload。'},
    }
    output_path = project_root / PD_RESULT_PATH
    output_path.parent.mkdir(parents=True, exist_ok=True)
    output_path.write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding='utf-8')
    print(json.dumps(report, ensure_ascii=False, indent=2))
else:
    print('PD GPU/backend 预检未启动：将 RUN_PD_GPU_PREFLIGHT 改为 True 后运行。')

PD 实验记录

完成 CPU 或 GPU/backend 实验后，每组新条件新增一行；没有真实 worker 交接数据时，保留为空，不用 CPU 模拟值代填。

配置组	GPU / backend	模型与 dtype	worker 配置	prompt / decode tokens	并发	TTFT / TPOT	吞吐	P95	交接时间	峰值显存	evidence level	decision	结果文件
示例：CPU 账本	CPU / 模拟	待填写	shared / split 计划	待填写	待填写	不适用	不适用	不适用	不适用	不适用	cpu_simulation	待填写	待填写

相关阅读

完成 prefill、decode 和资源池拆分的判断后，可以继续阅读推理系统论文与真实服务基准。

• DistServe 原论文：Disaggregating Prefill and Decoding for Goodput-optimized Large Language Model Serving
• vLLM 官方仓库
• 39. Inference Fallback and Tiers | 推理分层与回退策略
• 66. Inference Performance Comparison | 推理性能对比实验
• 70. Serving Scheduler Benchmark | 推理服务调度基准