Loading...
Loading...
Compare original and translation side by side
Skill by ara.so — Daily 2026 Skills collection.
flash-linear-attentionchunk_kda由ara.so开发的技能——Daily 2026技能合集。
flash-linear-attentionchunk_kdagit clone https://github.com/MoonshotAI/FlashKDA.git flash-kda
cd flash-kda
git submodule update --init --recursive
pip install -v .pip install -U flash-linear-attention # >= 0.5.0git clone https://github.com/MoonshotAI/FlashKDA.git flash-kda
cd flash-kda
git submodule update --init --recursive
pip install -v .pip install -U flash-linear-attention # >= 0.5.0flash_kda.fwdflash_kda.fwdimport torch
import flash_kda
flash_kda.fwd(
q, k, v, g, beta, scale, out,
A_log, dt_bias, lower_bound,
initial_state=None,
final_state=None,
cu_seqlens=None
)| Parameter | Dtype | Shape | Notes |
|---|---|---|---|
| bf16 | | Query; K must be 128 |
| bf16 | | Key; K must be 128 |
| bf16 | | Value; V must be 128 |
| bf16 | | Gate logits (sigmoid/activation applied internally) |
| bf16 | | Beta logits (sigmoid applied internally) |
| float | scalar | Attention scale factor |
| bf16 | | Pre-allocated output tensor |
| fp32 | | Per-head log-gate parameter |
| fp32 | | Per-head gate bias |
| float | scalar | Gate lower bound, range |
| bf16/fp32/None | | Optional initial recurrent state |
| bf16/fp32/None | | Optional output final state |
| int64 | | Optional cumulative seq lengths for varlen |
K == V == 128cu_seqlensBTinitial_statefinal_stateimport torch
import flash_kda
flash_kda.fwd(
q, k, v, g, beta, scale, out,
A_log, dt_bias, lower_bound,
initial_state=None,
final_state=None,
cu_seqlens=None
)| 参数 | 数据类型 | 形状 | 说明 |
|---|---|---|---|
| bf16 | | 查询向量;K必须为128 |
| bf16 | | 键向量;K必须为128 |
| bf16 | | 值向量;V必须为128 |
| bf16 | | 门控logits(内部会应用sigmoid/激活函数) |
| bf16 | | Beta logits(内部会应用sigmoid函数) |
| float | 标量 | 注意力缩放因子 |
| bf16 | | 预分配的输出张量 |
| fp32 | | 每头的log门控参数 |
| fp32 | | 每头的门控偏置 |
| float | 标量 | 门控下界,范围为 |
| bf16/fp32/None | | 可选的初始循环状态 |
| bf16/fp32/None | | 可选的输出最终状态 |
| int64 | | 可选的变长序列累积长度 |
K == V == 128cu_seqlensBTinitial_statefinal_statechunk_kdaimport torch
import logging
from fla.ops.kda import chunk_kdachunk_kdaimport torch
import logging
from fla.ops.kda import chunk_kdaundefinedundefinedimport torch
import flash_kda
def run_flash_kda(
q, k, v, g, beta,
A_log, dt_bias,
lower_bound=-5.0,
initial_state=None,
):
B, T, H, K = q.shape
V = v.shape[-1]
scale = K ** -0.5
out = torch.empty(B, T, H, V, dtype=torch.bfloat16, device=q.device)
final_state = torch.zeros(B, H, V, K, dtype=torch.float32, device=q.device)
flash_kda.fwd(
q, k, v, g, beta,
scale, out,
A_log, dt_bias, lower_bound,
initial_state=initial_state,
final_state=final_state,
cu_seqlens=None,
)
return out, final_state
B, T, H, K = 1, 4096, 8, 128
device = 'cuda'
dtype = torch.bfloat16
q = torch.randn(B, T, H, K, device=device, dtype=dtype)
k = torch.randn(B, T, H, K, device=device, dtype=dtype)
v = torch.randn(B, T, H, K, device=device, dtype=dtype) # V==K==128
g = torch.randn(B, T, H, K, device=device, dtype=dtype)
beta = torch.randn(B, T, H, device=device, dtype=dtype)
A_log = torch.full((H,), -0.1, device=device, dtype=torch.float32)
dt_bias = torch.zeros(H, K, device=device, dtype=torch.float32)
with torch.inference_mode():
out, state = run_flash_kda(q, k, v, g, beta, A_log, dt_bias)
print(out.shape) # [1, 4096, 8, 128]
print(state.shape) # [1, 8, 128, 128]import torch
import flash_kda
def run_flash_kda(
q, k, v, g, beta,
A_log, dt_bias,
lower_bound=-5.0,
initial_state=None,
):
B, T, H, K = q.shape
V = v.shape[-1]
scale = K ** -0.5
out = torch.empty(B, T, H, V, dtype=torch.bfloat16, device=q.device)
final_state = torch.zeros(B, H, V, K, dtype=torch.float32, device=q.device)
flash_kda.fwd(
q, k, v, g, beta,
scale, out,
A_log, dt_bias, lower_bound,
initial_state=initial_state,
final_state=final_state,
cu_seqlens=None,
)
return out, final_state
B, T, H, K = 1, 4096, 8, 128
device = 'cuda'
dtype = torch.bfloat16
q = torch.randn(B, T, H, K, device=device, dtype=dtype)
k = torch.randn(B, T, H, K, device=device, dtype=dtype)
v = torch.randn(B, T, H, K, device=device, dtype=dtype) # V==K==128
g = torch.randn(B, T, H, K, device=device, dtype=dtype)
beta = torch.randn(B, T, H, device=device, dtype=dtype)
A_log = torch.full((H,), -0.1, device=device, dtype=torch.float32)
dt_bias = torch.zeros(H, K, device=device, dtype=torch.float32)
with torch.inference_mode():
out, state = run_flash_kda(q, k, v, g, beta, A_log, dt_bias)
print(out.shape) # [1, 4096, 8, 128]
print(state.shape) # [1, 8, 128, 128]cu_seqlensimport torch
import flash_kdacu_seqlensimport torch
import flash_kdaundefinedundefinedinitial_stateimport torch
import flash_kda
H, K, V = 16, 128, 128
B = 2
scale = K ** -0.5
def inference_step(q, k, v, g, beta, A_log, dt_bias, state=None):
T = q.shape[1]
out = torch.empty(B, T, H, V, dtype=torch.bfloat16, device='cuda')
new_state = torch.zeros(B, H, V, K, dtype=torch.float32, device='cuda')
flash_kda.fwd(
q, k, v, g, beta, scale, out,
A_log, dt_bias, lower_bound=-5.0,
initial_state=state,
final_state=new_state,
cu_seqlens=None,
)
return out, new_state
A_log = torch.zeros(H, dtype=torch.float32, device='cuda')
dt_bias = torch.zeros(H, K, dtype=torch.float32, device='cuda')
state = None
for chunk_idx in range(4):
q = torch.randn(B, 256, H, K, dtype=torch.bfloat16, device='cuda')
k = torch.randn(B, 256, H, K, dtype=torch.bfloat16, device='cuda')
v = torch.randn(B, 256, H, V, dtype=torch.bfloat16, device='cuda')
g = torch.randn(B, 256, H, K, dtype=torch.bfloat16, device='cuda')
beta = torch.randn(B, 256, H, dtype=torch.bfloat16, device='cuda')
with torch.inference_mode():
out, state = inference_step(q, k, v, g, beta, A_log, dt_bias, state)
print(f"Chunk {chunk_idx}: out={out.shape}, state={state.shape}")initial_stateimport torch
import flash_kda
H, K, V = 16, 128, 128
B = 2
scale = K ** -0.5
def inference_step(q, k, v, g, beta, A_log, dt_bias, state=None):
T = q.shape[1]
out = torch.empty(B, T, H, V, dtype=torch.bfloat16, device='cuda')
new_state = torch.zeros(B, H, V, K, dtype=torch.float32, device='cuda')
flash_kda.fwd(
q, k, v, g, beta, scale, out,
A_log, dt_bias, lower_bound=-5.0,
initial_state=state,
final_state=new_state,
cu_seqlens=None,
)
return out, new_state
A_log = torch.zeros(H, dtype=torch.float32, device='cuda')
dt_bias = torch.zeros(H, K, dtype=torch.float32, device='cuda')
state = None
for chunk_idx in range(4):
q = torch.randn(B, 256, H, K, dtype=torch.bfloat16, device='cuda')
k = torch.randn(B, 256, H, K, dtype=torch.bfloat16, device='cuda')
v = torch.randn(B, 256, H, V, dtype=torch.bfloat16, device='cuda')
g = torch.randn(B, 256, H, K, dtype=torch.bfloat16, device='cuda')
beta = torch.randn(B, 256, H, dtype=torch.bfloat16, device='cuda')
with torch.inference_mode():
out, state = inference_step(q, k, v, g, beta, A_log, dt_bias, state)
print(f"Chunk {chunk_idx}: out={out.shape}, state={state.shape}")| Variable | Values | Effect |
|---|---|---|
| | Set to |
undefined| 变量名 | 取值范围 | 效果 |
|---|---|---|
| | 设置为 |
undefinedundefinedundefinedbash tests/test.shtests/test_fwd.pybash tests/test.shtests/test_fwd.pyimport logging
logging.basicConfig(level=logging.INFO)import logging
logging.basicConfig(level=logging.INFO)undefinedundefinedimport torch
cap = torch.cuda.get_device_capability()
assert cap >= (9, 0), f"FlashKDA requires SM90+, got SM{cap[0]}{cap[1]}"import torch
cap = torch.cuda.get_device_capability()
assert cap >= (9, 0), f"FlashKDA需要SM90+级GPU,当前为SM{cap[0]}{cap[1]}"undefinedundefinedundefinedundefinedtorch.inference_mode()torch.no_grad()torch.inference_mode()torch.no_grad()undefinedundefinedundefinedundefinedundefinedundefinedundefinedundefinedlower_boundlower_boundlower_bound = -5.0 # valid: range is [-5.0, 0]
lower_bound = -2.5 # valid
lower_bound = 0.0 # valid boundary
lower_bound = -10.0 # out of spec — use -5.0 as safe minimumlower_bound = -5.0 # 有效:范围为[-5.0, 0]
lower_bound = -2.5 # 有效
lower_bound = 0.0 # 有效边界
lower_bound = -10.0 # 超出规格——使用-5.0作为安全最小值bash setup_clangd.shbash setup_clangd.shundefinedundefined