[CPU] Support FP8 KV cache (#32733)

Co-authored-by: Zheng, Beilei <beilei.zheng@intel.com>
Co-authored-by: mingfeima <mingfei.ma@intel.com>
This commit is contained in:
Xuan Liao
2026-09-02 10:20:54 +08:00
committed by GitHub
co-authored by Zheng, Beilei mingfeima
parent e874ae64cd
commit 26f760d5c0
14 changed files with 952 additions and 262 deletions
@@ -1,4 +1,4 @@
"""Unit tests for FP4 KV cache quantization strategy pattern - no server, no model loading."""
"""Unit tests for KV cache quantization strategies - no server, no model loading."""
from sglang.test.ci.ci_register import register_cpu_ci
@@ -30,6 +30,7 @@ class TestKVCacheQuantRegistry(CustomTestCase):
self.assertIn("nvfp4", KV_CACHE_QUANT_REGISTRY)
self.assertIn("fp4_mx_block16", KV_CACHE_QUANT_REGISTRY)
self.assertIn("cpu_fp8_e4m3", KV_CACHE_QUANT_REGISTRY)
def test_factory_nvfp4(self):
from sglang.srt.layers.quantization.fp4_kv_cache_quant_method import (
@@ -102,6 +103,65 @@ class TestKVCacheQuantRegistry(CustomTestCase):
get_kv_cache_quant_method("unknown_method")
class TestCPUFP8KVCacheMethod(CustomTestCase):
def test_static_scale_quantize_and_store(self):
from sglang.srt.layers.quantization.fp4_kv_cache_quant_method import (
CPUFP8KVCacheMethod,
)
method = CPUFP8KVCacheMethod()
buffers = method.create_buffers(4, 2, 8, 1, "cpu")
loc = torch.tensor([1, 3])
cache_k = torch.randn(2, 2, 8, dtype=torch.bfloat16)
cache_v = torch.randn(2, 2, 8, dtype=torch.bfloat16)
method.quantize_and_store(
buffers["k_buffer"][0],
buffers["v_buffer"][0],
buffers["k_scale_buffer"],
buffers["v_scale_buffer"],
loc,
cache_k,
cache_v,
k_scale=0.5,
v_scale=0.25,
)
torch.testing.assert_close(
buffers["k_buffer"][0][loc].float(),
(cache_k / 0.5).to(torch.float8_e4m3fn).float(),
)
torch.testing.assert_close(
buffers["v_buffer"][0][loc].float(),
(cache_v / 0.25).to(torch.float8_e4m3fn).float(),
)
self.assertIsNone(buffers["k_scale_buffer"])
self.assertIsNone(buffers["v_scale_buffer"])
self.assertEqual(method.compute_cell_size(2, 8, 1, 4), 128)
def test_defaults_to_unit_scales(self):
from sglang.srt.layers.quantization.fp4_kv_cache_quant_method import (
CPUFP8KVCacheMethod,
)
method = CPUFP8KVCacheMethod()
buffers = method.create_buffers(1, 1, 8, 1, "cpu")
cache = torch.ones(1, 1, 8, dtype=torch.bfloat16)
method.quantize_and_store(
buffers["k_buffer"][0],
buffers["v_buffer"][0],
buffers["k_scale_buffer"],
buffers["v_scale_buffer"],
torch.tensor([0]),
cache,
cache,
)
expected = cache.to(torch.float8_e4m3fn)
torch.testing.assert_close(buffers["k_buffer"][0][0], expected[0])
torch.testing.assert_close(buffers["v_buffer"][0][0], expected[0])
class TestNVFP4KVCacheMethod(CustomTestCase):
"""Test NVFP4KVCacheMethod buffer creation and properties."""