[RAG 프로젝트][데이콘] 재정정보 AI 검색 알고리즘 경진대회

https://dacon.io/competitions/official/236295/overview/description

재정정보 AI 검색 알고리즘 경진대회 - DACON

 

지난번 KT AIVLE SCHOOL 빅프로젝트를 통해 KoTRA에 참가하여 우수상을 받았고 이를 기점으로 RAG를 더 공부하고 프로젝트를 더 도전해보기로 하였다. 

마침 데이콘에 좋은 대회가 열려서 참가한다. 

 

 

일단 대회에 대해서 간단하게 설명하자면 

목적 :

- 질의 응답 AI 알고리즘 개발

- 방대한 양의 재정 데이터가 일반 국민과 전문가 모두에게 쉽게 접근 가능하도록 만드는 것이 목적

 

즉 RAG를 통해 방대한 양의 재정 데이터에 접근하고 LLM이 질문에 대해 대답을 해주면 된다.

 

데이터를 살펴본다.

테스트 데이터는 다음과 같이 일반적인 보고서 형식의 pdf파일들이다. 

 

그리고 test.csv를 살펴보면, 질문(Question)과 이 질문의 소스인 pdf파일을 알려준다.

와우!! 질문에 대한 소스를 미리 알려주면 meta data를 통해 vector db 구축을 할 필요가 없어 보인다. 

이전에 챗봇 만들때는 사용자가 어떤 질문을 할 줄 모르니깐 하나의 vector DB에 데이터를 다 넣어두었는데

이번에는 pdf별 db를 따로 구축하면 되겠다. 

 

https://github.com/khw11044/FAISSPipeLine

 

위 코드를 살펴보면 

 

1. DB 구축

먼저 1.get_testdata.ipynb에서 test 데이터에 대한 VectorDB를 생성한다.

Chroma랑 FAISS 둘다 해봤는데 FAISS가 성능이 더 좋았다. 

 

utils/utils.py를 살펴보면 DB 구축 코드를 짜 두었다. 

임베딩 모델도 바꿔가면서 실험하면 될거 같다.

 

나중에 앙상블 리트리버를 사용하기 위해 text split를 한 Documents를 FAISS db에 데이터를 넣는거 말고도 pkl 파일로 저장하였다. 또한 Chroma는 pdf 한글 파일명으로 chroma 이름을 지을 수 있었는데 FAISS는 왜인지 오류가 발생했다. 

그래서 알파벳으로 이름을 매핑하여 저장하였다. 

 

2.get_traindata.ipynb에서는 

정답 데이터가 있는 traindata일부를 뽑아 나의 RAGpipeline의 성능을 테스트 해보았다. 

성능 테스트 계산은 

utils/cal.py에 있다. 

 

계산은 문자 단위의 F1 Score를 계산한다. 

 

2. RAGpipeline 

나는 class로 다음과 같이 간단히 코드를 작성하였다.

코드는 utils/RagPipeline.py에 있다.

 

class Ragpipeline:
    def __init__(self, source):
        # 1. LLM 바꿔보면서 실험하기 
        self.llm = ChatOllama(model="llama3.1", temperature=0.1)
        # self.llm = ChatOllama(model="llama3-ko-instruct", temperature=0)
        
        self.base_retriever = self.init_retriever(source)
        self.ensemble_retriever = self.init_ensemble_retriever(source)
        self.chain = self.init_chain()
        
        
    def init_retriever(self, source):
        # vector_store = Chroma(persist_directory=source, embedding_function=get_embedding())
        embeddings_model = get_embedding()
        vector_store = FAISS.load_local(source, embeddings_model, allow_dangerous_deserialization=True)
        retriever = vector_store.as_retriever(
                search_type="mmr",                                              # mmr 검색 방법으로 
                # search_kwargs={'fetch_k': 5, "k": 3, 'lambda_mult': 0.4},      # 상위 10개의 관련 context에서 최종 5개를 추리고 'lambda_mult'는 관련성과 다양성 사이의 균형을 조정하는 파라메타 default 값이 0.5
                search_kwargs={'fetch_k': 8, "k": 3}, 
            )
        
        return retriever
    
    def init_ensemble_retriever(self, source):
        retriever = self.base_retriever
        
        all_docs = pickle.load(open(f'{source}.pkl', 'rb'))
        bm25_retriever = BM25Retriever.from_documents(all_docs)
        bm25_retriever.k = 3   
        
        ensemble_retriever = EnsembleRetriever(
                retrievers=[bm25_retriever, retriever],
                weights=[0.4, 0.6],
                search_type='mmr'
            )
        
        return ensemble_retriever
        
    def init_chain(self):
        prompt = PromptTemplate.from_template(template)
        # 2. Chroma 할지, FAISS 할지, 앙상블 리트리버를 할지, 리트리버의 하이퍼파라메타 바꿔보면서 하기 
        retriever = self.base_retriever       # get_retriever()
        
        rag_chain = (
            {"context": retriever | format_docs, "question": RunnablePassthrough()}
            | prompt
            | self.llm
            | StrOutputParser()
        )
        
        return rag_chain
        
    
    def answer_generation(self, question: str) -> dict:
        full_response = self.chain.invoke(question)
        return full_response

 

search_kwargs의 파라미터도 바꿔 보고 search_type도 바꿔보고 

multi-query 리트리버나 parentDocument 리트리버 같은것도 사용해보고 하면 좋을 것 같다.