사용자의 질문을 여러 개 만드는 기법 : Query Translation (Part 1)

본 게시물은 Lance Martin 님의 유튜브 영상을 기반으로 작성되었습니다. 

사용자가 작성한 질문이 모호하거나 구체적으로 구조화되지 않을 경우, 문서에서 의미적 유사성을 기준으로 검색하는 과정에서 원하는 정보를 찾지 못하게 되는 경우가 존재한다. 이러한 문제를 해결하기 위해 사용자의 질문을 다양한 관점에서 재작성하거나 다른 표현으로 변환하여, 원래 질문의 의미를 보존하면서도 문서와의 내용과의 매칭 가능성을 높이는 것을 의미하는 것을 Query Translation 이라고 한다. 

 

위 3 가지의 기법은 Query Translation의 대표적인 기법이다. 이들은 각각 다르게 사용자의 질문을 변형시켜 검색 성능을 향상시키는 기법 질문을 재구성하거나 변형하는 방식이라는 공통점을 가지고 있다. 

 

1. Query Rewriting(쿼리 재작성): 사용자의 질문을 다양한 관점에서 재작성하여 검색 능력을 향상시키는 방식이며, RAG Fusion과 Multi-Query 기법이 가장 대표적인 방법이다. 
2. Sub-Question(하위 질문 생성): 복잡하거나 추상적인 질문을 구체적인 질문으로 분해하는 방법이다. 이를 통해 더 정확하고 세부적인 문서를 검색할 수 있다는 것이 특징이다. Google의 "Least-toMost" 기법은 복잡한 질문을 더 작은 단계로 나누어 해결하는 sub-question의 대표적인 방식이다. 
3. Abstract Query(추상적인 질문 생성): 질문을 더 높은 수준으로 추상화하여, 일반적이거나 광범위한 문서를 검색하는 방법이다. Google Deepmind에서 게재한 논문이며, 대표적인 방법으로는 "Stepback Prompting"이라는 기법이 대표적인 방식이다. 

---

 

1. Query Rewriting (Multi-Query) 

 

Multi-Query의 계락도

 

위 그림과 같이 Multi-Query 방식은 질문을 여러 가지 형태로 변환하여 다양한 관점에서 검색이 가능하게 만들어주는 일종의 Query Rewriting 방식이다. 이는 작성된 질문이 임베딩될 때, 고차원 임베딩 공간에서 문서와 일치하지 않을 수도 있다는 것을 염두한 것이다. 질문을 다양한 방식으로 다시 작성함으로써, 문서 임베딩의 뉘앙스와 유사한 뉘앙스를 가진 임베딩을 생성하여 적절한 문서를 찾을 가능성을 높이는 방법이다. 그럼 코드를 통해 조금 더 자세히 알아보도록 하겠습니다. 

 

1. 문서 업로드 및 벡터 스토어 생성 

import bs4
from langchain_community.document_loaders import WebBaseLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma

loader = WebBaseLoader(
    web_paths=("https://lilianweng.github.io/posts/2023-06-23-agent/",),
    bs_kwargs=dict(
        parse_only=bs4.SoupStrainer(
            class_=("post-content", "post-title", "post-header")
        )
    ),
)

blog_docs = loader.load()

text_splitter = RecursiveCharacterTextSplitter.from_tiktoken_encoder(
    chunk_size=300,
    chunk_overlap=50
)

splits = text_splitter.split_documents(blog_docs)

vectorstore = Chroma.from_documents(documents=splits,
                                    embedding=OpenAIEmbeddings())

retriever = vectorstore.as_retriever()

 

 

2. 다중 쿼리 생성을 위한 프롬프트 정의

from langchain.prompts import ChatPromptTemplate

template = """
You are an AI language model assistant. 
Your task is to generate five different versions of the given user question to retrieve relevant documents from a vector database. 
By generating multiple perspectives on the user question, your goal is to help the user overcome some of the limitations of the distance-based similarity search.
Provide these alternative questions separated by newlines. 
Original question: {question}
"""
    
prompt_perspectives = ChatPromptTemplate.from_template(template)

 

3. 다중 쿼리를 사용한 검색 및 문서 통합: 

from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI
from langchain.load import dumps, loads

generate_queries = (
    prompt_perspectives
    | ChatOpenAI(temperature=0)
    | StrOutputParser()
    | (lambda x: x.split("\n"))
)

def get_unique_union(documents: list[list]):
    """ Unique union of retrieved docs """
    flattened_docs = [dumps(doc) for sublist in documents for doc in sublist]
    unique_docs = list(set(flattened_docs))
    return [loads(doc) for doc in unique_docs]

retrieval_chain = generate_queries | retriever.map() | get_unique_union

question = "What is task decomposition for LLM agents?"
docs = retrieval_chain.invoke({"question":question})

len(docs)

 

(3)을 실행하면, LangSmith에서 생성된 5가지의 질문을 확인해볼 수 있다. 

 

 

4. 최종 RAG 

from operator import itemgetter
from langchain_openai import ChatOpenAI
from langchain_core.runnables import RunnablePassthrough

template = """
Answer the following question based on this context:
{context}

Question: 
{question}
"""

prompt = ChatPromptTemplate.from_template(template)
llm = ChatOpenAI(temperature=0)

# retrieval_chain = generate_queries | retriever.map() | get_unique_union
final_rag_chain = (
    {"context": retrieval_chain,
     "question": itemgetter("question")}
    | prompt
    | llm
    | StrOutputParser()
)

final_rag_chain.invoke({"question":question})

 

 

itemgetter("question")은 파이썬의 operator 모듈에서 제공하는 함수인 itemgetter를 사용하여 특정 키 값을 추출하도록 설정하는 역할을 수행한다. 예를 들어, {"context": ..., "question": "What is task decomposition for LLM agents?"}라는 입력이 주어졌다면, itemgetter("question")는 "What is task decomposition for LLM agents?" 라는 값을 반환하는 것이다. 

 

 

---

2. Query Rewriting (RAG-Fusion) 

RAG Fusion의 계략도

 

 

RAG Fusion은 위에서 설명한 Multi-Query 방식과 매우 흡사하다. 그러나 RAG Fusion은 Multi-Query 결과를 통합하는 추가적인 과정이 필요하며, 이는 Reciprocal Rank Fusion(RRF)을 통해 이루어진다. RRF는 서로 다른 서브 쿼리에서 검색된 문서들의 순위를 재정렬하여, 가장 관련성이 높고 포괄적인 결과를 제공하는 기법이다.

 

RAG Fusion은 아래 3 가지의 주요 단계를 포함한다. 

 

1. Query Generation: 사용자의 질문을 바탕으로 여러 개의 서브 쿼리를 생성하는 것으로 시작하며, 이는 사용자 질문의 의도를 완벽하게 파악하게 위한 다양한 관점을 제공한다.  
2. Sub-Query Retrieval: 생성된 독립적인 서브 쿼리는 검색 단계를 거쳐 답변을 생성하게 된다. 
3. Reciprocal Rank Fusion(RRF): 이렇게 얻은 여러 검색 결과를 RRF 방식으로 통합한다. 이때 각 문서의 순위를 기반으로 점수를 매기고, 상위 문서에 더 높은 가중치를 부여하여 순위를 재정렬한 후, LLM에게 전달 되어 최종적인 답변을 생성하게 된다. 

Reciprocal Rank Fusion(RRF) 

RRF (Reciprocal Rank Fusion)는 서로 다른 관련성 지표(Relevance Indicators)를 가진 여러 개의 결과를 하나로 결합하는 방법이다. 이때 각각의 문서(소스)에서 도출된 "점수"가 아니라 "순위"를 기반으로 통합하는 것이 가장 큰 특징이다. 이를 바탕으로 각 문서의 중요도를 순위를 기준으로 정의하여 어떤 문서가 가장 중요한지 판단한다. 

 

그럼 도대체 왜 "순위"를 기준으로 문서의 중요도를 나누게 되는 것일까?

 

우리가 논문을 검색할 때, 다양한 학술 데이터베이스(Google Scholar, Arxiv 등등)에서 동일한 키워드로 논문을 검색한다고 가정해보자. Google Scholar 같은 경우 검색 결과의 관련성을 0~1 사이의 소수점 점수로 나타내고, Arxiv는 관련성을 0~100 사이의 정수로 표현된다. 두 데이터베이스에 대한 점수 스케일이 서로 다르기 때문에 직접 비교하거나 통합하는데 어려움이 있어, 순위로 이에 대한 중요도를 나누는 것이다. 

 

특정 문서 d의 RRF 점수는 다음과 같이 계산된다.

 

RRF 수식

 

(1) 각 검색 엔진에서의 순위 : 각 검색 엔진에서 문서 d에 대해 순위를 매긴다. 예를 들어, 문서 d가 세 가지 검색 엔진에서 각각 다음 순위를 받았다고 가정해보겠습니다. 

• Google Scholar: 순위 r_1(d) = 1
• Arxiv: 순위 = r_2(d) = 5 
• PubMed: 순위 r_3(d) = 2 

(2) 보정 상수 k : RRF 공식에서 보정 상수 k를 사용하여 낮은 순위일수록 점수가 급격히 감소하지 않도록 한다. 일반적으로 k = 60을 사용한다. 

 

(3) 각 엔진에서의 기여 계산: 각 검색 엔진에서 문서 d에 대한 기여도를 계산한다

 (4) RRF 점수 계산: 모든 검색 엔진의 기여도를 합산하여 문서 d의 최종 RRF 점수를 계산하며, 이는 문서 d의 최종 중요도를 나타낸다. 이 러한 점수를 다른 문서와 비교하여 최종 순위를 결정하게 된다. 

 

그럼 구현 방법에 대해서 알아보도록 하겠다. 

 

1. RAG Fusion용 프롬프트 정의 및 다중 쿼리 생성

from langchain.prompts import ChatPromptTemplate

template = """
You are a helpful assistant that generates multiple search queries based on a single input query. \n
Generate multiple search queries related to: {question} \n

Output (4 queries):
"""

prompt_rag_fusion = ChatPromptTemplate.from_template(template)

from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI

generate_queries = (
    prompt_rag_fusion
    | ChatOpenAI(temperature=0)
    | StrOutputParser()
    | (lambda x: x.split("\n"))
)

 

2. Reciprocal Rank Fusion (RRF) 함수 정의 및 검색 수행

from langchain.load import dumps, loads

def reciprocal_rank_fusion(results: list[list], k=60):
    """ Reciprocal_rank_fusion that takes multiple lists of ranked documents 
        and an optional parameter k used in the RRF formula """
    
    # Initialize a dictionary to hold fused scores for each unique document
    fused_scores = {}

    # Iterate through each list of ranked documents
    for docs in results:
        # Iterate through each document in the list, with its rank (position in the list)
        for rank, doc in enumerate(docs):
            # Convert the document to a string format to use as a key (assumes documents can be serialized to JSON)
            doc_str = dumps(doc)
            # If the document is not yet in the fused_scores dictionary, add it with an initial score of 0
            if doc_str not in fused_scores:
                fused_scores[doc_str] = 0
            # Retrieve the current score of the document, if any
            previous_score = fused_scores[doc_str]
            # Update the score of the document using the RRF formula: 1 / (rank + k)
            fused_scores[doc_str] = previous_score + 1 / (rank + k)

    # Sort the documents based on their fused scores in descending order to get the final reranked results
    reranked_results = [
        (loads(doc), score)
        for doc, score in sorted(fused_scores.items(), key=lambda x: x[1], reverse=True)
    ]

    # Return the reranked results as a list of tuples, each containing the document and its fused score
    return reranked_results

retrieval_chain_rag_fusion = generate_queries | retriever.map() | reciprocal_rank_fusion
docs = retrieval_chain_rag_fusion.invoke({"question": question})
len(docs)

3. 최종 RAG 체인 정의

from langchain_core.runnables import RunnablePassthrough

template = """Answer the following question based on this context:

{context}

Question: {question}
"""

prompt = ChatPromptTemplate.from_template(template)

final_rag_chain = (
    {"context": retrieval_chain_rag_fusion,
     "question": itemgetter("question")}
    | prompt
    | llm
    | StrOutputParser()
)

final_rag_chain.invoke({"question":question})