논문 정리 : https://velog.io/@j_keun/LSTM-%EB%85%BC%EB%AC%B8-%EB%A6%AC%EB%B7%B0

1. 기존 RNN에는 어떤 문제가 있었는가?

기존 RNN(Recurrent Neural Network)은 이전 시점의 은닉 상태를 다음 시점으로 전달하므로, 순서가 있는 데이터를 처리할 수 있다. 예를 들어 문장, 음성, 주가처럼 앞의 정보가 뒤의 결과에 영향을 주는 데이터에 적합하다.

하지만 학습 과정에서 BPTT(Backporpagation Through Time)를 사용하면 기울기 소실과 기울기 폭주 문제가 발생한다.

• 기울기 소실: 과거 시점으로 오차를 전달할수록 값이 계속 작아져 거의 0이 되는 현상
• 기울기 폭주: 반대로 기울기가 계속 커져 학습이 불안정해지는 현상

특히 일반 RNN은 많은 시간 단계를 거쳐 오차를 전달할 때, 활성화 함수의 미분값과 가중치가 계속 곱해진다. 이 곱의 결과가 1보다 작으면 기울기가 빠르게 사라지고, 1보다 크면 지나치게 커진다.

➡ 기존 RNN은 최근 정보는 비교적 잘 반영하지만, 오래전의 중요한 정보를 현재 예측에 활용하는데 어려움이 있다.

2. 긴 Sequence를 학습할 때 왜 문제가 발생하는가?

긴 시퀀스에서 현재 시점의 오차가 과거의 중요한 입력까지 전달되어야 한다. 하지만 현재와 과거 사이에 시간 단계가 많아질수록, 역전파 과정의 기울기 곱셈도 길어진다.

예를 들어 다음 문장을 보자.

"나는 한국에서 태어나 자랐고, 현재 미국에서 일하고 이싿. 내 모국어는 ___이다."

정답을 예측하려면 문장 앞부분의 "한국"이라는 정보를 기억해야 한다. 하지만 그 사이에 많은 단어가 들어가면, 일반 RNN은 앞부분의 정보를 학습 신호로 유지하기 어렵다.

이를 수식적으로 보면, 과거 시점으로 전달되는 기울기는 여러 시점의 미분값과 가중치가 연속적으로 곱해진 결과다.

현재 오차
    ➡ t-1의 가중치 * 미분값
    ➡ t-2의 가중치 * 미분값
    ➡ ...
    ➡ 과거 시점

따라서 시간 간격이 갈수록 다음 문제가 발생한다.

• 과거의 중요한 정보가 학습 과정에서 사라진다.
• 현재의 예측 오류가 과거 입력의 가중치까지 도달하지 못한다.
• 모델이 무엇을 오래 기억해야 하는지 학습하기 어렵다.
• 잡음이나 관련 없는 입력이 많은 경우 중요한 정보를 유지하기 더 어렵다.

➡ 핵심은 "시퀀스가 길다"는 것만이 문제가 아니라, 오래전 정보가 현재 결과에 반드시 필요하지만 중간에 관련 없는 정보가 많이 있는 상황이다.

3. LSTM은 기존 RNN의 문제를 어떤 아이디어로 해결하려 했는가?

LSTM은 메모리 셀 내부에 오차가 일정하게 흐를 수 있는 경로를 만들고, 그 경로를 게이트로 제어하는 방식으로 문제를 해결했다.

1. 메모리 셀 내부에 정보를 오래 유지한다.
2. CEC(Constant Error Carousel)를 통해 기울기가 급격히 사라지거나 커지는 것을 줄인다.
3. 입력 게이트와 출력 게이트를 통해 기억할 정보와 사용할 정보를 선택한다.

기존 RNN에서는 새 입력이 계속 들어오면 이전 정보가 쉽게 덮어써지거나, 기억된 정보가 계속 출력되어 다른 계산을 방해할 수 있다. LSTM은 이를 막기 위해 게이트를 둔다.

새 입력
  ↓
입력 게이트: 이 정보를 저장할지 결정
  ↓
메모리 셀: 필요한 정보를 유지
  ↓
출력 게이트: 현재 시점에 꺼내 쓸지 결정
  ↓
다음 계층 또는 최종 예측

➡ LSTM은 필요한 정보는 유지하고, 필요 없는 정보는 저장하거나 출력하지 않도록 학습하는 구조다.

4. LSTM의 핵심 구조는 무엇인가?

4-1 메모리 셀

메모리 셀은 정보를 장기간 보관하기 위한 내부 저장 공간이다. 일반 RNN의 은닉 상태보다 장기 정보를 유지하기 위한 구조가 강화되어 있다.

논문에서는 메모리 셀의 내부 상태는 이전 상태에 새 후보 정보를 더하는 형태로 표현

현재 셀 상태
= 이전 셀 상태
+ 입력 게이트가 허용한 새 정보

➡ 이전 상태가 매 시점마다 강하게 변형되지 않으며, 긴 시간 동안 정보와 학습 신호를 유지할 수 있다.

4-2 CEC(Constant Error Carousel)

CEC는 메모리 셀 안에서 자신으로 이어지는 연결의 가중치를 1.0으로 유지하는 구조다 .

이 경로에서는 오차가 반복적으로 작아지거나 커지지 않고, 비교적 일정하기 순환할 수 있다.

논문에서는 장기 의존성 학습의 핵심 기반이 CEC라고 말한다.

셀 상태(t-1)
     ↓
   CEC 연결
     ↓
셀 상태(t)

➡ CEC는 먼 과거의 입력까지 학습 신호가 도달할 수 있도록 돕는 통로이다.

4-3 입력 게이트

입력 게이트는 새로 들어온 정보를 메모리 셀이 저장할지 결정한다.

• 값이 1에 가까우면: 현재 입력을 메모리에 반영
• 값이 0에 가까우면: 현재 입력을 무시하고 기존 메모리를 보호

예) 문장에서 인물의 국적 정보를 기억해야 한다면, 국적이 등장하는 시점에는 입력게이트를 열고, 이후의 관련 없는 단어가 들어올 때는 게이트를 닫아 기존 기억이 덮어써지지 않도록 할 수 있다.

4-4 출력 게이트

출력 게이트는 메모리 셀에 저장된 정보를 현재 시점의 계산에 사용할지 결정한다.

• 값이 1에 가까우면: 메모리 내용을 외부로 전달한다.
• 값이 0에 가까우면: 기억은 유지하되 현재 출력에는 사용하지 않는다.

➡ 입력 게이트가 "언제 저장할지"를 결정한다면, 출력 게이트는 "언제 꺼내 사용할지"를 결정한다고 생각하면 된다.

5. 논문에서 말하는 장기 의존성(Long-Term Dependency)이란 무엇인가?

장기 의존성이란 시퀀스의 앞부분에 등장한 정보가 많은 시간 단계가 지난 뒤의 결과를 결정하는 관계를 말한다.

예)

시점 1: 중요한 신호 A 등장
시점 2~99: 잡음 또는 관련 없는 입력
시점 100: 신호 A를 이용해 결과 예측

모델은 시점 1의 신호 A를 시점 100까지 기억해야 한다. 중간에 입력이 많고 잡음이 섞여 있어도 중요한 정보를 잃으면 안된다.

논문에서는 다음 조건일때 장기 의존성 문제가 더 발생한다고 설명한다.

• 중요한 입력과 결과 사이의 시간 간격이 길다.
• 중간에 관련 없는 입력이나 잡음이 많다.
• 중요한 정보가 어느 위치에 나타날지 미리 알기 어렵다.
• 현재 예측을 위해 여러 과거 정보를 함께 기억해야 한다.

➡ LSTM은 CEC와 게이트를 이용해 이러한 장기 의존성을 학습하도록 설계

6. 논문을 읽고 새롭게 알게 된 내용

6-1 LSTM의 핵심은 오차 흐름 제어다.

논문을 읽기 전에는 LSTM이 단순히 정보를 오래 저장하는 모델이라고 생각했다. 하지만 논문에서 다루는 더 핵심적인 문제는, 긴 시퀀스에서 과거까지 학습 신호가 전달되지 않는 기울기 소실 문제였다.

➡ LSTM은 기억 장치이면서 동시에 긴 시간 간격에서도 오차를 전달할 수 있도록 설계된 학습 구조임을 알게되었다.

6-2 LSTM도 모든 문제를 해결하지는 못한다.

논문에서는 LSTM의 한계도 설명해준다. 예를 들어,멀리 떨어진 두 입력의 XOR처럼 두 정보를 모두 기억해야만 오차가 줄어드는 비분해적 문제는 학습하기 어렵다. 하나의 정보만 기억한 상태에서는 성능이 좋아지지 않으므로, 기울기 기반 학습이 어느 방향으로 가야 하는지 찾기 어렵다고 설명한다.

또한 정밀하게 시간 단계를 새어야 하는 문제에는 별도의 카운팅 메커니즘이 필요하다고 설명한다.

7. 기타 추가 내용

7-1 LSTM의 장점

• 긴 시간 간격의 중요한 정보를 학습할 수 있다.
• 잡음이 있는 시퀀스에서도 필요한 정보를 선택적으로 유지할 수 있다.
• 시계열 예측, 음성 처리, 자연어 처리 등 순서가 중요한 데이터에 적용할 수 있다.
• BPTT와 유사한 수준의 계산 복잡도로 학습할 수 있다.
• 관련 정보가 시퀀스의 어느 위치에 등장하더라도 일반화 할 수 있다.

7-2 LSTM의 한계

• 모든 장기 의존성 문제를 자동으로 해결하지는 못한다.
• 매우 정확한 시간 단계 계산에는 한계가 있을 수 있다.
• 게이트와 메모리 셀 때문에 일반 RNN보다 구조와 파라미터가 복잡해질 수 있다.

원 논문: Hochreiter, S. & Schmidhuber, J. (1997). Long Short-Term Memory. Neural Computation, 9(8), 1735-1780.

목차

1. Introduction
2. Previous Work
3. Constant Error Backprop
4. The Concept of Long Short-Term Memory
5. Experiments
6. Discussion
7. Conclusion
8. Acknowledgments

1. Introduction: 문제의 출발점

RNN(Recurrent Neural Network)은 현재 입력과 이전 상태를 함께 사용하므로, 문장·음성·시계열처럼 순서가 있는 데이터를 처리할 수 있다. 그러나 현재 시점의 오류를 오래전 입력까지 전달해야 할 때 학습이 매우 느려지거나 실패할 수 있다.

논문은 그 원인을 불충분하고 점차 감소하는 오차 역전파(error backflow)에서 찾는다. 먼 과거로 갈수록 학습 신호가 약해져, 과거 정보가 현재 결과에 미치는 영향을 모델이 배우기 어렵다는 뜻이다.

논문이 해결하려는 목표

• 매우 긴 시간 간격의 입력과 결과를 연결한다.
• 중간에 잡음이 많아도 중요한 정보를 유지한다.
• 짧은 시간 간격의 학습 능력을 잃지 않는다.
• 전체 BPTT보다 실용적인 계산 비용을 유지한다.

장기 의존성의 예시

시점 1: 중요한 정보 X
시점 2~999: 잡음 또는 관련 없는 입력
시점 1000: X를 활용해야 정답을 낼 수 있는 출력

이 상황에서 모델은 X를 오래 기억해야 할 뿐 아니라, 현재 오류가 시점 1의 파라미터까지 전달되도록 학습해야 한다.

2. Previous Work: 기존 접근법의 한계

2절에서 저자들은 당시의 순환 신경망 학습 방법을 검토한다.

논문의 핵심 주장은 단순하다. 기존 방법은 짧은 시간 간격에서는 동작할 수 있지만, 최소 시간 간격이 긴 문제에서는 과거 입력까지 충분한 학습 신호를 전달하기 어렵다.

3. Constant Error Backprop: 일정한 오차 흐름의 필요성

3.1 Exponentially Decaying Error

원 논문은 시간 t에서 유닛 j로 역전파되는 지역 오차를 다음처럼 표현한다.

$$ \delta_j(t)=f'j(net_j(t))\sum_k \delta_k(t+1)w{kj} $$

시간을 거꾸로 이동할수록 미분값과 가중치가 계속 곱해진다. 이 곱이 평균적으로 1보다 작으면 오차는 지수적으로 감소하고, 1보다 크면 폭주한다.

현재 출력 오류
  → 현재 은닉 상태
  → 이전 은닉 상태
  → 더 이전 은닉 상태
  → 먼 과거의 입력

따라서 긴 시퀀스에서는 “과거 정보가 중요하다”는 사실을 알더라도, 그 정보를 저장해야 하는 가중치까지 학습 신호가 닿지 않을 수 있다.

3.2 Constant Error Flow: 단순한 접근

저자들은 자기 연결을 가진 선형 유닛 j를 생각한다. 시간 t에서 오차 흐름이 일정하려면 다음 조건이 필요하다.

$$ f'j(net_j(t))w{jj}=1.0 $$

가장 단순한 방법은 항등 함수와 자기 연결 가중치 1.0을 사용하는 것이다.

$$ f_j(x)=x, \qquad w_{jj}=1.0 $$

이때 상태는 다음처럼 유지된다.

$$ y_j(t+1)=y_j(t) $$

논문은 이 자기 연결 구조를 CEC(Constant Error Carousel)라고 부른다.

단순 CEC만으로는 부족한 이유

이 두 문제를 해결하기 위해 4절에서 입력 게이트와 출력 게이트가 등장한다.

4. The Concept of Long Short-Term Memory

4.1 Memory Cells and Gate Units

원 논문의 LSTM 메모리 셀은 다음 요소로 구성된다.

• 중심 선형 유닛과 고정 자기 연결: CEC
• 입력 게이트 $in_j$
• 출력 게이트 $out_j$
• 내부 상태 $s_{c_j}(t)$

게이트 활성화

$$ y_{in_j}(t)=f_{in_j}(net_{in_j}(t)) $$

$$ y_{out_j}(t)=f_{out_j}(net_{out_j}(t)) $$

게이트의 순입력은 이전 시점 네트워크 유닛의 출력으로부터 계산된다.

$$ net_{in_j}(t)=\sum_u w_{in_j,u}y_u(t-1) $$

$$ net_{out_j}(t)=\sum_u w_{out_j,u}y_u(t-1) $$

여기서 $u$는 입력 유닛, 게이트 유닛, 메모리 셀, 일반 은닉 유닛 등이 될 수 있다.

메모리 셀 상태와 출력

원 논문에서 메모리 셀의 내부 상태는 다음처럼 갱신된다.

$$ s_{c_j}(0)=0 $$

$$ s_{c_j}(t)=s_{c_j}(t-1)+y_{in_j}(t)g(net_{c_j}(t)) \quad (t>0) $$

메모리 셀의 외부 출력은 다음과 같다.

$$ y_{c_j}(t)=y_{out_j}(t)h(s_{c_j}(t)) $$

수식의 직관적 해석

이전 셀 상태 s(t-1)
        +
입력 게이트가 허용한 새 정보
        ↓
현재 셀 상태 s(t)
        ↓
출력 게이트가 허용한 만큼만 외부로 전달
        ↓
셀 출력 y(t)

4.2 왜 게이트 유닛인가?

입력 게이트는 관련 없는 입력이 CEC 내부 기억을 덮어쓰지 못하게 한다. 출력 게이트는 현재 필요하지 않은 기억이 다른 유닛을 교란하지 못하게 한다.

입력 게이트가 닫힘
→ 기존 기억 보호

출력 게이트가 닫힘
→ 기억은 유지하지만 현재 계산에는 노출하지 않음

이로써 LSTM은 다음을 분리한다.

무엇을 기록할지     → 입력 게이트
어떻게 오래 유지할지 → CEC
언제 사용할지       → 출력 게이트

4.3 현대 LSTM과의 차이

1997년 원 논문에는 Forget Gate가 없다. 현대 LSTM은 이후 Forget Gate를 추가해 이전 셀 상태를 얼마나 유지할지 직접 조절한다.

현대 표준 LSTM의 대표적인 셀 상태 수식은 다음과 같다.

$$ c_t=f_t \odot c_{t-1}+i_t \odot \tilde{c}_t $$

이 수식은 현대 구현을 설명할 때 사용해야 하며, 1997년 원 논문의 수식과 혼동하지 않아야 한다.

5. Experiments: LSTM은 어떤 과제를 풀었는가?

논문은 실험 과제를 난도 순으로 배치한다.

5.1 Experiment 1: Embedded Reber Grammar

과제 목표

Embedded Reber Grammar는 정해진 문법 규칙으로 생성되는 기호열을 한 글자씩 입력받고, 다음 기호를 예측하는 벤치마크다.

입력: B → T → ... → P → ...
목표: 매 시점마다 다음에 올 수 있는 기호 예측

문자열의 마지막 직전 기호를 올바르게 예측하려면 초반의 두 번째 기호 T 또는 P를 기억해야 한다.

왜 이 과제를 사용했는가?

• 최소 시간 지연이 약 9단계인 경계 사례라서 기존 방식도 완전히 실패하지는 않는다.
• 기존 RNN 연구에서 널리 사용된 벤치마크라 비교가 가능하다.
• 출력 게이트가 왜 필요한지 보여 주기에 적합하다.

비교 모델과 대표 결과

원 논문 표 1의 대표 결과를 블로그용으로 다시 정리하면 다음과 같다.

해석

LSTM은 단순히 문법을 외우는 것이 아니라, 필요한 초기 기호는 저장하되 그 기억이 이후의 쉬운 문법 전이 계산을 방해하지 않게 해야 한다. 이때 출력 게이트가 기억의 공개 시점을 조절한다.

5.2 Experiment 2: Noise-Free and Noisy Sequences

Task 2a: 잡음 없는 긴 시간 지연

입력은 다음 두 시퀀스 중 하나다.

$$ (x,a_1,a_2,\ldots,a_{p-1},x) $$

$$ (y,a_1,a_2,\ldots,a_{p-1},y) $$

모델은 매 시점 다음 기호를 예측한다. 마지막 기호 x 또는 y를 맞히려면 맨 처음의 x 또는 y를 p 단계 동안 기억해야 한다.

Task 2b: 규칙성이 약한 잡음 시퀀스

중간 기호가 규칙적으로 정렬되지 않고, x 또는 y와 무관한 기호가 무작위로 등장한다. 마지막의 예측 가능한 x 또는 y를 맞히기 위해, 모델은 초반의 핵심 기호를 유지해야 한다.

Task 2c: 매우 긴 지연과 다수의 방해 기호

가장 어려운 과제다. 시퀀스는 시작 표식 b, 핵심 기호 x 또는 y, 다수의 무작위 방해 기호, 트리거 e, 최종 답 x 또는 y로 구성된다.

b → x 또는 y → 방해 기호 다수 → e(트리거) → x 또는 y

모델은 트리거 e가 등장했을 때 초반의 핵심 기호가 무엇이었는지 출력해야 한다. 방해 기호는 임의 위치에 들어가며, 긴 경우 최소 시간 지연이 1,000단계에 이른다.

원 논문 표 3: Task 2c 대표 결과

해석

시간 지연이 길어지는 것 자체보다, 같은 종류의 방해 기호가 자주 반복되어 기억을 흔드는 상황에서 학습 시간이 더 증가한다. 그럼에도 논문은 최소 시간 지연 1,000단계 설정을 해결했다고 보고한다.

5.3 Experiment 3: Noise and Signal on Same Channel

Task 3a: Two-Sequence Problem

하나의 실수 입력 채널만 사용한다. 두 클래스 중 하나가 동일 확률로 선택되며, 첫 N개 값이 클래스 정보를 갖는다.

클래스 1: 처음 N개 값 = +1.0
클래스 2: 처음 N개 값 = -1.0
이후 값: 평균 0, 분산 0.2의 가우시안 잡음

시퀀스 마지막에서 클래스 1은 1.0, 클래스 2는 0.0을 출력해야 한다.

Task 3b: 중요한 신호도 잡음에 오염

3a와 달리 클래스 정보를 담은 첫 N개 값에도 잡음이 섞인다. 즉, 모델은 명확한 +1과 -1을 읽는 것이 아니라, 잡음이 섞인 값에서 클래스 신호를 추출해야 한다.

Task 3c: 잡음이 있는 실수 목표값의 조건부 기대값 학습

3c는 단순 추측으로 풀기 어려운 형태로 수정된다.

클래스 1의 잡음 없는 목표값: 0.2
클래스 2의 잡음 없는 목표값: 0.8
실제 학습 목표: 목표값에 가우시안 잡음을 더한 값

모델은 잡음이 섞인 목표값을 그대로 외우는 대신, 입력이 주어졌을 때의 조건부 기대값을 학습해야 한다.

원 논문 표 6: Task 3c 대표 결과

해석

이 실험은 신호와 잡음이 같은 채널에 있어도 LSTM이 장기 기억을 형성할 수 있는지, 그리고 단순한 분류를 넘어 연속적인 기대값을 예측할 수 있는지 확인한다.

5.4 Experiment 4: Adding Problem

과제 입력

각 입력 요소는 두 성분의 쌍이다.

$$ (X_t,M_t) $$

시퀀스마다 정확히 두 위치가 M_t=1.0으로 표시된다. 모델은 그 두 위치의 실수 $X_1$, $X_2$를 오랫동안 기억한 뒤, 시퀀스 마지막에서 아래 목표를 출력해야 한다.

$$ target=0.5+\frac{X_1+X_2}{4.0} $$

-1.0 마커는 시퀀스의 시작과 끝을 표시하며, 나머지 대부분의 위치는 0.0이다.

왜 중요한 과제인가?

• 이산 기호가 아닌 연속 실수값을 저장해야 한다.
• 두 개의 값을 오랫동안 보존해야 한다.
• 표시 위치가 고정되지 않는다.
• 마지막에 덧셈 계산까지 수행해야 한다.

원 논문 표 7: Adding Problem 결과

해석

논문은 LSTM이 긴 시간 동안 연속값을 큰 손실 없이 저장하고, 분산 표현과 수치 연산이 필요한 과제를 해결할 수 있음을 보인다.

5.5 Experiment 5: Multiplication Problem

과제 목표

입력 구성과 두 개의 마커 위치는 Adding Problem과 유사하다. 차이는 마지막 목표값이 덧셈이 아니라 곱셈이라는 점이다.

$$ target=X_1 \times X_2 $$

왜 덧셈 다음에 곱셈을 검증했는가?

CEC는 이전 상태에 새 값을 더하는 구조이므로, “덧셈 과제는 CEC의 누적 특성 때문에 쉽게 풀린 것 아니냐”는 의문이 생길 수 있다.

곱셈은 단순 누적으로 바로 해결할 수 없는 비적분적 연산이다. 이 과제는 LSTM이 단순 누적기가 아니라, 장기 기억을 바탕으로 더 복잡한 연산도 학습할 수 있는지 검증한다.

5.6 Experiment 6: Temporal Order

Task 6a: 멀리 떨어진 두 기호의 순서

시퀀스는 E로 시작하고 B라는 트리거 기호로 끝난다. 그 사이에는 주로 {a,b,c,d}에서 무작위로 선택된 방해 기호가 나타난다. 다만 두 위치에는 X 또는 Y가 등장한다.

정답 클래스는 X, Y의 시간 순서로 결정된다.

Task 6b: 멀리 떨어진 세 기호의 순서

6b에서는 중요 기호가 세 개가 되며, 가능한 순서 조합은 8개다.

예를 들어 X,Y,X와 Y,X,X는 서로 다른 클래스로 분류해야 한다. 오류 신호는 시퀀스 마지막에만 제공된다.

원 논문 표 9: Temporal Order 결과

해석

이 실험은 LSTM이 단순히 어떤 기호가 등장했는지를 기억하는 수준을 넘어, 멀리 떨어진 여러 입력의 시간적 순서까지 추출할 수 있음을 검증한다.

5.7 Summary of Experimental Conditions

5.7절은 앞선 실험의 조건을 원 논문 표 10, 표 11로 종합한다. 세부 표에는 과제 번호, 최소 길이, 지연, 셀 블록 수, 입력·출력 수, 가중치 수, 게이트 편향, 함수, 학습률 등이 기록되어 있다.

블로그 관점에서 중요한 비교 기준은 다음과 같다.

6. Discussion: 한계와 장점

6.1 Limitations of LSTM

강하게 지연된 XOR 문제

멀리 떨어진 두 입력의 XOR를 계산하는 문제는 두 정보 중 하나만 기억해서는 오차를 줄일 수 없다. 따라서 쉬운 하위 문제부터 점진적으로 해결하기 어려운 비분해적 문제다.

추가 게이트에 따른 구조 증가

메모리 셀 블록마다 입력 게이트와 출력 게이트가 추가된다. 원 논문은 완전 연결 구조에서 가중치 수가 증가할 수 있음을 인정하지만, 실험에서는 비교 모델과 비슷한 가중치 수를 사용하려 했다.

정밀한 시간 카운팅의 한계

99단계 전과 100단계 전을 반드시 구분해야 하는 문제에는 별도의 카운팅 장치가 필요할 수 있다. 반면 3단계 전과 11단계 전처럼 비교적 큰 시간 차이를 구분하는 것은 가능하다고 설명한다.

6.2 Advantages of LSTM

• CEC를 통해 매우 긴 시간 간격의 오차 흐름을 유지할 수 있다.
• 잡음, 연속값, 분산 표현을 다룰 수 있다.
• 중요한 입력 위치가 달라져도 일반화할 수 있다.
• 입력·출력 게이트 편향, 학습률 등 여러 설정 범위에서 비교적 안정적으로 동작한다.
• 가중치당·시간 단계당 갱신 복잡도가 본질적으로 BPTT와 같은 $O(1)$ 수준이다.

7. Conclusion: 논문의 핵심 결론

논문은 메모리 셀 내부의 CEC가 일정한 오차 흐름을 제공하고, 이것이 긴 시간 간격을 연결하는 기반이라고 결론 내린다.

CEC
→ 정보와 오차 흐름을 장기간 유지

입력 게이트
→ 관련 없는 입력이 기억을 덮어쓰지 못하게 보호

출력 게이트
→ 현재 필요하지 않은 기억이 다른 유닛을 방해하지 않도록 보호

LSTM의 핵심은 “모든 것을 오래 기억한다”가 아니다. 필요한 정보를 선택적으로 기록하고, 보존하고, 필요한 시점에만 출력하는 구조라는 점이다.

최종 정리

1997년 LSTM 논문은 긴 시퀀스에서 RNN이 실패하는 원인을 그래디언트 흐름 관점에서 분석하고, 이를 CEC와 게이트라는 구조로 해결하려 한 연구다.

기존 RNN의 그래디언트 소실·폭주
        ↓
CEC로 일정한 오차 흐름 확보
        ↓
입력 게이트로 기록 제어
        ↓
출력 게이트로 사용 제어
        ↓
긴 지연·잡음·연속값·시간 순서 실험으로 검증

현대 LSTM은 Forget Gate 등을 추가하며 발전했지만, 긴 시간 간격의 정보를 학습하기 위해 기억 경로와 게이트 제어를 분리한다는 원 논문의 핵심 아이디어는 그대로 이어지고 있다.

참고 자료

• Hochreiter, S., & Schmidhuber, J. (1997). Long Short-Term Memory. Neural Computation, 9(8), 1735-1780.
• 원 논문 PDF

목표 점수 target을 정확히 만들기 위해 다트를 던진다.

최적의 방법은 다음 우선순위로 결정한다.

1. 던지는 다트 수를 최소화한다.
2. 다트 수가 같으면 싱글 또는 불을 맞힌 횟수를 최대화한다.

[최소 다트 수, 최대 싱글 또는 불 횟수]를 반환한다.

핵심 아이디어

dp[score]를 정확히 score점을 만드는 최적의 결과라고 정의한다.

dp[score] = [최소 다트 수, 그때의 최대 싱글 또는 불 횟수]

마지막 다트로 얻은 점수가 points라면, 그 직전에는 score - points점을 만들었어야 한다.

dp[score] = dp[score - points] + 마지막 다트 정보

각 점수에서 가능한 모든 마지막 다트를 비교하면 된다.

다트 점수와 싱글 횟수

한 번의 다트로 만들 수 있는 점수는 다음과 같다.

• 싱글: 1 ~ 20점, 싱글 또는 불 횟수 1 증가
• 더블: 2 ~ 40점, 횟수 증가 없음
• 트리플: 3 ~ 60점, 횟수 증가 없음
• 불: 50점, 싱글 또는 불 횟수 1 증가

같은 점수를 만들 수 있는 방법이 여러 개여도 모두 비교 기준에 따라 처리한다. 예를 들어 6점은 싱글 6, 더블 3, 트리플 2로 만들 수 있고, 다트 수가 같으므로 싱글 6을 선택한다.

풀이 과정

1. 한 번에 만들 수 있는 점수와 싱글 또는 불 횟수 증가량을 만든다.
2. dp[0] = [0, 0]으로 초기화한다.
3. 1점부터 target점까지 순서대로 확인한다.
4. 가능한 마지막 다트를 하나씩 적용해 이전 점수의 결과를 갱신한다.
5. 다트 수가 더 적거나, 다트 수가 같으면서 싱글 또는 불 횟수가 더 많을 때만 갱신한다.

Python 코드

def solution(target):
    throws = [(50, 1)]

    for number in range(1, 21):
        throws.append((number, 1))       # 싱글
        throws.append((number * 2, 0))   # 더블
        throws.append((number * 3, 0))   # 트리플

    infinity = target + 1
    dp = [(infinity, -1) for _ in range(target + 1)]
    dp[0] = (0, 0)

    for score in range(1, target + 1):
        for points, single_or_bull in throws:
            if score < points:
                continue

            previous_darts, previous_singles = dp[score - points]

            if previous_darts == infinity:
                continue

            candidate_darts = previous_darts + 1
            candidate_singles = previous_singles + single_or_bull
            current_darts, current_singles = dp[score]

            if (
                candidate_darts < current_darts
                or (
                    candidate_darts == current_darts
                    and candidate_singles > current_singles
                )
            ):
                dp[score] = (candidate_darts, candidate_singles)

    return list(dp[target])

예시

target = 21

7 트리플로 한 번에 21점을 만들 수 있다.

다트 수: 1
싱글 또는 불 횟수: 0
결과: [1, 0]

target = 58

불 50점과 싱글 8점을 사용하면 2번의 다트로 58점을 만든다.

불 50 + 싱글 8 = 58
다트 수: 2
싱글 또는 불 횟수: 2
결과: [2, 2]

시간 복잡도

한 번의 다트로 만들 수 있는 경우는 불을 포함해 61개다.

T를 목표 점수라고 하자.

• 시간 복잡도: O(T * 61) = O(T)
• 공간 복잡도: O(T)

target <= 100,000이므로 충분히 빠르게 동작한다.

정리

각 점수마다 최소 다트 수와 최대 싱글 또는 불 횟수를 함께 저장한다. 다트 수를 우선 비교하고, 동점일 때만 싱글 또는 불 횟수를 비교하면 문제의 우선순위를 그대로 DP에 반영할 수 있다.

수열의 연속 부분 수열 하나를 선택하고, [1, -1, 1, -1, ...] 또는 [-1, 1, -1, 1, ...] 형태의 펄스 수열을 곱한다.

만들어진 연속 펄스 부분 수열의 합 중 최댓값을 구한다.

핵심 아이디어

펄스 수열의 부호 패턴은 두 가지뿐이다.

패턴 1:  1, -1,  1, -1, ...
패턴 2: -1,  1, -1,  1, ...

원본 수열의 인덱스 기준으로 두 패턴을 미리 곱한 변환 수열을 생각한다.

sequence:  2,  3, -6,  1, ...
변환 1:    2, -3, -6, -1, ...
변환 2:   -2,  3,  6,  1, ...

어떤 연속 구간을 선택하더라도 두 변환 수열 중 하나의 연속 부분합으로 표현된다. 따라서 두 변환 수열 각각에서 최대 연속 부분합을 구하고, 둘 중 큰 값을 선택하면 된다.

최대 연속 부분합은 카데인 알고리즘으로 한 번의 순회에 구할 수 있다.

카데인 알고리즘

현재 값이 value일 때, 현재 위치에서 끝나는 최대 연속 부분합은 두 선택지 중 큰 값이다.

current = max(value, current + value)

• 이전 구간을 버리고 현재 값부터 새로 시작한다.
• 이전 구간 뒤에 현재 값을 이어 붙인다.

두 변환 패턴의 값을 같은 순회에서 함께 계산할 수 있다.

풀이 과정

1. 짝수 인덱스에서는 첫 번째 패턴의 부호를 1, 홀수 인덱스에서는 -1로 둔다.
2. 두 번째 패턴은 첫 번째 패턴의 부호를 반대로 한 값이다.
3. 각 변환 값에 카데인 알고리즘을 적용한다.
4. 두 패턴에서 얻은 최대 부분합 중 큰 값을 반환한다.

Python 코드

def solution(sequence):
    current_first = 0
    current_second = 0
    answer = -float("inf")

    for index, value in enumerate(sequence):
        # 전체 수열 기준 [1, -1, 1, -1, ...] 패턴
        first_value = value if index % 2 == 0 else -value
        second_value = -first_value

        # 각 패턴에서 현재 인덱스로 끝나는 최대 연속 부분합
        current_first = max(first_value, current_first + first_value)
        current_second = max(second_value, current_second + second_value)

        answer = max(answer, current_first, current_second)

    return answer

예시

sequence = [2, 3, -6, 1, 3, -1, 2, 4]에서 인덱스 1부터 3까지의 구간을 선택해 보자.

원본 구간:      [3, -6, 1]
펄스 수열:      [1, -1, 1]
곱한 결과:      [3, 6, 1]
합:             10

이 구간은 두 번째 변환 패턴에서의 연속 부분합으로 계산되며, 전체 최댓값은 10이다.

시간 복잡도

N을 수열 길이라고 하자.

수열을 한 번만 순회하고, 두 개의 현재 부분합만 유지한다.

• 시간 복잡도: O(N)
• 공간 복잡도: O(1)

정리

펄스 수열의 시작 부호는 두 가지뿐이다. 두 부호 패턴으로 수열을 변환한 뒤 각각의 최대 연속 부분합을 구하면, 모든 연속 펄스 부분 수열의 합 중 최댓값을 선형 시간에 구할 수 있다.

도로는 수평 또는 수직 선분이며, 교차하거나 만나는 지점에서 서로 연결된다. 각 도로의 중앙에는 제한 속도를 가진 카메라가 있다.

1번 도시에서 출발해 각 도시까지 일정한 속도로 이동할 때, 경로에서 통과하는 모든 카메라 제한 속도를 만족하는 최대 속도를 구한다. 카메라를 전혀 지나지 않는 경로가 있으면 제한 없이 이동할 수 있으므로 0을 반환한다.

핵심 아이디어

1. 도로를 중요한 지점에서 분할해 그래프로 만든다

도로 위에서 이동 방향을 바꾸거나 제한 속도가 발생할 수 있는 지점은 다음과 같다.

• 도로의 양 끝점
• 도시 위치
• 도로의 교차점 또는 접점
• 각 도로 중앙의 카메라 위치

각 도로에 포함된 중요 지점을 도로 방향으로 정렬하고, 이웃한 지점을 간선으로 연결한다. 같은 좌표의 지점은 하나의 그래프 정점으로 합친다.

2. 카메라는 정점 제한으로 처리한다

카메라가 있는 지점을 지나면 제한 속도를 지켜야 한다. 카메라 정점과 연결되는 모든 간선의 제한 속도를 해당 카메라 제한 속도 이하로 설정하면, 그 정점을 통과하는 모든 경로에 제한을 적용할 수 있다.

한 좌표에 카메라가 여러 개라면 가장 낮은 제한 속도를 사용한다.

3. Widest Path를 사용한다

어떤 경로로 이동 가능한 최고 일정 속도는 그 경로에서 만나는 간선 제한 속도의 최솟값이다. 도시마다 이 최솟값을 최대화해야 하므로 최단 경로가 아니라 Widest Path 문제다.

다익스트라와 같은 방식으로 우선순위 큐를 사용하되, 거리가 작은 정점 대신 현재까지의 제한 속도가 큰 정점을 먼저 꺼낸다.

next_speed = min(current_speed, edge_limit)

Python 코드

from heapq import heappop, heappush


def solution(city, road):
    INF = 10**18
    road_count = len(road)

    # (x1, y1, x2, y2, limit) 형태로 도로를 저장한다.
    roads = [tuple(info) for info in road]
    road_points = [[] for _ in range(road_count)]
    camera_limit = {}

    def is_horizontal(current_road):
        return current_road[1] == current_road[3]

    def contains(current_road, x, y):
        x1, y1, x2, y2, _ = current_road
        return x1 <= x <= x2 and y1 <= y <= y2

    def add_point(road_index, point):
        road_points[road_index].append(point)

    # 도로의 끝점과 중앙 카메라 위치를 추가한다.
    for index, (x1, y1, x2, y2, limit) in enumerate(roads):
        add_point(index, (x1, y1))
        add_point(index, (x2, y2))

        camera = ((x1 + x2) // 2, (y1 + y2) // 2)
        add_point(index, camera)
        camera_limit[camera] = min(camera_limit.get(camera, INF), limit)

    # 도시가 포함된 모든 도로에 도시 위치를 분할 지점으로 추가한다.
    for x, y in city:
        for index, current_road in enumerate(roads):
            if contains(current_road, x, y):
                add_point(index, (x, y))

    # 모든 도로 쌍의 교차점 또는 접점을 찾는다.
    for first in range(road_count):
        x1, y1, x2, y2, _ = roads[first]
        first_is_horizontal = is_horizontal(roads[first])

        for second in range(first + 1, road_count):
            a1, b1, a2, b2, _ = roads[second]
            second_is_horizontal = is_horizontal(roads[second])
            point = None

            if first_is_horizontal != second_is_horizontal:
                horizontal = roads[first] if first_is_horizontal else roads[second]
                vertical = roads[second] if first_is_horizontal else roads[first]

                hx1, hy, hx2, _, _ = horizontal
                vx, vy1, _, vy2, _ = vertical

                if hx1 <= vx <= hx2 and vy1 <= hy <= vy2:
                    point = (vx, hy)

            elif first_is_horizontal and y1 == b1:
                left = max(x1, a1)
                right = min(x2, a2)
                if left == right:
                    point = (left, y1)

            elif not first_is_horizontal and x1 == a1:
                bottom = max(y1, b1)
                top = min(y2, b2)
                if bottom == top:
                    point = (x1, bottom)

            if point is not None:
                add_point(first, point)
                add_point(second, point)

    # 동일 좌표를 하나의 그래프 정점으로 합친다.
    point_to_index = {}

    def get_index(point):
        if point not in point_to_index:
            point_to_index[point] = len(point_to_index)
        return point_to_index[point]

    # 먼저 모든 중요 지점에 그래프 번호를 부여한다.
    for points in road_points:
        for point in points:
            get_index(point)

    graph = [[] for _ in range(len(point_to_index))]

    # 각 도로의 이웃한 중요 지점을 연결한다.
    for index, points in enumerate(road_points):
        current_road = roads[index]

        if is_horizontal(current_road):
            ordered_points = sorted(set(points), key=lambda point: point[0])
        else:
            ordered_points = sorted(set(points), key=lambda point: point[1])

        for left, right in zip(ordered_points, ordered_points[1:]):
            # 카메라 정점을 지나기 위해 지켜야 하는 가장 낮은 제한 속도
            limit = min(
                camera_limit.get(left, INF),
                camera_limit.get(right, INF),
            )

            left_index = get_index(left)
            right_index = get_index(right)
            graph[left_index].append((right_index, limit))
            graph[right_index].append((left_index, limit))

    start = get_index(tuple(city[0]))
    best_speed = [-1] * len(graph)
    best_speed[start] = INF
    heap = [(-INF, start)]

    # 제한 속도가 큰 경로부터 확정하는 Widest Path 탐색
    while heap:
        negative_speed, current = heappop(heap)
        current_speed = -negative_speed

        if current_speed < best_speed[current]:
            continue

        for next_node, limit in graph[current]:
            next_speed = min(current_speed, limit)

            if next_speed > best_speed[next_node]:
                best_speed[next_node] = next_speed
                heappush(heap, (-next_speed, next_node))

    answer = []
    for x, y in city[1:]:
        speed = best_speed[get_index((x, y))]
        answer.append(0 if speed == INF else speed)

    return answer

정당성 설명

그래프의 정점은 도시, 도로 끝점, 교차점, 카메라 위치를 모두 포함한다. 따라서 실제 도로에서 이동 방향을 바꾸거나 카메라 제한이 적용될 수 있는 모든 지점이 그래프에 표현된다. 각 도로에서 이웃한 중요 지점을 연결했으므로 그래프의 경로와 실제 도로 이동 경로는 서로 대응한다.

카메라 정점에 인접한 간선의 제한 속도는 그 지점의 최소 카메라 제한 속도 이하로 설정했다. 그러므로 그래프 경로의 최소 간선 제한 속도는 실제 이동 경로에서 지켜야 하는 가장 낮은 카메라 제한 속도와 같다.

Widest Path 탐색은 현재까지 확보한 제한 속도가 가장 큰 정점을 먼저 처리한다. 어떤 도시를 향하는 경로의 값은 경로 간선 제한 속도의 최솟값이며, 전이식 min(current_speed, edge_limit)은 이 값을 정확히 계산한다. 더 큰 값이 발견될 때만 갱신하므로, 최종 best_speed는 1번 도시에서 각 정점으로 갈 수 있는 최고 일정 속도다.

따라서 카메라를 지나지 않는 경우에는 INF를 0으로 바꾸고, 그 외에는 best_speed를 반환하면 요구한 답을 얻는다.

복잡도 분석

도로 수를 M, 중요 지점과 그래프 간선 수를 각각 V, E라고 하자.

• 도시를 도로에 추가: O(N * M)
• 도로 교차점 검사: O(M^2)
• 도로별 지점 정렬: O(V log V)
• Widest Path 탐색: O((V + E) log V)

전체 시간 복잡도는 O(M^2 + (V + E) log V)이며, 공간 복잡도는 O(V + E)다.

주의할 점

• 도로가 교차하는 좌표뿐 아니라, 도시와 카메라 위치에서도 도로를 분할해야 한다.
• 같은 카메라 위치에 여러 도로의 카메라가 겹치면 제한 속도는 최솟값을 사용한다.
• 카메라를 지나지 않는 경로는 제한 속도가 없으므로 내부적으로 큰 값 INF로 표현한 뒤 최종 결과에서 0으로 바꾼다.
• 최단 거리 다익스트라와 달리, 우선순위 큐에서는 현재 제한 속도가 큰 상태를 먼저 꺼낸다.

마무리

기하 문제처럼 보이지만 도로를 교차점과 특수 지점에서 분할하면 그래프 문제가 된다. 이후에는 경로의 최소 제한 속도를 최대화하는 Widest Path를 적용해 도시별 최고 일정 속도를 구할 수 있다.

이번에는 Denoising AutoEncoder, Sparse AutoEncoder, VAE(Variational AutoEncoder)를 살펴본다. 이어서 생성자와 판별자가 경쟁하며 새로운 데이터를 만드는 GAN(Generative Adversarial Network)을 구현한다.

1. Latent Representation 다시 확인하기

학습이 끝난 AutoEncoder에서는 Encoder만 따로 사용할 수 있다.

x, _ = next(iter(test_loader))
x = x[:8].to(DEVICE)

with torch.no_grad():
    z = model.encoder(x)

print("입력 Shape:", x.shape)
print("Latent Shape:", z.shape)
print("원본 값 개수:", 1 * 28 * 28)
print("Latent 값 개수:", z[0].numel())

실행 결과:

입력 Shape: torch.Size([8, 1, 28, 28])
Latent Shape: torch.Size([8, 32, 7, 7])
원본 값 개수: 784
Latent 값 개수: 1568

공간 해상도는 28 × 28에서 7 × 7로 줄었지만 Channel이 1에서 32로 늘어 전체 값의 개수는 더 많다.

AutoEncoder의 Latent Representation이 반드시 원본보다 적은 수의 값을 가져야 하는 것은 아니다. 합성곱 AutoEncoder는 공간 크기를 줄이는 대신 Channel을 늘려 원본 복원에 유용한 Feature Map으로 변환하기도 한다.

다만 차원 축소가 목적이라면 전체 원소 수가 실제로 작은 Bottleneck을 설계해야 한다.

2. Denoising AutoEncoder

Denoising AutoEncoder는 노이즈가 섞인 데이터를 입력으로 받고, 깨끗한 원본을 Target으로 사용한다.

노이즈가 섞인 이미지
        ↓
      Encoder
        ↓
Latent Representation
        ↓
      Decoder
        ↓
깨끗하게 복원된 이미지

일반 AutoEncoder와 학습 목표를 비교하면 다음과 같다.

노이즈 추가하기

def add_noise(x, noise_factor=0.35):
    noisy = x + noise_factor * torch.randn_like(x)
    return torch.clamp(noisy, 0.0, 1.0)

• torch.randn_like(x)는 x와 같은 Shape의 표준정규분포 노이즈를 만든다.
• noise_factor는 노이즈의 강도를 조절한다.
• torch.clamp()는 픽셀값이 [0, 1] 범위를 벗어나지 않도록 제한한다.

x, _ = next(iter(train_loader))
noisy_x = add_noise(x[:8])

위쪽은 원본, 아래쪽은 노이즈를 추가한 이미지다.

학습하기

denoising_model = ConvAutoencoder().to(DEVICE)
criterion = nn.MSELoss()
optimizer = optim.AdamW(
    denoising_model.parameters(),
    lr=1e-3,
)

def train_denoising_autoencoder(
    model,
    loader,
    criterion,
    optimizer,
    device,
    epochs=5,
):
    for epoch in range(epochs):
        model.train()
        running_loss = 0.0

        for clean_x, _ in loader:
            clean_x = clean_x.to(device)
            noisy_x = add_noise(clean_x)

            _, restored_x = model(noisy_x)
            loss = criterion(restored_x, clean_x)

            optimizer.zero_grad()
            loss.backward()
            optimizer.step()

            running_loss += (
                loss.item() * clean_x.size(0)
            )

        epoch_loss = (
            running_loss / len(loader.dataset)
        )

        print(
            f"Epoch {epoch + 1:02d}/{epochs} | "
            f"Loss: {epoch_loss:.6f}"
        )

중요한 부분은 다음 한 줄이다.

loss = criterion(restored_x, clean_x)

모델에는 noisy_x를 입력하지만 Loss는 복원 결과와 깨끗한 clean_x 사이에서 계산한다.

노트북의 실행 결과는 다음과 같다.

Epoch 01/5 | Loss: 0.029357
Epoch 02/5 | Loss: 0.008080
Epoch 03/5 | Loss: 0.007619
Epoch 04/5 | Loss: 0.007417
Epoch 05/5 | Loss: 0.007292

복원 결과

위쪽: 노이즈 이미지
가운데: 모델이 복원한 이미지
아래쪽: 깨끗한 원본 이미지

노이즈는 크게 줄었고 숫자 형태도 잘 복원했다. 다만 원본보다 획이 부드럽거나 일부 세부 정보가 손실될 수 있다.

3. Sparse AutoEncoder

일반 AutoEncoder는 Latent Representation의 많은 뉴런을 자유롭게 사용할 수 있다. Sparse AutoEncoder는 가능하면 적은 수의 뉴런만 강하게 활성화하도록 제약을 추가한다.

일반 Latent
[0.7, 0.5, 0.9, 0.4, 0.6, 0.8]

Sparse Latent
[0.8, 0.0, 1.3, 0.0, 0.9, 0.0]

모든 뉴런을 조금씩 사용하는 대신 일부 뉴런이 특정 특징을 강하게 표현하도록 유도한다.

전체 Loss = Reconstruction Loss + Sparsity Penalty

Latent 공간에 어떤 제약을 주느냐에 따라 학습되는 표현의 성질이 달라진다.

4. AutoEncoder를 이용한 이상 탐지

정상 데이터만 충분히 학습한 AutoEncoder는 정상 패턴을 잘 복원하도록 최적화된다.

정상 데이터
→ 학습한 패턴과 비슷함
→ 잘 복원됨
→ Reconstruction Error가 작음

이상 데이터
→ 학습한 패턴과 다름
→ 잘 복원하지 못함
→ Reconstruction Error가 큼

Reconstruction Error를 기준으로 Threshold를 정하면 이상 여부를 판단할 수 있다.

error = ((x - x_hat) ** 2).mean(dim=(1, 2, 3))
is_anomaly = error > threshold

하지만 AutoEncoder가 이상 데이터까지 잘 복원하는 경우도 있다. 따라서 실제 시스템에서는 다음 내용을 함께 고려해야 한다.

• 정상 데이터의 다양성을 충분히 포함했는가?
• Validation 데이터로 적절한 Threshold를 정했는가?
• 이상 유형별 Recall과 False Positive Rate는 어떤가?
• 다른 이상 탐지 방법과 비교했는가?

Reconstruction Error 하나만으로 모든 이상 탐지 문제를 해결할 수 있는 것은 아니다.

5. 일반 AutoEncoder의 생성 한계

일반 AutoEncoder는 입력을 복원하는 작업에는 적합하다.

x → Encoder → z → Decoder → x_hat

새 이미지를 생성하려면 입력 이미지 없이 시작점 z를 정해야 한다.

? → z → Decoder → 새로운 이미지

하지만 일반 AutoEncoder는 Latent Space가 규칙적인 확률분포가 되도록 학습하지 않는다.

• 학습 데이터가 존재하는 위치에서는 Decoder가 잘 동작할 수 있다.
• 학습된 위치 사이에는 Decoder가 충분히 학습하지 못한 빈 영역이 생길 수 있다.
• 임의의 z를 Decoder에 넣는다고 자연스러운 이미지가 보장되지는 않는다.

즉, 복원을 잘하는 것과 새로운 데이터를 잘 생성하는 것은 다른 문제다.

6. VAE

VAE(Variational AutoEncoder)는 Latent Space를 확률분포로 표현하는 생성 모델이다.

일반 AutoEncoder는 이미지 하나를 정확한 좌표 하나로 바꾼다.

이미지 → z = [2.0, 3.0]

VAE는 정확한 점 하나가 아니라 평균과 분산으로 범위를 표현한다.

이미지
  ↓
Encoder
  ↓
평균 μ, 분산 σ²
  ↓
분포에서 z Sampling
  ↓
Decoder
  ↓
복원 이미지

지도에 비유하면 다음과 같다.

각 데이터를 주변 범위까지 포함해 표현하고, 분포들이 지나치게 흩어지지 않도록 제한한다. 이를 통해 임의의 z를 Sampling해도 의미 있는 결과가 나올 가능성을 높인다.

7. Reparameterization Trick

Encoder는 μ와 logvar를 출력한다.

var = σ²
logvar = log(σ²)

표준편차는 다음처럼 계산한다.

std = torch.exp(0.5 * logvar)

그다음 표준정규분포에서 epsilon을 뽑아 z를 만든다.

eps = torch.randn_like(std)
z = mu + eps * std

eps는 “아주 작은 값”을 의미하지 않는다. 그리스 문자 epsilon을 변수 이름으로 사용한 것이며, 여기서는 평균 0, 표준편차 1인 정규분포에서 뽑은 무작위 표본이다.

실행 예제

mu = torch.tensor([[1.0, 2.0]])
logvar = torch.tensor([[0.0, 0.0]])

std = torch.exp(0.5 * logvar)
eps = torch.randn_like(std)
z = mu + eps * std

print("mu:", mu)
print("logvar:", logvar)
print("std:", std)
print("epsilon:", eps)
print("z:", z)

노트북의 실행 결과:

mu:      tensor([[1., 2.]])
logvar:  tensor([[0., 0.]])
std:     tensor([[1., 1.]])
epsilon: tensor([[-1.1235, 1.1595]])
z:       tensor([[-0.1235, 3.1595]])

왜 이 방식이 필요한가?

단순히 분포에서 직접 Sampling하면 무작위 연산을 통과해 μ와 σ를 만든 Encoder까지 Gradient를 전달하기 어렵다.

z = μ + ε × σ

처럼 표현하면 무작위성은 ε에 분리되고, z는 μ와 σ에 대한 미분 가능한 계산으로 만들어진다. 이를 Reparameterization Trick이라고 한다.

8. VAE Loss

VAE는 두 Loss를 함께 사용한다.

VAE Loss = Reconstruction Loss + KL Loss

수식으로 표현하면 다음과 같다.

L_VAE = L_reconstruction + L_KL

Reconstruction Loss

원본과 복원 이미지의 차이를 줄인다.

recon_loss = F.binary_cross_entropy(
    x_hat,
    x,
    reduction="sum",
)

KL Divergence

Encoder가 만든 분포가 기준 분포인 표준정규분포 N(0, I)에서 지나치게 멀어지지 않도록 한다.

kl_loss = -0.5 * torch.sum(
    1 + logvar - mu.pow(2) - logvar.exp()
)

KL Loss는 Latent Space가 무작위 Sampling에 적합한 규칙적인 형태를 갖도록 유도한다.

def vae_loss_function(x_hat, x, mu, logvar):
    recon_loss = F.binary_cross_entropy(
        x_hat,
        x,
        reduction="sum",
    )

    kl_loss = -0.5 * torch.sum(
        1 + logvar - mu.pow(2) - logvar.exp()
    )

    batch_size = x.size(0)

    total_loss = (
        recon_loss + kl_loss
    ) / batch_size

    recon_loss = recon_loss / batch_size
    kl_loss = kl_loss / batch_size

    return total_loss, recon_loss, kl_loss

reduction="sum"으로 784개 픽셀의 BCE를 모두 더한 뒤 Batch 크기로 나누므로 Loss 값이 기본 AutoEncoder의 평균 MSE보다 크게 나타난다. 서로 다른 계산 방식의 Loss 숫자를 직접 비교하면 안 된다.

9. 합성곱 VAE 구현

이번 VAE는 2차원 Latent Space를 사용한다.

LATENT_DIM = 2

전체 Shape 흐름

x: [B, 1, 28, 28]
        ↓ Encoder
h: [B, 64, 7, 7]
        ↓ Flatten
h: [B, 3136]
        ├─ fc_mu     → μ:      [B, 2]
        └─ fc_logvar → logvar: [B, 2]
                           ↓ Sampling
                         z: [B, 2]
                           ↓ fc_decode
                       [B, 3136]
                           ↓ Reshape
                       [B, 64, 7, 7]
                           ↓ Decoder
x_hat: [B, 1, 28, 28]

모델 코드

class ConvVAE(nn.Module):
    def __init__(self, latent_dim=2):
        super().__init__()

        self.encoder = nn.Sequential(
            nn.Conv2d(
                1, 32,
                kernel_size=3,
                stride=2,
                padding=1,
            ),
            nn.ReLU(),
            nn.Conv2d(
                32, 64,
                kernel_size=3,
                stride=2,
                padding=1,
            ),
            nn.ReLU(),
        )

        self.flatten_dim = 64 * 7 * 7

        self.fc_mu = nn.Linear(
            self.flatten_dim,
            latent_dim,
        )
        self.fc_logvar = nn.Linear(
            self.flatten_dim,
            latent_dim,
        )

        self.fc_decode = nn.Linear(
            latent_dim,
            self.flatten_dim,
        )

        self.decoder = nn.Sequential(
            nn.ConvTranspose2d(
                64, 32,
                kernel_size=4,
                stride=2,
                padding=1,
            ),
            nn.ReLU(),
            nn.ConvTranspose2d(
                32, 1,
                kernel_size=4,
                stride=2,
                padding=1,
            ),
            nn.Sigmoid(),
        )

    def encode(self, x):
        h = self.encoder(x)
        h = h.view(x.size(0), -1)

        mu = self.fc_mu(h)
        logvar = self.fc_logvar(h)

        return mu, logvar

    def reparameterize(self, mu, logvar):
        std = torch.exp(0.5 * logvar)
        eps = torch.randn_like(std)
        return mu + eps * std

    def decode(self, z):
        h = self.fc_decode(z)
        h = h.view(-1, 64, 7, 7)
        return self.decoder(h)

    def forward(self, x):
        mu, logvar = self.encode(x)
        z = self.reparameterize(mu, logvar)
        x_hat = self.decode(z)

        return x_hat, mu, logvar

10. VAE 학습 결과

optimizer = optim.AdamW(
    vae.parameters(),
    lr=1e-3,
)

학습 중에는 Total, Reconstruction, KL Loss를 각각 기록한다.

노트북의 주요 결과는 다음과 같다.

• Total Loss와 Reconstruction Loss가 감소하며 복원 성능이 좋아졌다.
• KL Loss는 단조롭게 감소하지 않고 Reconstruction 목표와 균형을 이루며 변한다.
• KL Loss가 지나치게 빠르게 0에 가까워지면 Decoder가 Latent 정보를 거의 사용하지 않는 Posterior Collapse가 발생할 수 있다.

따라서 KL Loss는 무조건 작을수록 좋다고 해석하면 안 된다.

VAE 복원 결과

위쪽은 원본, 아래쪽은 복원 이미지다. 숫자의 전체적인 형태는 복원하지만 기본 AutoEncoder보다 부드럽거나 흐리게 보이는 결과가 있다.

VAE는 Latent에서 확률적으로 z를 Sampling하며, 단순한 픽셀 단위 Loss와 Decoder 구조도 흐릿한 복원에 영향을 줄 수 있다.

11. 현대 생성 모델에서 VAE의 역할

고품질 이미지 생성은 단순 VAE만 사용하는 방식보다 Diffusion, Transformer, Flow 계열 모델과 결합하는 방향으로 발전했다.

Latent 기반 생성 모델은 고해상도 픽셀 공간을 직접 처리하는 대신 이미지를 작은 Latent Representation으로 압축한 뒤 그 공간에서 생성 작업을 수행할 수 있다.

Image
  ↓
VAE Encoder
  ↓
Latent Representation
  ↓
Diffusion 또는 Transformer
  ↓
Generated Latent
  ↓
VAE Decoder
  ↓
Generated Image

VAE는 확률적인 Latent Space와 Sampling을 이해하는 기초가 된다.

12. GAN

GAN(Generative Adversarial Network)은 생성자와 판별자라는 두 신경망이 서로 경쟁하며 학습하는 생성 모델이다.

Noise z
   ↓
Generator
   ↓
Fake Image ──┐
             ├─ Discriminator → Real 또는 Fake 판단
Real Image ──┘

GAN은 특정 학습 이미지를 그대로 복사하는 것이 아니라 학습 데이터의 분포와 특징을 익혀 새로운 Sample을 생성하는 것을 목표로 한다.

13. GAN의 경쟁 학습

판별자 학습

판별자는 실제 이미지에는 1, 생성자가 만든 이미지에는 0을 출력하도록 학습한다.

실제 이미지 x
→ D(x)
→ 1에 가깝게

가짜 이미지 G(z)
→ D(G(z))
→ 0에 가깝게

생성자 학습

생성자는 판별자가 가짜 이미지를 진짜라고 판단하도록 학습한다.

z → G(z) → D(G(z)) → 1에 가깝게

여기서 Target 1은 생성 이미지가 실제 정답 이미지라는 뜻이 아니다. 생성자가 판별자를 속이기 위해 원하는 목표값이다.

두 모델을 번갈아 학습하면서 생성자는 실제 데이터 분포와 비슷한 Sample을 만들고, 판별자는 진짜와 가짜를 더 잘 구별하도록 발전한다.

14. GAN 데이터 준비

BATCH_SIZE = 128

transform = transforms.Compose([
    transforms.ToTensor(),
    transforms.Normalize((0.5,), (0.5,)),
])

train_dataset = datasets.MNIST(
    root="./data",
    train=True,
    download=True,
    transform=transform,
)

train_loader = DataLoader(
    train_dataset,
    batch_size=BATCH_SIZE,
    shuffle=True,
    num_workers=4,
)

ToTensor()가 만든 [0, 1] 범위의 픽셀을 Normalize((0.5,), (0.5,))로 변환한다.

정규화 결과 = (x - 0.5) / 0.5

0 → -1
1 →  1

실행 결과:

images: torch.Size([128, 1, 28, 28])
min/max: -1.0 1.0

Generator 마지막 층의 Tanh도 [-1, 1] 범위를 출력하므로 실제 이미지와 생성 이미지의 범위를 맞춘다.

15. Generator

Generator는 이미지가 아니라 z라는 무작위 벡터를 입력받는다.

z [B, 100]
   ↓ Linear
[B, 256]
   ↓ Linear
[B, 512]
   ↓ Linear
[B, 784]
   ↓ Reshape
[B, 1, 28, 28]

LATENT_DIM = 100

class Generator(nn.Module):
    def __init__(self, latent_dim=100):
        super().__init__()

        self.net = nn.Sequential(
            nn.Linear(latent_dim, 256),
            nn.LeakyReLU(0.2, inplace=True),

            nn.Linear(256, 512),
            nn.BatchNorm1d(512),
            nn.LeakyReLU(0.2, inplace=True),

            nn.Linear(512, 28 * 28),
            nn.Tanh(),
        )

    def forward(self, z):
        x = self.net(z)
        return x.view(-1, 1, 28, 28)

G = Generator(LATENT_DIM).to(DEVICE)

LeakyReLU

일반 ReLU는 음수 입력을 모두 0으로 만든다. LeakyReLU(0.2)는 음수 영역에도 작은 기울기를 남겨 Gradient가 완전히 끊어지는 현상을 줄인다.

x >= 0 → x
x < 0  → 0.2x

생성 결과 Shape

z = torch.randn(
    8,
    LATENT_DIM,
    device=DEVICE,
)

fake = G(z)

print("Noise:", z.shape)
print("Fake:", fake.shape)
print("Range:", fake.min().item(), fake.max().item())

노트북 실행 결과:

Noise: torch.Size([8, 100])
Fake: torch.Size([8, 1, 28, 28])
Range: -0.9718 0.9251

16. Discriminator

Discriminator는 이미지 한 장을 받아 진짜 여부를 나타내는 Logit 하나를 출력한다.

[B, 1, 28, 28]
   ↓ Flatten
[B, 784]
   ↓ Linear
[B, 512]
   ↓ Linear
[B, 256]
   ↓ Linear
[B, 1] Logit

class Discriminator(nn.Module):
    def __init__(self):
        super().__init__()

        self.net = nn.Sequential(
            nn.Flatten(),

            nn.Linear(28 * 28, 512),
            nn.LeakyReLU(0.2, inplace=True),
            nn.Dropout(0.3),

            nn.Linear(512, 256),
            nn.LeakyReLU(0.2, inplace=True),
            nn.Dropout(0.3),

            nn.Linear(256, 1),
        )

    def forward(self, x):
        return self.net(x)

D = Discriminator().to(DEVICE)

마지막에 Sigmoid를 넣지 않는다. BCEWithLogitsLoss가 내부에서 Sigmoid와 BCE를 수치적으로 안정적인 방식으로 함께 계산하기 때문이다.

criterion = nn.BCEWithLogitsLoss()

확률이 필요할 때만 출력 Logit에 torch.sigmoid()를 적용한다.

with torch.no_grad():
    logits = D(fake)
    probabilities = torch.sigmoid(logits)

17. Optimizer와 Fixed Noise

Generator와 Discriminator는 서로 다른 Optimizer를 사용한다.

optimizer_D = torch.optim.Adam(
    D.parameters(),
    lr=2e-4,
    betas=(0.5, 0.999),
)

optimizer_G = torch.optim.Adam(
    G.parameters(),
    lr=2e-4,
    betas=(0.5, 0.999),
)

학습 변화를 비교할 고정 Noise도 만든다.

fixed_noise = torch.randn(
    64,
    LATENT_DIM,
    device=DEVICE,
)

Epoch마다 새로운 Noise를 사용하면 이미지가 달라진 이유가 학습 때문인지 입력 Noise 때문인지 구분하기 어렵다. 같은 fixed_noise를 반복 사용하면 같은 잠재 입력이 학습에 따라 어떻게 바뀌는지 비교할 수 있다.

18. Discriminator 학습 과정

optimizer_D.zero_grad()

# 실제 이미지의 Target은 1
real_logits = D(real)
real_targets = torch.ones_like(real_logits)
d_real_loss = criterion(
    real_logits,
    real_targets,
)

# 생성 이미지의 Target은 0
z = torch.randn(
    batch_size,
    LATENT_DIM,
    device=DEVICE,
)

fake = G(z)
fake_logits = D(fake.detach())
fake_targets = torch.zeros_like(fake_logits)
d_fake_loss = criterion(
    fake_logits,
    fake_targets,
)

d_loss = d_real_loss + d_fake_loss
d_loss.backward()
optimizer_D.step()

fake.detach()가 필요한 이유

fake는 Generator의 연산 결과이므로 원래는 Generator의 계산 그래프와 연결되어 있다.

Discriminator를 학습하는 단계에서는 Generator를 업데이트하면 안 된다.

fake.detach()

는 Tensor 값은 그대로 사용하면서 Generator 방향의 Gradient 연결을 끊는다.

G(z) ──X── Gradient 차단
  ↓
D(fake) → D만 학습

19. Generator 학습 과정

Generator는 판별자가 가짜 이미지를 1, 즉 진짜로 판단하도록 학습한다.

optimizer_G.zero_grad()

z = torch.randn(
    batch_size,
    LATENT_DIM,
    device=DEVICE,
)

fake = G(z)
fake_logits = D(fake)

g_targets = torch.ones_like(fake_logits)
g_loss = criterion(fake_logits, g_targets)

g_loss.backward()
optimizer_G.step()

이 단계에서는 fake.detach()를 사용하지 않는다. D(fake)에서 계산한 Gradient가 Discriminator 연산을 거쳐 Generator까지 전달되어야 하기 때문이다.

G Parameter
   ↑ Gradient
G(z) → D(G(z)) → BCE Target 1

Generator와 Discriminator 학습을 한 Batch 안에서 차례로 반복한다.

20. GAN 생성 결과

Epoch 1

숫자와 비슷한 형태가 나타나기 시작했지만 노이즈가 많고 경계가 불분명하다.

Epoch 10

Epoch 1보다 숫자 형태가 선명해지고 배경 노이즈도 줄었다. 다만 일부 이미지는 어떤 숫자인지 불명확하며 획이 깨진 Sample도 남아 있다.

이 예제는 Linear 층으로 만든 간단한 GAN이므로 학습 시간이 짧고 생성 품질에도 한계가 있다.

21. GAN Loss 해석

노트북의 Epoch별 마지막 Batch Loss는 다음과 같다.

일반적인 분류 모델은 Loss가 계속 낮아지면 학습이 잘된다고 해석하는 경우가 많다. GAN은 두 모델이 서로 경쟁하므로 G Loss와 D Loss가 단조롭게 감소하지 않아도 된다.

다음 내용을 함께 확인해야 한다.

1. 고정 Noise에서 생성 이미지가 점차 좋아지는가?
2. 여러 숫자가 생성되며 다양성이 유지되는가?
3. 거의 같은 이미지만 반복해서 생성하지 않는가?
4. Discriminator가 너무 강해 Generator의 학습이 멈추지 않는가?
5. 생성 이미지에 심한 Artifact가 남아 있지 않은가?

위 표의 값은 각 Epoch 전체 평균이 아니라 학습 코드가 출력한 마지막 Batch의 Loss다. 전체 경향을 비교하려면 Epoch 평균 Loss를 별도로 계산하는 편이 더 안정적이다.

22. Mode Collapse

Mode Collapse는 Generator가 다양한 Sample을 만들지 못하고 비슷한 결과만 반복해서 생성하는 현상이다.

예를 들어 MNIST에는 0부터 9까지 다양한 숫자가 있지만 Generator가 판별자를 잘 속이는 특정 모양의 3만 반복해서 만들 수 있다.

서로 다른 Noise z
  ↓
Generator
  ↓
거의 동일한 이미지들

생성 이미지가 선명해 보여도 다양성이 부족하면 좋은 생성 모델이라고 보기 어렵다.

23. 생성 품질 평가

GAN의 Loss만으로 이미지 품질을 완전히 판단하기 어렵다. 실제 생성 결과를 시각화하고 필요하면 생성 품질 지표를 사용한다.

대표적인 지표로 FID(Fréchet Inception Distance)가 있다.

• 실제 이미지와 생성 이미지의 Feature 분포를 비교한다.
• 일반적으로 값이 낮을수록 두 분포가 비슷하다고 해석한다.
• 충분한 Sample 수와 일관된 평가 설정이 필요하다.

MNIST처럼 단순한 데이터에서는 숫자 분류 모델을 이용해 생성 이미지의 인식률과 클래스 다양성을 함께 확인할 수도 있다.

24. AutoEncoder, VAE, GAN 비교

VAE와 GAN은 모두 생성 모델이지만 학습 방식은 다르다.

VAE
분포를 명시적으로 정리하며 Reconstruction과 KL Loss를 최소화

GAN
생성자와 판별자의 경쟁을 통해 실제 데이터 분포를 간접적으로 학습

25. 핵심 정리

1. Denoising AutoEncoder는 노이즈 이미지를 입력하고 깨끗한 원본을 Target으로 사용한다.
2. Sparse AutoEncoder는 적은 수의 Latent 뉴런만 강하게 사용하도록 제약한다.
3. 정상 데이터로 학습한 AutoEncoder의 Reconstruction Error를 이상 탐지에 활용할 수 있다.
4. 일반 AutoEncoder는 복원을 잘해도 임의의 Latent Vector에서 자연스러운 데이터를 생성한다고 보장할 수 없다.
5. VAE는 하나의 좌표 대신 μ와 σ로 Latent 분포를 표현한다.
6. Reparameterization Trick은 무작위성을 epsilon에 분리해 Encoder까지 Gradient가 흐르게 한다.
7. VAE Loss는 Reconstruction Loss와 KL Loss의 합이다.
8. KL Loss가 지나치게 0에 가까워지면 Latent를 사용하지 않는 Posterior Collapse가 생길 수 있다.
9. GAN의 Generator는 Noise로 가짜 이미지를 만들고, Discriminator는 진짜와 가짜를 구별한다.
10. BCEWithLogitsLoss를 사용하면 Discriminator 마지막 층에 Sigmoid를 따로 넣지 않는다.
11. Discriminator 학습에서 fake.detach()는 Generator 방향의 Gradient를 차단한다.
12. GAN Loss는 경쟁 관계 때문에 단조롭게 감소하지 않으며 생성 이미지와 다양성을 함께 평가해야 한다.

n x n 시계 격자에서 한 시계를 조작하면 자기 자신과 상하좌우 인접 시계가 시계 방향으로 90도 회전한다. 모든 시곗바늘을 12시 방향인 0으로 만들기 위한 최소 조작 횟수를 구한다.

시계 방향은 0, 1, 2, 3으로 표현하며 모든 변화는 mod 4 연산으로 처리할 수 있다.

핵심 관찰

1. 같은 시계는 최대 3번만 조작하면 된다

한 시계를 4번 조작하면 영향을 받는 모든 시계가 한 바퀴 돌아 원래 상태로 돌아온다.

따라서 최적해에서 각 시계의 조작 횟수는 0, 1, 2, 3번 중 하나만 고려하면 충분하다.

2. 첫 행을 제외한 행의 조작은 강제로 결정된다

r행의 시계를 조작할 수 있는 시점에는 r - 1행을 더 이상 바꿀 수 있는 방법이 거의 없다.

r - 1행의 c열 시계에 영향을 주는 아래 행의 조작은 오직 (r, c)의 조작뿐이다. 그러므로 (r - 1, c)를 0으로 만들기 위해 (r, c)를 몇 번 조작해야 하는지는 다음처럼 유일하게 결정된다.

count = (-board[r - 1][c]) % 4

첫 행의 조작 횟수만 정하면, 둘째 행부터 마지막 행까지의 조작 횟수는 모두 자동으로 정해진다.

풀이 과정

1. 첫 행의 각 칸을 0~3번 조작하는 모든 경우를 탐색한다.
2. 첫 행 조작을 격자에 적용한다.
3. r = 1부터 마지막 행까지 순회한다.
4. 바로 위 행의 각 시계를 0으로 만들기 위해 현재 행의 같은 열 시계를 필요한 횟수만큼 조작한다.
5. 마지막 행이 모두 0인 경우, 조작 횟수 최솟값을 갱신한다.

첫 행의 경우의 수는 최대 4^8 = 65,536개이므로 충분히 탐색할 수 있다.

Python 코드

from itertools import product


def solution(clockHands):
    n = len(clockHands)
    answer = float("inf")

    dr = [0, -1, 1, 0, 0]
    dc = [0, 0, 0, -1, 1]

    def rotate(board, row, col, count):
        """(row, col)을 count번 조작한 결과를 보드에 반영한다."""
        for direction in range(5):
            nr = row + dr[direction]
            nc = col + dc[direction]

            if 0 <= nr < n and 0 <= nc < n:
                board[nr][nc] = (board[nr][nc] + count) % 4

    # 첫 행의 모든 조작 조합을 시도한다.
    for first_row in product(range(4), repeat=n):
        board = [row[:] for row in clockHands]
        operation_count = sum(first_row)

        # 첫 행 조작을 적용한다.
        for col, count in enumerate(first_row):
            if count:
                rotate(board, 0, col, count)

        # 현재 행을 조작해 바로 위 행을 0으로 확정한다.
        for row in range(1, n):
            for col in range(n):
                count = (-board[row - 1][col]) % 4

                if count:
                    rotate(board, row, col, count)
                    operation_count += count

        # 마지막 행도 모두 0이면 유효한 해다.
        if all(clock == 0 for clock in board[-1]):
            answer = min(answer, operation_count)

    return answer

정당성 설명

첫 행의 각 칸은 0~3번만 조작하면 되므로, 알고리즘은 가능한 첫 행 조작 조합을 모두 탐색한다.

임의의 첫 행 조작 조합을 고정하자. r - 1행 c열에 영향을 주는 아직 선택되지 않은 조작은 (r, c)뿐이다. 따라서 이 시계를 0으로 만들려면 (r, c)를 (-board[r - 1][c]) mod 4번 조작해야 하며, 다른 횟수로는 r - 1행을 0으로 확정할 수 없다.

그러므로 고정된 첫 행 조합에 대해 알고리즘이 계산한 이후 행의 조작은 유일하며, 그 조합에서 가능한 해가 존재한다면 반드시 찾아낸다. 모든 첫 행 조합을 탐색하므로 모든 가능한 해를 검토하고, 그중 최소 조작 횟수를 반환한다.

복잡도 분석

첫 행의 조작 조합은 4^n개다. 각 조합마다 n^2개의 칸을 처리하고, 한 번의 조작 반영은 최대 5개 칸에만 영향을 준다.

• 시간 복잡도: O(4^n * n^2)
• 공간 복잡도: O(n^2)

n <= 8이므로 최대 65,536개의 첫 행 조합만 확인하면 된다.

주의할 점

• 조작 횟수는 board[row - 1][col]이 아니라 그 음수의 mod 4 값이다. 현재 값이 3이면 한 번 조작해 0으로 만들어야 한다.
• 현재 행을 처리할 때는 반드시 바로 위 행을 0으로 확정해야 한다.
• 마지막 행은 아래에서 보정할 행이 없으므로, 모든 값이 0인지 직접 검사해야 한다.
• 보드마다 독립적으로 시뮬레이션해야 하므로, 첫 행 조합을 바꿀 때 원본 배열을 깊은 복사해야 한다.

마무리

행 단위로 보면 첫 행만 선택이고 나머지는 강제 전이라는 구조가 보인다. 첫 행의 작은 경우의 수를 완전탐색하고, 이후 행을 위에서 아래로 확정해 나가면 최소 조작 횟수를 구할 수 있다.

이번에는 Transformer의 Multi-Head Attention, 위치 정보, Encoder와 Decoder 구조를 살펴본다. 이어서 MNIST 이미지에 합성곱 AutoEncoder를 적용해 원본 이미지를 복원하는 과정을 정리한다.

1. Padding Mask

배치 학습에서는 길이가 서로 다른 문장을 하나의 Tensor로 묶기 위해 짧은 문장 뒤에 <PAD> Token을 추가한다.

커피 한잔 어때
안녕 <PAD> <PAD>

<PAD>는 길이를 맞추기 위한 자리일 뿐 실제 의미가 없다. Attention이 이 위치를 참고하면 의미 없는 정보가 문맥에 섞이므로 Padding Mask로 제외해야 한다.

import torch

sample_scores = torch.tensor([[2.0, 1.0, 0.5, 3.0]])
padding_mask = torch.tensor([1, 1, 1, 0])

masked_scores = sample_scores.masked_fill(
    padding_mask == 0,
    float("-inf"),
)

weights_before = torch.softmax(sample_scores, dim=-1)
weights_after = torch.softmax(masked_scores, dim=-1)

print("Mask 전:", weights_before)
print("Mask 후:", weights_after)

실행 결과:

Mask 전: tensor([[0.2321, 0.0854, 0.0518, 0.6308]])
Mask 후: tensor([[0.6285, 0.2312, 0.1402, 0.0000]])

Mask가 적용된 Score는 -inf가 된다. softmax(-inf)의 결과는 0이므로 해당 위치의 Value가 최종 출력에 반영되지 않는다.

Score:  [2.0, 1.0, 0.5, -inf]
                         ↓ Softmax
Weight: [0.6285, 0.2312, 0.1402, 0.0000]

2. Transformer

Transformer는 2017년 논문 Attention Is All You Need에서 제안된 신경망 구조다.

RNN 계열 모델은 이전 시점의 계산이 끝나야 다음 시점을 처리할 수 있다. Transformer는 Self-Attention으로 모든 Token의 관계를 직접 계산하므로 병렬 연산에 유리하며, 멀리 떨어진 Token 사이의 관계도 직접 연결할 수 있다.

RNN
x1 → h1 → x2 → h2 → x3 → h3

Transformer
x1 ─┬─ x1, x2, x3과 관계 계산
x2 ─┼─ x1, x2, x3과 관계 계산
x3 ─┴─ x1, x2, x3과 관계 계산

대표적인 Transformer 기반 모델에는 BERT, GPT, T5 등이 있다.

3. Multi-Head Attention

Self-Attention을 Q, K, V 한 세트로만 계산하면 하나의 표현 공간에서 관계를 학습한다. Multi-Head Attention은 여러 Head가 서로 다른 Q, K, V 투영을 학습해 다양한 관계를 동시에 포착하도록 만든다.

학습 결과에 따라 어떤 Head는 가까운 Token 관계에, 다른 Head는 문법적 관계나 의미적 관계에 반응할 수 있다. 각 Head의 역할을 사람이 미리 지정하는 것은 아니다.

d_model = 512
num_heads = 8
head_dim = 512 / 8 = 64

d_model은 num_heads로 나누어떨어져야 한다.

PyTorch로 확인하기

import torch
import torch.nn as nn

embed_dim = 8
num_heads = 2
head_dim = embed_dim // num_heads

print("전체 Embedding 차원:", embed_dim)
print("Head 개수:", num_heads)
print("Head 하나의 차원:", head_dim)

# batch=1, token=3, embed_dim=8
x = torch.randn(1, 3, embed_dim)

mha = nn.MultiheadAttention(
    embed_dim=embed_dim,
    num_heads=num_heads,
    batch_first=True,
)

attn_output, attn_weights = mha(
    query=x,
    key=x,
    value=x,
    need_weights=True,
)

print("입력 Shape:", x.shape)
print("출력 Shape:", attn_output.shape)
print("Attention Weight Shape:", attn_weights.shape)

실행 결과:

전체 Embedding 차원: 8
Head 개수: 2
Head 하나의 차원: 4
입력 Shape: torch.Size([1, 3, 8])
출력 Shape: torch.Size([1, 3, 8])
Attention Weight Shape: torch.Size([1, 3, 3])

입력과 출력의 마지막 차원은 모두 8이다. 내부에서는 두 Head가 각각 4차원 공간에서 Attention을 계산한 뒤, 결과를 합쳐 다시 8차원 표현을 만든다.

Head별 Attention Weight 확인하기

nn.MultiheadAttention은 기본적으로 여러 Head의 Weight를 평균해 반환한다.

[batch, target_length, source_length]

average_attn_weights=False를 지정하면 Head별 Weight를 따로 확인할 수 있다.

attn_output, head_weights = mha(
    query=x,
    key=x,
    value=x,
    need_weights=True,
    average_attn_weights=False,
)

print(head_weights.shape)
print("Head 0:\n", head_weights[0, 0])
print("Head 1:\n", head_weights[0, 1])

실행 결과:

torch.Size([1, 2, 3, 3])

차원의 의미는 다음과 같다.

[batch, num_heads, target_length, source_length]
[1,     2,         3,             3]

Self-Attention에서는 Query와 Key가 같은 시퀀스에서 만들어지므로 보통 target_length와 source_length가 같다.

Attention Weight 시각화

• 세로축은 정보를 찾는 Query Token이다.
• 가로축은 참고 대상인 Key Token이다.
• 색이 밝을수록 해당 Key를 더 많이 참고한다.
• Weight는 학습 과정에서 계속 달라지므로 한 번 출력한 값만으로 Head의 역할을 단정해서는 안 된다.

4. Transformer Encoder 전체 흐름

Transformer Encoder의 한 Block은 다음 순서로 동작한다.

문장
  ↓
Tokenization
  ↓
Token ID
  ↓
Token Embedding + Positional Encoding
  ↓
Multi-Head Self-Attention
  ↓
Residual Connection + LayerNorm
  ↓
Feed Forward Network
  ↓
Residual Connection + LayerNorm
  ↓
Encoder 출력

각 단계는 입력과 출력의 Shape을 유지하면서 Token 표현을 점차 문맥에 맞게 바꾼다.

5. Token Embedding

먼저 Token을 정수 ID로 바꾼다.

vocab = {
    "<PAD>": 0,
    "커피": 1,
    "한잔": 2,
    "어때": 3,
}

tokens = ["커피", "한잔", "어때"]
token_ids = torch.tensor([[1, 2, 3]])

print(token_ids)
print(token_ids.shape)

tensor([[1, 2, 3]])
torch.Size([1, 3])

nn.Embedding은 각 Token ID를 d_model차원의 실수 벡터로 바꾼다.

vocab_size = len(vocab)
d_model = 8

embedding = nn.Embedding(
    vocab_size,
    d_model,
    padding_idx=0,
)

X = embedding(token_ids)
print(X.shape)

torch.Size([1, 3, 8])

Shape은 다음처럼 변한다.

[batch, sequence_length]
[1, 3]
      ↓ Embedding
[batch, sequence_length, d_model]
[1, 3, 8]

6. Positional Encoding

Self-Attention은 모든 Token의 관계를 한 번에 계산하므로 RNN처럼 계산 순서에서 위치 정보를 얻지 못한다. 따라서 입력에 Token의 위치 정보를 별도로 넣어야 한다.

원래 Transformer 논문은 고정된 Sin과 Cos 함수를 이용한 Positional Encoding을 사용한다.

최종 입력 = Token Embedding + Positional Encoding

예를 들어 다음처럼 같은 차원의 두 벡터를 더한다.

Token Embedding:      [0.20, -0.40, 0.70, ...]
Position 0 Encoding:  [0.00,  1.00, 0.00, ...]
최종 입력:            [0.20,  0.60, 0.70, ...]

모델이 각 숫자를 Token 정보와 위치 정보로 따로 읽는 것은 아니다. 같은 d_model 공간에서 두 벡터를 더해 의미와 위치가 함께 반영된 표현을 만든다.

Sin/Cos Positional Encoding 구현

import math

class PositionalEncoding(nn.Module):
    def __init__(self, d_model, max_len=100):
        super().__init__()

        position = torch.arange(
            max_len,
            dtype=torch.float32,
        ).unsqueeze(1)

        div_term = torch.exp(
            torch.arange(
                0,
                d_model,
                2,
                dtype=torch.float32,
            )
            * (-math.log(10000.0) / d_model)
        )

        pe = torch.zeros(max_len, d_model)
        pe[:, 0::2] = torch.sin(position * div_term)
        pe[:, 1::2] = torch.cos(position * div_term)

        self.register_buffer("pe", pe.unsqueeze(0))

    def forward(self, x):
        seq_len = x.size(1)
        return x + self.pe[:, :seq_len]

pe는 학습으로 바뀌는 Parameter가 아니라 공식으로 계산한 고정값이다. register_buffer()로 등록하면 다음 특성을 얻는다.

• model.to(device)를 호출할 때 모델과 함께 GPU나 MPS로 이동한다.
• state_dict()에 포함되어 저장된다.
• Optimizer의 학습 대상 Parameter에는 포함되지 않는다.

pos_encoding = PositionalEncoding(d_model)
X_pos = pos_encoding(X)

print("Embedding:", X.shape)
print("Position 추가:", X_pos.shape)

Embedding: torch.Size([1, 3, 8])
Position 추가: torch.Size([1, 3, 8])

위치 정보를 더해도 Shape은 변하지 않는다.

RoPE

RoPE(Rotary Positional Embedding)는 위치 벡터를 입력 Embedding에 단순히 더하지 않는다. Attention에 사용하는 Q와 K를 Token 위치에 따라 회전시켜 상대적인 위치 관계가 점수에 반영되도록 만든다.

Sin/Cos Positional Encoding
Embedding에 위치 벡터를 더한다.

RoPE
Q와 K의 Attention 계산에 위치 관계를 반영한다.

7. Multi-Head Self-Attention 적용

위치 정보가 더해진 X_pos를 Query, Key, Value에 모두 사용하면 Self-Attention이 된다.

num_heads = 2

mha = nn.MultiheadAttention(
    embed_dim=d_model,
    num_heads=num_heads,
    batch_first=True,
)

attn_out, attn_weights = mha(
    query=X_pos,
    key=X_pos,
    value=X_pos,
    need_weights=True,
    average_attn_weights=False,
)

print("입력:", X_pos.shape)
print("출력:", attn_out.shape)
print("Head별 Weight:", attn_weights.shape)

실행 결과:

입력: torch.Size([1, 3, 8])
출력: torch.Size([1, 3, 8])
Head별 Weight: torch.Size([1, 2, 3, 3])

Attention은 Token 사이의 정보를 섞지만 [batch, sequence_length, d_model] Shape은 유지한다.

8. Residual Connection

Residual Connection은 Attention 결과만 다음 단계로 보내지 않고 원래 입력을 다시 더하는 구조다.

y = x + F(x)

• x는 Attention에 들어가기 전의 원래 정보다.
• F(x)는 Attention이 계산한 변화 정보다.
• x + F(x)는 원래 정보를 유지하면서 새로운 문맥 정보를 추가한 결과다.

residual = X_pos + attn_out
print(residual.shape)

torch.Size([1, 3, 8])

원소별 덧셈을 수행하므로 x와 F(x)의 Shape이 같아야 한다. Residual Connection은 깊은 신경망에서 정보와 Gradient가 전달될 경로를 제공해 학습을 돕는다.

9. Layer Normalization

nn.LayerNorm(d_model)은 각 Token 벡터의 Feature 차원을 기준으로 값을 정규화한다.

입력 Shape이 [batch, sequence_length, d_model]이면 각 Token의 (d_model,) 벡터를 독립적으로 정규화한다.

norm1 = nn.LayerNorm(d_model)
x1 = norm1(residual)

print("평균:", x1[0, 0].mean())
print("분산:", x1[0, 0].var(unbiased=False))

실행 결과:

평균: tensor(0.)
분산: tensor(1.000)

LayerNorm은 각 샘플과 Token 안에서 정규화하므로 BatchNorm보다 배치 크기의 영향을 덜 받는다. 가변 길이 시퀀스를 다루는 Transformer에 잘 맞는다.

10. Feed Forward Network

Attention과 FFN은 서로 다른 역할을 한다.

ffn = nn.Sequential(
    nn.Linear(d_model, 32),
    nn.ReLU(),
    nn.Linear(32, d_model),
)

ffn_out = ffn(x1)

print("입력:", x1.shape)
print("출력:", ffn_out.shape)

입력: torch.Size([1, 3, 8])
출력: torch.Size([1, 3, 8])

동일한 FFN이 모든 Token 위치에 적용되지만, 각 Token은 다른 Token과 섞이지 않고 자신의 Feature만 변환한다.

Encoder Block에는 두 개의 Sub-layer가 있으며 각각 Residual Connection과 LayerNorm이 붙는다.

Self-Attention
  ↓
Residual + LayerNorm
  ↓
FFN
  ↓
Residual + LayerNorm

norm2 = nn.LayerNorm(d_model)
encoder_output = norm2(x1 + ffn_out)

print(encoder_output.shape)

torch.Size([1, 3, 8])

11. Encoder에서 Padding Mask 사용하기

PyTorch의 nn.MultiheadAttention은 key_padding_mask에서 다음 규칙을 사용한다.

False: 참고할 수 있는 실제 Token
True:  참고 대상에서 제외할 PAD Token

padded_ids = torch.tensor([
    [1, 2, 3],
    [1, 3, 0],
])

padding_mask = padded_ids.eq(0)

print(padded_ids)
print(padding_mask)

실행 결과:

tensor([[1, 2, 3],
        [1, 3, 0]])

tensor([[False, False, False],
        [False, False,  True]])

Mask를 Multi-Head Attention에 전달한다.

padded_x = pos_encoding(embedding(padded_ids))

masked_out, masked_weights = mha(
    padded_x,
    padded_x,
    padded_x,
    key_padding_mask=padding_mask,
    need_weights=True,
)

print(masked_weights[1])

두 번째 문장의 Attention Weight는 다음과 같다.

tensor([[0.392, 0.608, 0.000],
        [0.372, 0.628, 0.000],
        [0.414, 0.586, 0.000]])

마지막 열은 <PAD>가 있는 Key 위치다. 모든 Query에서 가중치가 0이므로 참고 대상에서 제외된 것을 확인할 수 있다.

key_padding_mask는 PAD를 Key와 Value의 참고 대상에서 제외한다. PAD 위치의 Query 출력까지 자동으로 삭제하는 것은 아니므로, 이후 Loss를 계산할 때도 PAD 위치를 제외해야 한다.

12. Causal Mask

Padding Mask와 Causal Mask는 목적이 다르다.

문장 나는 커피를 마신다를 학습할 때 각 위치가 볼 수 있는 범위는 다음과 같다.

Query  | 나는 | 커피를 | 마신다
-------|------|--------|-------
나는   |  O   |   X    |   X
커피를 |  O   |   O    |   X
마신다 |  O   |   O    |   O

추론할 때는 Token을 하나씩 생성하지만, 학습할 때는 정답 문장 전체를 알고 있다. Mask 없이 병렬 계산하면 앞쪽 위치가 미래의 정답 Token을 미리 볼 수 있으므로 Causal Mask가 필요하다.

seq_len = 4

causal_mask = torch.triu(
    torch.ones(seq_len, seq_len, dtype=torch.bool),
    diagonal=1,
)

print(causal_mask)

tensor([[False,  True,  True,  True],
        [False, False,  True,  True],
        [False, False, False,  True],
        [False, False, False, False]])

True인 오른쪽 위 영역이 미래 Token에 해당한다.

실제 Attention Weight는 다음처럼 미래 위치가 0이 된다.

tensor([[1.000, 0.000, 0.000, 0.000],
        [0.640, 0.360, 0.000, 0.000],
        [0.315, 0.311, 0.374, 0.000],
        [0.166, 0.159, 0.303, 0.372]])

13. Transformer Encoder Block 구현

지금까지 확인한 구성 요소를 하나의 Encoder Block으로 묶는다.

class SimpleTransformerEncoderBlock(nn.Module):
    def __init__(
        self,
        d_model=8,
        num_heads=2,
        d_ff=32,
        dropout=0.1,
    ):
        super().__init__()

        self.self_attention = nn.MultiheadAttention(
            d_model,
            num_heads,
            dropout=dropout,
            batch_first=True,
        )

        self.norm1 = nn.LayerNorm(d_model)
        self.norm2 = nn.LayerNorm(d_model)

        self.ffn = nn.Sequential(
            nn.Linear(d_model, d_ff),
            nn.ReLU(),
            nn.Dropout(dropout),
            nn.Linear(d_ff, d_model),
        )

        self.dropout1 = nn.Dropout(dropout)
        self.dropout2 = nn.Dropout(dropout)

    def forward(self, x, padding_mask=None):
        attn_out, weights = self.self_attention(
            x,
            x,
            x,
            key_padding_mask=padding_mask,
            need_weights=True,
        )

        x = self.norm1(
            x + self.dropout1(attn_out)
        )

        ffn_out = self.ffn(x)
        x = self.norm2(
            x + self.dropout2(ffn_out)
        )

        return x, weights

block = SimpleTransformerEncoderBlock()
out, weights = block(X_pos)

print("입력:", X_pos.shape)
print("출력:", out.shape)
print("Weight:", weights.shape)

실행 결과:

입력: torch.Size([1, 3, 8])
출력: torch.Size([1, 3, 8])
Weight: torch.Size([1, 3, 3])

이 구현은 Attention 뒤에 정규화하는 Post-Norm 구조다.

x → Attention → Add → LayerNorm → FFN → Add → LayerNorm

14. Transformer Decoder

Encoder는 입력 Token 전체를 Self-Attention으로 처리해 문맥이 반영된 표현을 만든다. Decoder는 이 정보를 이용해 출력 문장을 생성한다.

Decoder의 주요 구성은 다음과 같다.

1. Masked Self-Attention: 미래 출력 Token을 미리 보지 못하게 한다.
2. Cross-Attention: Encoder 출력 중 현재 생성에 필요한 부분을 참고한다.
3. FFN: 각 Decoder Token 표현을 비선형 변환한다.

입력 문장
   ↓
Encoder
   ↓
Encoder Memory ──────────────┐
                            │ K, V
출력 입력                    │
   ↓                        │
Masked Self-Attention        │
   ↓                        │
Cross-Attention ◀────────────┘
   ↓
FFN
   ↓
Decoder 출력

Cross-Attention

Cross-Attention에서는 Q와 K, V의 출처가 다르다.

Q = Decoder의 현재 표현
K = Encoder 출력
V = Encoder 출력

cross_attention = nn.MultiheadAttention(
    embed_dim=8,
    num_heads=2,
    batch_first=True,
)

encoder_memory = torch.randn(1, 5, 8)
decoder_state = torch.randn(1, 3, 8)

cross_out, cross_weights = cross_attention(
    query=decoder_state,
    key=encoder_memory,
    value=encoder_memory,
)

print("Q(Decoder):", decoder_state.shape)
print("K/V(Encoder):", encoder_memory.shape)
print("출력:", cross_out.shape)
print("Weight:", cross_weights.shape)

실행 결과:

Q(Decoder): torch.Size([1, 3, 8])
K/V(Encoder): torch.Size([1, 5, 8])
출력: torch.Size([1, 3, 8])
Weight: torch.Size([1, 3, 5])

[1, 3, 5]는 Decoder의 Token 3개가 Encoder Token 5개를 각각 얼마나 참고했는지 나타낸다.

GPT 같은 Decoder-only 모델은 별도의 Encoder와 Cross-Attention 없이 Causal Self-Attention을 중심으로 구성한다.

15. AutoEncoder

AutoEncoder는 입력을 작은 잠재 표현으로 바꾸고, 그 표현으로 원본을 다시 복원하는 신경망이다.

입력 x
  ↓
Encoder
  ↓
Latent Representation z
  ↓
Decoder
  ↓
복원 결과 x_hat

입력과 복원 결과의 차이가 작아지도록 학습한다.

입력:  숫자 7 이미지
정답:  같은 숫자 7 이미지
출력:  AutoEncoder가 복원한 숫자 7 이미지

비지도 학습과 자기지도 학습

이미지 분류 모델은 강아지, 고양이와 같은 사람이 만든 클래스 라벨을 필요로 한다. AutoEncoder는 별도의 클래스 라벨 없이 입력 데이터 자체를 정답으로 사용한다.

for x, _ in loader:
    _, x_hat = model(x)
    loss = criterion(x_hat, x)

DataLoader가 제공한 label을 _로 받아 사용하지 않고, 입력 x가 그대로 학습 Target이 된다. 전통적으로 비지도 학습으로 분류하며, 입력에서 정답을 자동으로 만드는 관점에서는 자기지도 학습의 한 형태로도 볼 수 있다.

16. Encoder, Latent Representation, Decoder

Encoder

Encoder는 입력에서 복원에 필요한 특징을 추출한다.

입력 이미지
[1, 28, 28]
      ↓ Encoder
잠재 표현
[32, 7, 7]

Latent Representation

Latent Representation은 Encoder가 만든 내부 표현이다. 원본을 그대로 복사하는 대신 Decoder가 원본을 다시 만들 때 필요한 특징을 담는다.

이번 합성곱 모델은 공간 크기를 28 × 28에서 7 × 7로 줄이는 대신 Channel을 1에서 32로 늘린다.

원본 원소 수:  1 × 28 × 28 = 784
Latent 원소 수: 32 × 7 × 7 = 1,568

따라서 이 예제는 가로와 세로 크기는 압축하지만 전체 원소 수가 더 작은 엄격한 차원 축소 구조는 아니다. 더 강한 Bottleneck이 필요하다면 Channel 수를 줄이거나 Linear 층으로 더 작은 Latent Vector를 만들 수 있다.

Decoder

Decoder는 Latent Representation만 보고 원본과 비슷한 데이터를 복원한다.

[32, 7, 7]
      ↓ Decoder
[1, 28, 28]

17. Reconstruction Loss

AutoEncoder는 원본과 복원본의 차이를 Reconstruction Loss로 계산한다.

예를 들어 한 픽셀의 원본값이 0.8, 복원값이 0.6이라면 제곱 오차는 다음과 같다.

(0.8 - 0.6)² = 0.04

전체 픽셀의 평균 제곱 오차를 줄이도록 Encoder와 Decoder의 Parameter를 함께 업데이트한다.

criterion = nn.MSELoss()

18. MNIST 데이터 준비

import torch
import torch.nn as nn
import torch.optim as optim
import matplotlib.pyplot as plt

from torch.utils.data import DataLoader
from torchvision import datasets, transforms
from torchvision.utils import make_grid

torch.manual_seed(2026)

ToTensor()는 MNIST 이미지를 Tensor로 바꾸고 픽셀 범위를 [0, 1]로 변환한다.

transform = transforms.ToTensor()

train_dataset = datasets.MNIST(
    root="./data",
    train=True,
    download=True,
    transform=transform,
)

test_dataset = datasets.MNIST(
    root="./data",
    train=False,
    download=True,
    transform=transform,
)

train_loader = DataLoader(
    train_dataset,
    batch_size=128,
    shuffle=True,
    num_workers=4,
)

test_loader = DataLoader(
    test_dataset,
    batch_size=128,
    shuffle=False,
    num_workers=4,
)

images, labels = next(iter(train_loader))

print("images:", images.shape)
print("labels:", labels.shape)

실행 결과:

images: torch.Size([128, 1, 28, 28])
labels: torch.Size([128])

images의 차원은 [batch, channel, height, width] 순서다.

19. 합성곱 AutoEncoder 구현

class ConvAutoencoder(nn.Module):
    def __init__(self):
        super().__init__()

        self.encoder = nn.Sequential(
            # [B, 1, 28, 28] → [B, 16, 14, 14]
            nn.Conv2d(
                1,
                16,
                kernel_size=3,
                stride=2,
                padding=1,
            ),
            nn.ReLU(),

            # [B, 16, 14, 14] → [B, 32, 7, 7]
            nn.Conv2d(
                16,
                32,
                kernel_size=3,
                stride=2,
                padding=1,
            ),
            nn.ReLU(),
        )

        self.decoder = nn.Sequential(
            # [B, 32, 7, 7] → [B, 16, 14, 14]
            nn.ConvTranspose2d(
                32,
                16,
                kernel_size=3,
                stride=2,
                padding=1,
                output_padding=1,
            ),
            nn.ReLU(),

            # [B, 16, 14, 14] → [B, 1, 28, 28]
            nn.ConvTranspose2d(
                16,
                1,
                kernel_size=3,
                stride=2,
                padding=1,
                output_padding=1,
            ),
            nn.Sigmoid(),
        )

    def forward(self, x):
        z = self.encoder(x)
        x_hat = self.decoder(z)
        return z, x_hat

Shape 변화

stride=2인 Conv2d는 공간 크기를 절반으로 줄인다. ConvTranspose2d는 반대로 공간 크기를 키운다.

마지막 Sigmoid는 복원 픽셀값을 0과 1 사이로 만든다. 입력도 ToTensor()로 [0, 1] 범위이므로 출력 범위가 잘 맞는다.

20. AutoEncoder 학습

model = ConvAutoencoder().to(DEVICE)

criterion = nn.MSELoss()
optimizer = optim.AdamW(
    model.parameters(),
    lr=1e-3,
)

학습 함수는 입력 x와 복원 결과 x_hat의 차이를 계산한다.

def train_autoencoder(
    model,
    loader,
    criterion,
    optimizer,
    device,
    epochs=5,
):
    history = []

    for epoch in range(epochs):
        model.train()
        running_loss = 0.0

        for x, _ in loader:
            x = x.to(device)

            _, x_hat = model(x)
            loss = criterion(x_hat, x)

            optimizer.zero_grad()
            loss.backward()
            optimizer.step()

            running_loss += loss.item() * x.size(0)

        epoch_loss = running_loss / len(loader.dataset)
        history.append(epoch_loss)

        print(
            f"Epoch {epoch + 1:02d}/{epochs} | "
            f"Loss: {epoch_loss:.6f}"
        )

    return history

노트북의 실행 결과는 다음과 같다.

Epoch 01/5 | Loss: 0.031372
Epoch 02/5 | Loss: 0.001786
Epoch 03/5 | Loss: 0.001211
Epoch 04/5 | Loss: 0.000970
Epoch 05/5 | Loss: 0.000810

Reconstruction Loss가 지속해서 감소하므로 원본을 더 비슷하게 복원하도록 학습되고 있음을 알 수 있다.

21. 복원 결과 확인

@torch.no_grad()
def show_reconstructions(model, loader, device, n=8):
    model.eval()

    x, _ = next(iter(loader))
    x = x[:n].to(device)

    _, x_hat = model(x)

    comparison = torch.cat(
        [x.cpu(), x_hat.cpu()],
        dim=0,
    )

    grid = make_grid(
        comparison,
        nrow=n,
        padding=2,
    )

    plt.figure(figsize=(14, 4))
    plt.imshow(
        grid.permute(1, 2, 0).squeeze(),
        cmap="gray",
    )
    plt.axis("off")
    plt.title("Top: Original / Bottom: Reconstruction")
    plt.show()

• 위쪽 행은 원본 이미지다.
• 아래쪽 행은 AutoEncoder가 복원한 이미지다.
• 전체적인 숫자 모양은 잘 복원하지만 가장자리 일부가 부드럽거나 흐리게 표현될 수 있다.

학습 Loss가 낮아도 새로운 데이터의 복원 품질이 항상 좋다는 뜻은 아니다. Test 데이터의 Reconstruction Loss와 복원 이미지를 함께 확인해야 한다.

22. AutoEncoder 활용 분야

일반 AutoEncoder는 입력을 복원하도록 학습하지만, 임의의 Latent Vector에서 항상 자연스러운 데이터를 생성할 수 있는 것은 아니다. 생성 목적이라면 Latent Space의 분포를 학습하는 VAE 같은 구조가 더 적합할 수 있다.

23. 핵심 정리

1. Padding Mask는 <PAD>의 Attention Weight를 0으로 만들어 의미 없는 위치를 참고하지 않게 한다.
2. Multi-Head Attention은 여러 표현 공간에서 Token 관계를 동시에 학습한다.
3. average_attn_weights=False를 사용하면 Head별 Attention Weight를 확인할 수 있다.
4. Self-Attention은 순서를 모르므로 Token Embedding에 Positional Encoding을 더한다.
5. Residual Connection은 원래 정보에 Attention이나 FFN의 변화 정보를 더한다.
6. LayerNorm은 각 Token의 Feature 차원을 기준으로 정규화한다.
7. FFN은 각 Token 위치에 독립적으로 같은 비선형 변환을 적용한다.
8. Causal Mask는 미래 Token을 가려 다음 Token 예측의 정답 누출을 방지한다.
9. Decoder의 Cross-Attention은 Decoder 표현을 Q로, Encoder 출력을 K와 V로 사용한다.
10. AutoEncoder는 입력 자체를 Target으로 사용해 Encoder와 Decoder를 함께 학습한다.
11. MNIST 합성곱 AutoEncoder는 28 × 28 → 7 × 7 → 28 × 28로 공간 크기를 줄였다가 복원한다.
12. AutoEncoder는 Reconstruction Loss뿐 아니라 실제 복원 결과도 함께 평가해야 한다.

이번에는 기본 RNN의 한계를 보완한 LSTM과 GRU를 살펴보고, 입력 시퀀스를 출력 시퀀스로 바꾸는 Seq2Seq, 필요한 정보에 직접 집중하는 Attention까지 이어서 정리한다.

1. LSTM

LSTM(Long Short-Term Memory)은 기본 RNN의 장기 의존성 문제를 완화하기 위해 만든 순환 신경망이다.

기본 RNN은 주로 하나의 hidden state로 과거 정보를 전달하지만, LSTM은 다음 두 상태를 함께 사용한다.

LSTM은 세 개의 게이트로 정보의 흐름을 조절한다.

LSTM에 들어오는 값

시점 t의 LSTM은 다음 정보를 받는다.

x_t       : 현재 시점의 입력
h_(t-1)   : 이전 시점의 hidden state
C_(t-1)   : 이전 시점의 cell state

게이트는 사람이 직접 정한 규칙이 아니다. 현재 입력과 이전 상태를 사용해 계산한 뒤, 학습 과정에서 가중치와 편향을 수정하면서 필요한 정보의 비율을 스스로 결정한다.

2. LSTM의 게이트 동작

Forget Gate

Forget Gate는 이전 cell state에서 각 정보를 얼마나 유지할지 결정한다.

f_t = sigmoid(W_f · [h_(t-1), x_t] + b_f)

sigmoid의 결과는 0과 1 사이의 값이다.

• 0에 가까우면 해당 정보를 거의 전달하지 않는다.
• 1에 가까우면 해당 정보를 대부분 유지한다.
• 중간값이면 그 비율만큼 정보를 전달한다.

따라서 Forget Gate는 단순히 기억과 삭제 중 하나만 고르는 스위치가 아니라, 정보를 얼마나 남길지 조절하는 벡터다.

Input Gate

Input Gate는 현재 입력에서 어떤 정보를 새 기억으로 저장할지 결정한다.

i_t       = sigmoid(W_i · [h_(t-1), x_t] + b_i)
C_tilde_t = tanh(W_c · [h_(t-1), x_t] + b_c)

• i_t는 새 정보를 얼마나 반영할지 나타낸다.
• C_tilde_t는 현재 입력으로 만든 새로운 기억의 후보값이다.

이전 기억과 새 기억을 합쳐 현재 cell state를 만든다.

C_t = f_t ⊙ C_(t-1) + i_t ⊙ C_tilde_t

여기서 ⊙는 같은 위치의 원소끼리 곱하는 연산을 의미한다.

Output Gate

Output Gate는 갱신된 cell state 중 현재 hidden state로 전달할 정보를 선택한다.

o_t = sigmoid(W_o · [h_(t-1), x_t] + b_o)
h_t = o_t ⊙ tanh(C_t)

전체 흐름은 다음과 같다.

이전 기억 C_(t-1)
        │
        ├── Forget Gate ── 과거 기억 중 남길 부분
        │
현재 입력 x_t
        ├── Input Gate  ── 새로 저장할 부분
        │
        ▼
새 Cell State C_t
        │
        └── Output Gate ── 현재 Hidden State h_t

장기 기억과 단기 기억의 기준

LSTM 내부에 6시간이 지나면 장기 기억과 같은 고정 규칙은 없다. 정보가 얼마나 오래되었는지보다 현재 예측에 얼마나 도움이 되는지가 중요하다.

예를 들어 전력 사용량이 매일 저녁 비슷한 시간에 증가한다면 오래전 시점에서 만들어진 정보도 유용할 수 있다. 반대로 1시간 전 정보라도 잡음에 가깝다면 게이트가 영향력을 줄일 수 있다.

3. 가정 전력 사용량 예측

LSTM을 이용해 한 가정의 다음 1시간 전력 사용량을 예측한다.

• 원본 데이터는 약 4년 동안 1분 간격으로 측정한 전력 사용량이다.
• Global_active_power 한 개의 Feature를 사용한다.
• 1분 단위 데이터를 1시간 단위 평균으로 변환한다.
• 직전 24시간을 입력으로 사용해 다음 1시간을 예측한다.

입력 X: t-24, t-23, ..., t-1 시점의 전력 사용량
정답 y: t 시점의 전력 사용량

데이터 불러오기

import pandas as pd

DATA_PATH = "./data/household_power_consumption.txt"

df = pd.read_csv(
    DATA_PATH,
    sep=";",
    na_values="?",
    low_memory=False,
)

print(df.shape)

실행 결과:

(2075259, 9)

주요 컬럼은 다음과 같다.

날짜와 시간 전처리

Date와 Time을 하나의 datetime으로 합친 뒤 시간 순서로 정렬한다.

df["datetime"] = pd.to_datetime(
    df["Date"] + " " + df["Time"],
    format="%d/%m/%Y %H:%M:%S",
    errors="coerce",
)

df = df.dropna(subset=["datetime"])
df = df.sort_values("datetime")
df = df.set_index("datetime")

예측에 사용할 Global_active_power만 남기고 숫자로 변환한다.

power = df[["Global_active_power"]].copy()
power["Global_active_power"] = pd.to_numeric(
    power["Global_active_power"],
    errors="coerce",
)
power = power.dropna()

print(power.shape)

(2049280, 1)

1시간 단위로 변환하기

hourly = power.resample("1h").mean()

print(
    "시간 단위 결측 개수:",
    hourly["Global_active_power"].isna().sum(),
)

hourly = hourly.ffill().dropna()
print("hourly shape:", hourly.shape)

실행 결과:

시간 단위 결측 개수: 421
hourly shape: (34589, 1)

resample("1h").mean()은 같은 시간에 속한 1분 단위 관측값의 평균을 계산한다. 시간 단위로 비어 있는 구간은 ffill()을 사용해 직전 관측값으로 채운다.

4. 전력 데이터를 시퀀스로 만들기

데이터는 시간 순서를 유지한 채 Train 70%, Validation 15%, Test 15%로 나눈다. StandardScaler는 미래 정보가 학습 데이터에 섞이지 않도록 Train 구간에만 fit()한다.

import numpy as np
from sklearn.preprocessing import StandardScaler

values = hourly["Global_active_power"].values.reshape(-1, 1)

n = len(values)
train_end = int(n * 0.70)
val_end = int(n * 0.85)

scaler = StandardScaler()
scaler.fit(values[:train_end])
scaled_values = scaler.transform(values).astype(np.float32)

최근 24시간과 다음 1시간 묶기

SEQ_LENGTH = 24

def make_sequence(data, seq_length):
    X = []
    y = []
    target_indices = []

    for i in range(seq_length, len(data)):
        X.append(data[i - seq_length:i])
        y.append(data[i])
        target_indices.append(i)

    return (
        np.array(X, dtype=np.float32),
        np.array(y, dtype=np.float32),
        np.array(target_indices),
    )

X, y, indices = make_sequence(scaled_values, SEQ_LENGTH)

print(X.shape)
print(y.shape)

실행 결과:

(34565, 24, 1)
(34565, 1)

각 차원의 의미는 다음과 같다.

X.shape = (샘플 수, 시퀀스 길이, Feature 수)
        = (34565, 24, 1)

첫 번째 샘플은 2006-12-16 17:00부터 2006-12-17 16:00까지의 24시간을 입력으로 사용하고, 2006-12-17 17:00의 값을 정답으로 사용한다.

시퀀스의 정답 시점을 기준으로 데이터를 나눈 결과는 다음과 같다.

5. PyTorch LSTM 모델

import torch
import torch.nn as nn

class PowerLSTM(nn.Module):
    def __init__(
        self,
        input_size=1,
        hidden_size=64,
        num_layers=2,
        dropout=0.2,
    ):
        super().__init__()

        self.lstm = nn.LSTM(
            input_size=input_size,
            hidden_size=hidden_size,
            num_layers=num_layers,
            batch_first=True,
            dropout=dropout if num_layers > 1 else 0.0,
        )
        self.fc = nn.Linear(hidden_size, 1)

    def forward(self, x):
        out, (h_n, c_n) = self.lstm(x)
        last_hidden = out[:, -1, :]
        prediction = self.fc(last_hidden)
        return prediction

모델의 처리 과정은 다음과 같다.

[batch, 24, 1]
        │
        ▼
2층 LSTM, hidden_size=64
        │
        ▼
마지막 시점의 hidden state: [batch, 64]
        │
        ▼
Linear(64, 1)
        │
        ▼
다음 1시간의 전력 사용량: [batch, 1]

LSTM의 출력 Shape

batch_first=True일 때 입력은 [batch, sequence, feature] 순서다.

out, (h_n, c_n) = model.lstm(sample_x)

print("입력:", sample_x.shape)
print("out:", out.shape)
print("h_n:", h_n.shape)
print("c_n:", c_n.shape)

실행 결과:

입력: torch.Size([4, 24, 1])
out:  torch.Size([4, 24, 64])
h_n:  torch.Size([2, 4, 64])
c_n:  torch.Size([2, 4, 64])

단방향 LSTM의 마지막 층에서는 out[:, -1, :]와 h_n[-1]이 같은 마지막 시점의 hidden state를 나타낸다.

6. 학습과 평가 결과

학습에는 평균 제곱 오차와 AdamW Optimizer를 사용한다.

criterion = nn.MSELoss()
optimizer = torch.optim.AdamW(
    model.parameters(),
    lr=1e-3,
    weight_decay=1e-4,
)

MAX_EPOCHS = 50
PATIENCE = 8
CLIP_NORM = 1.0

• Validation Loss가 좋아진 모델의 상태를 저장한다.
• 8번 연속 좋아지지 않으면 Early Stopping을 적용한다.
• clip_grad_norm_()으로 기울기 폭주를 완화한다.

노트북에서는 29 Epoch에 Early Stopping이 발생했고, 가장 좋은 가중치를 다시 불러와 Test 데이터를 평가했다.

원래 단위로 복원하기

모델의 출력은 표준화된 값이므로 inverse_transform()으로 원래 전력 단위로 복원한다.

pred_scaled, true_scaled = predict(model, test_loader, DEVICE)

pred = scaler.inverse_transform(pred_scaled).ravel()
true = scaler.inverse_transform(true_scaled).ravel()

Persistence Baseline과 비교하기

Persistence Baseline은 복잡한 모델 없이 마지막 입력값이 다음 시점에도 그대로 이어진다고 예측한다.

last_input_scaled = X_test[:, -1, :]
baseline_pred = scaler.inverse_transform(last_input_scaled).ravel()

노트북의 실행 결과는 다음과 같다.

RMSE와 MAE는 낮을수록 좋다. 이 실행에서는 LSTM이 단순 Baseline보다 두 지표 모두 낮았다. 특히 RMSE는 약 14.7%, MAE는 약 8.2% 감소했다.

그래프를 볼 때는 다음 내용을 함께 확인해야 한다.

1. 실제값의 상승과 하락 방향을 따라가는가?
2. 급격한 Peak를 지나치게 평평하게 예측하는가?
3. 예측이 실제값보다 한 시점 늦게 따라가는가?

위 결과에서는 전반적인 주기와 방향은 따라가지만, 일부 급격한 Peak를 실제보다 낮게 예측한다. 따라서 시계열 모델은 하나의 평가지표뿐 아니라 Baseline과 시각화 결과를 함께 확인해야 한다.

7. GRU

GRU(Gated Recurrent Unit)는 LSTM과 마찬가지로 장기 의존성 문제를 완화하기 위한 순환 신경망이다. LSTM보다 구조가 단순하며 별도의 cell state를 사용하지 않는다.

GRU는 두 게이트를 사용한다.

LSTM과 GRU 비교

어느 모델이 항상 더 좋다고 정할 수는 없다. 데이터 길이와 크기, 필요한 문맥 범위에 따라 Validation 성능과 학습 시간을 비교해 선택해야 한다.

PyTorch에서 GRU 사용하기

LSTM과 생성 방식은 거의 같다.

gru = nn.GRU(
    input_size=1,
    hidden_size=64,
    num_layers=2,
    batch_first=True,
    dropout=0.2,
)

out, h_n = gru(x)

LSTM은 다음처럼 두 상태를 반환한다.

out, (h_n, c_n) = lstm(x)

GRU는 별도의 cell state가 없으므로 out과 h_n만 반환한다.

8. Seq2Seq

Seq2Seq(Sequence-to-Sequence)는 입력 시퀀스를 받아 다른 출력 시퀀스를 생성하는 구조다. 기계 번역, 문서 요약, 질의응답 등에 사용할 수 있다.

I love coffee
      │
      ▼
   Encoder
      │
문장 정보 압축
      │
      ▼
   Decoder
      │
      ▼
<BOS> 나는 → 커피를 → 좋아해 <EOS>

• <BOS>는 문장 생성의 시작을 알리는 특별 Token이다.
• <EOS>는 문장의 끝을 알리는 특별 Token이다.
• Decoder는 이전에 생성한 Token을 이용해 다음 Token을 순서대로 예측한다.
• 이처럼 이전 출력을 다음 입력으로 사용하는 방식을 자기회귀(Autoregressive)라고 한다.

Encoder와 Decoder

고전적인 Seq2Seq의 Encoder와 Decoder에는 RNN, LSTM, GRU 등을 사용할 수 있다.

Seq2Seq의 문제점

초기 Seq2Seq는 Encoder의 마지막 hidden state 하나를 Decoder에 전달하는 방식을 주로 사용했다.

문장이 길어질수록 앞부분부터 뒷부분까지의 모든 정보를 하나의 고정된 크기 벡터에 압축해야 한다. 이 과정에서 중요한 정보가 손실되는 현상을 정보 병목(Bottleneck)이라고 한다.

또한 RNN 계열은 이전 시점의 계산이 끝나야 다음 시점을 계산할 수 있다.

t=1 계산 → t=2 계산 → t=3 계산 → ...

따라서 긴 시퀀스에서는 학습이 느리고 병렬 처리에도 불리하다.

9. Attention 메커니즘

Attention은 현재 처리하는 Token이 다른 Token을 얼마나 참고해야 하는지 점수로 계산하는 방법이다.

"나는 오늘 학교에서 수학 시험을 봤다"

어떤 Token이 중요한지는 항상 같지 않다. 현재 어떤 Token을 처리하는지에 따라 참고해야 할 대상이 달라진다. Attention은 이러한 관계를 학습된 점수로 표현한다.

Token을 Embedding Vector로 바꾸기

신경망은 문자열을 그대로 계산할 수 없으므로 Token을 ID로 바꾸고, nn.Embedding을 이용해 학습 가능한 실수 벡터로 변환한다.

import torch
import torch.nn as nn

torch.manual_seed(2026)

tokens = ["커피", "한잔", "어때"]
token_ids = torch.tensor([0, 1, 2])

embedding = nn.Embedding(
    num_embeddings=3,
    embedding_dim=4,
)

X = embedding(token_ids)

print("Token ID Shape:", token_ids.shape)
print("Embedding Shape:", X.shape)

실행 결과:

Token ID Shape: torch.Size([3])
Embedding Shape: torch.Size([3, 4])

Token ID 0, 1, 2는 단어의 의미나 크기를 나타내는 숫자가 아니다. Embedding Table에서 각 Token의 벡터를 찾기 위한 행 번호다.

10. Self-Attention과 Q, K, V

Self-Attention은 같은 시퀀스 안의 모든 Token이 서로의 관계를 계산하는 Attention이다.

커피 → 커피, 한잔, 어때를 참고
한잔 → 커피, 한잔, 어때를 참고
어때 → 커피, 한잔, 어때를 참고

Token이 3개라면 관계 점수표의 Shape은 3 × 3이 된다.

Query, Key, Value

Embedding X를 서로 다른 세 개의 선형층에 통과시켜 Query, Key, Value를 만든다.

embed_dim = 4

W_Q = nn.Linear(embed_dim, embed_dim, bias=False)
W_K = nn.Linear(embed_dim, embed_dim, bias=False)
W_V = nn.Linear(embed_dim, embed_dim, bias=False)

Q = W_Q(X)
K = W_K(X)
V = W_V(X)

print("X:", X.shape)
print("Q:", Q.shape)
print("K:", K.shape)
print("V:", V.shape)

실행 결과:

X: torch.Size([3, 4])
Q: torch.Size([3, 4])
K: torch.Size([3, 4])
V: torch.Size([3, 4])

Q와 K로 관계 점수 계산하기

현재 Token의 Query와 모든 Token의 Key를 내적해 관련도 점수를 계산한다.

scores = Q @ K.T
print(scores.shape)

Shape의 변화는 다음과 같다.

Q:   (3, 4)
K.T: (4, 3)
----------------
결과: (3, 3)

• 행은 정보를 찾는 Query Token이다.
• 열은 참고 대상인 Key Token이다.
• scores[0, 2]는 커피의 Query와 어때의 Key를 비교한 점수다.

Scaled Dot-Product Attention

Transformer에서는 점수가 지나치게 커지는 것을 줄이기 위해 Key 벡터의 차원 d_k의 제곱근으로 나눈다. 그다음 softmax로 각 행의 합이 1인 Attention Weight를 만든다.

import math

d_k = Q.size(-1)

scores = (Q @ K.T) / math.sqrt(d_k)
attention_weights = torch.softmax(scores, dim=-1)
attention_output = attention_weights @ V

print("Score Shape:", scores.shape)
print("Weight Shape:", attention_weights.shape)
print("Output Shape:", attention_output.shape)

수식으로 표현하면 다음과 같다.

Attention(Q, K, V)
    = softmax(QK^T / √d_k)V

처리 과정은 다음과 같이 정리할 수 있다.

Token IDs
    │
    ▼
Embedding X
    │
    ├── Linear ── Q
    ├── Linear ── K
    └── Linear ── V

Q × K.T
    │
    ▼
관련도 점수
    │
√d_k로 나누기
    │
Softmax
    │
    ▼
Attention Weight
    │
Weight × V
    │
    ▼
문맥을 반영한 출력

11. Attention과 Transformer의 연결

Attention을 사용하면 각 Token이 다른 Token을 직접 참고할 수 있다. RNN처럼 정보를 시점별로 차례로 전달할 필요가 없으므로 학습할 때 여러 Token의 관계 계산을 병렬화하기 쉽다.

다만 Self-Attention만으로는 Token의 순서를 자동으로 알 수 없다. 실제 Transformer는 Token Embedding에 위치 정보를 더하고, Multi-Head Attention과 Feed Forward Network 등을 함께 사용한다.

RNN 계열
앞 시점의 상태를 다음 시점으로 순차 전달

Transformer
각 Token이 다른 Token과의 관계를 직접 계산

이번 예제는 Transformer 전체 구조를 한 번에 구현하기보다, 그 핵심 연산인 Embedding → Q·K·V → 관계 점수가 어떻게 만들어지는지를 확인하는 단계다.

12. 모델별 핵심 비교

13. 핵심 정리

1. LSTM은 hidden state와 cell state를 사용한다.
2. Forget, Input, Output Gate가 과거 기억의 유지와 새 정보의 저장을 조절한다.
3. 전력 예측 예제는 최근 24시간을 이용해 다음 1시간을 예측한다.
4. LSTM의 출력은 out, h_n, c_n이며 각 Shape과 역할이 다르다.
5. 모델 성능은 RMSE와 MAE뿐 아니라 단순 Baseline, 예측 그래프와 함께 평가해야 한다.
6. GRU는 별도의 cell state 없이 Update Gate와 Reset Gate를 사용한다.
7. Seq2Seq는 Encoder와 Decoder로 구성되며, 긴 문장을 하나의 벡터로 압축할 때 정보 병목이 생길 수 있다.
8. Self-Attention은 Query와 Key로 관계 점수를 만들고, Value를 가중합해 문맥이 반영된 출력을 만든다.
9. Transformer의 핵심 Attention 연산은 softmax(QK^T / √d_k)V로 표현할 수 있다.

1. 텍스트 벡터화

벡터화(Vectorization)는 텍스트를 머신러닝과 딥러닝 모델이 계산할 수 있는 숫자 벡터로 바꾸는 과정이다.

텍스트를 숫자로 표현하는 대표적인 방법은 다음과 같다.

원-핫 인코딩

원-핫 인코딩(One-Hot Encoding)은 Vocabulary 크기만큼의 벡터를 만들고, 표현하려는 Token 위치만 1, 나머지는 모두 0으로 채운다.

import torch
import torch.nn.functional as F

token_id = torch.tensor(2)
vocab_size = 5

one_hot = F.one_hot(token_id, num_classes=vocab_size)

print(one_hot)
print(one_hot.shape)

실행 결과:

tensor([0, 0, 1, 0, 0])
torch.Size([5])

Vocabulary 크기가 10,000이면 Token 하나를 나타내기 위해 길이가 10,000인 벡터가 필요하다. 그중 대부분이 0이므로 이러한 벡터를 희소 벡터(Sparse Vector)라고 한다.

Bag of Words

Bag of Words(BOW)는 문서에 각 단어가 몇 번 등장했는지를 벡터로 표현한다.

Vocabulary: ["I", "love", "Python"]
문장: "I love Python Python"
BOW: [1, 1, 2]

단어 빈도를 간단하게 표현할 수 있지만 다음 두 문장을 구별하기 어렵다.

나는 너를 좋아한다
너는 나를 좋아한다

두 문장에 등장하는 단어가 같다면 BOW 벡터도 비슷해진다. 즉, 단어의 순서와 문맥을 반영하지 못한다.

TF-IDF

TF-IDF(Term Frequency-Inverse Document Frequency)는 단어가 한 문서에서 자주 나타날수록 가중치를 높이고, 전체 문서에서 흔하게 나타날수록 가중치를 낮춘다.

TF-IDF = TF × IDF

• TF: 특정 문서에서 해당 단어가 등장한 빈도다.
• IDF: 전체 문서에서 드물게 등장할수록 커지는 값이다.

모든 문서에 자주 나타나는 단어보다 특정 문서의 내용을 잘 나타내는 단어에 높은 가중치를 줄 수 있다. 다만 BOW와 마찬가지로 단어 순서와 문맥을 직접 학습하지는 않는다.

2. Word Embedding

Word Embedding은 Token을 저차원의 밀집 벡터(Dense Vector)로 표현하는 방법이다.

Token ID
   ↓
Embedding Layer
   ↓
[0.21, -0.72, 0.13, 0.84, ...]

원-핫 벡터는 각 단어가 서로 독립적이지만, Embedding은 학습을 통해 비슷한 문맥에서 사용되는 단어가 가까운 벡터를 갖도록 만들 수 있다.

nn.Embedding

PyTorch의 nn.Embedding은 Token ID에 대응하는 벡터를 Embedding Table에서 가져오는 층이다.

import torch.nn as nn

embedding = nn.Embedding(
    num_embeddings=10000,
    embedding_dim=128,
    padding_idx=0,
)

print(embedding)
print(embedding.weight.shape)

실행 결과:

Embedding(10000, 128, padding_idx=0)
torch.Size([10000, 128])

입력과 출력 Shape는 다음처럼 변한다.

input_ids = torch.tensor([
    [1, 2, 3],
    [4, 2, 5],
], dtype=torch.long)

embedding = nn.Embedding(
    num_embeddings=10,
    embedding_dim=4,
    padding_idx=0,
)

output = embedding(input_ids)

print("입력 Shape:", input_ids.shape)
print("출력 Shape:", output.shape)
print("학습 여부:", embedding.weight.requires_grad)
print("Parameter 수:", embedding.weight.numel())

실행 결과:

입력 Shape: torch.Size([2, 3])
출력 Shape: torch.Size([2, 3, 4])
학습 여부: True
Parameter 수: 40

입력 Shape [batch_size, sequence_length]의 뒤에 embedding_dim이 추가된다.

[2, 3] → [2, 3, 4]

Embedding 벡터의 의미를 사람이 직접 지정하는 것은 아니다. 처음에는 초기값으로 시작하고, 모델의 다른 가중치와 함께 학습된다.

3. Word2Vec

Word2Vec은 주변 문맥을 이용해 단어의 벡터를 학습한다. 비슷한 문맥에서 자주 사용되는 단어가 벡터 공간에서도 가까워지도록 만든다.

CBOW와 Skip-gram

문장이 다음과 같다고 가정한다.

나는 오늘 공원에서 귀여운 강아지를 산책시키며 행복한 시간을 보냈다.

중심 단어가 강아지를이고 Window 크기가 2라면 주변 문맥은 다음과 같다.

["공원에서", "귀여운", "산책시키며", "행복한"]

• CBOW는 주변 네 단어로 강아지를을 예측한다.
• Skip-gram은 강아지를로 주변 네 단어를 예측한다.

SGNS

SGNS(Skip-Gram with Negative Sampling)는 Skip-gram의 계산량을 줄이는 방법이다.

• 실제 중심 단어와 주변 단어의 조합은 관련이 있는 Positive Sample로 학습한다.
• 무작위로 선택한 단어 조합은 관련이 없는 Negative Sample로 학습한다.
• Vocabulary 전체에 대한 확률을 매번 계산하지 않고 일부 단어만 비교한다.

NSMC 데이터 전처리

한국어 Word2Vec 학습을 위해 네이버 영화 리뷰 데이터(NSMC)를 사용한다.

노트북의 macOS 환경에서는 MeCab을 다음과 같이 연결했다.

import os
from konlpy.tag import Mecab

os.environ["MECABRC"] = "/opt/homebrew/etc/mecabrc"

mecab = Mecab(
    dicpath="/opt/homebrew/lib/mecab/dic/mecab-ko-dic",
)

MECABRC와 사전 경로는 설치 방법과 운영체제에 따라 달라진다.

import urllib.request
import pandas as pd

urllib.request.urlretrieve(
    "https://raw.githubusercontent.com/e9t/nsmc/master/ratings.txt",
    filename="ratings.txt",
)

train_data = pd.read_table("ratings.txt")
train_data = train_data[:20000]

한글과 공백을 제외한 문자를 제거한다.

train_data["document"] = train_data["document"].str.replace(
    "[^ㄱ-ㅎㅏ-ㅣ가-힣 ]",
    "",
    regex=True,
)

MeCab으로 형태소를 분리하고 불용어를 제거한다.

stopwords = [
    "도", "는", "다", "의", "가", "이", "은", "한", "에", "하",
    "고", "을", "를", "인", "듯", "과", "와", "네", "들", "지",
    "임", "게",
]

tokenized_data = []

for sentence in train_data["document"]:
    tokens = mecab.morphs(sentence)
    tokens = [word for word in tokens if word not in stopwords]
    tokenized_data.append(tokens)

MeCab은 붙어 있는 한국어 문장을 형태소 단위로 나눈다.

mecab.morphs("아버지가방에들어가신다")

['아버지', '가', '방', '에', '들어가', '신다']

Gensim으로 Word2Vec 학습하기

from gensim.models import Word2Vec

model = Word2Vec(
    sentences=tokenized_data,
    vector_size=100,
    window=5,
    min_count=5,
    workers=2,
    sg=1,
    negative=5,
)

학습 결과는 3,990개 단어를 각각 100차원 벡터로 표현했다.

print(model.wv.vectors.shape)

(3990, 100)

most_similar()는 코사인 유사도를 기준으로 가까운 단어를 찾는다.

model.wv.most_similar("블록버스터", topn=5)

데이터를 20,000개 리뷰로 제한했기 때문에 결과 품질은 전체 데이터와 학습 설정에 따라 달라질 수 있다.

4. FastText와 한글 자모 분해

Word2Vec은 Vocabulary에 없는 단어의 벡터를 바로 만들기 어렵다. FastText는 단어를 문자 n-gram 단위로 나누어 학습하므로, 처음 보는 단어나 오타도 이미 학습한 부분 문자열을 조합해 표현할 수 있다.

playing
→ <pl, pla, lay, ayi, yin, ing, ng>

한국어는 조사와 어미가 붙어 단어 형태가 다양하다. 자모 단위로 분해하면 비슷한 철자를 가진 단어와 오타가 더 많은 부분을 공유한다.

hgtk로 한글 분해와 조합하기

import hgtk

print(hgtk.letter.decompose("김"))
print(hgtk.letter.compose("ㄱ", "ㅣ", "ㅁ"))

실행 결과:

('ㄱ', 'ㅣ', 'ㅁ')
김

받침이 없는 글자는 빈 문자열 대신 -를 넣어 항상 세 글자 단위로 표현한다.

def word_to_jamo(token):
    def to_special_token(jamo):
        return jamo if jamo else "-"

    decomposed_token = ""

    for char in token:
        try:
            cho, jung, jong = hgtk.letter.decompose(char)
            decomposed_token += (
                to_special_token(cho)
                + to_special_token(jung)
                + to_special_token(jong)
            )
        except Exception as error:
            if type(error).__name__ == "NotHangulException":
                decomposed_token += char

    return decomposed_token

print(word_to_jamo("남동생"))
print(word_to_jamo("여동생"))

ㄴㅏㅁㄷㅗㅇㅅㅐㅇ
ㅇㅕ-ㄷㅗㅇㅅㅐㅇ

MeCab 형태소 분석과 자모 분해를 연결한다.

def tokenize_by_jamo(sentence):
    return [word_to_jamo(token) for token in mecab.morphs(sentence)]

FastText 실습에는 네이버 쇼핑 리뷰 20만 건을 사용했다.

urllib.request.urlretrieve(
    "https://raw.githubusercontent.com/bab2min/corpus/master/sentiment/naver_shopping.txt",
    filename="ratings_total.txt",
)

total_data = pd.read_table(
    "ratings_total.txt",
    names=["ratings", "reviews"],
)

각 리뷰를 형태소로 나눈 뒤 자모 단위로 변환한다.

from tqdm import tqdm

tokenized_data = []

for review in tqdm(total_data["reviews"].to_list()):
    tokenized_sample = tokenize_by_jamo(review)
    tokenized_data.append(tokenized_sample)

분해한 Token을 텍스트 파일로 저장하고 FastText를 학습한다.

import fasttext

with open("tokenized_data.txt", "w", encoding="utf8") as output_file:
    for line in tokenized_data:
        output_file.write(" ".join(line) + "\n")

model = fasttext.train_unsupervised(
    "tokenized_data.txt",
    model="cbow",
)

model.save_model("fasttext.bin")

노트북에서는 남동쉥, 남동셍ㅋ, 제품^^처럼 변형되거나 오타가 있는 입력에서도 원래 단어와 관련된 결과가 나타나는지 확인했다.

남동쉥 → 남동생, 남친, 남매, 남짓, 남녀 ...
제품^^ → 제품, 제풍, 반제품, 완제품, 상품 ...

이는 FastText가 단어 전체만 외우지 않고 문자 조각을 함께 학습하기 때문이다.

5. 시퀀스 데이터

시퀀스 데이터(Sequence Data)는 값의 순서와 앞뒤 관계가 중요한 데이터다.

일반적인 MLP는 각 입력을 독립적으로 처리한다. 반면 RNN은 이전 시점의 정보를 Hidden State에 담아 다음 시점 계산에 전달한다.

6. RNN의 동작 원리

RNN(Recurrent Neural Network)은 현재 입력과 이전 Hidden State를 함께 사용해 새로운 Hidden State를 만든다.

h_t = tanh(W_xh x_t + W_hh h_(t-1) + b)

Hidden State는 가중치 자체가 아니라, 현재 시점까지 본 정보 중 다음 계산에 필요한 내용을 압축한 작업 메모리다.

RNN은 시점마다 새로운 Cell을 학습하는 것이 아니다. 같은 RNN Cell과 같은 가중치를 전체 시간축에서 반복해서 사용한다.

입력 Shape

batch_first=True인 RNN의 입력 Shape는 다음과 같다.

[batch_size, sequence_length, input_size]

예를 들어 [32, 20, 10]은 다음을 의미한다.

• 한 번에 32개 시퀀스를 처리한다.
• 시퀀스 하나는 연속된 20개 시점으로 구성된다.
• 각 시점은 10개 Feature를 가진다.

장기 의존성 문제

기본 RNN은 가까운 과거의 패턴을 처리하는 데 유용하지만, 오래전 정보를 마지막까지 유지하기 어렵다. 같은 가중치를 시간축에서 반복 적용하며 역전파하기 때문에 긴 시퀀스에서는 기울기 소실 또는 기울기 폭주가 발생하기 쉽다.

이 문제를 완화하기 위해 LSTM과 GRU 같은 구조가 사용된다.

7. RNN으로 다음 KOSPI 종가 예측하기

최근 20거래일의 정보를 입력으로 사용해 다음 거래일의 종가를 예측한다.

1일차 입력 + 초기 Hidden State → h1
2일차 입력 + h1                → h2
...
20일차 입력 + h19              → h20
h20 → Linear Layer → 21일차 종가 예측

데이터 확인과 정제

원본 데이터는 4,513행이며 다음 열을 가진다.

날짜를 datetime으로 바꾸고 필수 열의 결측치를 제거한 뒤 날짜순으로 정렬한다.

df["Date"] = pd.to_datetime(
    df["Date"],
    dayfirst=True,
    errors="coerce",
)

numeric_cols = ["Open", "High", "Low", "Close", "Adj Close", "Volume"]

for column in numeric_cols:
    df[column] = pd.to_numeric(df[column], errors="coerce")

required_cols = ["Date", "Open", "High", "Low", "Close"]
df_copy = df.dropna(subset=required_cols).copy()
df_copy = df_copy.sort_values("Date").reset_index(drop=True)

정제 후 2000년 1월 4일부터 2018년 1월 22일까지 총 4,452행이 남았다.

Feature Engineering

당일 가격과 최근 가격 흐름을 함께 표현하도록 파생변수를 만든다.

data = df_copy.copy()

data["Return_1d"] = data["Close"].pct_change()
data["Range_pct"] = (data["High"] - data["Low"]) / data["Close"]
data["OC_pct"] = (data["Close"] - data["Open"]) / data["Open"]

data["MA5"] = data["Close"].rolling(5).mean()
data["MA20"] = data["Close"].rolling(20).mean()

data["MA5_ratio"] = data["Close"] / data["MA5"]
data["MA20_ratio"] = data["Close"] / data["MA20"]
data["Volatility_10"] = data["Return_1d"].rolling(10).std()

data["Target_Close"] = data["Close"].shift(-1)
data["Target_Date"] = data["Date"].shift(-1)

사용하는 Feature는 총 10개다.

FEATURES = [
    "Open",
    "High",
    "Low",
    "Close",
    "Return_1d",
    "Range_pct",
    "OC_pct",
    "MA5_ratio",
    "MA20_ratio",
    "Volatility_10",
]

shift(-1)을 사용하면 현재 행의 Target에 다음 거래일 종가가 들어간다. 이동평균과 다음 날 Target을 만드는 과정에서 생긴 결측치를 제거한 결과 학습에 사용할 데이터는 4,432행이다.

8. 시계열 데이터 분리와 Scaling

시계열 데이터를 무작위로 나누면 미래 데이터가 Train에 포함되고 과거 데이터가 Test에 포함될 수 있다. 따라서 날짜순으로 다음과 같이 분리한다.

n = len(data)

train_end = int(n * 0.70)
val_end = int(n * 0.85)

Feature마다 값의 크기가 다르므로 StandardScaler로 평균 0, 표준편차 1에 가깝게 변환한다.

from sklearn.preprocessing import StandardScaler

scaler_x = StandardScaler()
scaler_y = StandardScaler()

scaler_x.fit(data.loc[:train_end - 1, FEATURES])
scaler_y.fit(data.loc[:train_end - 1, ["Target_Close"]])

X_all = scaler_x.transform(data[FEATURES]).astype(np.float32)
y_all = scaler_y.transform(data[["Target_Close"]]).astype(np.float32)

Scaler는 반드시 Train 데이터에만 fit()한다. Validation과 Test를 함께 사용하면 미래 데이터의 평균과 표준편차를 학습 단계에서 미리 알게 되는 데이터 누수가 발생한다.

X_all Shape: (4432, 10)
y_all Shape: (4432, 1)

9. 연속된 20일을 하나의 시퀀스로 만들기

RNN은 한 행씩 독립적으로 보는 대신 연속된 여러 행을 하나의 입력으로 받는다.

SEQUENCE_LENGTH = 20

def make_sequence(X, y, sequence_length):
    X_seq = []
    y_seq = []
    target_indices = []

    for index in range(sequence_length - 1, len(X)):
        start = index - sequence_length + 1
        X_seq.append(X[start:index + 1])
        y_seq.append(y[index])
        target_indices.append(index)

    return (
        np.asarray(X_seq, dtype=np.float32),
        np.asarray(y_seq, dtype=np.float32),
        np.asarray(target_indices, dtype=np.int64),
    )

Window는 하루씩 이동한다.

첫 번째 입력: 0~19번 행 → 19번 행에 연결된 다음 날 종가
두 번째 입력: 1~20번 행 → 20번 행에 연결된 다음 날 종가
세 번째 입력: 2~21번 행 → 21번 행에 연결된 다음 날 종가

전체 데이터에서 먼저 Window를 만들고 Target 시점의 인덱스를 기준으로 Train, Validation, Test를 나눈다. 이렇게 하면 Validation과 Test의 첫 번째 예측에서도 그 이전 날짜를 입력 이력으로 사용할 수 있다.

전체 Sequence Shape: (4413, 20, 10)
전체 Target Shape: (4413, 1)

분리 결과는 다음과 같다.

Train: (3083, 20, 10), (3083, 1)
Validation: (665, 20, 10), (665, 1)
Test: (665, 20, 10), (665, 1)

10. Dataset과 DataLoader

from torch.utils.data import Dataset, DataLoader

class SequenceDataset(Dataset):
    def __init__(self, X, y):
        self.X = torch.tensor(X, dtype=torch.float32)
        self.y = torch.tensor(y, dtype=torch.float32)

    def __len__(self):
        return len(self.X)

    def __getitem__(self, index):
        return self.X[index], self.y[index]

BATCH_SIZE = 64

train_loader = DataLoader(
    SequenceDataset(X_train, y_train),
    batch_size=BATCH_SIZE,
    shuffle=True,
)

val_loader = DataLoader(
    SequenceDataset(X_val, y_val),
    batch_size=BATCH_SIZE,
    shuffle=False,
)

test_loader = DataLoader(
    SequenceDataset(X_test, y_test),
    batch_size=BATCH_SIZE,
    shuffle=False,
)

Train의 shuffle=True는 20일로 구성된 Window들의 배치 순서를 섞는다. 각 Window 내부의 날짜 순서는 바뀌지 않는다.

입력 Batch Shape: torch.Size([64, 20, 10])
정답 Batch Shape: torch.Size([64, 1])

11. RNN 회귀 모델 만들기

class RNNRegressor(nn.Module):
    def __init__(self, input_size, hidden_size=64, num_layers=2, dropout=0.10):
        super().__init__()

        self.rnn = nn.RNN(
            input_size=input_size,
            hidden_size=hidden_size,
            num_layers=num_layers,
            nonlinearity="tanh",
            batch_first=True,
            dropout=dropout if num_layers > 1 else 0.0,
        )

        self.fc = nn.Linear(hidden_size, 1)

    def forward(self, x):
        out, h_n = self.rnn(x)
        last_hidden = out[:, -1, :]
        prediction = self.fc(last_hidden)
        return prediction

model = RNNRegressor(
    input_size=len(FEATURES),
    hidden_size=64,
    num_layers=2,
    dropout=0.10,
).to(DEVICE)

모델 구조는 다음과 같다.

RNNRegressor(
  (rnn): RNN(10, 64, num_layers=2, batch_first=True, dropout=0.1)
  (fc): Linear(in_features=64, out_features=1, bias=True)
)

out과 h_n

4개 샘플을 모델의 RNN 층에 전달한 결과는 다음과 같다.

입력:          [4, 20, 10]
RNN out:       [4, 20, 64]
RNN h_n:       [2, 4, 64]
마지막 시점:   [4, 64]

• out은 마지막 RNN Layer가 모든 시점에서 만든 Hidden State다.
• h_n은 각 RNN Layer의 마지막 Hidden State다.
• 다음 날 종가 예측에는 out[:, -1, :]로 마지막 시점의 정보를 사용한다.
• Linear(64, 1)은 64개 Hidden 값을 종가 예측값 하나로 바꾼다.

12. 학습 설정

criterion = nn.HuberLoss(delta=1.0)

optimizer = torch.optim.AdamW(
    model.parameters(),
    lr=1e-3,
    weight_decay=1e-4,
)

scheduler = torch.optim.lr_scheduler.ReduceLROnPlateau(
    optimizer=optimizer,
    mode="min",
    factor=0.5,
    patience=5,
)

MAX_EPOCHS = 200
PATIENCE = 15
CLIP_NORM = 1.0

loss.backward()
torch.nn.utils.clip_grad_norm_(
    model.parameters(),
    max_norm=CLIP_NORM,
)
optimizer.step()

Validation Loss가 가장 작을 때 모델 상태를 저장한다.

if val_loss < best_val_loss:
    best_val_loss = val_loss
    best_state = copy.deepcopy(model.state_dict())
    wait = 0
else:
    wait += 1

학습이 끝난 뒤에는 최적 상태를 다시 불러온 다음 Test 예측을 수행한다.

model.load_state_dict(best_state)

13. 학습 결과와 예측값 복원

노트북에서는 36번째 Epoch에서 Early Stopping이 실행됐다.

Epoch 001 | Train 0.046967 | Valid 0.001290 | LR 1.00e-03
Epoch 010 | Train 0.001861 | Valid 0.000449 | LR 1.00e-03
Epoch 020 | Train 0.001287 | Valid 0.000699 | LR 2.50e-04
Epoch 030 | Train 0.001110 | Valid 0.000472 | LR 1.25e-04
Early Stopping: epoch 36

모델은 표준화된 종가를 예측하므로 inverse_transform()으로 실제 종가 단위로 되돌린다.

pred_scaled, true_scaled = predict(model, test_loader, DEVICE)

pred_close = scaler_y.inverse_transform(pred_scaled).ravel()
true_close = scaler_y.inverse_transform(true_scaled).ravel()

실행 결과:

예측 개수: 665
예측값 앞 5개: [2130.90, 2109.48, 2097.60, 2090.25, 2100.17]
실제값 앞 5개: [2104.58, 2091.00, 2085.52, 2097.38, 2096.77]

앞의 몇 개 값이 비슷해 보인다는 것만으로 모델 성능을 판단할 수는 없다. 전체 Test 데이터에 대해 MAE와 RMSE를 계산하고, 실제값과 예측값의 시계열 그래프도 함께 확인해야 한다.

from sklearn.metrics import mean_absolute_error, mean_squared_error

mae = mean_absolute_error(true_close, pred_close)
rmse = mean_squared_error(true_close, pred_close) ** 0.5

print("MAE:", mae)
print("RMSE:", rmse)

14. 전체 흐름 정리

텍스트 데이터
  ↓
형태소 분석과 불용어 제거
  ↓
원-핫·BOW·TF-IDF 또는 Word Embedding
  ↓
Word2Vec / FastText로 단어 관계 학습

시계열 데이터
  ↓
Feature Engineering
  ↓
시간순 Train / Validation / Test 분리
  ↓
Train 기준 Scaling
  ↓
연속된 20일 Window 생성
  ↓
RNN → 마지막 Hidden State → Linear
  ↓
다음 거래일 종가 예측

• 원-핫 인코딩과 BOW는 단순하지만 고차원 희소 벡터가 되고 문맥을 충분히 반영하지 못한다.
• nn.Embedding은 Token ID를 학습 가능한 밀집 벡터로 바꾼다.
• Word2Vec은 주변 문맥으로 단어의 의미 관계를 학습한다.
• FastText는 문자 n-gram을 사용해 OOV와 오타에 비교적 강하다.
• RNN은 이전 Hidden State를 다음 시점 계산에 전달해 순서 정보를 처리한다.
• 시계열은 무작위 분할하지 않고 시간순으로 분리해야 한다.
• Scaling 기준은 Train 데이터에서만 계산해야 데이터 누수를 막을 수 있다.
• 기본 RNN은 장기 의존성과 기울기 문제를 가지며, LSTM과 GRU가 이를 보완한다.

병사 n명으로 순서대로 등장하는 적을 막는다.

• 한 라운드를 일반적으로 막으면 적 수만큼 병사가 줄어든다.
• 무적권은 최대 k번 사용할 수 있고, 사용한 라운드에서는 병사가 줄지 않는다.

최대로 막을 수 있는 라운드 수를 구한다.

핵심 아이디어

어떤 시점까지 막은 라운드들 중에서 무적권은 적 수가 가장 많은 k개 라운드에 쓰는 것이 항상 최적이다.

적 수가 작은 라운드에 무적권을 쓰고 큰 라운드에 병사를 쓰는 경우가 있다면, 두 라운드의 무적권 사용 여부를 바꾸면 병사 소모는 줄어들거나 같아진다.

따라서 각 라운드에서 다음 상태를 유지한다.

• 최소 힙: 지금까지 등장한 공격 중 무적권을 사용할 k개의 큰 공격
• 병사: 무적권을 쓰지 않는 나머지 공격을 막는 데 사용

힙 크기가 k를 넘으면, 힙에서 가장 작은 공격을 꺼내 병사로 막는다.

풀이 과정

1. 적 수를 최소 힙에 넣는다.
2. 힙의 크기가 k보다 크면, 가장 작은 적 수를 꺼내 병사에서 뺀다.
3. 병사가 음수가 되면 현재 라운드를 막을 수 없으므로, 현재 인덱스를 반환한다.
4. 모든 라운드를 처리하면 전체 라운드 수를 반환한다.

Python 코드

import heapq


def solution(n, k, enemy):
    invincible_rounds = []

    for round_index, enemy_count in enumerate(enemy):
        # 현재까지는 이 라운드에도 무적권을 쓴다고 가정한다.
        heapq.heappush(invincible_rounds, enemy_count)

        # 무적권 수를 넘으면 가장 작은 공격은 병사로 막는다.
        if len(invincible_rounds) > k:
            n -= heapq.heappop(invincible_rounds)

        if n < 0:
            return round_index

    return len(enemy)

예시

n = 7, k = 3, enemy = [4, 2, 4, 5, 3, 3, 1]인 경우를 보자.

5라운드까지 처리한 뒤 무적권을 사용할 공격은 가장 큰 5, 4, 4다.

무적권 사용: 5, 4, 4
병사로 처리: 2 + 3 = 5
남은 병사: 7 - 5 = 2

6라운드의 적 수 3까지 포함하면, 무적권을 제외하고 병사로 막아야 하는 적 수의 합이 10이 된다. 병사가 부족하므로 5라운드까지만 막을 수 있다.

시간 복잡도

N을 전체 라운드 수라고 하자.

힙에는 최대 k개의 원소만 유지한다.

• 힙 삽입과 삭제: O(log k)
• 전체 라운드 처리: O(N log k)
• 공간 복잡도: O(k)

정리

매 순간 무적권을 가장 큰 공격들에 배정한다고 생각하면 된다. 최소 힙에서 가장 작은 값을 병사로 처리하면서, 지금까지의 최적 무적권 배정을 계속 유지할 수 있다.

중복 없는 학습 단어들이 주어졌을 때, 각 단어를 다른 단어와 구분해 자동완성하려면 몇 글자를 입력해야 하는지 구한다.

모든 단어에 필요한 입력 글자 수의 합을 반환한다.

핵심 아이디어

단어를 사전순으로 정렬하면, 어떤 단어와 가장 긴 접두사를 공유할 수 있는 단어는 정렬된 목록에서 바로 앞 또는 바로 뒤에 있다.

따라서 현재 단어가 필요한 입력 글자 수는 다음처럼 구할 수 있다.

max(앞 단어와의 공통 접두사 길이, 뒤 단어와의 공통 접두사 길이) + 1

단, 현재 단어 자체가 다른 단어의 접두사라면 더 입력할 문자가 없으므로 단어 전체를 입력해야 한다.

required = min(len(word), longest_common_prefix + 1)

왜 인접한 단어만 비교할까?

사전순 정렬에서 같은 접두사를 가진 단어들은 항상 연속해서 모인다.

예를 들어 word와 wor...로 시작하는 단어들은 모두 한 구간에 모인다. 현재 단어와 가장 긴 접두사를 공유하는 단어는 그 구간에서 바로 앞이나 바로 뒤에 있으므로, 두 이웃만 비교하면 충분하다.

풀이 과정

1. 단어 목록을 사전순으로 정렬한다.
2. 각 단어와 앞 단어의 최장 공통 접두사 길이를 구한다.
3. 각 단어와 뒤 단어의 최장 공통 접두사 길이를 구한다.
4. 둘 중 큰 값에 1을 더한다.
5. 현재 단어 길이를 넘지 않도록 제한한 값을 답에 더한다.

Python 코드

def common_prefix_length(first, second):
    limit = min(len(first), len(second))
    index = 0

    while index < limit and first[index] == second[index]:
        index += 1

    return index


def solution(words):
    words.sort()
    total = 0

    for index, word in enumerate(words):
        longest_common_prefix = 0

        if index > 0:
            longest_common_prefix = max(
                longest_common_prefix,
                common_prefix_length(word, words[index - 1])
            )

        if index + 1 < len(words):
            longest_common_prefix = max(
                longest_common_prefix,
                common_prefix_length(word, words[index + 1])
            )

        # word가 다른 단어의 접두사인 경우에는 단어 전체를 입력해야 한다.
        total += min(len(word), longest_common_prefix + 1)

    return total

예시

words = ["go", "gone", "guild"]는 이미 사전순으로 정렬되어 있다.

go는 gone의 접두사다. 공통 접두사 길이는 2이지만 go 뒤에 입력할 글자가 없으므로, 단어 전체인 2글자를 입력해야 한다.

총 입력 글자 수는 2 + 3 + 2 = 7이다.

시간 복잡도

N을 단어 개수, L을 모든 단어 길이의 합, M을 가장 긴 단어 길이라고 하자.

• 단어 정렬: 최대 O(N log N * M)

• 인접 단어 공통 접두사 계산: 각 단어는 앞뒤 단어와만 비교하므로 O(L)

• 시간 복잡도: O(N log N * M + L)

• 공간 복잡도: 정렬 구현을 포함해 O(N)

정리

자동완성에 필요한 글자 수는 해당 단어와 가장 비슷한 다른 단어를 구분할 수 있는 위치로 결정된다. 사전순 정렬 후 앞뒤 단어의 공통 접두사만 비교하면, 트라이 없이도 필요한 총 입력 횟수를 구할 수 있다.

A팀의 출전 순서는 이미 정해져 있고, B팀은 출전 순서를 자유롭게 정할 수 있다.

B팀 선수가 A팀 선수보다 큰 숫자를 낼 때만 1점을 얻는다. B팀이 얻을 수 있는 최대 승점을 구한다.

핵심 아이디어

실제 출전 순서보다 어떤 숫자가 어떤 숫자를 이기는지가 중요하다. 따라서 A와 B를 모두 오름차순 정렬한다.

B의 작은 숫자부터 확인하면서, 아직 이기지 않은 A 숫자 중 가장 작은 숫자를 이길 수 있는지 확인한다.

• B의 숫자 > A의 가장 작은 미매칭 숫자: 이 A 선수를 이기고 승점을 얻는다.
• 그렇지 않다: 이 B 선수는 어떤 남은 A 선수도 이길 수 없으므로 승점 없이 넘긴다.

이길 수 있는 B 선수를 더 큰 A 숫자에 쓰면 더 작은 A 숫자를 이길 기회를 잃을 수 있다. 따라서 가능한 가장 작은 A 숫자와 매칭하는 선택이 항상 최적이다.

풀이 과정

1. A와 B를 오름차순 정렬한다.
2. A에서 아직 이기지 않은 가장 작은 숫자의 인덱스를 a_index로 둔다.
3. 정렬된 B를 앞에서부터 확인한다.
4. 현재 B 숫자가 A[a_index]보다 크면 승점을 1 늘리고 a_index를 다음 A 선수로 옮긴다.
5. 모든 B 선수를 확인한 뒤 승점을 반환한다.

Python 코드

def solution(A, B):
    A.sort()
    B.sort()

    score = 0
    a_index = 0

    for b_number in B:
        # 현재 B 선수가 가장 작은 미매칭 A 선수를 이길 수 있다.
        if a_index < len(A) and b_number > A[a_index]:
            score += 1
            a_index += 1

    return score

예시

A = [5, 1, 3, 7] -> [1, 3, 5, 7]
B = [2, 2, 6, 8] -> [2, 2, 6, 8]

따라서 B팀은 최대 3점을 얻는다.

시간 복잡도

N을 팀 인원 수라고 하자.

• A, B 정렬: O(N log N)

• B 배열 순회: O(N)

• 시간 복잡도: O(N log N)

• 공간 복잡도: 정렬 구현에 따라 O(N)

정리

이길 수 있는 B 숫자는 남은 A 숫자 중 가장 작은 수를 이기는 데 사용한다. 이 단순한 그리디 기준을 정렬된 두 배열에 적용하면 B팀의 최대 승점을 구할 수 있다.

주어진 단어 조각을 원하는 만큼 사용해 문자열 t를 완성한다.

문자열을 완성하는 데 필요한 단어 조각 수의 최솟값을 구하고, 만들 수 없다면 -1을 반환한다.

핵심 아이디어

dp[i]를 t의 앞에서부터 i글자까지 완성하는 데 필요한 최소 조각 수라고 정의한다.

dp[0] = 0

어떤 위치 i까지 만들 수 있고, 그 위치부터 시작하는 조각 piece가 t와 일치한다면 다음 위치를 갱신한다.

dp[i + len(piece)] = min(
    dp[i + len(piece)],
    dp[i] + 1
)

각 조각을 무한히 사용할 수 있으므로, 같은 조각을 여러 위치에서 반복해서 사용해도 된다.

풀이 과정

1. 길이가 len(t) + 1인 DP 배열을 만들고, 도달할 수 없는 값은 큰 값으로 초기화한다.
2. dp[0] = 0으로 시작한다.
3. 문자열의 각 위치에서, 해당 위치부터 일치하는 모든 단어 조각을 확인한다.
4. 일치하는 조각의 끝 위치를 최소 조각 수로 갱신한다.
5. dp[len(t)]가 여전히 큰 값이면 -1을 반환한다.

Python 코드

def solution(strs, t):
    length = len(t)
    infinity = length + 1

    # dp[i]: t의 앞 i글자를 만드는 데 필요한 최소 조각 수
    dp = [infinity] * (length + 1)
    dp[0] = 0

    for start in range(length):
        if dp[start] == infinity:
            continue

        for piece in strs:
            end = start + len(piece)

            if end <= length and t.startswith(piece, start):
                dp[end] = min(dp[end], dp[start] + 1)

    return -1 if dp[length] == infinity else dp[length]

예시

strs = ["ba", "na", "n", "a"], t = "banana"인 경우를 보자.

dp[0] = 0
"ba" 사용  -> dp[2] = 1
"na" 사용  -> dp[4] = 2
"na" 사용  -> dp[6] = 3

따라서 "ba" + "na" + "na"로 문자열을 만들 수 있고, 필요한 조각 수는 3이다.

시간 복잡도

N을 t의 길이, S를 단어 조각 개수, L을 조각의 최대 길이라고 하자.

각 위치에서 모든 조각을 확인하고, 문자열 비교는 최대 L글자를 확인한다.

• 시간 복잡도: O(N * S * L)
• 공간 복잡도: O(N)

이 문제에서는 N <= 20,000, S <= 100, L <= 5이므로 충분히 빠르게 동작한다.

정리

문자열을 앞에서부터 완성하는 최소 비용 문제로 바꾸면 된다. dp[i]가 도달 가능한 위치인지 확인하고, 그 위치에 이어 붙일 수 있는 단어 조각으로 다음 상태를 갱신한다.

1.1 프로젝트 명과 개발 배경

Windows 실행 파일은 .exe처럼 보이는 확장자만으로 신뢰할 수 없다. 또한 다운로드 폴더나 공유 폴더에는 문서·이미지·실행 파일이 섞여 있어, 사용자는 어떤 파일부터 확인해야 하는지 빠르게 판단하기 어렵다.

이 프로젝트는 이러한 문제를 대상으로 다음 질문에 답하는 보조 분석 서비스를 목표로 한다.

1. 입력 파일이 실제 Windows PE 구조를 가진 파일인가?
2. 유효한 PE라면 1차 정적 특징 기준에서 위험 신호가 낮은가, 의심되는가, 높은가?
3. 위험 방향의 결과라면 2차 원시 바이트 모델은 어떤 악성 유형을 가장 높게 보는가?
4. 두 모델이 불일치하거나 유형 점수가 낮을 때, 이를 단정하지 않고 어떻게 표현할 것인가?

따라서 본 서비스는 Windows Defender·EDR·샌드박스를 대체하는 보안 제품이 아니다. 파일 실행 전 선별과 결과 해석을 돕는 교육용·보조 정적 분석 서비스로 범위를 명확히 둔다.

1.2 문제 정의

초기 문제는 “악성코드를 탐지할 수 있는 분류 모델을 만든다”였지만, 최종적으로는 아래처럼 구체화했다.

정상 파일과 악성 의심 PE 파일을 읽기 전용 정적 분석으로 우선 선별하고, 위험 파일에 한해 유형 분류와 불확실성 정보를 함께 제공한다.

이 정의에는 두 가지 제약을 포함한다.

• 정적 분석만으로 실제 실행 행위, 감염 여부, 정보 유출, 네트워크 통신을 확정하지 않는다.
• 모델의 점수는 보정(calibration) 전 분류 점수이므로 실제 악성 확률이나 자동 차단 근거로 표현하지 않는다.

2. 기획 변경 및 문제 해결 과정

2.1 초기 접근법: 파일 이미지화 기반 분류

초기 기획은 실행 파일의 바이트를 Grayscale 이미지 등 2차원 이미지로 변환한 뒤, 컴퓨터 비전 분류 모델로 정상·악성 또는 악성 유형을 구분하는 방식이었다.

이 방식은 바이트를 0~255의 픽셀 값으로 대응시키고, 일정한 폭으로 줄바꿈해 이미지로 재배열한다. CNN이나 Vision Transformer는 이미지에서 질감·반복 패턴·국소 구조를 학습하므로, 원시 바이트를 즉시 이미지 모델에 넣을 수 있다는 장점이 있다.

초기에는 서로 다른 귀납 편향(inductive bias)을 비교하기 위해 아래 모델을 후보로 검토했다.

2.2 초기 방식의 한계

이미지화 접근은 모델을 빠르게 실험할 수 있지만, 파일 보안 분석 서비스의 핵심 질문에는 다음 한계가 있었다.

특히 Grad-CAM 결과는 “모델이 이미지의 어디에 반응했는가”에는 도움이 되지만, 그 위치가 실제 악성 행위나 특정 악성 코드 패턴이라는 사실을 보장하지 않는다. 이 문제는 단순히 더 높은 정확도의 이미지 모델을 선택해서 해결되지 않는다.

2.3 피드백 수용과 최종 기획 전환

초기 피드백의 핵심은 다음과 같았다.

단순히 파일을 이미지로 변환해 분류하는 방식만으로는 기존 접근과 차별성이 부족하며, 파일 구조와 원시 데이터의 의미를 더 직접적으로 반영할 필요가 있다.

이에 따라 프로젝트는 “이미지 분류 성능 비교”에서 “PE 파일을 안전하게 선별하고, 위험 결과를 단계적으로 해석하는 서비스”로 범위를 재정의했다.

2.4 MalConv2 학습 문제와 데이터 관점의 개선

2차 다중 클래스 모델에서는 원시 바이트 입력의 용량과 클래스 분포가 중요한 과제였다. 원시 PE 파일은 길이가 균일하지 않고, 악성 유형별 표본 수 차이가 클 수 있어 다음 문제가 발생한다.

• 긴 원시 바이트 입력은 GPU 메모리·I/O·학습 시간을 크게 사용한다.
• 다중 클래스 데이터가 불균형하면 빈도가 높은 유형 중심으로 학습되어 macro F1이 낮아질 수 있다.
• 악성 유형별 파일 길이·패킹 방식·수집 시점 차이가 모델이 학습하는 분포에 영향을 준다.

이에 팀은 BODMAS를 2차 분류의 기준 데이터셋으로 유지했다. 현재 웹 서비스에 연결된 2차 모델 번들도 config.json의 DATASET_NAME과 class_mapping.txt 기준으로 BODMAS의 benign 1개와 악성 유형 16개, 총 17개 클래스 체계를 사용한다. 다만 BODMAS 내부에서 유형별 표본 수가 고르지 않아, 일부 클래스가 충분한 학습 신호를 갖지 못하는 문제가 있었다.

이를 보완하기 위해 RawMal-TF를 함께 사용했다. RawMal-TF는 악성코드 타입과 패밀리 라벨이 부여된 원시 Windows PE 악성 바이너리와 사전 추출 특징 벡터를 제공하는 공개 데이터셋이다. 팀은 RawMal-TF에서 BODMAS의 부족한 악성 유형에 대응하는 샘플을 선별하고, BODMAS의 17개 클래스 매핑에 맞춰 라벨을 정렬한 뒤 학습 데이터에 보완했다. 즉, BODMAS는 최종 분류 체계와 번들의 기준이며, RawMal-TF는 클래스 불균형을 완화하기 위한 보완 데이터 소스다.

현재 저장된 2차 학습 실행 정보는 다음을 확인할 수 있다.

3. 시스템 아키텍처 및 파이프라인

3.1 전체 데이터 흐름

flowchart TD
    A[사용자: 단일 파일 또는 폴더 업로드] --> B[FastAPI: 임시 사본 저장]
    B --> C{PE 구조 검증\nMZ / e_lfanew / PE\0\0 / COFF}
    C -- 비 PE --> U[미지원 결과]
    C -- 손상 또는 읽기 실패 --> E[오류 결과]
    C -- 유효 PE --> D[1차 XGBoost\n341개 PE 정적 특징]
    D -- 점수 < 0.10 --> N[Normal]
    D -- 0.10 이상 --> F[2차 MalConv2 계열\n원시 바이트 전체]
    F --> G{최고 클래스 점수}
    G -- 0.70 미만 --> Q[Unknown\n분류 불확실]
    G -- benign --> R[판정 충돌\n1차 위험 / 2차 정상]
    G -- 악성 유형 --> S[악성 유형 의심]
    S --> T[선택: 악성 유형 안내 / Gemini 조언]
    S --> V[선택: 단계적 전체 가림 재추론\n64KB → 4KB → 256B]
    V --> W[상위 3개 영향 후보 구간]

분석 작업은 FastAPI의 메모리 기반 작업 관리자가 순차적으로 처리한다. 폴더 분석에서 한 파일이 비-PE·손상·추론 오류여도 다음 파일의 분석은 계속한다. 프런트엔드는 REST API로 작업을 만들고, 초기 분석 진행률은 SSE(Server-Sent Events)로 수신한다.

3.2 입력 검증과 읽기 전용 정책

분석 전에는 확장자를 신뢰하지 않고 다음을 순서대로 확인한다.

1. DOS 헤더의 MZ 서명
2. DOS 헤더 0x3C 위치의 e_lfanew가 가리키는 PE 헤더 오프셋 범위
3. NT 헤더의 PE\0\0 시그니처
4. COFF 헤더의 Machine 필드

검증 결과는 다음처럼 분리한다.

3.3 1차 모델: XGBoost 기반 정상/악성 선별

채택 이유

1차는 모든 유효 PE 파일을 빠르게 통과하는 관문이므로, 구조화된 정적 특징에 강하고 추론 비용이 낮은 XGBoost를 채택했다.

• 표 형식 특징에 적합: 파일 크기·엔트로피·헤더·섹션·Import/Export·데이터 디렉터리·Overlay·바이트 히스토그램처럼 서로 성격이 다른 수치 특징을 함께 다룰 수 있다.
• 비선형 상호작용 학습: 단일 특징 하나가 아니라 “섹션 수, 엔트로피, 실행 권한, Import 구조” 등의 조합에서 나타나는 위험 신호를 트리 앙상블로 학습한다.
• 실무형 선별 비용에 적합: 2차 원시 바이트 딥러닝보다 가벼워, 모든 PE에 먼저 적용하고 위험 방향 결과만 다음 단계로 넘길 수 있다.
• 특징 계약 검증: 실제 서비스는 학습 당시의 341개 특징 이름·순서·특징 추출기 해시를 번들에서 확인한다. 웹 서버에서 특징을 임의로 다시 구현해 잘못된 열 순서로 예측하는 위험을 줄인다.

341개 정적 특징 구성

XGBoost는 파일의 원시 바이트 전체를 바로 신경망에 넣지 않는다. PE 헤더·섹션·Import/Export·Overlay·바이트 히스토그램에서 만든 341개 수치 특징을 입력으로 받고, 여러 결정트리가 앞선 트리의 오차를 순차적으로 보완한 뒤 위험 점수를 계산한다. 이 점수를 임계값과 비교해 Normal·Suspicious·Malware로 나누며, 뒤의 두 결과만 MalConv2 2차 분류로 전달한다.

1차 라우팅 정책

임계값은 validation 분할에서 결정됐으며, 1차 점수는 확률 보정 전 값이다. 따라서 UI에서는 점수 자체보다 구간 결과와 후속 단계 여부를 중심으로 해석한다.

3.4 2차 모델: MalConv2 계열 악성 유형 분류

채택 이유

2차 모델은 1차에서 Suspicious 또는 Malware로 분류된 파일만 대상으로 한다. 이 단계는 사람이 설계한 PE 정적 특징만으로 충분하지 않을 수 있는 원시 바이트 문맥을 학습해, 세부 유형을 구분하는 역할을 맡는다.

• 원시 바이트 직접 입력: 이미지를 만들거나 수작업 특징만으로 치환하지 않고 파일 바이트 0~255를 직접 입력한다.
• 긴 파일 처리: LowMemConv 계열의 청크 스캔 방식을 사용한다. 현재 65,536 bytes는 파일을 그 길이로 자르는 최대 길이가 아니라 내부 스캔 청크 크기다.
• 다중 클래스 분류: benign과 16개 악성 유형, 총 17개 클래스의 softmax 점수를 계산한다.
• 모델 구조 검증: 가중치·클래스 매핑·설정의 일치를 확인한 뒤 추론한다. 현재 체크포인트에 context_net 가중치가 있으면 Global Context 계열 MalConv 구조를 선택해 strict loading을 수행한다.

MalConv2-GCG는 파일을 이미지로 바꾸지 않고 원시 바이트를 토큰으로 입력한다. Context Path는 파일 전체의 문맥을 요약하고, Feature Path는 1D Convolution으로 지역 바이트 패턴을 추출한다. 전역 문맥 게이트는 유형 분류에 덜 중요한 채널을 낮추고 중요한 패턴을 상대적으로 강조한다. 이후 시간 축 최대 풀링과 완전연결 계층을 거쳐 17개 클래스 점수를 계산한다.

원시 바이트 전처리와 내부 구조

파일 원시 바이트 0~255
        ↓
토큰 1~256으로 이동 (0은 Padding 전용으로 예약)
        ↓
Byte Embedding
        ↓
병렬 1D Convolution / Gating 또는 Context 경로
        ↓
Temporal Max Pooling
        ↓
Fully Connected Layer
        ↓
17개 클래스 Softmax

현재 서비스는 원본 파일 전체 바이트를 보존해 모델 어댑터에 전달한다. 바이트 값에 1을 더해 1~256 토큰으로 바꾸고, 0은 모델의 padding에만 사용한다. 이 처리는 실제 0x00 바이트와 패딩이 혼동되는 문제를 방지한다.

2차 결과와 최종 상태 정책

2차 0.70 임계값은 현재 검증 데이터 기반 확률 보정 전의 Unknown 처리 기준이다. 이 값은 설정 파일로 분리되어 있어 향후 calibration 또는 재평가 결과에 따라 API를 바꾸지 않고 조정할 수 있다.

3.5 결과 해석 보조 기능

악성 유형 의심 결과에는 아래 보조 기능을 제공한다.

3.6 FastAPI 백엔드 구현 구조

백엔드는 React 화면에서 업로드한 단일 파일 또는 폴더 파일 목록을 받아, PE 검증과 두 단계 모델 추론을 순서대로 수행한다. 분석 시간이 파일마다 다를 수 있으므로 업로드 요청을 받자마자 분석 작업 ID를 반환하고, 이후 진행 상태와 결과를 작업 단위로 조회하는 방식으로 구성했다.

React 브라우저
  │ 파일·폴더 업로드 / 작업 상태·진행률 요청
  ▼
FastAPI Router
  │ /api/v1/analyses/file, /folder
  ▼
AnalysisManager
  ├─ PE Validator
  ├─ 1차 XGBoost
  ├─ 필요 시 2차 MalConv2-GCT
  └─ 최종 상태 조합
  ▼
작업 결과 반환 → React 결과 화면

앱 시작점과 API 연결

main.py는 FastAPI 애플리케이션을 만들고, React 개발 서버의 요청을 위한 CORS와 분석 라우터를 등록한다. lifespan에는 프로세스 전체가 공유할 분석 관리자를 연결해 요청마다 모델 관리 객체가 새로 만들어지지 않도록 했다.

app = FastAPI(
    title=resolved_settings.title,
    version=resolved_settings.version,
    lifespan=create_analysis_lifespan(manager_factory),
)

app.add_middleware(CORSMiddleware, ...)
app.include_router(analyze_router, prefix="/api/v1")

분석 작업 생성 API

단일 파일 분석 API는 파일을 받은 뒤 전체 추론이 끝날 때까지 HTTP 연결을 붙잡지 않는다. 대신 작업을 생성해 202 Accepted와 작업 정보를 반환한다. 폴더 분석도 같은 방식으로 여러 파일과 상대 경로를 작업 대기열에 넣는다.

@router.post("/file", response_model=AnalysisJob,
             status_code=status.HTTP_202_ACCEPTED)
async def create_single_file_analysis(
    manager: AnalysisManagerDependency,
    file: UploadFile = File(...),
) -> AnalysisJob:
    return await manager.create_job(
        input_kind="file",
        uploads=[file],
        relative_paths=[file.filename or "uploaded-file"],
    )

읽기 전용 PE 검증

모델 호출 이전에 MZ 서명, DOS 헤더의 PE 오프셋(e_lfanew), PE\0\0 시그니처를 확인한다. 따라서 확장자만 .exe로 변경한 문서 파일은 모델까지 전달되지 않는다.

if file_bytes[:2] != b"MZ":
    return PeValidationResult(is_valid=False)

pe_offset = int.from_bytes(file_bytes[0x3C:0x40], "little")

if file_bytes[pe_offset : pe_offset + 4] != b"PE\x00\x00":
    return PeValidationResult(is_valid=False)

1차·2차 모델 연결과 최종 상태 결정

유효한 PE 파일은 먼저 1차 XGBoost로 분석한다. 1차 결과가 Normal이면 종료하고, Suspicious 또는 Malware이면 원시 바이트 시퀀스를 준비해 2차 MalConv2-GCT로 전달한다. 이후 두 모델 결과를 조합해 Normal, 악성 유형 의심, 판정 충돌, 분류 불확실 중 하나의 최종 상태로 반환한다.

validation = validate_pe(file_bytes)
stage1 = self.stage1.analyze_file(file_path)

if stage1.needs_stage2:
    sequence = prepare_malconv2_byte_sequence(file_bytes)
    stage2 = self._get_stage2()
    family_class, family_confidence, is_unknown = \
        stage2.predict_stage2(sequence)

final_status = self._final_status(
    stage1.result, family_class, is_unknown
)

이 흐름은 main.py, analyze.py, pe_validator.py, analysis_service.py를 기준으로 정리했다.

4. 모델 분석 및 시각화 가이드

4.1 모델별 핵심 구조·장단점 비교

5. 결론 및 회고

5.1 프로젝트를 통해 얻은 기술적 인사이트

1. 입력 표현은 모델 선택만큼 중요하다.

   • 이미지 모델의 성능을 비교하는 것보다, 바이트를 왜 이미지로 바꾸는지와 그 과정에서 어떤 의미가 사라지는지를 먼저 검토해야 했다.

2. 악성코드 분석은 단일 정확도보다 역할 분리가 중요하다.

   • 1차는 빠르게 위험 파일을 선별하고, 2차는 비용이 큰 원시 바이트 다중 클래스 분류를 담당한다. 두 모델의 목적이 다르므로 동일한 지표만으로 선택하지 않았다.

3. 불확실성을 결과로 설계해야 한다.

   • 2차 최고 클래스가 benign인 경우를 단순 정상으로 덮지 않고 판정 충돌로, 최고 점수가 낮은 경우를 Unknown·분류 불확실로 표현했다. 이는 모델 결과를 과신하지 않기 위한 제품 정책이다.

4. 설명 가능성은 곧 행위 증명이 아니다.

   • Grad-CAM이나 단계적 가림 분석은 모델이 반응한 위치를 보조적으로 보여 줄 수 있다. 그러나 해당 위치가 실제 감염 사실이나 악성 행위를 증명한다고 해석해서는 안 된다.

5. 운영 가능한 ML 서비스에는 모델 외 계층이 필요하다.

   • PE 검증, 업로드 임시 파일 관리, 오류 분리, 모델 번들 무결성 검사, 진행률, API 계약, UI 상태 표현, 외부 LLM 전송 최소화가 모델 추론만큼 중요했다.

5.2 향후 발전 방향

5.3 최종 회고

이 프로젝트의 핵심 성과는 “이미지 분류 모델 하나를 학습했다”는 데 있지 않다. 피드백을 통해 초기 접근의 한계를 인정하고, PE 구조 검증 → 1차 선별 → 2차 원시 바이트 분류 → 불확실성 표현 → 방어적 안내라는 전체 분석 경험으로 문제를 재정의했다는 데 있다.

향후에는 동적 분석과 외부 보안 도구 검증을 결합해 실제 악성 행위 판단에 가까워질 수 있다. 그러나 현재 단계에서도 모델 결과를 과장하지 않고, 어떤 파일을 먼저 검토해야 하는지와 무엇을 추가 확인해야 하는지를 제공하는 정적 분석 보조 서비스로서 명확한 역할을 가진다.

이번에는 자연어 처리의 기본 단위인 Corpus, Document, Token, Vocabulary를 살펴보고, 직접 만든 사전과 Hugging Face의 klue/bert-base 토크나이저를 비교한다.

1. 자연어와 자연어 처리

자연어는 사람이 일상에서 말하고 쓰는 언어다. 문법과 어휘뿐 아니라 문맥, 표현 방식, 뉘앙스처럼 규칙만으로 처리하기 어려운 요소를 포함한다.

자연어 처리는 텍스트 또는 음성 데이터를 모델이 다룰 수 있는 형태로 바꾸고, 그 안에서 의미 있는 결과를 찾거나 새로운 문장을 만드는 기술이다.

최근에는 텍스트뿐 아니라 이미지, 음성, 영상과 결합하는 멀티모달 AI로도 확장되고 있다.

2. Corpus와 Document

Corpus

Corpus(말뭉치)는 자연어 처리 모델을 분석하거나 학습하기 위해 모은 전체 텍스트 데이터 집합이다.

예를 들어 영화 리뷰 10만 개를 모았다면, 리뷰 10만 개 전체가 하나의 Corpus가 된다.

Document

Document는 Corpus를 구성하는 개별 텍스트 하나다. 반드시 긴 글일 필요는 없다.

Corpus: 영화 리뷰 10만 개
 ├── Document 1: "재미있고 배우들의 연기가 좋다."
 ├── Document 2: "전개가 느려서 아쉬웠다."
 └── Document 3: "다시 보고 싶은 영화다."

뉴스 기사 하나, 리뷰 하나, 댓글 하나, 짧은 문장 하나도 모두 Document가 될 수 있다.

3. Tokenization

Tokenization(토큰화)은 텍스트를 모델이 처리할 수 있는 작은 단위인 Token으로 나누는 과정이다.

"자연어 처리를 공부합니다."
            ↓ Tokenization
["자연", "##어", "처리", "##를", "공부", "##합니다", "."]
            ↓ Token ID 변환
[3941, 2051, 4211, 2138, 4244, 11800, 18]
            ↓ Tensor 변환
tensor([3941, 2051, 4211, 2138, 4244, 11800, 18])

토큰은 항상 단어 하나와 같지 않다. 단어 전체일 수도 있고, 단어의 일부, 구두점, 공백과 관련된 단위일 수도 있다.

단순 공백 기준 토큰화

가장 간단한 방법은 공백을 기준으로 문장을 나누는 것이다.

text = "나는 자연어를 공부한다"
tokens = text.split()

print(tokens)

['나는', '자연어를', '공부한다']

이 방식은 이해하기 쉽지만, 한국어의 조사나 어미를 분리하지 못하고 구두점 처리도 세밀하지 않다. 예를 들어 자연어를은 하나의 Token으로 남고, 처음 보는 단어는 사전에 없을 가능성이 커진다.

4. Vocabulary와 Token ID

Vocabulary는 토크나이저가 사용할 수 있는 Token의 집합이다. 모델은 문자열 자체가 아니라 Vocabulary에 등록된 정수 ID를 입력으로 받는다.

vocab = {
    "<PAD>": 0,
    "<UNK>": 1,
    "나는": 2,
    "자연어를": 3,
    "공부한다": 4,
}

text = "나는 자연어를 공부한다"
tokens = text.split()

token_ids = [vocab.get(token, vocab["<UNK>"]) for token in tokens]
input_ids = torch.tensor(token_ids, dtype=torch.long)

print("원문:", text)
print("Token:", tokens)
print("Token ID:", token_ids)
print("PyTorch Tensor:", input_ids)
print("dtype:", input_ids.dtype)

실행 결과는 다음과 같다.

원문: 나는 자연어를 공부한다
Token: ['나는', '자연어를', '공부한다']
Token ID: [2, 3, 4]
PyTorch Tensor: tensor([2, 3, 4])
dtype: torch.int64

2, 3, 4라는 숫자 자체에는 언어적 의미가 없다. 이 숫자는 Vocabulary 안에서 특정 Token을 찾기 위한 식별자다. 이후 모델은 이 ID를 Embedding 층에 전달해 학습 가능한 벡터로 바꾼다.

Special Token

모델은 보통 한 번에 여러 문장을 Batch로 처리한다. 문장 길이가 다르면 Tensor 모양을 맞추기 어려우므로 짧은 문장 뒤에 <PAD>를 추가한다.

문장 A: [2, 3, 4]
문장 B: [2, 4]

Padding 후
문장 A: [2, 3, 4]
문장 B: [2, 4, 0]  # 0은 <PAD>

5. OOV 문제와 <UNK>

OOV(Out-Of-Vocabulary)는 입력된 Token이 Vocabulary에 존재하지 않는 상황이다.

vocab = {
    "<PAD>": 0,
    "<UNK>": 1,
    "나는": 2,
    "파이토치를": 3,
    "공부한다": 4,
}

tokens = ["나는", "트랜스포머를", "공부한다"]
token_ids = [vocab.get(token, vocab["<UNK>"]) for token in tokens]

print(token_ids)

[2, 1, 4]

트랜스포머를은 Vocabulary에 없으므로 <UNK>의 ID인 1로 바뀐다. 단어 단위 사전만 사용하면 새로운 단어, 오타, 활용형을 자주 <UNK>로 처리하게 되어 정보가 사라질 수 있다.

노트북 원본은 tokens를 중괄호 {}로 만든 set으로 작성했다. set은 순서를 보장하지 않으므로, 문장의 Token 순서를 유지해야 할 때는 위 예제처럼 대괄호 []를 사용하는 list가 알맞다.

6. Subword Tokenization

현대 NLP 모델은 OOV 문제를 줄이기 위해 Subword Tokenization을 많이 사용한다. 단어 전체를 하나로만 처리하지 않고, 자주 등장하는 부분 단위로 나눈다.

"자연어 처리를 공부합니다."
        ↓
['자연', '##어', '처리', '##를', '공부', '##합니다', '.']

##는 BERT 계열 WordPiece 토크나이저에서 앞 Token에 이어 붙는 부분이라는 표시다. 예를 들어 자연과 ##어를 합치면 자연어가 되고, 공부와 ##합니다를 합치면 공부합니다가 된다.

처음 보는 단어라도 이미 알고 있는 부분 Token의 조합으로 표현할 가능성이 높다. 이 때문에 단어 전체만 사용하는 방식보다 <UNK> 발생을 줄일 수 있다.

7. Hugging Face AutoTokenizer 사용하기

Hugging Face는 자연어 처리뿐 아니라 이미지, 음성, 멀티모달 모델과 관련 도구를 제공하는 생태계다. AutoTokenizer를 사용하면 모델 저장소에 맞는 Tokenizer 설정과 Vocabulary를 불러올 수 있다.

from transformers import AutoTokenizer

# Hugging Face Hub의 klue/bert-base 토크나이저 불러오기
tokenizer = AutoTokenizer.from_pretrained("klue/bert-base")

text = "자연어 처리를 공부합니다."
tokens = tokenizer.tokenize(text)
token_ids = tokenizer.convert_tokens_to_ids(tokens)

print("Token:", tokens)
print("Token ID:", token_ids)

실행 결과는 다음과 같다.

Token: ['자연', '##어', '처리', '##를', '공부', '##합니다', '.']
Token ID: [3941, 2051, 4211, 2138, 4244, 11800, 18]

klue/bert-base의 Vocabulary 크기와 주요 Special Token ID는 다음과 같이 확인했다.

print("Vocabulary 크기:", tokenizer.vocab_size)
print("[PAD] ID:", tokenizer.pad_token_id)
print("[UNK] ID:", tokenizer.unk_token_id)
print("[CLS] ID:", tokenizer.cls_token_id)
print("[SEP] ID:", tokenizer.sep_token_id)

Vocabulary 크기: 32000
[PAD] ID: 0
[UNK] ID: 1
[CLS] ID: 2
[SEP] ID: 3

노트북에서는 [PAD] ID를 출력할 때 tokenizer.pad_token_type_id를 사용했다. 이 값은 이 모델에서 우연히 0으로 출력되지만, PAD Token의 Vocabulary ID를 확인하려면 의미에 맞는 tokenizer.pad_token_id를 사용해야 한다.

tokenize()와 모델 입력 만들기의 차이

tokenizer.tokenize()는 사람이 확인하기 좋은 Token 문자열 목록을 반환한다. 실제 모델 입력을 만들 때는 보통 토큰화와 ID 변환, Special Token 추가, Padding, Attention Mask 생성을 한 번에 처리한다.

encoded = tokenizer(
    "자연어 처리를 공부합니다.",
    return_tensors="pt",
)

print(encoded["input_ids"])
print(encoded["attention_mask"])

input_ids는 모델에 전달할 Token ID이고, attention_mask는 실제 Token과 Padding 위치를 구분하는 값이다. 이처럼 모델용 입력을 만들 때는 tokenizer(...) 호출이 더 편리하다.

8. macOS와 Windows 환경 차이

토큰화와 transformers API는 운영체제와 관계없이 거의 동일하다. 하지만 가상환경 활성화 명령어, 경로 표기, 사용할 수 있는 GPU 가속 방식은 macOS와 Windows에서 다르다.

8.1 한눈에 비교하기

경로를 문자열로 이어 붙이기보다 pathlib.Path를 사용하면 운영체제마다 다른 구분자를 신경 쓰지 않아도 된다.

from pathlib import Path

model_dir = Path("models") / "klue-bert"
print(model_dir)

8.2 공통 준비

transformers와 PyTorch는 프로젝트마다 독립된 가상환경에 설치하는 편이 좋다. 프로젝트마다 라이브러리 버전이 달라도 서로 영향을 주지 않는다.

프로젝트 폴더
├── .venv/          # 현재 프로젝트 전용 Python 환경
├── notebook.ipynb
└── requirements.txt

Hugging Face의 from_pretrained()는 처음 실행할 때 모델 또는 토크나이저 파일을 Hub에서 내려받아 로컬 캐시에 저장한다. 이후에는 같은 버전이 캐시에 있으면 로컬 파일을 사용한다.

8.3 macOS: Apple Silicon과 MPS

Apple Silicon Mac에서는 PyTorch의 MPS(Metal Performance Shaders) 백엔드를 사용할 수 있다. 노트북의 실행 결과도 mps였다.

python3 -m venv .venv
source .venv/bin/activate

python -m pip install --upgrade pip
python -m pip install torch torchvision transformers

설치 후에는 MPS 사용 가능 여부를 확인한다.

import torch

print("MPS 사용 가능:", torch.backends.mps.is_available())
print("MPS 빌드 여부:", torch.backends.mps.is_built())

• is_built()가 False이면 현재 PyTorch 빌드에 MPS 지원이 포함되지 않은 상태다.
• is_built()는 True인데 is_available()이 False이면 현재 장치나 macOS 환경에서 MPS를 사용할 수 없는 상태다.
• MPS를 사용할 수 없으면 cpu로 실행하면 된다. 기능이 틀린 것이 아니라 가속 장치만 달라지는 것이다.

macOS에서 cuda는 NVIDIA CUDA를 뜻하므로 일반적인 Apple Silicon Mac에서는 선택되지 않는다. 따라서 torch.cuda.is_available()이 False여도 오류가 아니다.

8.4 Windows: NVIDIA GPU와 CUDA

Windows에서는 NVIDIA GPU가 있고 CUDA용 PyTorch를 설치했다면 보통 cuda를 사용한다. GPU가 없거나 CPU 전용 PyTorch를 설치했다면 cpu로 실행된다.

py -m venv .venv
.\.venv\Scripts\Activate.ps1

py -m pip install --upgrade pip
py -m pip install transformers

PyTorch는 Windows 환경에서 GPU와 CUDA 버전에 맞는 설치 명령을 선택해야 한다. PyTorch의 설치 선택 화면에서 Windows, Pip, 사용 중인 CUDA 또는 CPU를 고른 뒤 제시된 명령을 사용한다.

# CPU 환경 예시
py -m pip install torch torchvision transformers

# NVIDIA GPU 환경은 PyTorch 공식 선택 화면에서
# 현재 CUDA 조합에 맞는 명령을 확인해 설치한다.

설치가 끝나면 Python에서 GPU를 확인한다.

import torch

print("CUDA 사용 가능:", torch.cuda.is_available())
if torch.cuda.is_available():
    print("GPU 이름:", torch.cuda.get_device_name(0))

torch.cuda.is_available()이 False라면 NVIDIA GPU가 없거나, GPU 드라이버·CUDA 조합과 PyTorch 패키지가 맞지 않거나, CPU 전용 빌드를 설치했을 가능성이 있다. 이 경우 PyTorch 공식 설치 선택 화면에서 환경에 맞는 명령을 다시 확인한다.

8.5 운영체제와 관계없이 동작하는 장치 선택 코드

노트북의 흐름을 조금 보완하면 macOS, Windows, CPU 환경에서 모두 안전하게 사용할 수 있다.

import torch

if torch.cuda.is_available():
    DEVICE = torch.device("cuda")       # 주로 Windows 또는 Linux의 NVIDIA GPU
elif torch.backends.mps.is_available():
    DEVICE = torch.device("mps")        # Apple Silicon Mac
elif hasattr(torch, "xpu") and torch.xpu.is_available():
    DEVICE = torch.device("xpu")        # 지원되는 Intel GPU 환경
else:
    DEVICE = torch.device("cpu")

print("사용 장치:", DEVICE)

hasattr(torch, "xpu") 검사는 PyTorch 버전에 따라 torch.xpu 속성이 없는 환경에서도 오류 없이 다음 조건으로 넘어가게 한다.

9. PyTorch Tensor와 실행 장치

Token ID는 정수 인덱스이므로 torch.long, 즉 일반적으로 torch.int64 자료형 Tensor로 만든다.

input_ids = torch.tensor(token_ids, dtype=torch.long)

노트북에서는 실행 장치를 다음 순서로 선택했고, 현재 환경에서는 mps가 선택됐다.

if torch.cuda.is_available():
    DEVICE = torch.device("cuda")
elif torch.backends.mps.is_available():
    DEVICE = torch.device("mps")
elif hasattr(torch, "xpu") and torch.xpu.is_available():
    DEVICE = torch.device("xpu")
else:
    DEVICE = torch.device("cpu")

print(DEVICE)

mps

mps는 Apple Silicon Mac의 GPU 가속을 위한 PyTorch 장치다. 학습이나 추론 시 모델과 입력 Tensor를 같은 장치에 올려야 한다.

model = model.to(DEVICE)
encoded = {key: value.to(DEVICE) for key, value in encoded.items()}

모델은 mps에 있는데 입력은 CPU Tensor인 상태처럼 장치가 다르면 오류가 발생한다. model과 input_ids, attention_mask 등 입력 Tensor를 모두 같은 DEVICE로 옮긴다.

10. 자주 만나는 환경 오류

가상환경에서 pip가 다른 Python을 가리키는 문제를 줄이려면 pip install ...보다 python -m pip install ... 형식을 사용하는 편이 안전하다. python이 바로 현재 활성화된 가상환경의 Python을 가리키기 때문이다.

11. 전체 흐름 정리

Corpus
  ↓
Document 선택
  ↓
Tokenization
  ↓
Vocabulary에서 Token ID 조회
  ↓
Padding·Attention Mask 생성
  ↓
PyTorch Tensor로 변환
  ↓
NLP 모델 입력

• Corpus는 학습·분석에 사용하는 전체 텍스트 집합이고, Document는 그 안의 개별 텍스트다.
• Token은 모델이 다루는 입력 단위이며, 단어 하나와 항상 같지 않다.
• Vocabulary는 Token과 정수 ID의 대응표다.
• <UNK>는 사전에 없는 Token을 처리하지만, Subword Tokenization은 OOV 문제를 줄이는 데 도움이 된다.
• AutoTokenizer는 모델에 맞는 Vocabulary와 Tokenization 규칙을 불러온다.
• 세밀한 NLP 모델 입력에는 Token ID뿐 아니라 Special Token, Padding, Attention Mask도 함께 필요하다.
• macOS에서는 MPS, Windows의 NVIDIA GPU 환경에서는 CUDA를 우선 확인하고, GPU가 없거나 사용할 수 없으면 CPU로 실행한다.

4 x 4 보드에서 같은 그림 카드 두 장을 선택해 제거한다. 방향키 이동, Ctrl + 방향키 이동, Enter 입력은 각각 1회 조작으로 센다.

현재 커서 위치에서 모든 카드 쌍을 제거하는 최소 조작 횟수를 구한다.

핵심 아이디어

이 문제에는 두 종류의 탐색이 필요하다.

1. BFS: 현재 남은 카드 상태에서 한 커서 위치에서 다른 위치까지의 최소 이동 횟수를 구한다.
2. DFS + 메모이제이션: 어떤 카드 쌍부터 제거할지, 그 카드 쌍에서 어느 카드를 먼저 선택할지 탐색한다.

카드 종류는 최대 6개이므로 제거 순서는 최대 6!개다. 각 쌍은 두 카드 중 어느 쪽을 먼저 선택할지 2가지 경우가 있으므로, 모든 경우를 탐색해도 충분하다.

남은 카드 상태를 비트마스크로 표현하기

카드 숫자는 1부터 6까지다. 어떤 카드 종류가 제거되었는지를 비트로 표현한다.

mask의 k번 비트가 1: 숫자 k 카드 쌍은 이미 제거됨
mask의 k번 비트가 0: 숫자 k 카드 쌍이 아직 남아 있음

보드를 직접 수정하지 않아도 mask만으로 현재 칸에 카드가 남아 있는지 판단할 수 있다.

def has_card(row, col, removed_mask):
    value = board[row][col]
    return value != 0 and (removed_mask & (1 << value)) == 0

Ctrl 이동 처리

Ctrl + 방향키는 해당 방향으로 이동하다가 다음 중 하나를 만나면 멈춘다.

• 가장 가까운 남은 카드
• 보드의 끝

따라서 한 칸씩 전진하면서 카드 또는 경계를 만날 때까지 확인한다.

DFS 상태

dfs(removed_mask, row, col)은 현재 커서 위치와 제거된 카드 상태에서, 남은 카드를 모두 제거하는 최소 조작 횟수다.

아직 남은 카드 종류 하나를 골라 두 장을 제거한다.

• 첫 번째 카드 → 두 번째 카드 순서
• 두 번째 카드 → 첫 번째 카드 순서

두 경우를 모두 계산한다. 카드 한 쌍을 선택하려면 Enter가 두 번 필요하므로 이동 횟수에 2를 더한다.

Python 코드

from collections import deque
from functools import lru_cache


def solution(board, r, c):
    positions = {}

    for row in range(4):
        for col in range(4):
            value = board[row][col]

            if value != 0:
                positions.setdefault(value, []).append((row, col))

    card_numbers = list(positions)
    full_mask = sum(1 << number for number in card_numbers)
    directions = [(-1, 0), (1, 0), (0, -1), (0, 1)]

    def has_card(row, col, removed_mask):
        value = board[row][col]
        return value != 0 and (removed_mask & (1 << value)) == 0

    def ctrl_move(row, col, dr, dc, removed_mask):
        while True:
            next_row = row + dr
            next_col = col + dc

            # 해당 방향의 끝 칸에서 멈춘다.
            if not (0 <= next_row < 4 and 0 <= next_col < 4):
                return row, col

            row, col = next_row, next_col

            if has_card(row, col, removed_mask):
                return row, col

    def move_distance(start_row, start_col, target_row, target_col, removed_mask):
        visited = [[False] * 4 for _ in range(4)]
        visited[start_row][start_col] = True
        queue = deque([(start_row, start_col, 0)])

        while queue:
            row, col, distance = queue.popleft()

            if (row, col) == (target_row, target_col):
                return distance

            for dr, dc in directions:
                # 일반 방향키 이동
                next_row = row + dr
                next_col = col + dc

                if (
                    0 <= next_row < 4
                    and 0 <= next_col < 4
                    and not visited[next_row][next_col]
                ):
                    visited[next_row][next_col] = True
                    queue.append((next_row, next_col, distance + 1))

                # Ctrl + 방향키 이동
                next_row, next_col = ctrl_move(
                    row,
                    col,
                    dr,
                    dc,
                    removed_mask
                )

                if not visited[next_row][next_col]:
                    visited[next_row][next_col] = True
                    queue.append((next_row, next_col, distance + 1))

    @lru_cache(None)
    def dfs(removed_mask, row, col):
        if removed_mask == full_mask:
            return 0

        minimum = float("inf")

        for number in card_numbers:
            bit = 1 << number

            if removed_mask & bit:
                continue

            first, second = positions[number]
            next_mask = removed_mask | bit

            # first 카드를 먼저 선택하는 경우
            first_to_second = (
                move_distance(row, col, *first, removed_mask)
                + move_distance(*first, *second, removed_mask)
                + 2
                + dfs(next_mask, *second)
            )

            # second 카드를 먼저 선택하는 경우
            second_to_first = (
                move_distance(row, col, *second, removed_mask)
                + move_distance(*second, *first, removed_mask)
                + 2
                + dfs(next_mask, *first)
            )

            minimum = min(minimum, first_to_second, second_to_first)

        return minimum

    return dfs(0, r, c)

코드 설명

카드 쌍을 제거하기 전까지 두 카드는 보드에 남아 있어야 한다. 따라서 첫 카드에서 두 번째 카드로 이동할 때도 removed_mask는 아직 바꾸지 않는다.

move_distance(*first, *second, removed_mask)

두 카드를 모두 선택하고 Enter를 두 번 누른 뒤에만 next_mask로 바꾼다.

+ 2
+ dfs(next_mask, *second)

이 순서가 중요하다. 먼저 카드를 제거해 버리면 Ctrl 이동이 실제 게임 규칙과 달라진다.

예시

첫 번째 예시에서 카드 쌍 제거 순서와 각 쌍의 선택 순서에 따라 Ctrl 이동 결과가 달라진다.

예를 들어 어떤 카드 쌍을 먼저 제거하면 이후 빈 칸이 늘어나므로, Ctrl 이동이 더 멀리 이동할 수 있다. DFS는 가능한 제거 순서를 모두 비교하고, BFS는 각각의 현재 보드 상태에서 실제 최소 이동 횟수를 계산한다.

따라서 두 카드가 남아 있는 순서까지 고려한 최솟값을 구할 수 있다.

시간 복잡도

카드 종류를 K라고 하자. K <= 6이다.

DFS 상태는 제거된 카드 조합과 커서 위치로 이루어진다.

• 제거 상태 수: 최대 2^K
• 커서 위치 수: 16
• 한 상태에서 확인할 카드 순서: 최대 2K
• BFS 한 번: 보드 칸이 16개이므로 O(1)

제한된 4 x 4 보드와 최대 6개 카드 쌍에서는 충분히 빠르게 동작한다.

• 시간 복잡도: O(2^K * 16 * K)
• 공간 복잡도: O(2^K * 16)

정리

카드 제거 순서가 Ctrl 이동 경로를 바꾸므로, 제거 순서와 카드 선택 순서를 모두 탐색해야 한다. 제거 상태는 비트마스크로 저장하고, 각 상태의 커서 이동은 BFS로 계산하면 최소 조작 횟수를 구할 수 있다.

트리에서 서로 다른 세 정점 a, b, c를 골랐을 때, 세 쌍의 거리의 중간값을 f(a, b, c)라고 한다.

모든 세 정점 조합 중 f의 최댓값을 구한다.

핵심 아이디어

트리에서 가장 먼 두 정점 사이의 거리를 지름이라고 하자. 어떤 세 정점을 골라도 두 정점 사이의 거리는 지름을 넘을 수 없으므로, f의 최댓값도 지름을 넘을 수 없다.

지름의 양 끝을 u, v, 지름 길이를 D라고 하자.

답이 D가 되는 경우

u에서 거리가 D인 정점이 둘 이상 있다면, 그중 두 정점과 u를 고를 수 있다.

u와 x의 거리 = D
u와 y의 거리 = D
x와 y의 거리 <= D

세 거리의 중간값은 D이므로 답은 D다.

u에서는 가장 먼 정점이 하나뿐이어도, v에서 거리가 D인 정점이 둘 이상이라면 같은 이유로 답은 D다.

답이 D - 1이 되는 경우

지름 양 끝 모두에서 가장 먼 정점이 하나뿐이면, 지름 길이 D를 두 번 포함하는 세 정점을 만들 수 없다. 이때 최댓값은 D - 1이 된다.

따라서 다음만 확인하면 된다.

1. 지름의 한 끝 u를 찾는다.
2. u에서 가장 먼 정점들의 개수를 센다.
3. 그 개수가 2 이상이면 D를 반환한다.
4. 그렇지 않으면 반대 끝 v에서도 같은 검사를 한다.
5. 둘 다 하나뿐이면 D - 1을 반환한다.

지름의 한 끝 찾기

트리의 임의의 정점에서 BFS를 수행해 가장 먼 정점을 찾으면, 그 정점은 지름의 한 끝이 된다.

그 정점에서 다시 BFS를 수행하면 지름 길이와 반대쪽 끝을 찾을 수 있다.

풀이 과정

1. 간선 정보로 인접 리스트를 만든다.
2. 1번 정점에서 BFS를 수행해 지름의 한 끝 u를 찾는다.
3. u에서 BFS를 수행해 지름 길이 D와 최장 거리 정점 개수를 구한다.
4. 최장 거리 정점이 2개 이상이면 D를 반환한다.
5. 반대쪽 지름 끝 v에서 BFS를 수행한다.
6. v에서도 최장 거리 정점이 2개 이상이면 D, 아니면 D - 1을 반환한다.

Python 코드

from collections import deque


def solution(n, edges):
    graph = [[] for _ in range(n)]

    for left, right in edges:
        left -= 1
        right -= 1
        graph[left].append(right)
        graph[right].append(left)

    def bfs(start):
        distance = [-1] * n
        distance[start] = 0
        queue = deque([start])

        farthest_node = start
        max_distance = 0
        farthest_count = 1

        while queue:
            node = queue.popleft()

            for next_node in graph[node]:
                if distance[next_node] != -1:
                    continue

                distance[next_node] = distance[node] + 1
                queue.append(next_node)

                if distance[next_node] > max_distance:
                    max_distance = distance[next_node]
                    farthest_node = next_node
                    farthest_count = 1
                elif distance[next_node] == max_distance:
                    farthest_count += 1

        return farthest_node, max_distance, farthest_count

    # 임의의 정점에서 가장 먼 정점은 지름의 한 끝이다.
    diameter_end, _, _ = bfs(0)

    # 지름의 한 끝에서 지름 길이와 최장 거리 정점 개수를 구한다.
    other_end, diameter, farthest_count = bfs(diameter_end)

    if farthest_count >= 2:
        return diameter

    # 반대쪽 끝에서도 최장 거리 정점이 여러 개인지 확인한다.
    _, _, farthest_count = bfs(other_end)

    if farthest_count >= 2:
        return diameter

    return diameter - 1

예시

예시 1: 일직선 트리

1 - 2 - 3 - 4

지름은 1과 4 사이의 거리 3이다. 지름 끝인 1에서 가장 먼 정점은 4 하나뿐이고, 4에서도 1 하나뿐이다.

따라서 답은 3 - 1 = 2다.

예시 2: 별 모양 트리

1   2
 \ /
  5
 / \
3   4

어떤 잎 정점에서 시작해도 다른 잎 정점 여러 개가 거리 2로 가장 멀다. 지름 길이 2가 두 번 포함되는 세 정점을 만들 수 있으므로 답은 2다.

시간 복잡도

정점 수를 N이라고 하자.

BFS는 간선과 정점을 각각 한 번씩 방문하므로 O(N)이다. BFS를 세 번 수행한다.

• 시간 복잡도: O(N)
• 공간 복잡도: O(N)

재귀 DFS 대신 BFS를 사용하므로, 정점 수가 25만 개인 긴 일자 트리에서도 재귀 깊이 제한에 걸리지 않는다.

정리

최댓값은 지름 길이 D 또는 D - 1 중 하나다. 지름 끝에서 최장 거리 정점이 여러 개인지 확인하면, 세 정점이 지름 길이를 두 번 만들 수 있는지 판단할 수 있다.

처리 시간이 서로 다른 여러 CPU 코어에 작업을 순서대로 배정한다.

• 시작 시각에는 모든 코어가 비어 있으므로, 앞 번호 코어부터 작업을 하나씩 받는다.
• 어떤 코어의 작업이 끝나면 즉시 다음 작업을 받는다.
• 같은 시각에 여러 코어가 비면 번호가 작은 코어부터 작업을 받는다.

n번째, 즉 마지막 작업을 처리하는 코어 번호를 반환한다.

핵심 아이디어

시간 time이 주어졌을 때, 그 시각까지 배정된 작업 수를 계산할 수 있다.

처음 시각에 배정되는 작업 수: 코어 개수
이후 각 코어가 처리한 작업 수: time // 코어 처리 시간

따라서 전체 작업 수는 다음과 같다.

len(cores) + sum(time // core for core in cores)

시간이 증가할수록 이 값은 줄어들지 않는다. 그러므로 n번째 작업이 배정되는 최소 시각을 이분 탐색으로 찾을 수 있다.

마지막 코어 찾기

최소 시각을 time이라고 하자.

time - 1까지 배정된 작업 수를 구하면, time에 동시에 비는 코어들 중 몇 번째 코어가 n번째 작업을 받는지 알 수 있다.

remaining = n - (time - 1초까지 배정된 작업 수)

앞 번호 코어부터 확인하면서 time % core == 0인 코어를 만날 때마다 remaining을 하나씩 줄인다. 값이 0이 되는 코어가 답이다.

풀이 과정

1. 작업 수가 코어 수 이하라면 시작 시각에 배정되므로 답은 n번 코어다.
2. 이분 탐색으로 n번째 작업이 배정되는 최소 시각을 찾는다.
3. 그 직전 시각까지 배정된 작업 수를 구한다.
4. 최소 시각에 비는 코어를 번호순으로 확인해 마지막 작업의 코어를 찾는다.

Python 코드

def solution(n, cores):
    core_count = len(cores)

    # 시작 시각에 앞 번호 코어부터 하나씩 작업을 받는다.
    if n <= core_count:
        return n

    def assigned_work_count(time):
        return core_count + sum(time // core for core in cores)

    left = 0
    right = max(cores) * (n - core_count)

    # n번째 작업이 배정되는 최소 시각을 찾는다.
    while left < right:
        mid = (left + right) // 2

        if assigned_work_count(mid) >= n:
            right = mid
        else:
            left = mid + 1

    time = left
    assigned_before = assigned_work_count(time - 1)
    remaining = n - assigned_before

    # 같은 시각에 비는 코어는 번호순으로 다음 작업을 받는다.
    for index, core in enumerate(cores, start=1):
        if time % core == 0:
            remaining -= 1

            if remaining == 0:
                return index

예시

n = 6, cores = [1, 2, 3]인 경우를 보자.

6번째 작업은 시각 2에 비는 코어 중 두 번째인 2번 코어에 배정되므로 답은 2다.

시간 복잡도

C를 코어 수, T를 탐색 시간의 최댓값이라고 하자.

• 작업 수를 계산하는 데 O(C)

• 이분 탐색은 O(log T)

• 마지막 코어를 찾는 데 O(C)

• 시간 복잡도: O(C log T)

• 공간 복잡도: O(1)

정리

작업 하나씩을 직접 시뮬레이션할 필요가 없다. n번째 작업이 배정되는 시각을 먼저 찾고, 그 시각에 비는 코어를 번호순으로 세면 마지막 작업을 처리하는 코어를 구할 수 있다.

각 칸에 S, L, R이 적힌 격자에서 빛은 현재 칸의 지시에 따라 직진, 좌회전, 우회전한 뒤 다음 칸으로 이동한다. 격자 밖으로 나가면 반대편으로 이어지는 토러스 구조다.

모든 빛의 경로 사이클 길이를 구해 오름차순으로 반환해야 한다.

핵심 아이디어

빛의 경로는 위치만으로 결정되지 않는다. 같은 칸에 있어도 어느 방향으로 들어왔는지에 따라 다음 이동이 달라진다.

따라서 아래 세 값을 하나의 상태로 관리한다.

(행, 열, 방향)

행이 N, 열이 M이면 가능한 상태는 총 N * M * 4개다. 각 상태에서 다음 상태는 정확히 하나로 정해진다.

또한 이동 규칙은 역방향으로도 상태를 하나만 복원할 수 있으므로, 모든 상태는 어떤 경로 사이클에 포함된다. 방문하지 않은 상태에서 시작해 같은 상태로 돌아올 때까지 이동하면 한 사이클의 길이를 셀 수 있다.

이동 규칙

방향을 0: 상, 1: 우, 2: 하, 3: 좌로 정의한다.

dr = [-1, 0, 1, 0]
dc = [0, 1, 0, -1]

현재 칸의 문자에 따라 방향을 바꾼다.

• S: 방향 유지
• L: (direction - 1) % 4
• R: (direction + 1) % 4

다음 위치는 나머지 연산으로 계산하면 격자 경계를 자연스럽게 넘어갈 수 있다.

next_row = (row + dr[direction]) % row_count
next_col = (col + dc[direction]) % col_count

Python 코드

def solution(grid):
    row_count = len(grid)
    col_count = len(grid[0])

    # visited[row][col][direction]
    visited = [
        [[False] * 4 for _ in range(col_count)]
        for _ in range(row_count)
    ]

    dr = [-1, 0, 1, 0]  # 상, 우, 하, 좌
    dc = [0, 1, 0, -1]
    cycles = []

    for start_row in range(row_count):
        for start_col in range(col_count):
            for start_direction in range(4):
                if visited[start_row][start_col][start_direction]:
                    continue

                row = start_row
                col = start_col
                direction = start_direction
                length = 0

                # 시작 상태로 되돌아올 때까지 한 경로를 추적한다.
                while not visited[row][col][direction]:
                    visited[row][col][direction] = True
                    length += 1

                    if grid[row][col] == "L":
                        direction = (direction - 1) % 4
                    elif grid[row][col] == "R":
                        direction = (direction + 1) % 4

                    row = (row + dr[direction]) % row_count
                    col = (col + dc[direction]) % col_count

                cycles.append(length)

    return sorted(cycles)

정당성 설명

각 상태 (행, 열, 방향)에서 현재 칸의 문자와 격자 크기는 고정되어 있으므로, 다음 상태는 정확히 하나로 결정된다.

방문하지 않은 상태에서 이동을 시작하면 유한한 상태 집합 안에서 계속 이동하므로 언젠가 이미 방문한 상태에 도달한다. 이 문제의 이동은 방향 전환과 토러스 이동으로 이루어져 역방향 상태도 유일하게 정할 수 있다. 따라서 경로 중간의 다른 사이클으로 합류하는 경우가 없고, 처음 방문 상태로 되돌아온 구간 전체가 하나의 경로 사이클이다.

알고리즘은 사이클을 추적하며 포함된 모든 상태를 방문 처리한다. 이후 해당 상태에서 다시 시작하지 않으므로 각 상태는 정확히 한 번만 어떤 사이클 길이에 포함된다. 따라서 모든 경로 사이클의 길이를 중복이나 누락 없이 구한다.

복잡도 분석

상태 수는 4 * N * M개다.

• 경로 탐색: O(N * M)
• 결과 정렬까지 포함한 시간 복잡도: O(N * M * log(N * M))
• 공간 복잡도: O(N * M)

각 상태는 한 번만 방문한다. 다만 사이클 길이를 오름차순으로 반환해야 하므로, 최악의 경우 결과 배열 정렬 비용이 추가된다.

주의할 점

• visited[row][col]처럼 위치만 방문 처리하면 서로 다른 방향의 상태가 섞여 잘못된 결과가 나온다.
• 방향 전환은 현재 칸에서 수행한 뒤 이동해야 한다.
• Python에서 음수 인덱스 대신 (direction - 1) % 4를 사용하면 좌회전을 명확하게 표현할 수 있다.
• 행과 열의 경계 이동에는 % row_count, % col_count를 각각 사용해야 한다.

마무리

이 문제는 격자를 그래프로 직접 만들 필요 없이, 위치와 방향을 합친 상태 그래프로 생각하면 간단해진다. 전체 상태를 한 번씩만 순회하는 시뮬레이션으로 모든 빛의 경로 사이클을 찾을 수 있다.