1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
| import torch
import torch.nn as nn
class ARTower(nn.Module):
"""
Cosmos 3 自回归推理塔
论文 Section 3.2.1: 多模态输入分为AR路径
- 文本token + ViT编码图像patch → AR序列
- 因果自注意力左到右处理
- DeepStack多层聚合
"""
def __init__(self, hidden_dim=4096, num_heads=32, num_layers=32):
super().__init__()
self.vit_encoder = None
self.text_embed = nn.Embedding(32000, hidden_dim)
self.deepstack_layers = nn.ModuleList([
nn.Linear(hidden_dim, hidden_dim) for _ in range(4)
])
self.causal_transformer = nn.TransformerDecoder(
nn.TransformerDecoderLayer(
d_model=hidden_dim,
nhead=num_heads,
dim_feedforward=hidden_dim * 4,
dropout=0.0,
activation='gelu',
batch_first=True
),
num_layers=num_layers
)
self.m rope = self._init_mrope(hidden_dim, num_heads)
def _init_mrope(self, dim, num_heads):
"""
3D多模态旋转位置编码
将文本、图像、视频、音频token对齐到
共享物理时间轴，支持FPS调制
"""
freqs = []
for i, (t, h, w) in enumerate([(8, 1, 1), (1, 16, 1), (1, 1, 16)]):
freq = 1.0 / (10000 ** (torch.arange(0, dim // 6, 2).float() / (dim // 6)))
freq = freq.unsqueeze(0)
freqs.append(freq)
return freqs
def forward(self, text_tokens, image_patches):
"""
AR路径前向传播
Args:
text_tokens: (B, L_text) 文本token ID
image_patches: (B, N_patch, hidden_dim) ViT编码的图像patch
Returns:
ar_repr: AR推理表征 (B, L_total, hidden_dim)
"""
text_feat = self.text_embed(text_tokens)
ar_seq = torch.cat([image_patches, text_feat], dim=1)
ar_repr = self.causal_transformer(ar_seq, ar_seq)
for layer in self.deepstack_layers:
residual = ar_repr
ar_repr = layer(ar_repr) + residual
return ar_repr
class DiffusionTower(nn.Module):
"""
Cosmos 3 扩散生成塔
论文 Section 3.2.2: 双向注意力生成
- 视频/音频/动作 → VAE压缩为潜在
- 双向注意力全局时空一致
- 迭代去噪生成
"""
def __init__(self, hidden_dim=4096, num_heads=32, num_layers=32):
super().__init__()
self.vae_encoder = None
self.bidirectional = nn.TransformerEncoder(
nn.TransformerEncoderLayer(
d_model=hidden_dim,
nhead=num_heads,
dim_feedforward=hidden_dim * 4,
dropout=0.0,
activation='gelu',
batch_first=True
),
num_layers=num_layers
)
def forward(self, noisy_latents, ar_repr):
"""
扩散塔前向传播
Args:
noisy_latents: 加噪的视频/音频/动作潜在 (B, L_diff, hidden_dim)
ar_repr: AR塔的推理表征 (B, L_ar, hidden_dim)
Returns:
denoised: 去噪后的潜在 (B, L_diff, hidden_dim)
"""
denoised = self.bidirectional(
noisy_latents,
mask=None
)
return denoised
if __name__ == "__main__":
batch_size = 2
hidden_dim = 512
ar_tower = ARTower(hidden_dim=hidden_dim, num_heads=8, num_layers=4)
text_tokens = torch.randint(0, 32000, (batch_size, 128))
image_patches = torch.randn(batch_size, 196, hidden_dim)
ar_repr = ar_tower(text_tokens, image_patches)
print(f"AR推理表征: {ar_repr.shape}")
diff_tower = DiffusionTower(hidden_dim=hidden_dim, num_heads=8, num_layers=4)
noisy_latents = torch.randn(batch_size, 100, hidden_dim)
denoised = diff_tower(noisy_latents, ar_repr)
print(f"去噪输出: {denoised.shape}")
print("\n✅ Cosmos 3 双塔架构测试通过")
|
