LangChain - Advanced RAG Technique for Better Retrieval Performance 정리

 
아래 유튜브 영상을 정리하였음.
https://youtu.be/KQjZ68mToWo?si=09NX4cfbE9lYTJ9l

 

 
일반적인 RAG Step

• Indexing Step: Data Load -> Split -> Embedding -> Store in VectorDB
• Retrieval Step: Ask Question -> Embedding Question -> Retrieve Similar Documents -> add as a prompt -> LLM

Langchain이 Vector Store에서 필요한 데이터를 더 잘 가져올 수 있도록 하는 기법

• MultiQueryRetriever
• Contextual Compression
• Ensemble Retriever
• Self-Querying Retriever
• Time-weighted Vector Store Retriever

 

Chunk Size Experiment

import os
from langchain.embeddings.openai import OpenAIEmbeddings

env_vars = {
    "OPENAI_API_KEY": "your-token",
}

def pretty_print_docs(docs):
    print(f"\n{'-' * 100}\n".join([f"Document {i+1}:\n" + d.page_content for i, d in enumerate(docs)]))

for key, value in env_vars.items():
    os.environ[key] = value

embedding = OpenAIEmbeddings(chunk_size=1)

chunk size 설정이 왜 중요한지를 설명하기 위한 예시. OpenAI에서 제공하는 Embedding Function을 사용했다.

from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.vectorstores import Chroma

from langchain.schema import Document

# Load blog post
from langchain.document_loaders import TextLoader

loader = TextLoader("./dogs.txt")
data = loader.load()
loader = TextLoader("./restaurant.txt")
data2 = loader.load()

docs = data + data2
#text_splitter = RecursiveCharacterTextSplitter(chunk_size=120, chunk_overlap=10)
#docs = text_splitter.split_documents(data)

RAG에 사용하기 위해 두 개의 텍스트 파일을 준비하고, langchain의 TextLoader를 사용해서 메모리에 올린다.

• dogs.txt: friction Q&A format of a Dog school. dog school 이름이 뭐고, 위치가 어디고, 어떤 프로그램을 진행하는지 등등.. 의 정보가 담겨 있다.
• restaurant.txt: 특정 음식점 관련 Q&A. 이름, 위치, vegan식 제공 여부 등등..
• 즉, 두 파일은 전혀 다른 내용을 담고 있다.

비교를 위해, 여기서는 Text Split을 진행하지 않는다. 따라서 위 경우 거대한 text Chunk 하나만 있는 셈.

vector1 = embedding.embed_query("How is the whether??")
vector2 = embedding.embed_query("What is the Name of the Dogschool?")
vector3 = embedding.embed_query("What food do you offer?")

data_vectors = [embedding.embed_query(doc.page_content) for doc in docs]
print(len(data_vectors)) ## return 2 (2 vectors)

Question Embedding

• vector 1번은 dogs.txt와 restaurant.txt 둘 다 관련없는 질문
• vector 2번은 dogs.txt에서, vector 3번은 restaurant.txt 에서 답을 찾을 수 있는 질문.

따라서 vector 2번은 dogs.txt의 embedding vector와 유사도가 높아야 하고, vector 3번은 restaurant.txt와 유사도가 높아야 한다. 1은 양쪽과 다 낮아야 함.

from sklearn.metrics.pairwise import cosine_similarity
import matplotlib.pyplot as plt
import numpy as np

cosine_sims_1 = [cosine_similarity([vector1], [data_vector])[0][0] for data_vector in data_vectors]
cosine_sims_2 = [cosine_similarity([vector2], [data_vector])[0][0] for data_vector in data_vectors]
cosine_sims_3 = [cosine_similarity([vector3], [data_vector])[0][0] for data_vector in data_vectors]

x = np.arange(len(data_vectors))

plt.scatter(x, cosine_sims_1, label='Weather', alpha=0.7)
plt.scatter(x, cosine_sims_2, label='Dogschool', alpha=0.7)
plt.scatter(x, cosine_sims_3, label='Restaurant', alpha=0.7)

plt.ylabel('Cosine Similarity')
plt.title('Consine Similarity between query and data vectors')
plt.legend()

plt.show()

 
vector 유사도 결과는 위와 같다.

• 오른쪽 dot은 dogs.txt 대상으로 한 vector 유사도. dogschool 관련 질문을 한 vector 2는 높은 유사도를, 나머지 둘은 낮은 유사도를 보였다
• 왼쪽 dot은 restaurant.txt 대상으로 한 vector 유사도.
  • 관련있는 질문인 vector 3은 유사도가 낮게 측정됐고, 관련없는 질문인 vector 1과 2는 상대적으로 유사도가 높게 나왓다.

single file에서도 chunk 단위로 text를 분할해두면, embedding할 때 파일 내의 특정 부분에서 더 높은 유사도를 가진 문장을 검출할 수 있음.

• 따라서 chunk로 분할하는 text splitting은 필요하다.

 
text split을 수행한 뒤 vector 유사도 결과는 위와 같다.

• smaller chunks 기준으로 vector 유사도를 비교할 경우, vector store에서 보다 정확한 값을 찾아낼 수 있다.
• 단, chunk size가 너무 작을 경우 LLM에게 제공하는 context 정보가 부족할 수 있음.

ParentDocumentRetriever

from langchain.storage import InMemoryStore
from langchain.retrievers import ParentDocumentRetriever

child_splitter = RecursiveCharacterTextSplitter(chunk_size=120, chunk_overlap=20)
parent_splitter = RecursiveCharacterTextSplitter(chunk_size=400, chunk_overlap=20)
vectorstore = Chroma(
    collection_name="full_documents", embedding_function=embedding
)
store = InMemoryStore()
retriever = ParentDocumentRetriever(
    vectorstore=vectorstore,
    docstore=store,
    child_splitter=child_splitter,
    parent_splitter=parent_splitter
)

retriever.add_documents(docs, ids=None)

chunk 개수를 두 개로 나눈 것.

• parent: 큰 사이즈의 chunk. context 정보를 더 많이 담고 있음.
• child: 더 세분화된 chunk

Vector Store로는 InMemory Chroma 사용.
ParentDocumentRetriever로 child / parent를 명시한 뒤, retriever에 앞서 정의한 docs (TextLoader로 가져온 두 개의 txt파일) 넣어준다.

vectorstore.similarity_search("What is the name of the dog school?") # 유샤도가 높은 Documents의 list 반환
retriever.get_relevant_documents("What is the name of the dog school?") # 위 함수랑 똑같은 기능. 보다 표준화된 인터페이스가 get_relevant_documents

 

MultiQueryRetriever

어감 (뉘앙스)이 어떻게 embedding되느냐에 따라 응답 결과가 크게 달라질 수 있음.

따라서 사용자의 query와 비슷한 형태의 질문을 여러 개 만들어서 수행 + 응답결과 리턴.

from langchain.chat_models import ChatOpenAI
from langchain.retrievers.multi_query import MultiQueryRetriever

llm = ChatOpenAI(
        temperature=0,
        max_tokens=800,
        model_kwargs={"top_p": 0, "frequency_penalty": 0, "presence_penalty": 0},
    )


retriever = MultiQueryRetriever.from_llm(
    retriever=vectorstore.as_retriever(), llm=llm
)

unique_docs = retriever.get_relevant_documents("What is the name of the dog school?")
len(unique_docs) # return 4 = input query와 유사도가 높은 docs

from typing import List

from langchain.chains import LLMChain
from langchain.output_parsers import PydanticOutputParser
from langchain.prompts import PromptTemplate
from pydantic import BaseModel, Field


class LineList(BaseModel):
    lines: List[str] = Field(description="Lines of text")


class LineListOutputParser(PydanticOutputParser):
    def __init__(self) -> None:
        super().__init__(pydantic_object=LineList)

    def parse(self, text: str) -> LineList:
        lines = text.strip().split("\n")
        return LineList(lines=lines)


output_parser = LineListOutputParser()

QUERY_PROMPT = PromptTemplate(
    input_variables=["question"],
    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}""",
)

llm_chain = LLMChain(llm=llm, prompt=QUERY_PROMPT, output_parser=output_parser)
llm_chain.invoke("What is the name of the dog school?") # returns 5 different questions

여기의 핵심은 Prompt. 결국 prompt에 'input query와 같은 의미를 같은 다른 question을 여러 개 만들어라' 고 명시함.

• 이렇게 쿼리 보내면, list of questions를 응답해줄 것.
• 이 query를 활용해서 similarity search 수행하면, 나오는 docs 조합을 활용해서 응답하도록 만든다.

Contextual Compression

크게 두 가지가 필요함.

• basic Retriever
• document Compressor

base Retriever에 query 보내고, 응답받은 document를 compressor에 전달한다. compressor에서 shorten it, by reducing the content of documents or omitting documents altogether.

vectorstore = Chroma(
    collection_name="full_documents", embedding_function=embedding
)
vectorstore.add_documents(docs)
retriever = vectorstore.as_retriever()

from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import LLMChainExtractor

compressor = LLMChainExtractor.from_llm(llm)
compression_retriever = ContextualCompressionRetriever(base_compressor=compressor, base_retriever=retriever)

compressed_docs = compression_retriever.get_relevant_documents(query=question)
pretty_print_docs(compressed_docs)

 
원본 docs는 위처럼 4개의 docs를 리턴한다.

• langchain에서 제공하는 LLMChainExtractor Class를 사용함.
• compressor retriever를 정의한다.
• get_relevant_documents 함수를 사용하면 된다.

 
Compressor 가 응답하는 docs는 위와 같다. 원본 문서보다 훨씬 간결한 내용을 담고 있음.

• 따라서 모든 docs를 LLM에 전부 던질 필요 없다. compressed된 정보만 전달할 수 있음.
• 단, compress하기 위해서 LLM api call이 발생하므로, openAI chatgpt의 경우 api 사용량이 증가함.

EmbeddingFilter

from langchain.embeddings import OpenAIEmbeddings
from langchain.retrievers.document_compressors import EmbeddingsFilter

embeddings_filter = EmbeddingsFilter(embeddings=embedding, similarity_threshold=0.5)
compression_retriever = ContextualCompressionRetriever(base_compressor=embeddings_filter, base_retriever=retriever)

compressed_docs = compression_retriever.get_relevant_documents(query=question)
pretty_print_docs(compressed_docs)

그래서 대안으로 쓸 수 있는 게 EmbeddingFilter.

• LLM에 전달하는 게 아니라 EmbeddingModel에 전달함. cheaper / faster.
• 코사인 유사도 계산. 유사도 값이 파라미터로 넘긴 threshold보다 작으면 버린다.
• compressor처럼 document 자체를 변경하는 게 아니라 필터링만 수행하는 것.
  • LLM 재요약만큼 성능이 좋지는 않지만, 여러 개의 filter 조건을 걸 수 있다는 것이 장점

from langchain.document_transformers import EmbeddingsRedundantFilter
from langchain.retrievers.document_compressors import DocumentCompressorPipeline
from langchain.text_splitter import CharacterTextSplitter

splitter = CharacterTextSplitter(chunk_size=300, chunk_overlap=0, separator=". ")
redundant_filter = EmbeddingsRedundantFilter(embeddings=embedding)
relevant_filter = EmbeddingsFilter(embeddings=embedding, similarity_threshold=0.76)
pipeline_compressor = DocumentCompressorPipeline(
    transformers=[splitter, redundant_filter, relevant_filter]
)

compression_retriever = ContextualCompressionRetriever(base_compressor=pipeline_compressor, base_retriever=retriever)

compressed_docs = compression_retriever.get_relevant_documents(query=question)
pretty_print_docs(compressed_docs)

redundantFilter: filter out redundant documents.
위 예시코드의 경우 중복 제거하는 필터 + threshold 필터 설정.

• 여러 개의 필터 적용 시 Pipeline object를 정의해서 transformers 필드에 리스트 형태로 파라미터를 넘긴다.
• filter는 애플리케이션 latency가 증가하는 요인이므로, 필요성과 유의점을 잘 알고 써야 한다.

Ensemble Retriever

Document 추출을 위한 알고리즘을 여러 개 써서 best documents를 가져오기 위한 방식.

from langchain.retrievers import BM25Retriever, EnsembleRetriever


bm25_retriever = BM25Retriever.from_documents(docs)
bm25_retriever.k = 2

chroma_vectorstore = Chroma.from_documents(docs, embedding)
chroma_retriever = chroma_vectorstore.as_retriever()

ensemble_retriever = EnsembleRetriever(
    retrievers=[bm25_retriever, chroma_retriever], weights=[0.5, 0.5]
)

docs = ensemble_retriever.get_relevant_documents(query=question)
docs

 
예시: BM25Retriever - good for retrieving keywords

• BM25Retriever로 keyword search, Vector search for high Dimensional Vector 가능.
• ensemble_retriever의 경우 weights를 리스트 파라미터로 넘겨줘야 하는데, 리스트 값은 합쳐서 1이 되어야 한다.
  • 각각의 retriever 응답값의 가중치 설정.

Self-Querying Retriever

document의 underlying metadata를 사용할 수 있는 방식. (일반적인 retrieval에서는 document의 content만 쓰임.)

• metadata를 document에 정의해두고, metadata 기준 docs filtering이 가능함.
• 잘 쓰면 훌륭한 성능을 낼 수 있다.

아래 예시의 경우 metadata로 type, feature, price 세 개의 필드가 있음.

from langchain.schema import Document
from langchain.vectorstores import Chroma

docs = [
    Document(
        page_content="Bello-Basistraining offers a comprehensive foundation for dog obedience, focusing on basic commands and socialization.",
        metadata={"type": "Basic Training", "feature": "Foundational Skills", "price": "Affordable"},
    ),
    Document(
        page_content="Pfote-Agilitykurs provides a fun and energetic way to keep dogs fit and mentally stimulated through obstacle courses.",
        metadata={"type": "Agility Training", "feature": "Physical Fitness", "price": "Moderate"},
    ),
    Document(
        page_content="Wuff-Verhaltensberatung specializes in addressing behavioral issues, offering tailored strategies for each dog.",
        metadata={"type": "Behavioral Consultation", "feature": "Customized Solutions", "price": "Premium"},
    ),
    Document(
        page_content="Schwanzwedeln-Therapiehundausbildung prepares dogs for roles in therapeutic and support settings, focusing on empathy and gentleness.",
        metadata={"type": "Therapy Dog Training", "feature": "Emotional Support", "price": "High"},
    ),
    Document(
        page_content="Schnüffler-Suchhundetraining trains dogs in scent detection, useful for search and rescue operations.",
        metadata={"type": "Search and Rescue Training", "feature": "Advanced Skills", "price": "Variable"},
    ),
    Document(
        page_content="Hunde-Haftpflichtversicherung offers comprehensive coverage for potential damages or injuries caused by your dog.",
        metadata={"type": "Dog Liability Insurance", "feature": "Financial Protection", "price": "Varies"},
    ),
]

vectorstore = Chroma.from_documents(docs, embedding)

from langchain.vectorstores import Chroma
from langchain.chains.query_constructor.base import AttributeInfo
from langchain.retrievers.self_query.base import SelfQueryRetriever

metadata_field_info = [
    AttributeInfo(
        name="type",
        description="The type of dog training service (e.g., Basic Training, Agility Training, Behavioral Consultation)",
        type="string",
    ),
    AttributeInfo(
        name="feature",
        description="Special features or benefits of the service",
        type="string",
    ),
    AttributeInfo(
        name="price",
        description="Price category of the service (e.g., Affordable, Moderate, Premium)",
        type="string",
    ),
]

document_content_description = "Description of a dog training service"
retriever = SelfQueryRetriever.from_llm(
    llm,
    vectorstore,
    document_content_description,
    metadata_field_info,
)
retriever.invoke("What Premium priced trainings do you offer?")

retriever 정의할 때, 마지막 필드인 metadata_field_info가 중요함

• 어떤 metadata가 저장되어 있는지 전달하는 것. type / feature / price 정보를 AttributeInfo 객체로 정의한다.

 
retriever의 응답 결과를 보면, Price가 Premium인 Document만 가져온 걸 확인할 수 있다.

Time-weighted vector store retriever

23.12.18 기준 모든 Vector store에서 지원되는 기능은 아님. faiss에서만 시연 성공했다고 함.

import faiss

from datetime import datetime, timedelta
from langchain.docstore import InMemoryDocstore
from langchain.embeddings import OpenAIEmbeddings
from langchain.retrievers import TimeWeightedVectorStoreRetriever
from langchain.schema import Document
from langchain.vectorstores import FAISS


# decay_rate = .0000000000000000000000001
decay_rate = .999

embedding_size = 1536 # openAIEmbedding에서 사용하는 embedding size는 1536이라고 함.
index = faiss.IndexFlatL2(embedding_size)
vectorstore = FAISS(embedding, index, InMemoryDocstore({}), {})
retriever = TimeWeightedVectorStoreRetriever(vectorstore=vectorstore, decay_rate=decay_rate, k=1)

yesterday = datetime.now() - timedelta(days=1)
retriever.add_documents([Document(page_content="hello world", metadata={"last_accessed_at": yesterday})])
retriever.add_documents([Document(page_content="hello foo")])

decay rate?

• time-weight이므로 retriever가 문서 가져올 때 timestamp 정보를 사용함.
• 따라서 add_document 보면 metadata로 last_accessed_at 이라는 정보를 넘겨줌.

 
decay rate를 높게 잡으면, last_accessed_at 값이 있는 'hello world' docs가 retriever에 잡히지 않는다.

• 위 예시 보면, hello world로 아예 docs와 동일한 query를 요청했음에도 응답으로 돌아오는 docs는 hello foo다. hello world 문서는 last_accessed_at timestamp에서 decay되었기 때문.