[코드] 오토인코더 - Training Deep AutoEncoders for Collaborative Filtering

input_layer.py

class UserItemRecDataProvider:
	
    def _build_maps(self):
    '''
    self._user_id_map ,self._item_id_map 에 딕셔너리 형태로 [원래 id] = 새로운 아이디 넣기
    '''
    	for line in src.readlines(): #한줄씩 읽기
        parts = line.strip().split(self._delimiter) #strip() 맨앞과 맨뒤 \n제거
        
        
    def iterate_one_epoch(self):
    '''
    배치로 학습할수 있게 처리하기
    '''
    	mini_batch = torch.sparse.FloatTensor(i_torch, v_torch, torch.Size([self._batch_size, self._vector_dim]))
        # 행렬 형태로 (i_torch[a], v_torch[a])의 위치에 torch...[a]의 값이 저장된다.
     	yield  mini_batch #객체로 저장한다. 해당 함수인 iterate_one_epoch의 객체로 저장된다.
        
    def iterate_one_epoch_eval(self, for_inf=False):
    '''
    ?
    '''
    	
        
        

 

model.py

def activation(input, kind):
'''
활성화 함수 정하기
'''


def MSEloss(inputs, targets, size_average=False):
'''
MSEloss를 계산하여 반환
'''
	num_ratings = torch.sum(mask.float()) # sum(행렬) -> 행렬에 있는 값 다 더하기
    
    
class AutoEncoder(nn.Module):
'''
파라미터 설정 및 인코더, 디코더 함수 만들기
'''
    def __init__(self, layer_sizes, nl_type='selu', is_constrained=True, dp_drop_prob=0.0, last_layer_activations=True):
    	#인코더의 w 설정
        self.encode_w = nn.ParameterList(
          [nn.Parameter(torch.rand(layer_sizes[i + 1], layer_sizes[i])) for i in range(len(layer_sizes) - 1)])
        #디코더의 w 설정
        self.decode_w = nn.ParameterList(
          [nn.Parameter(torch.rand(reversed_enc_layers[i + 1], reversed_enc_layers[i])) for i in range(len(reversed_enc_layers) - 1)])


    def encode(self, x):
    	# 인코더 w들에 대해 반복문을 돌며 활성함수 거치기
    	for ind, w in enumerate(self.encode_w):
      		x = activation(input=F.linear(input=x, weight=w, bias=self.encode_b[ind]), kind=self._nl_type)
    	if self._dp_drop_prob > 0:
      		x = self.drop(x)
    	return x
     
     
    def decode(self, z):
      if self.is_constrained:  # 인코더의 w 재사용
        for ind, w in enumerate(list(reversed(self.encode_w))):
          z = activation(input=F.linear(input=z, weight=w.transpose(0, 1), bias=self.decode_b[ind]), #마지막 레이어는 선형
                       kind=self._nl_type if ind!=self._last or self._last_layer_activations else 'none')
          #if self._dp_drop_prob > 0 and ind!=self._last: # 마지막 레이어는 드롭아웃 X
          #  z = self.drop(z)
      else:
        for ind, w in enumerate(self.decode_w):
          z = activation(input=F.linear(input=z, weight=w, bias=self.decode_b[ind]),
                       kind=self._nl_type if ind!=self._last or self._last_layer_activations else 'none')
          #if self._dp_drop_prob > 0 and ind!=self._last:
          #  z = self.drop(z)
          
      return z


	def forward(self, x):
    	# 인코더 거친 뒤 디코더 거치기
    	return self.decode(self.encode(x))
    
    

 

run.py

 

#인자값을 받기 위해 객체 생성 (description - 인자 도움말 전에 표시할 텍스트 (기본값: none))
parser = argparse.ArgumentParser(description='RecoEncoder')

#인자 추가 (metavar - 사용 메시지에 사용되는 인자의 이름) (help - 인자 설명)
parser.add_argument('--lr', type=float, default=0.00001, metavar='N', help='learning rate')

#인자 문자열을 객체로 변환한 뒤 namespace형으로 설정
args = parser.parse_args()
print(args)

docs.python.org/ko/3.7/library/argparse.html

 

# gpu 사용을 확인한다. 
# device = torch.device('cuda' if torch.cuda.is_available() else 'cpu') 의 코드로도 사용 가능
use_gpu = torch.cuda.is_available()
if use_gpu:
    print('GPU is available.') 
else: 
    print('GPU is not available.')

 

def do_eval(encoder, evaluation_data_layer):
  encoder.eval()
  denom = 0.0
  total_epoch_loss = 0.0
  for i, (eval, src) in enumerate(evaluation_data_layer.iterate_one_epoch_eval()):
    inputs = Variable(src.cuda().to_dense() if use_gpu else src.to_dense())
    targets = Variable(eval.cuda().to_dense() if use_gpu else eval.to_dense())
    outputs = encoder(inputs)
    loss, num_ratings = model.MSEloss(outputs, targets)
    total_epoch_loss += loss.item()
    denom += num_ratings.item()
  return sqrt(total_epoch_loss / denom)