mirror of
https://github.com/comfyanonymous/ComfyUI.git
synced 2025-01-11 10:25:16 +00:00
222 lines
8.4 KiB
Python
222 lines
8.4 KiB
Python
import torch
|
|
import torch.nn as nn
|
|
import torch.nn.functional as F
|
|
from dataclasses import dataclass
|
|
from typing import Optional, Any
|
|
|
|
from comfy.ldm.modules.attention import optimized_attention
|
|
import comfy.model_management
|
|
import comfy.ldm.common_dit
|
|
|
|
import comfy.model_management
|
|
|
|
@dataclass
|
|
class Llama2Config:
|
|
vocab_size: int = 128320
|
|
hidden_size: int = 4096
|
|
intermediate_size: int = 14336
|
|
num_hidden_layers: int = 32
|
|
num_attention_heads: int = 32
|
|
num_key_value_heads: int = 8
|
|
max_position_embeddings: int = 8192
|
|
rms_norm_eps: float = 1e-5
|
|
rope_theta: float = 500000.0
|
|
|
|
class RMSNorm(nn.Module):
|
|
def __init__(self, dim: int, eps: float = 1e-5, device=None, dtype=None):
|
|
super().__init__()
|
|
self.eps = eps
|
|
self.weight = nn.Parameter(torch.empty(dim, device=device, dtype=dtype))
|
|
|
|
def forward(self, x: torch.Tensor):
|
|
return comfy.ldm.common_dit.rms_norm(x, self.weight, self.eps)
|
|
|
|
|
|
def rotate_half(x):
|
|
"""Rotates half the hidden dims of the input."""
|
|
x1 = x[..., : x.shape[-1] // 2]
|
|
x2 = x[..., x.shape[-1] // 2 :]
|
|
return torch.cat((-x2, x1), dim=-1)
|
|
|
|
|
|
def precompute_freqs_cis(head_dim, seq_len, theta, device=None):
|
|
theta_numerator = torch.arange(0, head_dim, 2, device=device).float()
|
|
inv_freq = 1.0 / (theta ** (theta_numerator / head_dim))
|
|
|
|
position_ids = torch.arange(0, seq_len, device=device).unsqueeze(0)
|
|
|
|
inv_freq_expanded = inv_freq[None, :, None].float().expand(position_ids.shape[0], -1, 1)
|
|
position_ids_expanded = position_ids[:, None, :].float()
|
|
freqs = (inv_freq_expanded.float() @ position_ids_expanded.float()).transpose(1, 2)
|
|
emb = torch.cat((freqs, freqs), dim=-1)
|
|
cos = emb.cos()
|
|
sin = emb.sin()
|
|
return (cos, sin)
|
|
|
|
|
|
def apply_rope(xq, xk, freqs_cis):
|
|
cos = freqs_cis[0].unsqueeze(1)
|
|
sin = freqs_cis[1].unsqueeze(1)
|
|
q_embed = (xq * cos) + (rotate_half(xq) * sin)
|
|
k_embed = (xk * cos) + (rotate_half(xk) * sin)
|
|
return q_embed, k_embed
|
|
|
|
|
|
class Attention(nn.Module):
|
|
def __init__(self, config: Llama2Config, device=None, dtype=None, ops: Any = None):
|
|
super().__init__()
|
|
self.num_heads = config.num_attention_heads
|
|
self.num_kv_heads = config.num_key_value_heads
|
|
self.hidden_size = config.hidden_size
|
|
self.head_dim = self.hidden_size // self.num_heads
|
|
|
|
ops = ops or nn
|
|
self.q_proj = ops.Linear(config.hidden_size, config.hidden_size, bias=False, device=device, dtype=dtype)
|
|
self.k_proj = ops.Linear(config.hidden_size, self.num_kv_heads * self.head_dim, bias=False, device=device, dtype=dtype)
|
|
self.v_proj = ops.Linear(config.hidden_size, self.num_kv_heads * self.head_dim, bias=False, device=device, dtype=dtype)
|
|
self.o_proj = ops.Linear(config.hidden_size, config.hidden_size, bias=False, device=device, dtype=dtype)
|
|
|
|
def forward(
|
|
self,
|
|
hidden_states: torch.Tensor,
|
|
attention_mask: Optional[torch.Tensor] = None,
|
|
freqs_cis: Optional[torch.Tensor] = None,
|
|
):
|
|
batch_size, seq_length, _ = hidden_states.shape
|
|
|
|
xq = self.q_proj(hidden_states)
|
|
xk = self.k_proj(hidden_states)
|
|
xv = self.v_proj(hidden_states)
|
|
|
|
xq = xq.view(batch_size, seq_length, self.num_heads, self.head_dim).transpose(1, 2)
|
|
xk = xk.view(batch_size, seq_length, self.num_kv_heads, self.head_dim).transpose(1, 2)
|
|
xv = xv.view(batch_size, seq_length, self.num_kv_heads, self.head_dim).transpose(1, 2)
|
|
|
|
xq, xk = apply_rope(xq, xk, freqs_cis=freqs_cis)
|
|
|
|
xk = xk.repeat_interleave(self.num_heads // self.num_kv_heads, dim=1)
|
|
xv = xv.repeat_interleave(self.num_heads // self.num_kv_heads, dim=1)
|
|
|
|
output = optimized_attention(xq, xk, xv, self.num_heads, mask=attention_mask, skip_reshape=True)
|
|
return self.o_proj(output)
|
|
|
|
class MLP(nn.Module):
|
|
def __init__(self, config: Llama2Config, device=None, dtype=None, ops: Any = None):
|
|
super().__init__()
|
|
ops = ops or nn
|
|
self.gate_proj = ops.Linear(config.hidden_size, config.intermediate_size, bias=False, device=device, dtype=dtype)
|
|
self.up_proj = ops.Linear(config.hidden_size, config.intermediate_size, bias=False, device=device, dtype=dtype)
|
|
self.down_proj = ops.Linear(config.intermediate_size, config.hidden_size, bias=False, device=device, dtype=dtype)
|
|
|
|
def forward(self, x):
|
|
return self.down_proj(F.silu(self.gate_proj(x)) * self.up_proj(x))
|
|
|
|
class TransformerBlock(nn.Module):
|
|
def __init__(self, config: Llama2Config, device=None, dtype=None, ops: Any = None):
|
|
super().__init__()
|
|
self.self_attn = Attention(config, device=device, dtype=dtype, ops=ops)
|
|
self.mlp = MLP(config, device=device, dtype=dtype, ops=ops)
|
|
self.input_layernorm = RMSNorm(config.hidden_size, eps=config.rms_norm_eps, device=device, dtype=dtype)
|
|
self.post_attention_layernorm = RMSNorm(config.hidden_size, eps=config.rms_norm_eps, device=device, dtype=dtype)
|
|
|
|
def forward(
|
|
self,
|
|
x: torch.Tensor,
|
|
attention_mask: Optional[torch.Tensor] = None,
|
|
freqs_cis: Optional[torch.Tensor] = None,
|
|
):
|
|
# Self Attention
|
|
residual = x
|
|
x = self.input_layernorm(x)
|
|
x = self.self_attn(
|
|
hidden_states=x,
|
|
attention_mask=attention_mask,
|
|
freqs_cis=freqs_cis,
|
|
)
|
|
x = residual + x
|
|
|
|
# MLP
|
|
residual = x
|
|
x = self.post_attention_layernorm(x)
|
|
x = self.mlp(x)
|
|
x = residual + x
|
|
|
|
return x
|
|
|
|
class Llama2_(nn.Module):
|
|
def __init__(self, config, device=None, dtype=None, ops=None):
|
|
super().__init__()
|
|
self.config = config
|
|
self.vocab_size = config.vocab_size
|
|
|
|
self.embed_tokens = ops.Embedding(
|
|
config.vocab_size,
|
|
config.hidden_size,
|
|
device=device,
|
|
dtype=dtype
|
|
)
|
|
self.layers = nn.ModuleList([
|
|
TransformerBlock(config, device=device, dtype=dtype, ops=ops)
|
|
for _ in range(config.num_hidden_layers)
|
|
])
|
|
self.norm = RMSNorm(config.hidden_size, eps=config.rms_norm_eps, device=device, dtype=dtype)
|
|
# self.lm_head = ops.Linear(config.hidden_size, config.vocab_size, bias=False, device=device, dtype=dtype)
|
|
|
|
def forward(self, x, attention_mask=None, intermediate_output=None, final_layer_norm_intermediate=True, dtype=None):
|
|
x = self.embed_tokens(x, out_dtype=dtype)
|
|
|
|
freqs_cis = precompute_freqs_cis(self.config.hidden_size // self.config.num_attention_heads,
|
|
x.shape[1],
|
|
self.config.rope_theta,
|
|
device=x.device)
|
|
|
|
mask = None
|
|
if attention_mask is not None:
|
|
mask = 1.0 - attention_mask.to(x.dtype).reshape((attention_mask.shape[0], 1, -1, attention_mask.shape[-1])).expand(attention_mask.shape[0], 1, attention_mask.shape[-1], attention_mask.shape[-1])
|
|
mask = mask.masked_fill(mask.to(torch.bool), float("-inf"))
|
|
|
|
causal_mask = torch.empty(x.shape[1], x.shape[1], dtype=x.dtype, device=x.device).fill_(float("-inf")).triu_(1)
|
|
if mask is not None:
|
|
mask += causal_mask
|
|
else:
|
|
mask = causal_mask
|
|
|
|
intermediate = None
|
|
if intermediate_output is not None:
|
|
if intermediate_output < 0:
|
|
intermediate_output = len(self.layers) + intermediate_output
|
|
|
|
for i, layer in enumerate(self.layers):
|
|
x = layer(
|
|
x=x,
|
|
attention_mask=mask,
|
|
freqs_cis=freqs_cis,
|
|
)
|
|
if i == intermediate_output:
|
|
intermediate = x.clone()
|
|
|
|
x = self.norm(x)
|
|
if intermediate is not None and final_layer_norm_intermediate:
|
|
intermediate = self.norm(intermediate)
|
|
|
|
return x, intermediate
|
|
|
|
|
|
class Llama2(torch.nn.Module):
|
|
def __init__(self, config_dict, dtype, device, operations):
|
|
super().__init__()
|
|
config = Llama2Config(**config_dict)
|
|
self.num_layers = config.num_hidden_layers
|
|
|
|
self.model = Llama2_(config, device=device, dtype=dtype, ops=operations)
|
|
self.dtype = dtype
|
|
|
|
def get_input_embeddings(self):
|
|
return self.model.embed_tokens
|
|
|
|
def set_input_embeddings(self, embeddings):
|
|
self.model.embed_tokens = embeddings
|
|
|
|
def forward(self, input_ids, *args, **kwargs):
|
|
return self.model(input_ids, *args, **kwargs)
|