注意力后端
所有注意力实现执行的都是相同的计算:每个 token 都会与其他所有 token 进行比较。区别在于如何执行该计算。基础注意力机制的扩展性较差,因为它会在内存中实例化完整的注意力矩阵,从而产生拖慢推理速度的瓶颈。经过优化的实现则通过重新组织数学运算来减少内存访问,从而实现更快、成本更低的推理。
AttentionInterface 提供了经过优化的注意力实现。它将注意力实现与模型实现解耦,从而简化了对不同函数的实验。借助这一一致的接口,您可以轻松添加新的后端。
| 注意力后端 | 描述 |
|---|---|
"flash_attention_3" | 通过进一步重叠操作并更紧密地融合前向和反向传播,提升了 FlashAttention-2 的性能 |
"flash_attention_2" | 将计算划分为更小的块,并使用快速的片上内存 |
"flex_attention" | 一个用于指定自定义注意力模式(稀疏、块局部、滑动窗口)的框架,无需手动编写底层内核 |
"sdpa" | 内置的 缩放点积注意力 PyTorch 实现 |
"paged|flash_attention_3" | FlashAttention-3 的分页版本 |
"paged|flash_attention_2" | FlashAttention-2 的分页版本 |
"paged|sdpa" | SDPA 的分页版本 |
"paged|eager" | Eager 的分页版本 |
设置注意力后端
使用 from_pretrained() 中的 attn_implementation 参数,可以实例化一个使用特定注意力函数的模型。
import torch
from transformers import AutoModelForCausalLM
model = AutoModelForCausalLM.from_pretrained(
"meta-llama/Llama-3.2-1B", attn_implementation="flash_attention_2"
)使用 set_attn_implementation() 可在运行时切换注意力后端,而无需重新加载模型。
model.set_attn_implementation("sdpa")内核
在运行时,使用 Kernels 库直接从 Hub 下载并加载已编译的计算内核。这可以避免因 PyTorch 或 CUDA 版本不匹配而导致的打包问题。
Kernels 在检测到后会自动注册到 AttentionInterface。您无需显式安装 FlashAttention 包。按名称请求 FlashAttention 时,也会回退到 Hub kernel,参见 FlashAttention fallback。
import torch
from transformers import AutoModelForCausalLM
model = AutoModelForCausalLM.from_pretrained(
"meta-llama/Llama-3.2-1B", attn_implementation="kernels-community/flash-attn2"
)SDPA 上下文管理器
PyTorch 的缩放点积注意力(SDPA)会自动为 CUDA 后端选择最快的注意力函数。对于其他后端,它默认使用 PyTorch C++ 实现。
使用 torch.nn.attention.sdpa_kernel 上下文管理器强制 SDPA 使用特定的实现。
import torch
from torch.nn.attention import SDPBackend, sdpa_kernel
from transformers import AutoModelForCausalLM
model = AutoModelForCausalLM.from_pretrained(
"meta-llama/Llama-3.2-1B", attn_implementation="sdpa"
)
with sdpa_kernel(SDPBackend.FLASH_ATTENTION):
outputs = model.generate(**inputs)特定主干的注意力
多模态模型为每种模态使用不同的骨干网络。通过为每个骨干网络分配特定的注意力函数来优化性能。例如,某些视觉骨干网络在 fp32 下表现更好,而 FlashAttention 并不支持 fp32。
使用字典将视觉主干映射到不同的注意力函数,同时文本主干继续使用 FlashAttention。注意力实现中的键必须与子配置名称匹配。
from transformers import AutoModelForImageTextToText
attention_implementation_per_backbone = {"vision_config": "sdpa", "text_config": "flash_attention_2"}
for key in attention_implementation_per_backbone:
assert key in model.config.sub_configs, f"Invalid key in `attention_implementation`"
model = AutoModelForImageTextToText.from_pretrained(
"facebook/chameleon-7b", attn_implementation=attention_implementation_per_backbone
)从字典中省略某些骨干网络,即可使用默认的注意力函数(SDPA)。
model = AutoModelForImageTextToText.from_pretrained(
"facebook/chameleon-7b", attn_implementation={"text_config": "flash_attention_2"}
)使用单个字符串为所有骨干网络设置相同的注意力函数。
model = AutoModelForImageTextToText.from_pretrained(
"facebook/chameleon-7b", attn_implementation="eager"
)使用空键全局设置注意力函数。
model = AutoModelForImageTextToText.from_pretrained(
"facebook/chameleon-7b", attn_implementation={"": "eager"}
)创建新的注意力函数
通过向注意力注册表添加函数来自定义或创建新的注意力函数,使用 AttentionInterface.register()。模型通过 attn_implementation 参数使用这些函数。
在注册自定义注意力函数时,请同时注册一个匹配的注意力掩码函数。如果自定义的
attn_implementation名称未在 AttentionMaskInterface 中注册,Transformers 将跳过掩码创建,并向注意力层传递attention_mask=None。此时,您的注意力函数必须自行处理因果、填充、打包或滑动窗口等约束,否则这些约束可能会被静默忽略。
本示例通过自定义注意力函数,为每一层打印一条语句。它通过注册 masking_utils.sdpa_mask 作为注意力掩码函数,保留了原始实现中的掩码。
import torch
from transformers import AutoModelForCausalLM, AttentionInterface, AttentionMaskInterface
from transformers.integrations.sdpa_attention import sdpa_attention_forward
from transformers.masking_utils import sdpa_mask
def my_new_sdpa(*args, **kwargs):
print("I just entered the attention computation")
return sdpa_attention_forward(*args, **kwargs)
AttentionInterface.register("my_new_sdpa", my_new_sdpa)
AttentionMaskInterface.register("my_new_sdpa", sdpa_mask) # must have the same name as the registered attention function
model = AutoModelForCausalLM.from_pretrained("meta-llama/Llama-3.2-1B", attn_implementation="my_new_sdpa")
model(torch.ones(1, 5, dtype=int))您还可以向注意力函数添加新的参数。支持 AttentionInterface 的模型会将 kwargs 传递到注意力层和注意力函数中。请在模型的前向函数中通过 kwargs 传递参数。自定义注意力函数必须遵循此签名和返回格式。
import torch
from transformers import AutoModelForCausalLM, AttentionInterface, AttentionMaskInterface
from transformers.integrations.sdpa_attention import sdpa_attention_forward
from transformers.masking_utils import sdpa_mask
def custom_attention(
module: torch.nn.Module, # required arg
query: torch.Tensor, # required arg
key: torch.Tensor, # required arg
value: torch.Tensor, # required arg
attention_mask: Optional[torch.Tensor], # required arg
a_new_kwargs = None, # You can now add as many kwargs as you need
another_new_kwargs = None, # You can now add as many kwargs as you need
**kwargs, # You need to accept **kwargs as models will pass other args
) -> tuple[torch.Tensor, Optional[torch.Tensor]]
... # do your magic!
return attn_output, attn_weights # attn_weights are optional here
AttentionInterface.register("custom", custom_attention)
AttentionMaskInterface.register("custom", sdpa_mask) # to leave the existing mask untouched
model = AutoModelForCausalLM.from_pretrained(model_id, attn_implementation="custom")
model(torch.ones(1, 5, dtype=int), a_new_kwargs=..., another_new_kwargs=...)检查模型的 modeling code,以确认它向注意力函数传递了哪些参数和 kwargs。
AttentionMaskInterface
AttentionMaskInterface 是一个注册表,create_*_mask 函数会查询该注册表,将掩码转换为当前激活的注意力后端所期望的格式。FlexAttention 需要 BlockMask,SDPA 需要 4D 张量,而 FlashAttention 需要基础的 2D padding 掩码。您可以使用 AttentionMaskInterface.register() 来注册自定义后端,或覆盖现有后端的格式化器。
import torch
from transformers import AttentionMaskInterface
from transformers.masking_utils import sdpa_mask
def my_new_sdpa_mask(*args, **kwargs):
print("I just entered the attention mask computation")
return sdpa_mask(*args, **kwargs)
AttentionMaskInterface.register("my_new_sdpa_mask", my_new_sdpa_mask)如果当前 attn_implementation 没有注册对应的格式化器,则会跳过掩码创建,并将 attention_mask=None 传递给注意力层。
注册的函数必须与此签名匹配。
def custom_attention_mask(
batch_size: int, # required arg
q_length: int, # required arg
kv_length: int, # required arg
q_offset: int = 0, # required arg
kv_offset: int = 0, # required arg
mask_function: Callable = causal_mask_function, # required arg
attention_mask: Optional[torch.Tensor] = None, # required arg
**kwargs, # a few additional args may be passed as kwargs, especially the model's config is always passed
) -> Optional[torch.Tensor]:mask_function 参数是一个 Callable,其作用类似于 PyTorch 的 mask_mod 函数。它接收 4 个索引 (batch_idx, head_idx, q_idx, kv_idx),并返回一个布尔值,指示该位置是否参与注意力计算。这与 构建注意力掩码 中 or_mask_function 和 and_mask_function 所使用的基本形式相同。
如果
mask_function无法创建掩码,请使用此 变通方法 来处理 torch.export。
构建注意力掩码
使用 transformers.masking_utils 中的 create_*_mask 函数来构建注意力掩码。每个函数都会从模型配置中读取当前生效的注意力后端,在 AttentionMaskInterface 中查找该后端的掩码格式化器,并返回该后端所期望的格式。您无需自行对掩码进行取反、扩展或类型转换。
选择与注意力模式相匹配的函数。
| 函数 | 用例 |
|---|---|
create_causal_mask | decoder-only 模型中,每个 token 仅关注自身及之前的 token |
create_bidirectional_mask | 编码器模型,或从解码器到编码器状态的交叉注意力 |
create_sliding_window_causal_mask | 具有滑动窗口注意力模式的解码器模型 |
create_chunked_causal_mask | 将序列分块为固定大小块的解码器模型 |
create_bidirectional_sliding_window_mask | 采用滑动窗口注意力模式的编码器模型 |
旧版可调用掩码辅助函数 -
get_extended_attention_mask、create_extended_attention_mask_for_decoder、invert_attention_mask- 会发出弃用警告,并将在未来版本中移除。请改用create_*_mask函数。
在解码器的前向传播中调用 create_causal_mask。传入 config、输入嵌入、用户提供的 2D attention_mask 以及 cache。该函数利用嵌入来读取 batch size、query 长度、dtype 和 device,并利用 cache 来计算 key 长度。
from transformers.masking_utils import create_causal_mask
attention_mask = create_causal_mask(
config=self.config,
inputs_embeds=inputs_embeds,
attention_mask=attention_mask,
past_key_values=past_key_values,
)调用 create_bidirectional_mask 用于编码器的自注意力。由于编码器不进行缓存,因此丢弃 past_key_values。
from transformers.masking_utils import create_bidirectional_mask
attention_mask = create_bidirectional_mask(
config=self.config,
inputs_embeds=embedding_output,
attention_mask=attention_mask,
)对于交叉注意力,请将编码器状态作为 encoder_hidden_states 传入,以便掩码使用编码器的键和值长度,而不是解码器的查询长度。
encoder_attention_mask = create_bidirectional_mask(
config=self.config,
inputs_embeds=embedding_output,
attention_mask=encoder_attention_mask,
encoder_hidden_states=encoder_hidden_states,
)使用 or_mask_function 和 and_mask_function 参数,可以在基础掩码之上添加额外的约束。使用 or_mask_function 允许额外的位置参与注意力计算,使用 and_mask_function 则进一步限制基础模式。两者均遵循 AttentionMaskInterface 中描述的 4 索引 mask_function 签名。它们接收 (batch_idx, head_idx, q_idx, kv_idx) 作为参数,并返回布尔值。
or_mask_function和and_mask_function可以表达任意注意力模式,但它们的执行速度比内置模式慢,并且与 ExecuTorch 不兼容。在较小的模型(约 200M 参数)上,这种开销最为明显,因为掩码创建在正向传播时间中占据更大的比例。只有在标准的create_*_mask函数无法表达你的需求时,才应使用它们。
例如,可以在因果掩码上叠加一个在所有位置都返回 True 的函数,从而将其转换为完全双向的掩码。与因果模式的并集使得每个 token 都可以关注其他所有 token。
mask_kwargs = {
"config": self.config,
"inputs_embeds": inputs_embeds,
"attention_mask": attention_mask,
"past_key_values": past_key_values,
"position_ids": position_ids,
"or_mask_function": lambda *args: torch.tensor(True, dtype=torch.bool),
}
attention_mask = create_causal_mask(**mask_kwargs)在生成过程中,generate() 会通过 create_masks_for_generate 构建掩码,该函数会根据模型配置分发到相应的 create_*_mask。在模型类中重写此方法,即可为生成过程接入自定义的掩码策略。
传入自定义的 4D 注意力掩码
当 create_*_mask 函数无法表达所需的注意力模式时,请传入自定义的 4D 掩码。4D 掩码的形状为 (batch_size, 1, query_length, kv_length),其中 1 表示将相同的掩码广播到所有注意力头。Transformers 会检测并直接使用该掩码,跳过 create_*_mask。
4D 掩码使用两种取值约定之一。
| dtype | attend | 屏蔽 |
|---|---|---|
| boolean | True | False |
| float | 0.0 | -inf |
float 约定在 softmax 之前将掩码加到注意力分数上。分数加上 0.0 保持不变,因此该位置会被计入。分数加上 -inf 在 softmax 之后变为零,因此该位置会被排除。
所接受的约定取决于注意力后端。
sdpa接受布尔型或浮点型掩码。eager会将掩码加到分数上,因此仅接受浮点型掩码。flash_attention_2和flex_attention使用各自的格式(一个 2D 填充掩码和一个 BlockMask),不接受原始的 4D 掩码。
一个常见的错误是在浮点 4D 掩码中复用 2D 填充掩码的 1/0 约定。由于掩码会被加到分数上,0.0 会保留该位置,而 1.0 只会引入一个微小的偏差。
下面的示例对比了一个错误的掩码和一个正确的掩码。两者都从相同的 1/0 因果模式开始。
import torch
from transformers import AutoModelForCausalLM, AutoTokenizer
model = AutoModelForCausalLM.from_pretrained("TinyLlama/TinyLlama-1.1B-Chat-v1.0", attn_implementation="sdpa")
tokenizer = AutoTokenizer.from_pretrained("TinyLlama/TinyLlama-1.1B-Chat-v1.0")
input_ids = tokenizer("my favorite condiment on a", return_tensors="pt").input_ids
seq_len = input_ids.shape[1]
# 1 attends, 0 masks
causal = torch.tril(torch.ones(seq_len, seq_len))
# wrong: 1.0/0.0 floats are added to the scores, so 0.0 keeps a token and 1.0 barely changes it
wrong_mask = causal[None, None]
# correct: 0.0 attends, -inf masks
correct_mask = torch.where(causal.bool(), 0.0, float("-inf"))[None, None]错误的掩码会保留所有位置,因为 0.0 才是用于掩码的值,因此它永远不会排除任何内容。
wrong_mask correct_mask
(1 attends, 0 masks) (0 attends, -inf masks)
k0 k1 k2 k3 k4 k0 k1 k2 k3 k4
q0 1 0 0 0 0 q0 0 -inf -inf -inf -inf
q1 1 1 0 0 0 q1 0 0 -inf -inf -inf
q2 1 1 1 0 0 q2 0 0 0 -inf -inf
q3 1 1 1 1 0 q3 0 0 0 0 -inf
q4 1 1 1 1 1 q4 0 0 0 0 0双向注意力
Decoder-only 模型默认使用因果(单向)注意力,即每个 token 仅关注自身及其之前的 token。将 is_causal=False 设置为 False 可切换为双向注意力,使每个 token 都能关注所有其他 token。这允许您将 decoder-only 模型用作文本编码器,例如用于生成嵌入。
此功能仅适用于因果(解码器)模型。它不会将编码器模型转换为解码器模型。
在模型配置中设置 is_causal=False,以便将双向注意力设为每次前向传播的默认行为。
from transformers import AutoModel, AutoConfig
config = AutoConfig.from_pretrained("meta-llama/Llama-3.2-1B")
config.is_causal = False
model = AutoModel.from_pretrained("meta-llama/Llama-3.2-1B", config=config)
# all forward passes now use bidirectional attention
outputs = model(**inputs)在 forward 调用中传入 is_causal,而不是在模型配置中设置,即可在因果注意力和双向注意力之间切换,而无需两次加载模型。该关键字参数会临时覆盖配置,并在调用结束后恢复。
from transformers import AutoModel
model = AutoModel.from_pretrained("meta-llama/Llama-3.2-1B")
# run with bidirectional attention
outputs = model(**inputs, is_causal=False)
# run with default causal attention
outputs = model(**inputs)