Autoregressive Language Model - 한 단어씩 미래를 예측하는 방법

# 가상의 non-autoregressive 모델
입력: "오늘 날씨가"
출력: model.generate_all_at_once(입력)
결과: "좋네요 밖에 나가고 싶어요" (동시에 모든 토큰 생성)

# GPT 같은 autoregressive 모델
입력: "오늘 날씨가"
출력: model.generate_step_by_step(입력)

# 실제 내부 동작
step1: "좋네요" 생성
step2: "좋네요"를 보고 "밖에" 생성
step3: "좋네요 밖에"를 보고 "나가고" 생성
step4: "좋네요 밖에 나가고"를 보고 "싶어요" 생성

# 단순화된 autoregressive 텍스트 생성
def generate_text_autoregressive(prompt, model, max_length=20):
    """
    한 토큰씩 생성하는 autoregressive 방식
    """
    # 1. 프롬프트로 시작
    generated = prompt

    # 2. 최대 길이까지 반복
    for _ in range(max_length):
        # 3. 현재까지 생성된 텍스트를 기반으로 다음 토큰 예측
        next_token = model.predict_next_token(generated)

        # 4. 예측된 토큰을 추가
        generated += " " + next_token

        # 5. 종료 토큰이면 중단
        if next_token == "<END>":
            break

    return generated

# 사용 예시
prompt = "오늘 날씨가"
result = generate_text_autoregressive(prompt, model)
# "오늘 날씨가 좋네요 밖에 나가고 싶어요"

import numpy as np

def predict_next_token_with_probability(context, model, temperature=1.0):
    """
    다음 토큰을 확률적으로 선택
    """
    # 1. 모델이 각 토큰에 대한 확률 분포 반환
    logits = model.get_logits(context)  # [vocab_size]

    # 2. Temperature로 확률 조정
    # temperature가 높을수록 더 다양한 선택
    probabilities = softmax(logits / temperature)

    # 3. 확률 분포에서 샘플링
    next_token = np.random.choice(vocab, p=probabilities)

    return next_token

# 사용 예시
context = "오늘 날씨가"

# Temperature = 0.1 (보수적, 가장 확률 높은 것 선택)
token1 = predict_next_token_with_probability(context, model, temperature=0.1)
# 결과: "좋네요" (거의 항상 같은 결과)

# Temperature = 1.0 (중립적)
token2 = predict_next_token_with_probability(context, model, temperature=1.0)
# 결과: "좋네요", "맑네요", "흐리네요" 등 다양

# Temperature = 2.0 (창의적, 예측 불가능)
token3 = predict_next_token_with_probability(context, model, temperature=2.0)
# 결과: "좋네요", "이상해요", "최고야" 등 매우 다양

def gpt_style_generation(prompt, model, max_tokens=50, temperature=0.7):
    """
    GPT 스타일의 autoregressive 생성
    """
    # 토큰화
    tokens = tokenize(prompt)  # ["오늘", "날씨", "가"]

    # 생성 루프
    for i in range(max_tokens):
        # 1. 현재까지의 모든 토큰을 모델에 입력
        # 중요: 매번 처음부터 다시 계산!
        logits = model.forward(tokens)  # [seq_len, vocab_size]

        # 2. 마지막 위치의 logits만 사용 (다음 토큰 예측용)
        next_token_logits = logits[-1]  # [vocab_size]

        # 3. Temperature 적용 및 샘플링
        probabilities = softmax(next_token_logits / temperature)
        next_token = sample(probabilities)

        # 4. 생성된 토큰 추가
        tokens.append(next_token)

        # 5. 종료 조건 체크
        if next_token == END_TOKEN:
            break

    # 토큰을 텍스트로 변환
    return detokenize(tokens)

# 실제 사용
result = gpt_style_generation(
    "오늘 날씨가",
    model,
    max_tokens=20,
    temperature=0.7
)
print(result)
# "오늘 날씨가 정말 좋네요. 밖에 나가서 산책하기 딱 좋은 날씨예요."

# 매 단계마다 전체 시퀀스를 다시 처리
step1: model.forward(["오늘", "날씨", "가"])           → "좋네요" 예측
step2: model.forward(["오늘", "날씨", "가", "좋네요"])  → "밖에" 예측
step3: model.forward(["오늘", "날씨", "가", "좋네요", "밖에"]) → "나가고" 예측

# ❌ Greedy Decoding (항상 최고 확률 선택)
def greedy_generation(prompt, model):
    tokens = tokenize(prompt)

    for _ in range(max_tokens):
        logits = model.forward(tokens)
        # 항상 가장 높은 확률의 토큰 선택
        next_token = argmax(logits[-1])
        tokens.append(next_token)

    return detokenize(tokens)

# 결과: 항상 똑같은 출력
# "오늘 날씨가 좋습니다. 오늘 날씨가 좋습니다. 오늘 날씨가..."
# (반복에 빠지기 쉬움)

# ✅ Sampling (확률적 선택)
def sampling_generation(prompt, model, temperature=0.8):
    tokens = tokenize(prompt)

    for _ in range(max_tokens):
        logits = model.forward(tokens)
        # 확률 분포에서 샘플링
        probs = softmax(logits[-1] / temperature)
        next_token = np.random.choice(vocab, p=probs)
        tokens.append(next_token)

    return detokenize(tokens)

# 결과: 매번 다른 출력 (더 자연스러움)
# "오늘 날씨가 정말 좋네요. 산책하기 좋은 날이에요."

def top_k_sampling(logits, k=50):
    """
    상위 k개의 토큰만 고려
    """
    # 1. 상위 k개 토큰의 인덱스와 값 추출
    top_k_indices = np.argpartition(logits, -k)[-k:]
    top_k_logits = logits[top_k_indices]

    # 2. 나머지는 확률 0으로 설정
    filtered_logits = np.full_like(logits, -float('inf'))
    filtered_logits[top_k_indices] = top_k_logits

    # 3. 샘플링
    probs = softmax(filtered_logits)
    return np.random.choice(len(logits), p=probs)

def top_p_sampling(logits, p=0.9):
    """
    누적 확률이 p가 될 때까지의 토큰만 고려 (nucleus sampling)
    """
    # 1. 확률로 변환하고 내림차순 정렬
    probs = softmax(logits)
    sorted_indices = np.argsort(probs)[::-1]
    sorted_probs = probs[sorted_indices]

    # 2. 누적 확률 계산
    cumulative_probs = np.cumsum(sorted_probs)

    # 3. p를 초과하는 지점 찾기
    cutoff_index = np.searchsorted(cumulative_probs, p)

    # 4. 선택된 토큰들만 남기기
    selected_indices = sorted_indices[:cutoff_index + 1]
    selected_probs = probs[selected_indices]
    selected_probs /= selected_probs.sum()  # 정규화

    # 5. 샘플링
    return np.random.choice(selected_indices, p=selected_probs)

# 실전 사용
def generate_with_advanced_sampling(prompt, model, method='top_p'):
    tokens = tokenize(prompt)

    for _ in range(max_tokens):
        logits = model.forward(tokens)[-1]

        if method == 'top_k':
            next_token = top_k_sampling(logits, k=50)
        elif method == 'top_p':
            next_token = top_p_sampling(logits, p=0.9)

        tokens.append(next_token)

    return detokenize(tokens)

# 예시 확률 분포
vocab = ["좋네요", "맑네요", "흐리네요", "춥네요", "더워요", "이상해요", ...]
probs = [0.5,     0.3,     0.1,      0.05,    0.03,    0.02,     ...]

# Top-k (k=3): 상위 3개만 고려
# → "좋네요", "맑네요", "흐리네요" 중에서만 선택

# Top-p (p=0.9): 누적 확률 90%까지만 고려
# → "좋네요"(0.5) + "맑네요"(0.3) + "흐리네요"(0.1) = 0.9
# → 이 3개 중에서만 선택 (결과적으로 top-k와 비슷하지만 동적)

import openai

# Autoregressive 특성을 이용한 스트리밍
response = openai.ChatCompletion.create(
    model="gpt-4",
    messages=[{"role": "user", "content": "Python으로 피보나치 수열 구현해줘"}],
    stream=True  # 한 토큰씩 받기
)

for chunk in response:
    # 각 chunk는 새로운 토큰
    token = chunk.choices[0].delta.get("content", "")
    print(token, end="", flush=True)
    # 실시간으로 "def fibonacci(n):" 한 글자씩 출력됨

# ❌ 비효율적인 프롬프트
prompt = """
다음 질문에 답해주세요. 답변은 자세하게 해주세요.
가능한 많은 예시를 들어주세요.

질문: Python에서 리스트와 튜플의 차이는?
"""

# ✅ 효율적인 프롬프트
prompt = "Python 리스트 vs 튜플: 차이점과 사용 사례"

# Autoregressive 모델의 특성상:
# - 입력 토큰이 많으면 → 매 생성 단계마다 더 많은 연산
# - 출력 토큰이 많으면 → 더 많은 생성 단계 필요

# Context window: 4096 토큰 제한
# 입력: 3500 토큰
# 생성 가능: 596 토큰

# ❌ 문제 상황
long_context = "..." * 3500  # 매우 긴 컨텍스트
response = model.generate(long_context)
# 처음엔 빠르지만, 생성할수록 느려짐
# 왜? 매 단계마다 전체 컨텍스트(3500 + 생성된 토큰)를 처리해야 하니까

# ✅ 해결책: 컨텍스트 압축 또는 요약
summary = summarize(long_context)  # 500 토큰으로 압축
response = model.generate(summary)
# 훨씬 빠르고 효율적

# 문제 상황
prompt = "안녕하세요. 저는"
result = model.generate(prompt, temperature=0.0)  # greedy decoding

# 출력:
# "안녕하세요. 저는 안녕하세요. 저는 안녕하세요. 저는..."

# 왜 이런 일이?
# - Temperature가 0이면 항상 최고 확률 선택
# - "안녕하세요. 저는" 다음에 "안녕하세요"가 높은 확률
# - 루프에 빠짐

# ✅ 방법 1: Temperature 높이기
result = model.generate(prompt, temperature=0.7)

# ✅ 방법 2: Repetition Penalty 적용
result = model.generate(prompt, repetition_penalty=1.2)
# 이미 생성된 토큰의 확률을 낮춤

# ✅ 방법 3: Top-p Sampling
result = model.generate(prompt, top_p=0.9)

# 시나리오: 수학 문제 풀이
prompt = "123 + 456 = "

# step 1: "579" (정답!)
# step 2: " 다음" (정답!)
# step 3: " 문제는" (정답!)
# step 4: " 579" (오답! 이전 출력을 반복)
# step 5: " + " (오답이 계속 누적됨)
# step 6: " 579" (완전히 틀린 방향으로...)

# 문제: 한 번의 실수가 이후 모든 생성에 영향

# ✅ 방법 1: 구조화된 출력 강제
# JSON 형식으로 답변 요청
prompt = """
다음 계산의 결과를 JSON으로 답해주세요:
123 + 456 = ?

{
  "calculation": "123 + 456",
  "result":
"""

# ✅ 방법 2: Few-shot Learning
prompt = """
Q: 100 + 200 = ?
A: 300

Q: 50 + 75 = ?
A: 125

Q: 123 + 456 = ?
A:
"""

# 긴 대화에서 발생
conversation = [
    "User: 내 이름은 김철수야",
    "AI: 안녕하세요 김철수님!",
    # ... 100개의 메시지 ...
    "User: 내 이름이 뭐였지?",
    "AI: 죄송하지만 기억이 나지 않네요."  # Context window 초과로 잊어버림
]

# 왜?
# - Context window 제한 (예: 4096 토큰)
# - 초반 대화가 window 밖으로 밀려남

# ✅ 방법 1: 중요한 정보를 시스템 프롬프트에
system_prompt = "사용자의 이름은 김철수입니다."

# ✅ 방법 2: Conversation Summarization
def manage_context(conversation, max_tokens=3000):
    if count_tokens(conversation) > max_tokens:
        # 오래된 대화 요약
        summary = summarize(conversation[:50])
        return [summary] + conversation[50:]
    return conversation

# 문제 상황
prompt = "Python에서 리스트를 정렬하려면"

# 모델이 첫 토큰으로 "Java"를 선택했다고 가정 (매우 낮은 확률이지만 가능)
# step 1: "Java" (실수!)
# step 2: "에서는" (Java로 이어감)
# step 3: "Collections.sort" (완전히 다른 방향)

# Autoregressive 특성상 초반 실수를 되돌릴 수 없음

# ✅ Beam Search: 여러 가능성을 동시에 탐색
def beam_search(prompt, model, beam_width=5):
    # 초기 beam
    beams = [([], 0.0)]  # (토큰들, 누적 로그 확률)

    for step in range(max_length):
        candidates = []

        # 각 beam에서 가능한 다음 토큰들 생성
        for tokens, score in beams:
            logits = model.forward(tokens)
            top_tokens = get_top_k(logits, k=beam_width)

            for token, prob in top_tokens:
                new_tokens = tokens + [token]
                new_score = score + log(prob)
                candidates.append((new_tokens, new_score))

        # 상위 beam_width개만 유지
        beams = sorted(candidates, key=lambda x: x[1], reverse=True)[:beam_width]

    # 최고 점수의 시퀀스 반환
    return beams[0][0]

# 여러 가능성을 동시에 고려해서 초반 실수 방지

# 문제: 긴 응답을 생성할수록 느려짐
import time

def measure_generation_speed(prompt, model, max_tokens=100):
    tokens = tokenize(prompt)
    timings = []

    for i in range(max_tokens):
        start = time.time()
        logits = model.forward(tokens)  # 매번 전체 시퀀스 처리!
        next_token = sample(logits[-1])
        tokens.append(next_token)
        timings.append(time.time() - start)

    # 결과
    # step 1: 0.05초 (10 토큰 처리)
    # step 10: 0.10초 (20 토큰 처리)
    # step 50: 0.50초 (60 토큰 처리)
    # 점점 느려짐!

# ✅ KV Cache를 사용한 최적화
def generate_with_kv_cache(prompt, model, max_tokens=100):
    tokens = tokenize(prompt)

    # 초기: 전체 프롬프트 처리하고 cache 저장
    cache = model.forward_and_cache(tokens)

    for i in range(max_tokens):
        # 이전 토큰들은 cache 사용, 마지막 토큰만 처리
        logits, cache = model.forward_with_cache(tokens[-1:], cache)
        next_token = sample(logits[-1])
        tokens.append(next_token)

    # 결과
    # 모든 step이 거의 같은 시간 (0.05초)
    # 엄청난 속도 향상!

# Autoregressive (GPT)
input:  "오늘 날씨가 [?]"
output: "좋네요" → "밖에" → "나가고" (순차 생성)

# BERT (양방향)
input:  "오늘 날씨가 [MASK] 나가고 싶어요"
output: "좋아서" (양방향 문맥 고려, 한 번에 예측)

# Autoregressive (순차 생성)
step1: 토큰1 생성
step2: 토큰1 기반 → 토큰2 생성
step3: 토큰1,2 기반 → 토큰3 생성

# Diffusion (반복 개선)
step1: 노이즈에서 시작
step2: 조금 더 명확하게
step3: 조금 더 명확하게
...
step100: 최종 이미지

# Speculative Decoding
# 아이디어: 작은 모델로 여러 토큰을 빠르게 예측하고,
#          큰 모델로 검증

def speculative_decoding(prompt, small_model, large_model, k=5):
    tokens = tokenize(prompt)

    while len(tokens) < max_length:
        # 1. 작은 모델로 k개 토큰 빠르게 생성
        candidates = small_model.generate(tokens, k=k)
        # 예: ["좋네요", "밖에", "나가고", "싶어요", "."]

        # 2. 큰 모델로 한 번에 검증
        logits = large_model.forward(tokens + candidates)

        # 3. 각 위치에서 실제로 선택할 토큰 결정
        accepted = []
        for i, candidate in enumerate(candidates):
            if should_accept(logits[len(tokens) + i], candidate):
                accepted.append(candidate)
            else:
                break

        # 4. 수락된 토큰만 추가
        tokens.extend(accepted)

        if len(accepted) < k:
            break

    return tokens

# 효과: 이론적으로 2-3배 빠름

# NAT (Non-Autoregressive Transformer)
# 한 번에 전체 시퀀스 생성 시도

def nat_generation(prompt, model):
    # 1. 길이 예측
    predicted_length = model.predict_length(prompt)

    # 2. 전체 위치에 대해 동시에 예측
    tokens = model.generate_all_positions(
        prompt,
        length=predicted_length
    )

    return tokens

# 장점: 매우 빠름 (병렬 처리)
# 단점: 품질이 autoregressive보다 떨어짐
#      (문맥 일관성 부족)

# ❌ 비용 낭비
for item in items:
    response = model.generate(f"Summarize: {item}")
    # 매번 새로운 API 호출 = 비쌈

# ✅ 배치 처리
batch_prompt = "\n\n".join([f"Item {i}: {item}" for i, item in enumerate(items)])
response = model.generate(f"Summarize each item:\n{batch_prompt}")
# 한 번의 API 호출로 모두 처리

# ❌ 비효율적
prompt = """
당신은 도움이 되는 AI 어시스턴트입니다.
사용자의 질문에 정확하고 자세하게 답변해주세요.
가능한 많은 예시를 들어주세요.
설명은 초보자도 이해할 수 있게 해주세요.

질문: Python에서 리스트 컴프리헨션이란?
"""
# 토큰 낭비

# ✅ 효율적
prompt = "Python 리스트 컴프리헨션 설명 (초보자용, 예시 포함)"
# 같은 의미, 토큰 절약

# 사용자 경험 개선
def stream_response(prompt, model):
    print("답변 생성 중: ", end="")

    for token in model.generate_stream(prompt):
        print(token, end="", flush=True)
        # 실시간으로 토큰 출력
        # 사용자는 기다림이 덜 지루함

    print()  # 줄바꿈

class ConversationManager:
    def __init__(self, max_tokens=4000):
        self.max_tokens = max_tokens
        self.messages = []

    def add_message(self, role, content):
        self.messages.append({"role": role, "content": content})

        # 토큰 수 체크
        if self.count_tokens() > self.max_tokens:
            self._compress()

    def _compress(self):
        # 오래된 메시지 요약 또는 제거
        if len(self.messages) > 10:
            # 처음 5개 메시지 요약
            summary = self._summarize(self.messages[:5])
            self.messages = [summary] + self.messages[5:]

    def count_tokens(self):
        return sum(len(m["content"].split()) for m in self.messages)