[책스터디] 10-(2). 실습 : 의미검색 구현하기

☀️  Summary 

●  텍스트에 의미를 담아 숫자로 변환하는 방법 : 원핫인코딩, BoW, TF-IDF, Word2Vec(밀집임베딩)
●  문장임베딩방식 : 교차인코더, 바이인코더(BERT,풀링층), Sentence-Transformers 
●  의미검색, 키워드검색, 하이브리드검색 

 

 

 

3.   의미검색 구현하기 

---

 

3.1  의미검색구현하기

 

❍  의미검색

 ↪︎  단순히 키워드 매칭을 통한 검색이 아니라, 밀집 임베딩을 이용해 문장이나 문서의 의미를 고려한 검색을 수행하는 것 

 

❍  faiss

 ↪︎  메타가 개발한 벡터 연산 라이브러리로, 코사인유사도, 유클리드 거리 등 기본적인 방법을 지원할 뿐 아니라, 벡터 검색 속도를 향상해주는 ANN알고리즘도 다양하게 제공한다. 

 

 

❍  실습

 

◇ 1) 모델과 데이터셋 불러오기 

from datasets import load_dataset
from sentence_transformers import SentenceTransformer

# 기사 본문과 기사본문에 관련된 질문을 모은 데이터셋 
klue_mrc_dataset = load_dataset('klue', 'mrc', split='train')

# 한국어 임베딩 모델
sentence_model = SentenceTransformer('snunlp/KR-SBERT-V40K-klueNLI-augSTS')

 

 

◇ 2) 실습 데이터에서 1000개만 선택하고 문장 임베딩으로 변환 

klue_mrc_dataset = klue_mrc_dataset.train_test_split(train_size=1000, shuffle=False)['train']

# 기사 본문데이터를 저장하고 있는 context칼럼을 문장 임베딩 모델의 입력으로 넣어 문장 임베딩으로 변환 
embeddings = sentence_model.encode(klue_mrc_dataset['context'])

embeddings.shape
# 출력 결과
# (1000, 768)

 

 

◇ 3) KNN 검색 인덱스를 생성하고 문장 임베딩 저장

 ↪︎ 데이터를 저장할 테이블을 생성한다고 생각하면 된다.

 ↪︎ 의미검색을 구현하기 위해서는, 검색 쿼리 문장을 문장임베딩으로 변환하고, 인덱스에서 검색하면 된다. 

import faiss

# 인덱스 만들기
index = faiss.IndexFlatL2(embeddings.shape[1])

# 인덱스에 임베딩 저장하기
index.add(embeddings)

 

 

 

◇ 4) 의미검색

 ↪︎ 검색쿼리문장을 문장임베딩 모델의 encode 메서드를 사용해 문장 임베딩으로 변환하고 인덱스의 search 메서드로 쿼리임베딩과 가장 가까운 3개 문서를 반환받는다. 

query = "이번 연도에는 언제 비가 많이 올까?"
query_embedding = sentence_model.encode([query])
distances, indices = index.search(query_embedding, 3)

for idx in indices[0]:
  print(klue_mrc_dataset['context'][idx][:50])

# 출력 결과
# 올여름 장마가 17일 제주도에서 시작됐다. 서울 등 중부지방은 예년보다 사나흘 정도 늦은   (정답)
# 연구 결과에 따르면, 오리너구리의 눈은 대부분의 포유류보다는 어류인 칠성장어나 먹장어, 그 (오답)
# 연구 결과에 따르면, 오리너구리의 눈은 대부분의 포유류보다는 어류인 칠성장어나 먹장어, 그 (오답)

 

 

◇ 5) 의미검색의 한계 

 ↪︎ 키워드가 동일하지 않아도, 의미가 유사하면 내용이 관련 없더라도 검색 결과로 나올 수 있다. 

query = klue_mrc_dataset[3]['question'] # 로버트 헨리 딕이 1946년에 매사추세츠 연구소에서 개발한 것은 무엇인가?
query_embedding = sentence_model.encode([query])
distances, indices = index.search(query_embedding, 3)

for idx in indices[0]:
  print(klue_mrc_dataset['context'][idx][:50])

# 출력 결과
# 태평양 전쟁 중 뉴기니 방면에서 진공 작전을 실시해 온 더글러스 맥아더 장군을 사령관으로 (오답)
# 태평양 전쟁 중 뉴기니 방면에서 진공 작전을 실시해 온 더글러스 맥아더 장군을 사령관으로 (오답)
# 미국 세인트루이스에서 태어났고, 프린스턴 대학교에서 학사 학위를 마치고 1939년에 로체스 (정답)

 

 ↪︎ 로버트헨리딕이 개발한 것에 대한 질문 쿼리에 대해, 관련없는 답변이 먼저 상위에 나오는 결과 (키워드는 동일하지 않지만, 의미상 유사하다고 판단된 경우) ☞ 이러한 한계는 하이브리드검색으로 개선

 

 

 

3.2 라마인덱스에서 Sentence-Transformers 모델 사용하기 

 

from llama_index.core import VectorStoreIndex, ServiceContext
from llama_index.core import Document
from llama_index.embeddings.huggingface import HuggingFaceEmbedding

# 허깅페이스 관련 클래스에 저장해둔 모델을 불러오기 
embed_model = HuggingFaceEmbedding(model_name="snunlp/KR-SBERT-V40K-klueNLI-augSTS")
service_context = ServiceContext.from_defaults(embed_model=embed_model, llm=None)

# 로컬 모델 활용하기
# service_context = ServiceContext.from_defaults(embed_model="local")


text_list = klue_mrc_dataset[:100]['context']
documents = [Document(text=t) for t in text_list]

index_llama = VectorStoreIndex.from_documents(
    documents,
    service_context=service_context,
)

 

 ↪︎  라마인덱스는 Sentence-Transformers의 임베딩 모델을 통합할 수 있는 기능을 지원 

 

 

 

 

 

 

4.   검색 방식을 조합해 성능 높이기

---

 

4.1  키워드 검색방식 : BM25

 

❍  키워드검색

 ↪︎  의미검색과 달리, 동일한 키워드가 많이 포함될수록 유사도를 높게 평가하는 검색방식을 말한다. 

 ↪︎  관련성이 떨어지는 검색 결과가 나타날 가능성이 낮다는 장점이 있지만, 동일한 키워드를 사용하지 않으면 의미가 유사하더라도 검색하지 못한다는 단점이 있다. 따라서 주로 의미검색과 키워드검색을 조합한 하이브리드검색을 활용한다. 

 

 

❍  BM25

 ↪︎  TF-IDF와 유사한 통계기반 스코어링 방법으로, TF-IDF에 문서의 길이에 대한 가중치를 추가한 알고리즘이다. 

 ↪︎  간단하고 계산량이 적으며, 뛰어난 성능을 보여 대표적인 검색엔진인 Elasticsearch의 기본 알고리즘으로 사용되기도 한다. 

 

↪︎  1) 포화효과 고려 : 특정 문서 내에 토큰이 자주 나오더라도 TF 항이 일정 값 이상으로 커지지 않는다. 

↪︎  2) 문서 길이 고려 : 짧은 문서에 토큰 q가 등장한 경우, 더 중요도를 높게 판단한다. 

 

 

4.2 상호순위조합 이해하기 

 

❍  상호순위조합

 ↪︎  하이브리드 검색을 위해서는 통계기반 점수와 임베딩 유사도 점수를 하나로 합쳐야 한다. 그러나, 점수마다 분포가 다르므로 이를 맞춰주는 것이 필요하다. 

 ↪︎  상호순위조합은 각 점수에서의 '순위'를 활용해 점수를 산출한다. 순위에 따라 점수 (1/(k+순위)) 를 부여한다. k는 각 모델에서 고려할 순위이다. (교재 10.17 그림 참고하기) 

 

 

 

 

 

 

5.   하이브리드검색구현

---

 

5.1  BM25구현하기 

 

 

❍  Class정의

 

import math
import numpy as np
from typing import List
from transformers import PreTrainedTokenizer
from collections import defaultdict

class BM25:
  def __init__(self, corpus:List[List[str]], tokenizer:PreTrainedTokenizer):
    self.tokenizer = tokenizer
    self.corpus = corpus
    self.tokenized_corpus = self.tokenizer(corpus, add_special_tokens=False)['input_ids']
    self.n_docs = len(self.tokenized_corpus)
    self.avg_doc_lens = sum(len(lst) for lst in self.tokenized_corpus) / len(self.tokenized_corpus)
    self.idf = self._calculate_idf()
    self.term_freqs = self._calculate_term_freqs()

  def _calculate_idf(self):
    idf = defaultdict(float)
    for doc in self.tokenized_corpus:
      for token_id in set(doc):
        idf[token_id] += 1
    for token_id, doc_frequency in idf.items():
      idf[token_id] = math.log(((self.n_docs - doc_frequency + 0.5) / (doc_frequency + 0.5)) + 1)
    return idf

  def _calculate_term_freqs(self):
    term_freqs = [defaultdict(int) for _ in range(self.n_docs)]
    for i, doc in enumerate(self.tokenized_corpus):
      for token_id in doc:
        term_freqs[i][token_id] += 1
    return term_freqs

  def get_scores(self, query:str, k1:float = 1.2, b:float=0.75):
    query = self.tokenizer([query], add_special_tokens=False)['input_ids'][0]
    scores = np.zeros(self.n_docs)
    for q in query:
      idf = self.idf[q]
      for i, term_freq in enumerate(self.term_freqs):
        q_frequency = term_freq[q]
        doc_len = len(self.tokenized_corpus[i])
        score_q = idf * (q_frequency * (k1 + 1)) / ((q_frequency) + k1 * (1 - b + b * (doc_len / self.avg_doc_lens)))
        scores[i] += score_q
    return scores

  def get_top_k(self, query:str, k:int):
    scores = self.get_scores(query)
    top_k_indices = np.argsort(scores)[-k:][::-1]
    top_k_scores = scores[top_k_indices]
    return top_k_scores, top_k_indices

 

 

 ↪︎  get_scores : 점수 계산 

      ↪︎ idf와 term_freqs를 통해, 검색하려는 쿼리와 각 문서 사이의 점수를 계산 

 

 ↪︎  get_top_k : 상위 k개의 점수와 인덱스 추출 

      ↪︎ 쿼리와 문서 사이의 검수가 가장 높은 k개의 문서의 인덱스와 점수를 반환 

 

 ↪︎ _calculate_term_freqs : 각 토큰이 각 문서 내에서 몇 번 등장하는지 집계 

      ↪︎ self.n_docs 문서수 만큼의 딕셔너리를 만들고 각 문서 내에 어떤 토큰이 몇 번 등장하는지 집계한다. 

 

 

 

❍  BM25 점수 결과 확인

from transformers import AutoTokenizer
tokenizer = AutoTokenizer.from_pretrained('klue/roberta-base')

bm25 = BM25(['안녕하세요', '반갑습니다', '안녕 서울'], tokenizer)
bm25.get_scores('안녕')
# array([0.44713859, 0.        , 0.52354835])

 

 ↪︎  '안녕' 이라는 검색 쿼리에 대해, '반갑습니다'는 일치하는 토큰이 없어 유사도가 0이 된다. 

 

 

 

❍  BM25 검색 결과 한계 

 

# BM25 검색 준비
bm25 = BM25(klue_mrc_dataset['context'], tokenizer)

query = "이번 연도에는 언제 비가 많이 올까?"
_, bm25_search_ranking = bm25.get_top_k(query, 100)

for idx in bm25_search_ranking[:3]:
  print(klue_mrc_dataset['context'][idx][:50])

# 출력 결과
# 갤럭시S5 언제 발매한다는 건지언제는 “27일 판매한다”고 했다가 “이르면 26일 판매한다 (오답)
# 인구 비율당 노벨상을 세계에서 가장 많이 받은 나라, 과학 논문을 가장 많이 쓰고 의료 특 (오답)
# 올여름 장마가 17일 제주도에서 시작됐다. 서울 등 중부지방은 예년보다 사나흘 정도 늦은  (정답)

 

 ↪︎  질문에 대해, 정답이 되는 문장을 세번째로 출력한다. 검색 쿼리 문장과 정답 기사 사이에 일치하는 키워드가 적어 가장 먼저 검색되지 않은 것이다. 

 

 

 

❍  BM25검색 장점

 

query = klue_mrc_dataset[3]['question']  # 로버트 헨리 딕이 1946년에 매사추세츠 연구소에서 개발한 것은 무엇인가?
_, bm25_search_ranking = bm25.get_top_k(query, 100)

for idx in bm25_search_ranking[:3]:
  print(klue_mrc_dataset['context'][idx][:50])

# 출력 결과
# 미국 세인트루이스에서 태어났고, 프린스턴 대학교에서 학사 학위를 마치고 1939년에 로체스 (정답)
# ;메카동(メカドン)                                                      (오답)
# :성우 : 나라하시 미키(ならはしみき)
# 길가에 버려져 있던 낡은 느티나
# ;메카동(メカドン)                                                      (오답)
# :성우 : 나라하시 미키(ならはしみき)
# 길가에 버려져 있던 낡은 느티나

 

 ↪︎  반면, 의미검색에서 한계 예시에서 보였던 쿼리 검색 문장에 대한 결과는 정답 문장을 제일 첫번째로 등장하는 것을 볼 수 있다. 기사 본문을 출력하면 '매사추세츠 연구소'라는 표현이 많이 등장하는데, BM25의 일치하는 키워드 바탕의 관련 기사 검색의 장점을 잘 보여준다. 

 

 

 

 

 

5.2 상호순위조합 구현하기 

 

❍  상호순위조합함수구현

 

from collections import defaultdict

def reciprocal_rank_fusion(rankings:List[List[int]], k=5):
    rrf = defaultdict(float)
    for ranking in rankings:
        for i, doc_id in enumerate(ranking, 1):
            rrf[doc_id] += 1.0 / (k + i)  # 각 문서 인덱스에 순위기반 점수를 더함 
    return sorted(rrf.items(), key=lambda x: x[1], reverse=True) # 점수를 종합한 딕셔너리를 점수에 따라 높은 순으로 정렬해 반환

 

 ↪︎  reciprocal_rank_fusion : 각 검색 방식으로 계산해 정해진 문서의 순위를 입력으로 받아, 상호 순위 조합 점수가 높은 순대로 정렬해 반환 

 ↪︎  rankings 인자 : 여러 검색 방식에서 정해진 유사한 문서의 인덱스 리스트를 입력으로 받는다. 

 ↪︎ 여러 검색 방식의 점수를 종합하고 높은 점수를 받은 순서대로 정렬해 결과를 반환한다. 

 

rankings = [[1, 4, 3, 5, 6], [2, 1, 3, 6, 4]]
reciprocal_rank_fusion(rankings)

# [(1, 0.30952380952380953),
#  (3, 0.25),
#  (4, 0.24285714285714285),
#  (6, 0.2111111111111111),
#  (2, 0.16666666666666666),
#  (5, 0.1111111111111111)]

 

↪︎  예시 데이터로 함수의 구현 결과 확인 

 

 

 

❍  하이브리드검색 구현하기

 

# 의미 검색에서 반복적으로 수행하던 검색쿼리 문장 임베딩 변환과 인덱스 검색 부분을 한번에 수행할 수 있도록 정의한 함수 
def dense_vector_search(query:str, k:int):
  query_embedding = sentence_model.encode([query])
  distances, indices = index.search(query_embedding, k)
  return distances[0], indices[0]

# 검색 쿼리 문장과 상호 순위 조합에 사용할 파라미터 k를 입력으로 받음 
def hybrid_search(query, k=20):
 # 의미검색 수행
  _, dense_search_ranking = dense_vector_search(query, 100)
 # 키워드검색 수행
  _, bm25_search_ranking = bm25.get_top_k(query, 100)

# 두 검색 방식의 순위를 조합하고 결과를 반환 
  results = reciprocal_rank_fusion([dense_search_ranking, bm25_search_ranking], k=k)
  return results

 

 

 

❍  예시데이터에 대한 하이브리드 검색 결과 확인 

 

query = "이번 연도에는 언제 비가 많이 올까?"
print("검색 쿼리 문장: ", query)
results = hybrid_search(query)
for idx, score in results[:3]:
  print(klue_mrc_dataset['context'][idx][:50])

print("=" * 80)
query = klue_mrc_dataset[3]['question'] # 로버트 헨리 딕이 1946년에 매사추세츠 연구소에서 개발한 것은 무엇인가?
print("검색 쿼리 문장: ", query)

results = hybrid_search(query)
for idx, score in results[:3]:
  print(klue_mrc_dataset['context'][idx][:50])

# 출력 결과
# 검색 쿼리 문장:  이번 연도에는 언제 비가 많이 올까?
# 올여름 장마가 17일 제주도에서 시작됐다. 서울 등 중부지방은 예년보다 사나흘 정도 늦은  (정답)
# 갤럭시S5 언제 발매한다는 건지언제는 “27일 판매한다”고 했다가 “이르면 26일 판매한다  (오답)
# 연구 결과에 따르면, 오리너구리의 눈은 대부분의 포유류보다는 어류인 칠성장어나 먹장어, 그 (오답)
# ================================================================================
# 검색 쿼리 문장:  로버트 헨리 딕이 1946년에 매사추세츠 연구소에서 개발한 것은 무엇인가?
# 미국 세인트루이스에서 태어났고, 프린스턴 대학교에서 학사 학위를 마치고 1939년에 로체스 (정답)
# 1950년대 말 매사추세츠 공과대학교의 동아리 테크모델철도클럽에서 ‘해커’라는 용어가 처음 (오답)
# 1950년대 말 매사추세츠 공과대학교의 동아리 테크모델철도클럽에서 ‘해커’라는 용어가 처음 (오답)

 

 

 ↪︎  하이브리드 검색을 사용하니, 검색쿼리문장에 대해 모두 정답 문장이 첫번째 결과로 오는 것을 확인할 수 있다.