LoGO 해외로고 프로젝트 - RAG 2.

LoGO 해외로고, 해외진출을 희망하는 대한민국 기업을 위한 정보 검색 서비스에서 RAG 부분

두번째 게시물이다.

 

https://github.com/khw11044/KT_BIGPRO_RAG

GitHub - khw11044/KT_BIGPRO_RAG

 

위 깃헙링크에서 코드를 따라하면 되겠다.

 

해당 게시물에서는 빈 프로젝트 폴더에서 시작해서 하나하나 코딩을 해본다.

두번째 게시물은 RAG pipeline을 시작하고 retriever를 선정한다.

1. API Key를 로드한다.

# API KEY를 환경변수로 관리하기 위한 설정 파일
from dotenv import load_dotenv
# API KEY 정보로드
load_dotenv()

2. RAG Pipeline 1

RAG Pipeline을 위해 필요한 라이브러리를 임포트한다.

# langchain_openai에서 ChatOpenAI(LLM)과 OpenAIEmbeddings(임베딩모델: text를 vector화하는 모델)을 load
from langchain_openai import ChatOpenAI, OpenAIEmbeddings

# PDF파일등 데이터를 Chroma형식의 vectorDB에 저장하고 리트리버가 수집한 데이터에 접근하기 위해 Chroma를 load
from langchain_community.vectorstores import Chroma

# 우리가 만든 config.py에서 모델등 옵션들을 수정
from utils.config import config, metadata_field_info

 

다음 리트리버가 검색할 VectorDB를 정의하고 리트리버를 정한다.

 

class Ragpipeline:
    def __init__(self):
        # chatGPT API를 통해 llm 모델 로드
        self.llm = ChatOpenAI(
            model=config["llm_predictor"]["model_name"],  # chatgpt 모델 이름
            temperature=config["llm_predictor"]["temperature"],  # 창의성 0~1
        )

        self.vector_store   = self.init_vectorDB()
        self.retriever      = self.init_retriever()  

    def init_vectorDB(self, persist_directory=config["chroma"]["persist_dir"]):
        """vectorDB 설정"""
        embeddings = OpenAIEmbeddings(model=config["embed_model"]["model_name"])  
        vector_store = Chroma(
            persist_directory=persist_directory, 
            embedding_function=embeddings,                      
            collection_name = 'india',                          
            collection_metadata = {'hnsw:space': 'cosine'},    
        )
        return vector_store

    def init_retriever(self):
        """ Retriever 초기화 """               
        # base retriever 1         
        retriever = self.vector_store.as_retriever(
            search_type   = "similarity",                       # similarity, score_threshold, mmr 
            search_kwargs = {"k": config["retriever_k"]},

        )

        return retriever

 

여기서 init_retriever는 크게 3가지를 사용해볼 수 있다.

 

1. similarity

def init_retriever(self):
        """ Retriever 초기화 """               
        # base retriever 1         
        retriever = self.vector_store.as_retriever(
            search_type   = "similarity",                       # similarity, score_threshold, mmr 
            search_kwargs = {"k": config["retriever_k"]},

        )

        return retriever

 

2. similarity_score_threshold

 

def init_retriever(self):
        """ Retriever 초기화 """               
        # base retriever 2 
        retriever = self.vector_store.as_retriever(
            search_type="similarity_score_threshold",
            search_kwargs={"score_threshold": 0.75,
                           "k": config["retriever_k"]}
        )

 

3. mmr

 

def init_retriever(self):
        """ Retriever 초기화 """               
        # base retriever 3 
        retriever = self.vector_store.as_retriever(
            search_type="mmr",                                              # mmr 검색 방법으로 
            search_kwargs={'fetch_k': 10, "k": 5, 'lambda_mult': 0.4},      # 상위 10개의 관련 context에서 최종 5개를 추리고 'lambda_mult'는 관련성과 다양성 사이의 균형을 조정하는 파라메타 default 값이 0.5

        )

 

자 그러면, '인도 통관 및 운송'에 대해 검색을 해보자.

pipeline = Ragpipeline()
base_retriever = pipeline.init_retriever()

query = '인도 통관 및 운송'   # 질문할 문장
result = base_retriever.get_relevant_documents(query)

for i, doc in enumerate(result):
    print(i)
    print(f"문서 내용: {doc.page_content}") # 문서 내용 표시
    print(f'출처 : {doc.metadata}')
    print('---'*10)

 

통관 및 운송에 대해 컨텐츠를 잘 가져오는것을 확인할 수 있다.

3. RAG Pipeline 2

다음은 Multi-Query와 Self-Query retriever를 사용해 본다.

• Multi-Query Retriever는 llm을 이용해 하나의 질문에서 여러개의 질문을 파생하여 관련 컨텐츠를 찾아낸다.
• Self Querying Retriever는 metadata를 이용해서 필터링된 정보를 찾아낸다.

관련된 블로그를 참고해 보자.

https://asidefine.tistory.com/298

LangChain RAG Retriever 방법 정리 (Multi-Query, Parent Document, Ensemble Retriever, ... )

 

langchain 라이브러리에서 해당 리트리버들을 임포트한다.

 

from langchain.retrievers.multi_query import MultiQueryRetriever
from langchain.retrievers.self_query.base import SelfQueryRetriever

 

다음 base retriever를 mmr로 고정하고 이를 기반으로 MultiQueryRetriever와 SelfQueryRetriever를 정한다.

 

class Ragpipeline:
    def __init__(self):
        # chatGPT API를 통해 llm 모델 로드
        self.llm = ChatOpenAI(
            model=config["llm_predictor"]["model_name"],  # chatgpt 모델 이름
            temperature=config["llm_predictor"]["temperature"],  # 창의성 0~1
        )

        # 초기화 리스트들 

        self.vector_store   = self.init_vectorDB()                  # 1. RAG가 접근할 vectorDB를 초기화합니다.
        self.retriever      = self.init_retriever()                 # 2. LLM이 질문에 대한 답변을 생성하기 전 질문에 관련된 컨텐츠 기반 답변을 생성하기 위해, 컨텐츠를 검색해 찾는 리트리버를 초기화 합니다. 
        self.mq_retriever   = self.init_multi_query_retriever()
        self.sq_retriever   = self.init_self_query_retriever()

    def init_vectorDB(self, persist_directory=config["chroma"]["persist_dir"]):
        """vectorDB 설정"""
        embeddings = OpenAIEmbeddings(model=config["embed_model"]["model_name"])  # VectorDB에 저장될 데이터를 임베딩할 모델을 선언합니다.
        vector_store = Chroma(
            persist_directory=persist_directory,  # 기존에 vectordb가 있으면 해당 위치의 vectordb를 load하고 없으면 새로 생성합니다.
            embedding_function=embeddings,                      # 새롭게 데이터가 vectordb에 넣어질때 사용할 임베딩 방식을 정합니다, 저희는 위에서 선언한 embeddings를 사용합니다.
            collection_name = 'india',                          # india라는 이름을 정해줌으로써 나중에 vector store 관리 가능 
            collection_metadata = {'hnsw:space': 'cosine'},     # cosine 말고 l2 가 default / collection_metadata를 통해 유사도 검색에 사용될 공간('hnsw:space')을 'cosine'으로 지정하여, 코사인 유사도를 사용
        )
        return vector_store

    def init_retriever(self):
        """ Retriever 초기화 """               

        # base retriever 3 
        retriever = self.vector_store.as_retriever(
            search_type="mmr",                                              # mmr 검색 방법으로 
            search_kwargs={'fetch_k': 10, "k": 5, 'lambda_mult': 0.4},      # 상위 10개의 관련 context에서 최종 5개를 추리고 'lambda_mult'는 관련성과 다양성 사이의 균형을 조정하는 파라메타 default 값이 0.5

        )

        return retriever

    # MultiQueryRetriever 생성
    def init_multi_query_retriever(self):
        """사용자의 질문을 여러 개의 유사 질문으로 재생성 """
        retriever_from_llm = MultiQueryRetriever.from_llm(
            llm=self.llm,
            retriever=self.retriever
        )

        return retriever_from_llm

    # SelfQueryRetriever 생성
    def init_self_query_retriever(self):
        """metadata를 이용해서 필터링해서 정보를 반환"""
        document_content_description = "Report or Explain"      # document_content_description: 문서 내용 설명
        retriever = SelfQueryRetriever.from_llm(    
            self.llm,
            self.vector_store,
            document_content_description,
            metadata_field_info,                                # metadata_field_info: 메타데이터 필드 정보
            verbose = True
        )

        return retriever

 

이때, metadata_field_info는 utils폴더의 config.py 파일에 있다.

 

from utils.config import config, metadata_field_info

 

config.py

 

metadata_field_info = [
    AttributeInfo(
        name="category",
        description="The category of the documents. One of ['1.법률 및 규제', '2.경제 및 시장 분석', '3.정책 및 무역', '4.컨퍼런스 및 박람회, 전시회']",
        type="string",
    ),
    AttributeInfo(
        name="filename",
        description="The name of the document",
        type="string",
    ),
    AttributeInfo(
        name="page",
        description="The page of the document",
        type="integer",
    ),
    AttributeInfo(
        name="year",
        description="The Date the document was uploaded",
        type="integer",
    )
]

 

우리는 벡터 스토어에 컨텐츠와 함께 메타데이터를 넣을 때 위와 같이 카테고리, 파일 이름, 페이지, 년도의 정보를 함께 넣어주었다.
기존 방식이 유사 문서를 탐색하는 것과 달리, Self Querying 방식은 metadata를 이용해서 필터링해서 정보를 반환해주기 때문에, 빠르게 필터링된 내용을 가져 올 수 있다.

그럼 이제 한번 사용해보자.

먼저 Multi-Query Retriever

 

pipeline = Ragpipeline()

mq_retriever = pipeline.init_multi_query_retriever()

result = mq_retriever.get_relevant_documents(query)

for i, doc in enumerate(result):
    print(i)
    print(f"문서 내용: {doc.page_content}") # 문서 내용 표시
    print(f'출처 : {doc.metadata}')
    print('---'*10)

 

관련 내용을 잘 가져옴을 확인할 수 있다.

 

다음은 Self-Query Retriever

 

sq_retriever = pipeline.init_self_query_retriever()

result = sq_retriever.get_relevant_documents(query)

for i, doc in enumerate(result):
    print(i)
    print(f"문서 내용: {doc.page_content}") # 문서 내용 표시
    print(f'출처 : {doc.metadata}')
    print("---" * 50)

4. RAG Pipeline 3

다음으로 Ensemble Retriever를 사용해본다.

EnsembleRetriever는 여러 retriever를 입력으로 받아 get_relevant_documents() 메서드의 결과를 앙상블하고, Reciprocal Rank Fusion 알고리즘을 기반으로 결과를 재순위화합니다.

서로 다른 알고리즘의 장점을 활용함으로써, EnsembleRetriever는 단일 알고리즘보다 더 나은 성능을 달성할 수 있습니다.

가장 일반적인 패턴은 sparse retriever (예: BM25)와 dense retriever (예: embedding similarity)를 결합하는 것인데, 이는 두 retriever의 장점이 상호 보완적이기 때문입니다. 이를 "hybrid search" 라고도 합니다.

Sparse retriever는 키워드를 기반으로 관련 문서를 찾는 데 효과적이며, dense retriever는 의미적 유사성을 기반으로 관련 문서를 찾는 데 효과적입니다.

앙상블 리트리버를 사용하기 위해 VectorDB가 아닌 단어 그대로가 유지된 데이터가 필요하다. 따라서 같은 청크 사이즈로 잘라 pickle 파일로 관리한 DB를 불러와 사용한다.

필요 라이브러리를 불러온다.

# Ensemble retriever
from langchain.retrievers import EnsembleRetriever
from langchain_community.retrievers import BM25Retriever
import pickle

 

pickle파일의 내용을 불러오고 BM25 리트리버를 준비시킨다. 검색 결과의 개수는 1로 설정한다.

 

def init_bm25_retriever(self):

        all_docs = pickle.load(open(config["pkl_path"], 'rb'))              # pkl파일에 내용들에서 bm25

        bm25_retriever = BM25Retriever.from_documents(all_docs)
        bm25_retriever.k = 1                                                # BM25Retriever의 검색 결과 개수를 1로 설정합니다.

        return bm25_retriever

 

그러면 이제 앙상블 리트리버를 정의해보자.

 

def init_ensemble_retriever(self):

        bm25_retriever = self.bm25_retriever
        chroma_retriever = self.retriever

        ensemble_retriever = EnsembleRetriever(
            retrievers=[bm25_retriever, chroma_retriever],
            weights=[0.4, 0.6],
            search_type=config["ensemble_search_type"],  # mmr
        )

        return ensemble_retriever

 

서로 다른 2개의 리트리버를 불러와서 가중치와 search_type을 정한다.

그럼 최종 코드를 작성한다.

 

class Ragpipeline:
    def __init__(self):
        # chatGPT API를 통해 llm 모델 로드
        self.llm = ChatOpenAI(
            model=config["llm_predictor"]["model_name"],  # chatgpt 모델 이름
            temperature=config["llm_predictor"]["temperature"],  # 창의성 0~1
        )

        # 초기화 리스트들 

        self.vector_store   = self.init_vectorDB()                  # 1. RAG가 접근할 vectorDB를 초기화합니다.
        self.retriever      = self.init_retriever()                 # 2. LLM이 질문에 대한 답변을 생성하기 전 질문에 관련된 컨텐츠 기반 답변을 생성하기 위해, 컨텐츠를 검색해 찾는 리트리버를 초기화 합니다. 
        self.bm25_retriever = self.init_bm25_retriever()
        self.ensemble_retriever = self.init_ensemble_retriever()
        self.mq_retriever   = self.init_multi_query_retriever()
        self.sq_retriever   = self.init_self_query_retriever()

    def init_vectorDB(self, persist_directory=config["chroma"]["persist_dir"]):
        """vectorDB 설정"""
        embeddings = OpenAIEmbeddings(model=config["embed_model"]["model_name"])  # VectorDB에 저장될 데이터를 임베딩할 모델을 선언합니다.
        vector_store = Chroma(
            persist_directory=persist_directory,                # 기존에 vectordb가 있으면 해당 위치의 vectordb를 load하고 없으면 새로 생성합니다.
            embedding_function=embeddings,                      # 새롭게 데이터가 vectordb에 넣어질때 사용할 임베딩 방식을 정합니다, 저희는 위에서 선언한 embeddings를 사용합니다.
            collection_name = 'india',                          # india라는 이름을 정해줌으로써 나중에 vector store 관리 가능 
            collection_metadata = {'hnsw:space': 'cosine'},     # cosine 말고 l2 가 default / collection_metadata를 통해 유사도 검색에 사용될 공간('hnsw:space')을 'cosine'으로 지정하여, 코사인 유사도를 사용
        )
        return vector_store

    def init_retriever(self):
        """ Retriever 초기화 """               

        # base retriever 3 
        retriever = self.vector_store.as_retriever(
            search_type="mmr",                                              # mmr 검색 방법으로 
            search_kwargs={'fetch_k': 10, "k": 5, 'lambda_mult': 0.4},      # 상위 10개의 관련 context에서 최종 5개를 추리고 'lambda_mult'는 관련성과 다양성 사이의 균형을 조정하는 파라메타 default 값이 0.5

        )

        return retriever

    def init_bm25_retriever(self):

        all_docs = pickle.load(open(config["pkl_path"], 'rb'))              # pkl파일에 내용들에서 bm25

        bm25_retriever = BM25Retriever.from_documents(all_docs)
        bm25_retriever.k = 1                                                # BM25Retriever의 검색 결과 개수를 1로 설정합니다.

        return bm25_retriever

    def init_ensemble_retriever(self):

        bm25_retriever = self.bm25_retriever
        chroma_retriever = self.retriever

        ensemble_retriever = EnsembleRetriever(
            retrievers=[bm25_retriever, chroma_retriever],
            weights=[0.4, 0.6],
            search_type=config["ensemble_search_type"],  # mmr
        )

        return ensemble_retriever

    # MultiQueryRetriever 생성
    def init_multi_query_retriever(self):
        """사용자의 질문을 여러 개의 유사 질문으로 재생성 """
        retriever_from_llm = MultiQueryRetriever.from_llm(
            llm=self.llm,
            retriever=self.retriever
        )

        return retriever_from_llm

    # SelfQueryRetriever 생성
    def init_self_query_retriever(self):
        """metadata를 이용해서 필터링해서 정보를 반환"""
        document_content_description = "Report or Explain"      # document_content_description: 문서 내용 설명
        retriever = SelfQueryRetriever.from_llm(    
            self.llm,
            self.vector_store,
            document_content_description,
            metadata_field_info,                                # metadata_field_info: 메타데이터 필드 정보
            verbose = True
        )

        return retriever

 

그럼 앙상블 리트리버를 사용해보자.

 

pipeline = Ragpipeline()
ensemble_retriever = pipeline.init_ensemble_retriever()

query = '인도 통관 및 운송'   # 질문할 문장

result = ensemble_retriever.get_relevant_documents(query)

for i, doc in enumerate(result):
    print(i)
    print(f"문서 내용: {doc.page_content}") # 문서 내용 표시
    print(f'출처 : {doc.metadata}')
    print("---" * 50)

 

잘 검색하는 거 같다.

 

5. RAG Pipeline 4

그렇다면 최종적으로 Multi-Query Retriever의 base retriever를 ensemble retriever로 하면 어떻까?

아님 반대로 Ensemble retriever의 dense retriever를 multi-query retriever로 하면 어떻게 될까?

 

def init_mq_ensemble_retriever(self):

        mq_ensemble_retriever = MultiQueryRetriever.from_llm(
            llm=self.llm,
            retriever=self.ensemble_retriever
        )

        return mq_ensemble_retriever

def init_ensemble_mq_retriever(self):

        bm25_retriever = self.bm25_retriever
        chroma_retriever = self.mq_retriever

        ensemble_retriever = EnsembleRetriever(
            retrievers=[bm25_retriever, chroma_retriever],
            weights=[0.4, 0.6],
            search_type=config["ensemble_search_type"],  # mmr
        )

        return ensemble_retriever

 

위와 같은 코드를 추가할 수 있고 실제로 작동하는 것을 확인할 수 있다.

 

최종 코드

class Ragpipeline:
    def __init__(self):
        # chatGPT API를 통해 llm 모델 로드
        self.llm = ChatOpenAI(
            model=config["llm_predictor"]["model_name"],  # chatgpt 모델 이름
            temperature=config["llm_predictor"]["temperature"],  # 창의성 0~1
        )

        # 초기화 리스트들 
        self.vector_store   = self.init_vectorDB()                  # 1. RAG가 접근할 vectorDB를 초기화합니다.
        self.retriever      = self.init_retriever()                 # 2. LLM이 질문에 대한 답변을 생성하기 전 질문에 관련된 컨텐츠 기반 답변을 생성하기 위해, 컨텐츠를 검색해 찾는 리트리버를 초기화 합니다. 
        self.bm25_retriever = self.init_bm25_retriever()
        self.ensemble_retriever = self.init_ensemble_retriever()
        self.mq_retriever   = self.init_multi_query_retriever()
        self.sq_retriever   = self.init_self_query_retriever()

        self.mq_ensemble_retriever = self.init_mq_ensemble_retriever()
        self.ensemble_mq_retriever = self.init_ensemble_mq_retriever()

    def init_vectorDB(self, persist_directory=config["chroma"]["persist_dir"]):
        """vectorDB 설정"""
        embeddings = OpenAIEmbeddings(model=config["embed_model"]["model_name"])  # VectorDB에 저장될 데이터를 임베딩할 모델을 선언합니다.
        vector_store = Chroma(
            persist_directory=persist_directory,  # 기존에 vectordb가 있으면 해당 위치의 vectordb를 load하고 없으면 새로 생성합니다.
            embedding_function=embeddings,                      # 새롭게 데이터가 vectordb에 넣어질때 사용할 임베딩 방식을 정합니다, 저희는 위에서 선언한 embeddings를 사용합니다.
            collection_name = 'india',                          # india라는 이름을 정해줌으로써 나중에 vector store 관리 가능 
            collection_metadata = {'hnsw:space': 'cosine'},     # cosine 말고 l2 가 default / collection_metadata를 통해 유사도 검색에 사용될 공간('hnsw:space')을 'cosine'으로 지정하여, 코사인 유사도를 사용
        )
        return vector_store

    def init_retriever(self):
        """ Retriever 초기화 """               

        # base retriever 3 
        retriever = self.vector_store.as_retriever(
            search_type="mmr",                                              # mmr 검색 방법으로 
            search_kwargs={'fetch_k': 10, "k": 5, 'lambda_mult': 0.4},      # 상위 10개의 관련 context에서 최종 5개를 추리고 'lambda_mult'는 관련성과 다양성 사이의 균형을 조정하는 파라메타 default 값이 0.5

        )

        return retriever

    def init_bm25_retriever(self):

        all_docs = pickle.load(open(config["pkl_path"], 'rb'))

        bm25_retriever = BM25Retriever.from_documents(all_docs)
        bm25_retriever.k = 1                                            # BM25Retriever의 검색 결과 개수를 1로 설정합니다.

        return bm25_retriever

    def init_ensemble_retriever(self):

        bm25_retriever = self.bm25_retriever
        chroma_retriever = self.retriever

        ensemble_retriever = EnsembleRetriever(
            retrievers=[bm25_retriever, chroma_retriever],
            weights=[0.4, 0.6],
            search_type=config["ensemble_search_type"],  # mmr
        )

        return ensemble_retriever

    # MultiQueryRetriever 생성
    def init_multi_query_retriever(self):
        """사용자의 질문을 여러 개의 유사 질문으로 재생성 """
        retriever_from_llm = MultiQueryRetriever.from_llm(
            llm=self.llm,
            retriever=self.retriever
        )

        return retriever_from_llm

    # SelfQueryRetriever 생성
    def init_self_query_retriever(self):
        """metadata를 이용해서 필터링해서 정보를 반환"""
        document_content_description = "Report or Explain"      # document_content_description: 문서 내용 설명
        retriever = SelfQueryRetriever.from_llm(    
            self.llm,
            self.vector_store,
            document_content_description,
            metadata_field_info,                                # metadata_field_info: 메타데이터 필드 정보
            verbose = True
        )

        return retriever

    # 멀티쿼리 - 앙승블
    def init_mq_ensemble_retriever(self):

        mq_ensemble_retriever = MultiQueryRetriever.from_llm(
            llm=self.llm,
            retriever=self.ensemble_retriever
        )

        return mq_ensemble_retriever

    def init_ensemble_mq_retriever(self):

        bm25_retriever = self.bm25_retriever
        chroma_retriever = self.mq_retriever

        ensemble_retriever = EnsembleRetriever(
            retrievers=[bm25_retriever, chroma_retriever],
            weights=[0.4, 0.6],
            search_type=config["ensemble_search_type"],  # mmr
        )

        return ensemble_retriever

 

한번 확인해보자.

 

pipeline = Ragpipeline()
mq_ensemble_retriever = pipeline.init_mq_ensemble_retriever()

query = '인도 통관 및 운송'   # 질문할 문장

result = mq_ensemble_retriever.get_relevant_documents(query)

for i, doc in enumerate(result):
    print(i)
    print(f"문서 내용: {doc.page_content}") # 문서 내용 표시
    print(f'출처 : {doc.metadata}')
    print("---" * 50)