PowerMachine commited on
Commit
87d5a70
·
verified ·
1 Parent(s): fff13a9

Delete models/rope.py with huggingface_hub

Browse files
Files changed (1) hide show
  1. models/rope.py +0 -196
models/rope.py DELETED
@@ -1,196 +0,0 @@
1
- """
2
- RoPE (Rotary Position Embeddings) — Codificação posicional rotacional.
3
-
4
- Referência: Su et al., 2021 — "RoFormer: Enhanced Transformer with Rotary
5
- Position Embedding".
6
-
7
- Propriedade chave: <RoPE(q, m), RoPE(k, n)> = <RoPE(q, m-n), k>
8
- → a atenção depende apenas da diferença posicional relativa.
9
-
10
- Ver docs/MATH_ANALYSIS.md seção 4.1.
11
-
12
- Autor: CNN-BiGRU Project
13
- """
14
- from __future__ import annotations
15
-
16
- import math
17
- from typing import Optional, Tuple
18
-
19
- import torch
20
- import torch.nn as nn
21
-
22
-
23
- def precompute_rope_frequencies(
24
- head_dim: int,
25
- max_seq_len: int,
26
- base: float = 10000.0,
27
- device: Optional[torch.device] = None,
28
- dtype: torch.dtype = torch.float32,
29
- ) -> torch.Tensor:
30
- """
31
- Pré-computa as frequências angulares theta_i para RoPE.
32
-
33
- Fórmula: theta_i = base^(-2i / head_dim) para i = 0, 1, ..., head_dim/2 - 1
34
-
35
- Args:
36
- head_dim: dimensão de cada cabeça (deve ser par)
37
- max_seq_len: comprimento máximo da sequência
38
- base: base da exponencial (padrão 10000)
39
- device: device do tensor
40
- dtype: dtype do tensor
41
-
42
- Returns:
43
- cos_sin: tensor [max_seq_len, head_dim] com [cos(theta_0*m), cos(theta_1*m), ...,
44
- sin(theta_0*m), sin(theta_1*m), ...] intercalados por par
45
- (formato compatível com apply_rope).
46
-
47
- Matematicamente:
48
- Para cada posição m em [0, max_seq_len) e cada par de dimensões 2i:
49
- angle = m * theta_i = m * base^(-2i/head_dim)
50
- A matriz de rotação no plano (2i, 2i+1) é:
51
- [[cos(angle), -sin(angle)],
52
- [sin(angle), cos(angle)]]
53
- """
54
- if head_dim % 2 != 0:
55
- raise ValueError(f"head_dim deve ser par; recebido {head_dim}")
56
-
57
- # theta_i = base^(-2i / head_dim), i = 0, ..., head_dim/2 - 1
58
- i = torch.arange(0, head_dim, 2, device=device, dtype=dtype) # [head_dim/2]
59
- theta = 1.0 / (base ** (i / head_dim)) # [head_dim/2]
60
-
61
- # Posições m
62
- positions = torch.arange(max_seq_len, device=device, dtype=dtype) # [max_seq_len]
63
-
64
- # angle[m, i] = m * theta_i
65
- angles = torch.outer(positions, theta) # [max_seq_len, head_dim/2]
66
-
67
- # Duplicar para intercalar: [cos_0, cos_1, ..., sin_0, sin_1, ...]
68
- cos = torch.cos(angles) # [max_seq_len, head_dim/2]
69
- sin = torch.sin(angles)
70
-
71
- # Formato final: [max_seq_len, head_dim] com pares (cos_i, sin_i)
72
- cos_sin = torch.stack([cos, sin], dim=-1).reshape(max_seq_len, head_dim)
73
- return cos_sin
74
-
75
-
76
- def apply_rope(
77
- x: torch.Tensor,
78
- cos_sin: torch.Tensor,
79
- ) -> torch.Tensor:
80
- """
81
- Aplica RoPE ao tensor x.
82
-
83
- Args:
84
- x: [batch, n_heads, seq_len, head_dim]
85
- cos_sin: [seq_len, head_dim] — tensor pré-computado
86
-
87
- Returns:
88
- x_rotated: [batch, n_heads, seq_len, head_dim]
89
-
90
- Matemática (para cada batch, head, posição m, par de dims (2i, 2i+1)):
91
- x'[2i] = x[2i] * cos(angle) - x[2i+1] * sin(angle)
92
- x'[2i+1] = x[2i] * sin(angle) + x[2i+1] * cos(angle)
93
-
94
- O tensor cos_sin tem formato [seq_len, head_dim] com:
95
- cos_sin[m, 2i] = cos(m * theta_i)
96
- cos_sin[m, 2i+1] = sin(m * theta_i)
97
- """
98
- if x.dim() != 4:
99
- raise ValueError(
100
- f"x deve ser [batch, n_heads, seq_len, head_dim]; recebido {x.shape}"
101
- )
102
-
103
- B, H, T, D = x.shape
104
- if D % 2 != 0:
105
- raise ValueError(f"head_dim deve ser par; recebido {D}")
106
-
107
- # Reshape x para [B, H, T, D/2, 2] — pares de dimensões
108
- x_pairs = x.float().reshape(B, H, T, D // 2, 2)
109
-
110
- # Reshape cos_sin para [T, D/2, 2]
111
- cos_sin_t = cos_sin[:T].to(x.device).float().reshape(T, D // 2, 2)
112
- cos = cos_sin_t[..., 0] # [T, D/2]
113
- sin = cos_sin_t[..., 1]
114
-
115
- # Broadcast: [1, 1, T, D/2]
116
- cos = cos.unsqueeze(0).unsqueeze(0)
117
- sin = sin.unsqueeze(0).unsqueeze(0)
118
-
119
- # Aplicar rotação
120
- x_real = x_pairs[..., 0] # [B, H, T, D/2]
121
- x_imag = x_pairs[..., 1]
122
- rotated_real = x_real * cos - x_imag * sin
123
- rotated_imag = x_real * sin + x_imag * cos
124
-
125
- # Reagrupar para [B, H, T, D]
126
- rotated = torch.stack([rotated_real, rotated_imag], dim=-1).reshape(B, H, T, D)
127
- return rotated.to(x.dtype)
128
-
129
-
130
- class RotaryPositionEmbedding(nn.Module):
131
- """
132
- Módulo PyTorch para RoPE — mantém o cache de frequências como buffer.
133
-
134
- Uso:
135
- rope = RotaryPositionEmbedding(head_dim=64, max_seq_len=512)
136
- # Em forward:
137
- q = rope(q) # q: [batch, n_heads, seq_len, head_dim]
138
- k = rope(k)
139
- """
140
-
141
- def __init__(self, head_dim: int, max_seq_len: int = 512, base: float = 10000.0):
142
- super().__init__()
143
- if head_dim % 2 != 0:
144
- raise ValueError(f"head_dim deve ser par; recebido {head_dim}")
145
- self.head_dim = head_dim
146
- self.max_seq_len = max_seq_len
147
- self.base = base
148
-
149
- # Pré-computar e registrar como buffer (move com .to(device))
150
- cos_sin = precompute_rope_frequencies(head_dim, max_seq_len, base)
151
- self.register_buffer("cos_sin", cos_sin, persistent=False)
152
-
153
- def forward(self, x: torch.Tensor, offset: int = 0) -> torch.Tensor:
154
- """
155
- Aplica RoPE a x.
156
-
157
- Args:
158
- x: [batch, n_heads, seq_len, head_dim]
159
- offset: deslocamento posicional (para geração autoregressiva com cache KV,
160
- onde a posição do novo token é len(cached) + i)
161
-
162
- Returns:
163
- x_rotated: mesmo shape que x
164
- """
165
- if x.dim() != 4:
166
- raise ValueError(
167
- f"x deve ser [batch, n_heads, seq_len, head_dim]; recebido {x.shape}"
168
- )
169
- T = x.size(2)
170
- if offset + T > self.max_seq_len:
171
- # Estender o cache dinamicamente
172
- self._extend_cache(offset + T)
173
- # Slice das posições relevantes
174
- cos_sin = self.cos_sin[offset:offset + T]
175
- return apply_rope(x, cos_sin)
176
-
177
- def _extend_cache(self, new_max: int) -> None:
178
- """Estende o cache de frequências para new_max posições."""
179
- new_max_pow2 = 1
180
- while new_max_pow2 < new_max:
181
- new_max_pow2 *= 2
182
- new_cos_sin = precompute_rope_frequencies(
183
- self.head_dim, new_max_pow2, self.base,
184
- device=self.cos_sin.device, dtype=self.cos_sin.dtype
185
- )
186
- self.cos_sin = new_cos_sin.to(self.cos_sin.device)
187
- self.max_seq_len = new_max_pow2
188
- # Re-registrar como buffer (substitui o anterior)
189
- self.register_buffer("cos_sin", self.cos_sin, persistent=False)
190
-
191
-
192
- __all__ = [
193
- "precompute_rope_frequencies",
194
- "apply_rope",
195
- "RotaryPositionEmbedding",
196
- ]