Generation for NLP: Embedding

SeongGyun Hong·2024년 11월 13일
EmbeddingNLPnaverboostcamp

NaverBoostCamp

목록 보기
31/64

1. Embedding

간단한 과정을 말하자면

  • 먼저 토큰화를 진행하고
  • 해당 토큰에 대해서 Continuous 한 Vector로 변환하는 과정을 말하는데
  • 결국 핵심은 이산적인 벡터를 연속적인 Vector Space로 사영하는 과정이라고 하겠다.
  • 의미론적 정보(Semantic Information)가 Encode되어 Vector Representation으로 나타나지는것

임베딩이 왜 필요한가요?
이산 벡터 사이에는 연산이 의미가 없어짐.
Continuous Vector로 변환한 뒤에야 벡터 연산이 의미가 있어짐.
최근 많이 쓰이는 것은 코사인 유사도!
RAG등의 시스템에서도 많이 쓰이는데, 이때 벡터 스페이스로 옮겨온 시멘틱 정보들을 코사인 유사도 등으로 많이 의미적 유사도를 구하곤 함.

2. 임베딩의 중요성

  • IR, QA, STS 등에서 많이 사용된다.
  • 특히 RAG에서 매우 잘 사용된다.
    • PineCone, CromaDB
  • 임베딩의 성능에 따라서 Similarity Search의 성능이 매우 달라진다.

Sentence-BERT(Bi-Encdoder)
문장 두개를 넣고 유사한가요? 하는 과정으로 학습시키는 것

Contrastive Learning(+Sel-Supervised Learning)
Negative Sample, Positive Sample을 사용하여 학습시킴

3. 최근 사용되는 모델

최근에는 사전 학습 된 거대 모델을 많이 사용하려고 함.

  • E5
  • BGE
  • Jina Embedding

Sentence Transformer 라이브러리를 이용하여 많이 쓰임.
jina-embeddings-v2-base-code을 많이 쓰더라...

from sentence_transformers import SentenceTransformer

# 모델 로드
model = SentenceTransformer('sentence-transformers/all-MiniLM-L6-v2')

# 임베딩하고 싶은 문장들
sentences = [
    "이것은 첫 번째 문장입니다.",
    "두 번째 문장을 임베딩합니다.",
    "세 번째 문장도 벡터로 변환됩니다."
]

# 간단하게 임베딩 생성
embeddings = model.encode(sentences)
print(embeddings.shape)  # [3, 384]

이외에도 Transformers를 사용해서 마지막 Hidden State에서 모든 토큰에 대해서 평균으로 임베딩을 획득하거나, EOS또는 첫 토큰을 활용해서 임베딩을 얻을 수 있다.

  • 모델에 토큰 텍스트를 통과 시킨 후 풀링하여 임베딩 벡터를 획득하는 식
  1. 평균풀링방식
from transformers import AutoTokenizer, AutoModel
import torch
import torch.nn.functional as F

# 평균 풀링 함수 정의
def mean_pooling(model_output, attention_mask):
    token_embeddings = model_output[0]
    input_mask_expanded = attention_mask.unsqueeze(-1).expand(token_embeddings.size()).float()
    return torch.sum(token_embeddings * input_mask_expanded, 1) / torch.clamp(input_mask_expanded.sum(1), min=1e-9)

# 모델과 토크나이저 로드
tokenizer = AutoTokenizer.from_pretrained('sentence-transformers/all-MiniLM-L6-v2')
model = AutoModel.from_pretrained('sentence-transformers/all-MiniLM-L6-v2')

# 임베딩할 문장들
sentences = [
    "이것은 첫 번째 문장입니다.",
    "두 번째 문장을 임베딩합니다.",
    "세 번째 문장도 벡터로 변환됩니다."
]

# 토큰화
encoded_input = tokenizer(sentences, padding=True, truncation=True, return_tensors='pt')

# 임베딩 생성
with torch.no_grad():
    model_output = model(**encoded_input)

# 평균 풀링 적용
sentence_embeddings = mean_pooling(model_output, encoded_input['attention_mask'])

# 정규화
sentence_embeddings = F.normalize(sentence_embeddings, p=2, dim=1)
print(sentence_embeddings.shape)  # [3, 384]
  1. CLS 또는 첫 토큰 사용
from transformers import AutoTokenizer, AutoModel
import torch

tokenizer = AutoTokenizer.from_pretrained('bert-base-uncased')
model = AutoModel.from_pretrained('bert-base-uncased')

def get_first_token_embedding(text):
    inputs = tokenizer(text, return_tensors='pt')
    outputs = model(**inputs)
    # 첫 번째 토큰([CLS]) 임베딩 추출
    first_token_embedding = outputs.last_hidden_state[:, 0, :]
    return first_token_embedding
  1. EOS 토큰 사용
    def get_eos_token_embedding(text):
   inputs = tokenizer(text, return_tensors='pt')
   outputs = model(**inputs)
   
   # EOS 토큰 위치 찾기
   eos_token_id = tokenizer.eos_token_id
   eos_position = (inputs.input_ids == eos_token_id).nonzero()[0, 1]
   
   # EOS 토큰 임베딩 추출
   eos_embedding = outputs.last_hidden_state[0, eos_position, :]
   return eos_embedding

4. Embedding 모델에 대한 벤치마크

  • MTEB
    • 사람과의 유사도를 기반으로 평가

5. LLM-based Embedding Model

  • 기존의 S-BERT와 같은 Encoder 기반 모델들의 단점을 극복

  • 성능 면에서 매우 탁월하며 General Information에 강점을 가짐

    어떻게 학습하나요?

    LLM 기반으로 학습 데이터를 수집한다... OpenAI API를 활용하는 때가 많다...

    마지막 토큰에 해당하는 EOS 토큰의 경우 앞에 모든 토큰들에 대해서 어텐션이 걸려있어서 해당 어텐션이 걸린 EOS의 히든벡터를 쓰겠다는 것.
    그런데 그렇다고 날로 생성되는 EOS 를쓰면 되지 않느냐 할 수 있는데, 안 된다...
    추가적으로 Pos Neg 학습을 시켜줘야 함.

아래는 예시

from transformers import GPT2Tokenizer, GPT2Model
import torch
import torch.nn as nn
import torch.nn.functional as F

# 1. 토크나이저 및 모델 초기화
tokenizer = GPT2Tokenizer.from_pretrained("gpt2")
model = GPT2Model.from_pretrained("gpt2")

# 2. Positive 및 Negative 샘플 정의
positive_samples = [
    "오늘 날씨가 참 좋네요.",
    "하늘이 맑고 산책하기 좋은 날이에요.",
    "오늘 같은 날은 나가서 바람 쐬기 좋겠어요.",
    "공원이 정말 상쾌하네요.",
    "햇빛이 따뜻해서 기분이 좋아요."
]

negative_samples = [
    "오늘 회의가 너무 길었어요.",
    "집에 가는 길이 너무 막혀서 힘들었어요.",
    "날씨가 흐려서 우울해요.",
    "비가 와서 나가기 싫어요.",
    "오늘 하루가 피곤했어요."
]

# 3. 문장을 토큰화하여 텐서 형식으로 변환
positive_inputs = [tokenizer(p, return_tensors="pt", add_special_tokens=True) for p in positive_samples]
negative_inputs = [tokenizer(n, return_tensors="pt", add_special_tokens=True) for n in negative_samples]

# 4. NCE Loss와 Optimizer 설정
optimizer = torch.optim.AdamW(model.parameters(), lr=1e-5)
epochs = 3
criterion = nn.CrossEntropyLoss()  # NCE Loss에 활용

# 5. 학습 루프
for epoch in range(epochs):
    total_loss = 0
    model.train()
    
    for pos_input, neg_input in zip(positive_inputs, negative_inputs):
        # Positive와 Negative 샘플 각각에 대해 임베딩 계산
        pos_outputs = model(**pos_input)
        neg_outputs = model(**neg_input)
        
        # EOS 토큰의 위치의 임베딩 추출
        pos_embedding = pos_outputs.last_hidden_state[0, -1, :]
        neg_embedding = neg_outputs.last_hidden_state[0, -1, :]
        
        # NCE Loss 계산을 위해 점수를 정규화
        # Positive 샘플에는 높은 확률을, Negative 샘플에는 낮은 확률을 기대
        scores = torch.cat([pos_embedding.unsqueeze(0), neg_embedding.unsqueeze(0)], dim=0)
        labels = torch.tensor([0]).long()  # Positive 샘플을 정답(0)으로 설정
        
        # NCE Loss 계산
        loss = criterion(scores, labels)
        total_loss += loss.item()
        
        # 역전파 및 최적화
        optimizer.zero_grad()
        loss.backward()
        optimizer.step()
    
    print(f"Epoch {epoch + 1}, Loss: {total_loss / len(positive_samples)}")

# 6. 학습된 임베딩 결과 확인
model.eval()
with torch.no_grad():
    for pos_input, neg_input in zip(positive_inputs, negative_inputs):
        pos_outputs = model(**pos_input)
        neg_outputs = model(**neg_input)
        
        pos_embedding = pos_outputs.last_hidden_state[0, -1, :]
        neg_embedding = neg_outputs.last_hidden_state[0, -1, :]
        
        pos_score = F.cosine_similarity(pos_embedding.unsqueeze(0), pos_embedding.unsqueeze(0)).item()
        neg_score = F.cosine_similarity(pos_embedding.unsqueeze(0), neg_embedding.unsqueeze(0)).item()
        
        print("Positive Sample Similarity:", pos_score)
        print("Negative Sample Similarity:", neg_score)

# NCE Loss를 사용하여 Positive와 Negative 샘플 간의 점수를 학습
# CrossEntropyLoss를 통해 Positive는 0, Negative는 1로 라벨링하고,
# Positive 샘플이 높은 확률을 가지도록 학습시키며, Negative는 낮은 확률을 가지도록 함
# EOS 토큰의 임베딩을 통해 문장의 의미적 정보를 효과적으로 학습할 수 있음.

대부분의 MTEB에서의 상위권 모델들은 Decoder-Only LLM을 튜닝시킨 모델들이다.

6. 한국어 임베딩 성능

  • allganize/RAG-Evaluation-Dataset-KO (RAG Dataset)을 정제하여 벵치마크 생성

  • Marker에서 데이터 정제하여 만든 한국어 벤치마크 데이터셋

7. 임베딩에 따른 RAG 성능 비교

임베딩 성능에 따라서 RAG의 할루시네이션이나 신뢰도 등의 성능이 크게 차이난다.
최근에는 벡터서치가 유일한 답안이 아니라는 전제 하에 BM25와 결합하여서도 많이 씀.

from transformers import RagRetriever, RagTokenizer, RagTokenForGeneration
from rank_bm25 import BM25  # BM25-S 라이브러리가 BM25 모듈을 포함하는 것으로 가정
import torch

# 1. 예시 문서 설정 (작은 데이터셋으로 예시)
# 검색할 문서 집합입니다. 실제 프로젝트에서는 대규모 데이터셋을 사용합니다.
documents = [
    "오늘은 날씨가 맑습니다.",
    "오늘은 비가 내리고 있습니다.",
    "내일은 흐리고 비가 올 예정입니다.",
    "봄에는 꽃이 피고 여름에는 햇볕이 강합니다.",
    "겨울에는 눈이 내리고 날씨가 춥습니다."
]

# 2. BM25-S 기반의 검색기 생성 및 초기화
# 각 문서를 토큰화하여 BM25-S 검색 인덱스에 추가합니다.
# BM25-S는 일반 BM25보다 빠른 검색 속도를 제공합니다.
tokenized_documents = [doc.split(" ") for doc in documents]
bm25 = BM25(tokenized_documents)

# 3. RAG 모델 및 토크나이저 초기화
# RAG 모델을 'rag-token-nq'로 설정하여 질문 응답 생성 작업에 사용할 준비를 합니다.
rag_tokenizer = RagTokenizer.from_pretrained("facebook/rag-token-nq")
rag_model = RagTokenForGeneration.from_pretrained("facebook/rag-token-nq")

# 4. 쿼리 정의 및 BM25-S 기반 초기 검색
query = "내일 날씨가 어떤가요?"
tokenized_query = query.split(" ")
# BM25-S를 통해 쿼리와 관련된 문서를 초기 검색합니다.
# n값을 조절하여 상위 몇 개의 결과를 반환할지 결정할 수 있습니다.
top_k = 3
bm25_scores = bm25.get_scores(tokenized_query)
top_n_indices = sorted(range(len(bm25_scores)), key=lambda i: bm25_scores[i], reverse=True)[:top_k]

# 5. ReRanker를 통한 최종 정렬
# 검색된 상위 문서들을 모델로 Reranking 하여 최종 문서 순서를 결정합니다.
retrieved_docs = [documents[i] for i in top_n_indices]
ranked_docs = []

for doc in retrieved_docs:
    inputs = rag_tokenizer(query, doc, return_tensors="pt", truncation=True)
    relevance_score = rag_model(**inputs, labels=inputs["input_ids"]).loss.item()  # ReRank 점수로 활용
    ranked_docs.append((doc, relevance_score))

# ReRank 점수를 기준으로 정렬 (점수가 낮을수록 좋은 결과)
ranked_docs = sorted(ranked_docs, key=lambda x: x[1])

# 6. RAG 모델을 사용하여 답변 생성
# 최종적으로 ReRanked 문서들을 RAG 모델에 입력으로 넣어 답변을 생성합니다.
context_docs = [doc[0] for doc in ranked_docs]
input_ids = rag_tokenizer(query, context_docs, return_tensors="pt", truncation=True, padding="max_length").input_ids

# RAG 모델로 답변 생성
with torch.no_grad():
    generated_ids = rag_model.generate(input_ids=input_ids)
    answer = rag_tokenizer.batch_decode(generated_ids, skip_special_tokens=True)[0]

# 7. 결과 출력
print("Query:", query)
print("Retrieved Documents:", retrieved_docs)
print("ReRanked Documents:", ranked_docs)
print("Answer:", answer)

# -- 상세 주석 --
# BM25-S는 초기 검색 단계에서 빠른 문서 검색을 제공
# RAG 모델에 직접 적용하기 전
# Reranking을 수행하여 최종적으로 모델의 답변 정확도를 높임
# ReRanking은 각 문서에 대한 모델의 손실을 기반으로 relevance 점수를 측정하여, 검색 정확도를 높이는 방법
# 이후 RAG 모델은 질문(query)과 ReRanked 문서들을 기반으로 최종 답변을 생성하여 반환

대개 Encoder 기반 임베딩 모델은 CLS를... Decoder 기반 임베딩 모델은 EOS를 많이 쓴다.

profile
SeongGyun Hong
헤매는 만큼 자기 땅이다.
이전 포스트

Generation for NLP: Long Context

다음 포스트

Generation for NLP: AI Agent

0개의 댓글