검색기(Retriever)는 Retrieval-Augmented Generation (RAG) 시스템의 핵심 단계로, 데이터베이스에서 사용자의 질문에 관련된 정보를 찾아내는 역할을 합니다. 이 단계는 시스템의 성능과 사용자가 얻는 정보의 정확성에 큰 영향을 미칩니다.
검색기는 관련 정보를 신속하게 찾고 제공하여 RAG 시스템의 전반적인 효율성과 사용자 만족도를 높이는 중요한 역할을 합니다.
Sparse Retriever와 Dense Retriever는 정보 검색 시스템에서 사용되는 두 가지 주요 방법입니다.
from langchain_community.vectorstores import FAISS
# 벡터스토어를 생성합니다.
vectorstore = FAISS.from_documents(documents=split_documents, embedding=embeddings)
# Dense Retriever 생성
faiss_retriever = vectorstore.as_retriever()
from langchain_community.retrievers import BM25Retriever
# Sparse Retriever 생성
bm25_retriever = BM25Retriever.from_documents(split_documents)
결론: Sparse Retriever와 Dense Retriever는 각각의 장단점을 가지고 있으며, 특정 프로젝트의 요구사항에 맞게 선택되어야 합니다. Dense Retriever는 복잡한 질문에 대해 더 높은 정확성을 제공하는 반면, Sparse Retriever는 간단한 키워드 기반의 검색에 더 적합합니다.
VectorStore 지원 검색기는 vector store를 사용하여 문서를 검색하는 retriever입니다. 이 검색기는 벡터 스토어에 구현된 유사도 검색(similarity search) 또는 Maximal Marginal Relevance(MMR)과 같은 검색 메서드를 활용하여 저장된 텍스트 데이터를 쿼리합니다.
from dotenv import load_dotenv
from langchain_teddynote import logging
from langchain_community.vectorstores import FAISS
from langchain_openai import OpenAIEmbeddings
from langchain_text_splitters import CharacterTextSplitter
from langchain_community.document_loaders import TextLoader
# 환경 변수에서 API 키 로드
load_dotenv()
# LangSmith 추적 시작
logging.langsmith("CH11-Retriever")
# 텍스트 로더를 사용하여 파일을 로드합니다.
loader = TextLoader("./data/appendix-keywords.txt")
# 문서를 로드합니다.
documents = loader.load()
# 텍스트를 청크 크기 300으로 분할
text_splitter = CharacterTextSplitter(chunk_size=300, chunk_overlap=0)
split_docs = text_splitter.split_documents(documents)
# OpenAI 임베딩 생성
embeddings = OpenAIEmbeddings()
# FAISS 벡터 데이터베이스 생성
db = FAISS.from_documents(split_docs, embeddings)
as_retriever
메서드는 생성된 VectorStore를 기반으로 VectorStoreRetriever
를 초기화합니다. 이 메서드는 다양한 검색 옵션을 설정하여 사용자가 요구하는 문서 검색을 수행할 수 있도록 합니다.
# 데이터베이스를 검색기로 사용
retriever = db.as_retriever()
invoke()
메서드invoke()
메서드는 검색 쿼리를 받아 관련 문서를 반환하는 역할을 합니다.
# 관련 문서 검색
docs = retriever.invoke("임베딩(Embedding)은 무엇인가요?")
for doc in docs:
print(doc.page_content)
print("=========================================================")
MMR 알고리즘을 사용하여 검색할 때, search_type
을 "mmr"
로 설정하고 다양한 매개변수를 조정하여 검색 결과의 다양성을 조절할 수 있습니다.
# MMR 검색 유형을 지정
retriever = db.as_retriever(
search_type="mmr", search_kwargs={"k": 2, "fetch_k": 10, "lambda_mult": 0.6}
)
# 관련 문서 검색
docs = retriever.invoke("임베딩(Embedding)은 무엇인가요?")
for doc in docs:
print(doc.page_content)
print("=========================================================")
유사도 점수 임계값을 설정하여 특정 임계값 이상의 점수를 가진 문서만 검색하는 방법입니다.
retriever = db.as_retriever(
search_type="similarity_score_threshold",
search_kwargs={"score_threshold": 0.8},
)
# 관련 문서 검색
for doc in retriever.invoke("Word2Vec 은 무엇인가요?"):
print(doc.page_content)
print("=========================================================")
검색 설정을 동적으로 조정하기 위해 ConfigurableField
를 사용하여 다양한 검색 옵션을 지정할 수 있습니다.
from langchain_core.runnables import ConfigurableField
# 검색 설정을 동적으로 적용
retriever = db.as_retriever(search_kwargs={"k": 1}).configurable_fields(
search_type=ConfigurableField(
id="search_type",
name="Search Type",
description="The search type to use",
),
search_kwargs=ConfigurableField(
id="search_kwargs",
name="Search Kwargs",
description="The search kwargs to use",
),
)
# 검색 설정을 지정하여 문서 검색
config = {"configurable": {"search_kwargs": {"k": 3}}}
docs = retriever.invoke("임베딩(Embedding)은 무엇인가요?", config=config)
for doc in docs:
print(doc.page_content)
print("=========================================================")
Query와 Passage에 대해 서로 다른 임베딩 모델을 사용하는 경우, 각각 다른 모델을 설정하여 벡터 유사도 검색을 수행할 수 있습니다.
from langchain_community.vectorstores import FAISS
from langchain_text_splitters import CharacterTextSplitter
from langchain_community.document_loaders import TextLoader
from langchain_upstage import UpstageEmbeddings
# 문서 로드 및 분할
loader = TextLoader("./data/appendix-keywords.txt")
documents = loader.load()
text_splitter = CharacterTextSplitter(chunk_size=300, chunk_overlap=0)
split_docs = text_splitter.split_documents(documents)
# 문서용 Upstage 임베딩 생성
doc_embedder = UpstageEmbeddings(model="solar-embedding-1-large-passage")
db = FAISS.from_documents(split_docs, doc_embedder)
# 쿼리용 Upstage 임베딩 생성 및 검색 수행
query_embedder = UpstageEmbeddings(model="solar-embedding-1-large-query")
query_vector = query_embedder.embed_query("임베딩(Embedding)은 무엇인가요?")
db.similarity_search_by_vector(query_vector, k=2)
ContextualCompressionRetriever는 대량의 문서에서 사용자의 질의와 관련된 정보를 효율적으로 검색하고 불필요한 부분을 제거하여, 최적의 응답을 제공하는 것을 목표로 하는 시스템입니다. 이 접근법은 전체 문서를 반환하는 대신, 질의에 따라 문서를 압축하고 관련 정보만 반환하여 성능을 향상시킵니다.
기본 Retriever 설정:
문서 압축:
LLMChainExtractor
를 사용하여 검색된 문서를 압축합니다.LLM을 활용한 문서 필터링:
LLMChainFilter
는 문서의 내용을 변경하지 않고, 검색된 문서 중에서 어떤 문서를 반환할지 선택적으로 필터링합니다.EmbeddingsFilter를 활용한 유사도 기반 필터링:
EmbeddingsFilter
는 문서와 쿼리의 임베딩을 비교하여, 유사도 임계값 이상의 문서만 반환합니다. 이를 통해 계산 비용을 절약하면서도 관련성을 유지합니다.파이프라인 생성:
DocumentCompressorPipeline
을 구성할 수 있습니다. 예를 들어, 텍스트를 작은 청크로 분할한 후, 중복 문서를 제거하고, 유사도를 기준으로 문서를 필터링하는 압축 파이프라인을 생성합니다.from dotenv import load_dotenv
from langchain_teddynote import logging
from langchain_community.document_loaders import TextLoader
from langchain_community.vectorstores import FAISS
from langchain_openai import OpenAIEmbeddings
from langchain_text_splitters import CharacterTextSplitter
from langchain_teddynote.document_compressors import LLMChainExtractor, LLMChainFilter
from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import EmbeddingsFilter, DocumentCompressorPipeline
from langchain_community.document_transformers import EmbeddingsRedundantFilter
from langchain_openai import ChatOpenAI
# 환경 변수에서 API 키 로드
load_dotenv()
# LangSmith 추적 시작
logging.langsmith("CH11-Retriever")
# 문서 로드 및 분할
loader = TextLoader("./data/appendix-keywords.txt")
texts = loader.load_and_split(CharacterTextSplitter(chunk_size=300, chunk_overlap=0))
# 기본 retriever 설정
retriever = FAISS.from_documents(texts, OpenAIEmbeddings()).as_retriever()
# LLM 초기화 및 문서 압축기 생성
llm = ChatOpenAI(temperature=0, model="gpt-4o-mini")
compressor = LLMChainExtractor.from_llm(llm)
compression_retriever = ContextualCompressionRetriever(
base_compressor=compressor,
base_retriever=retriever,
)
# 예시 쿼리
docs = compression_retriever.invoke("Semantic Search 에 대해서 알려줘.")
pretty_print_docs(docs)
이와 같은 ContextualCompressionRetriever는 대규모 데이터베이스에서 효율적으로 정보를 검색하고, 관련성 있는 정보를 압축하여 반환함으로써 LLM 호출 비용을 절감하고, 응답의 품질을 향상시킵니다.
앙상블 검색기는 여러 검색 알고리즘을 결합하여 더 나은 검색 결과를 제공하는 시스템입니다. 이 검색기는 다양한 검색기에서 제공하는 정보를 통합하여 단일 검색기보다 더 정확하고 강력한 결과를 도출하는 데 중점을 둡니다.
다양한 검색기 통합:
결과 재순위화:
가중치 기반 조정:
from dotenv import load_dotenv
from langchain_teddynote import logging
from langchain.retrievers import BM25Retriever, EnsembleRetriever
from langchain.vectorstores import FAISS
from langchain_openai import OpenAIEmbeddings
# API 키 정보 로드
load_dotenv()
# LangSmith 추적 시작
logging.langsmith("CH11-Retriever")
# 샘플 문서 리스트
doc_list = [
"I like apples",
"I like apple company",
"I like apple's iphone",
"Apple is my favorite company",
"I like apple's ipad",
"I like apple's macbook",
]
# bm25 retriever와 faiss retriever 초기화
bm25_retriever = BM25Retriever.from_texts(doc_list)
bm25_retriever.k = 1 # 검색 결과 개수를 1로 설정
embedding = OpenAIEmbeddings() # OpenAI 임베딩 사용
faiss_vectorstore = FAISS.from_texts(doc_list, embedding)
faiss_retriever = faiss_vectorstore.as_retriever(search_kwargs={"k": 1})
# 앙상블 retriever 초기화
ensemble_retriever = EnsembleRetriever(
retrievers=[bm25_retriever, faiss_retriever],
weights=[0.7, 0.3], # 각 검색기에 가중치 부여
)
# 검색 결과 가져오기
query = "my favorite fruit is apple"
ensemble_result = ensemble_retriever.invoke(query)
bm25_result = bm25_retriever.invoke(query)
faiss_result = faiss_retriever.invoke(query)
# 검색 결과 출력
print("[Ensemble Retriever]")
for doc in ensemble_result:
print(f"Content: {doc.page_content}\n")
print("[BM25 Retriever]")
for doc in bm25_result:
print(f"Content: {doc.page_content}\n")
print("[FAISS Retriever]")
for doc in faiss_result:
print(f"Content: {doc.page_content}\n")
앙상블 검색기에서 가중치를 동적으로 변경할 수 있습니다. ConfigurableField 클래스를 사용하여 가중치를 런타임에 조정할 수 있습니다.
from langchain_core.runnables import ConfigurableField
# ConfigurableField를 사용하여 가중치 동적 변경 가능
ensemble_retriever = EnsembleRetriever(
retrievers=[bm25_retriever, faiss_retriever],
).configurable_fields(
weights=ConfigurableField(
id="ensemble_weights",
name="Ensemble Weights",
description="Ensemble Weights",
)
)
# BM25 retriever에 가중치를 더 부여
config = {"configurable": {"ensemble_weights": [1, 0]}}
docs = ensemble_retriever.invoke("my favorite fruit is apple", config=config)
print(docs)
# FAISS retriever에 가중치를 더 부여
config = {"configurable": {"ensemble_weights": [0, 1]}}
docs = ensemble_retriever.invoke("my favorite fruit is apple", config=config)
print(docs)
앙상블 검색기는 다수의 검색 알고리즘을 통합하여 검색 정확도를 높이는 데 유용합니다. 각 검색기의 결과에 가중치를 부여하거나 런타임에 검색기 설정을 동적으로 변경할 수 있어, 다양한 검색 시나리오에서 효과적인 검색 결과를 제공합니다.
긴 문맥 재정렬(LongContextReorder)는 검색된 문서의 순서를 조정하여, 모델이 긴 문맥 중에서 더 중요한 정보를 효과적으로 활용할 수 있도록 하는 방법입니다. 긴 컨텍스트에서 모델의 성능 저하를 방지하기 위해 문서의 순서를 재배열하는 기법으로, 중요한 문서가 컨텍스트의 시작과 끝에 위치하도록 합니다.
%pip install --upgrade --quiet sentence-transformers > /dev/null
from langchain.prompts import PromptTemplate
from langchain_community.document_transformers import LongContextReorder
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
# 임베딩을 가져옵니다.
embeddings = OpenAIEmbeddings()
texts = [
"이건 그냥 내가 아무렇게나 적어본 글입니다.",
"사용자와 대화하는 것처럼 설계된 AI인 ChatGPT는 다양한 질문에 답할 수 있습니다.",
"아이폰, 아이패드, 맥북 등은 애플이 출시한 대표적인 제품들입니다.",
"챗GPT는 OpenAI에 의해 개발되었으며, 지속적으로 개선되고 있습니다.",
"챗지피티는 사용자의 질문을 이해하고 적절한 답변을 생성하기 위해 대량의 데이터를 학습했습니다.",
"애플 워치와 에어팟 같은 웨어러블 기기도 애플의 인기 제품군에 속합니다.",
"ChatGPT는 복잡한 문제를 해결하거나 창의적인 아이디어를 제안하는 데에도 사용될 수 있습니다.",
"비트코인은 디지털 금이라고도 불리며, 가치 저장 수단으로서 인기를 얻고 있습니다.",
"ChatGPT의 기능은 지속적인 학습과 업데이트를 통해 더욱 발전하고 있습니다.",
"FIFA 월드컵은 네 번째 해마다 열리며, 국제 축구에서 가장 큰 행사입니다.",
]
# 검색기를 생성합니다.
retriever = Chroma.from_texts(texts, embedding=embeddings).as_retriever(
search_kwargs={"k": 10}
)
query = "ChatGPT에 대해 무엇을 말해줄 수 있나요?"
# 관련성 점수에 따라 정렬된 관련 문서를 가져옵니다.
docs = retriever.get_relevant_documents(query)
# 문서를 재정렬합니다.
reordering = LongContextReorder()
reordered_docs = reordering.transform_documents(docs)
# 재정렬된 문서 출력
for doc in reordered_docs:
print(doc.page_content)
from langchain.prompts import ChatPromptTemplate
from operator import itemgetter
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnableLambda
# 프롬프트 템플릿 정의
template = """Given this text extracts:
{context}
-----
Please answer the following question:
{question}
Answer in the following languages: {language}
"""
prompt = ChatPromptTemplate.from_template(template)
# Chain 정의
chain = (
{
"context": itemgetter("question")
| retriever
| RunnableLambda(reorder_documents),
"question": itemgetter("question"),
"language": itemgetter("language"),
}
| prompt
| ChatOpenAI()
| StrOutputParser()
)
# 쿼리와 언어 입력
answer = chain.invoke(
{"question": "ChatGPT에 대해 무엇을 말해줄 수 있나요?", "language": "KOREAN"}
)
# 답변 출력
print(answer)
긴 문맥 재정렬(LongContextReorder)은 긴 문맥에서 중요한 정보를 모델이 더 잘 활용할 수 있도록 문서 순서를 재배열하는 기법입니다. 이 기법을 사용하면, 모델의 성능 저하를 방지하고, 관련성이 높은 문서가 더 잘 활용되도록 도와줍니다.
Parent Document Retriever는 문서를 효율적으로 검색하고 관리하는 도구로, 문서를 작은 청크로 나눈 후 검색할 때 원본 문서의 맥락을 유지하면서 검색 성능을 향상시키는 데 중점을 둡니다. 이 도구는 특히 긴 문서나 여러 섹션으로 나뉜 문서를 다룰 때 유용합니다.
작은 청크와 맥락 유지:
부모 문서의 역할:
효율적인 검색:
%pip install -qU deeplake
from langchain.storage import InMemoryStore
from langchain_community.document_loaders import TextLoader
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain.retrievers import ParentDocumentRetriever
# 여러 개의 텍스트 파일 로드
loaders = [
TextLoader("./data/ai-story.txt"),
TextLoader("./data/appendix-keywords.txt"),
]
docs = [] # 문서를 저장할 리스트
for loader in loaders:
docs.extend(loader.load())
# 자식 분할기를 생성
child_splitter = RecursiveCharacterTextSplitter(chunk_size=300)
# DB 생성
vectorstore = Chroma(
collection_name="full_documents", embedding_function=OpenAIEmbeddings()
)
store = InMemoryStore()
# Retriever 생성
retriever = ParentDocumentRetriever(
vectorstore=vectorstore,
docstore=store,
child_splitter=child_splitter,
)
# 문서 추가
retriever.add_documents(docs, ids=None, add_to_docstore=True)
# 유사도 검색 수행
sub_docs = vectorstore.similarity_search("Word2Vec")
# 검색된 첫 번째 문서의 내용 출력
print(sub_docs[0].page_content)
# 전체 문서 기반 검색 수행
retrieved_docs = retriever.get_relevant_documents("Word2Vec")
# 검색된 문서의 일부 내용 출력
print(f"문서의 길이: {len(retrieved_docs[0].page_content)}")
print(retrieved_docs[0].page_content[2000:2500])
# 부모 문서와 자식 문서 생성
parent_splitter = RecursiveCharacterTextSplitter(chunk_size=900)
child_splitter = RecursiveCharacterTextSplitter(chunk_size=300)
# 자식 청크를 인덱싱하는 데 사용할 벡터 저장소
vectorstore = Chroma(
collection_name="split_parents", embedding_function=OpenAIEmbeddings()
)
store = InMemoryStore()
# ParentDocumentRetriever 초기화
retriever = ParentDocumentRetriever(
vectorstore=vectorstore,
docstore=store,
child_splitter=child_splitter,
parent_splitter=parent_splitter,
)
# 문서 추가
retriever.add_documents(docs)
# 유사도 검색 수행
sub_docs = vectorstore.similarity_search("Word2Vec")
print(sub_docs[0].page_content)
# 전체 문서 기반 검색 수행
retrieved_docs = retriever.get_relevant_documents("Word2Vec")
print(retrieved_docs[0].page_content)
Parent Document Retriever는 문서를 작은 청크로 나누어 검색 성능을 높이는 동시에, 검색된 청크가 속한 원본 문서를 반환하여 전체 맥락을 유지하는 도구입니다. 이를 통해 문서 검색의 정확도와 효율성을 동시에 향상시킬 수 있습니다.
MultiQueryRetriever는 주어진 쿼리에 대해 다양한 관점에서 여러 쿼리를 자동으로 생성하여 검색 결과의 범위와 깊이를 확장하는 도구입니다. 이는 단일 쿼리가 포착하지 못하는 세부적인 차이나 의미를 보완하여, 검색 결과를 더욱 풍부하게 만들어줍니다.
쿼리 확장:
풍부한 검색 결과:
자동 프롬프트 튜닝:
from dotenv import load_dotenv
from langchain_community.document_loaders import WebBaseLoader
from langchain.vectorstores import FAISS
from langchain_openai import OpenAIEmbeddings
from langchain_text_splitters import RecursiveCharacterTextSplitter
# API 키 정보 로드
load_dotenv()
# 블로그 포스트 로드
loader = WebBaseLoader(
"https://teddylee777.github.io/openai/openai-assistant-tutorial/", encoding="utf-8"
)
# 문서 분할
text_splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=0)
docs = loader.load_and_split(text_splitter)
# 임베딩 정의
openai_embedding = OpenAIEmbeddings()
# 벡터DB 생성
db = FAISS.from_documents(docs, openai_embedding)
retriever = db.as_retriever()
# 단일 쿼리로 검색
query = "OpenAI Assistant API의 Functions 사용법에 대해 알려주세요."
relevant_docs = retriever.get_relevant_documents(query)
# 검색된 문서의 개수와 내용을 확인
print(f"검색된 문서 개수: {len(relevant_docs)}")
print(relevant_docs[1].page_content)
from langchain.retrievers.multi_query import MultiQueryRetriever
from langchain_openai import ChatOpenAI
# 언어 모델 초기화
llm = ChatOpenAI(temperature=0)
# MultiQueryRetriever 초기화
multiquery_retriever = MultiQueryRetriever.from_llm(
retriever=db.as_retriever(), llm=llm
)
# 다중 쿼리로 검색
question = "OpenAI Assistant API의 Functions 사용법에 대해 알려주세요."
relevant_docs = multiquery_retriever.get_relevant_documents(query=question)
# 검색된 문서의 개수와 내용을 확인
print(f"===============\n검색된 문서 개수: {len(relevant_docs)}\n===============")
print(relevant_docs[0].page_content)
import logging
# 로깅 설정
logging.getLogger("langchain.retrievers.multi_query").setLevel(logging.INFO)
# 다중 쿼리로 검색
relevant_docs = multiquery_retriever.get_relevant_documents(query=question)
# 검색된 문서의 개수와 내용을 확인
print(f"===============\n검색된 문서 개수: {len(relevant_docs)}\n===============")
print(relevant_docs[0].page_content)
from langchain_core.runnables import RunnablePassthrough
from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI
# 프롬프트 템플릿 생성
prompt = PromptTemplate.from_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.
Your response should be a list of values separated by new lines, eg: `foo\nbar\nbaz\n`
#ORIGINAL QUESTION:
{question}
"""
)
# 언어 모델 인스턴스를 생성
llm = ChatOpenAI(temperature=0)
# 체인 생성
chain = {"question": RunnablePassthrough()} | prompt | llm | StrOutputParser()
# 질문에 대한 다중 쿼리 생성
question = "OpenAI Assistant API의 Functions 사용법에 대해 알려주세요."
multi_queries = chain.invoke({"question": question})
print(multi_queries)
# MultiQueryRetriever 초기화
multiquery_retriever = MultiQueryRetriever.from_llm(
llm=chain, retriever=db.as_retriever()
)
# 다중 쿼리로 문서 검색
relevant_docs = multiquery_retriever.get_relevant_documents(query=question)
# 검색된 문서의 개수와 내용을 확인
print(f"===============\n검색된 문서 개수: {len(relevant_docs)}\n===============")
print(relevant_docs[0].page_content)
MultiQueryRetriever는 다양한 관점에서 여러 쿼리를 생성하여 벡터 기반 검색의 한계를 극복하고, 더욱 풍부하고 정확한 검색 결과를 제공합니다. 이를 통해 프롬프트 튜닝을 자동화하고, 단일 쿼리보다 더 많은 관련 문서를 검색할 수 있습니다.
다중 벡터저장소 검색기(MultiVectorRetriever)를 사용하여 문서당 여러 벡터를 생성하고 검색 성능을 향상시키는 방법에 대해 자세히 설명하겠습니다.
MultiVectorRetriever는 문서당 여러 벡터를 생성하고 저장함으로써, 다양한 쿼리에 대해 더 정확하고 효율적인 검색 결과를 제공합니다. 여러 벡터를 사용함으로써 한 문서의 다양한 측면이나 세부 사항을 더 잘 포착할 수 있으며, 검색 시 사용자의 요구에 더 잘 부합하는 문서를 반환할 수 있습니다.
작은 청크 생성:
요약 임베딩:
가설 질문 활용:
수동 추가 방식:
텍스트 로드 및 전처리
from langchain_community.document_loaders import TextLoader
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain.retrievers.multi_vector import MultiVectorRetriever
loaders = [
TextLoader("./data/ai-story.txt"),
TextLoader("./data/appendix-keywords.txt"),
]
docs = []
for loader in loaders:
docs.extend(loader.load())
작은 청크 생성
import uuid
vectorstore = Chroma(
collection_name="full_documents", embedding_function=OpenAIEmbeddings()
)
store = InMemoryByteStore()
id_key = "doc_id"
retriever = MultiVectorRetriever(
vectorstore=vectorstore,
byte_store=store,
id_key=id_key,
)
doc_ids = [str(uuid.uuid4()) for _ in docs]
청크 분할
parent_text_splitter = RecursiveCharacterTextSplitter(chunk_size=4000)
child_text_splitter = RecursiveCharacterTextSplitter(chunk_size=400)
parent_docs = []
for i, doc in enumerate(docs):
_id = doc_ids[i]
parent_doc = parent_text_splitter.split_documents([doc])
for _doc in parent_doc:
_doc.metadata[id_key] = _id
parent_docs.extend(parent_doc)
child_docs = []
for i, doc in enumerate(docs):
_id = doc_ids[i]
child_doc = child_text_splitter.split_documents([doc])
for _doc in child_doc:
_doc.metadata[id_key] = _id
child_docs.extend(child_doc)
벡터 저장소에 문서 추가
retriever.vectorstore.add_documents(parent_docs)
retriever.vectorstore.add_documents(child_docs)
retriever.docstore.mset(list(zip(doc_ids, docs)))
유사도 검색 수행
result_docs = vectorstore.similarity_search("Word2Vec 의 정의?")
print(result_docs[0].page_content)
가설 질문 생성 및 저장
from langchain.output_parsers.openai_functions import JsonKeyOutputFunctionsParser
from langchain.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
functions = [
{
"name": "hypothetical_questions",
"description": "Generate hypothetical questions",
"parameters": {
"type": "object",
"properties": {
"questions": {
"type": "array",
"items": {
"type": "string"
},
},
},
"required": ["questions"],
},
}
]
chain = (
{"doc": lambda x: x.page_content}
| ChatPromptTemplate.from_template(
"Generate a list of exactly 3 hypothetical questions that the below document could be used to answer. Answer in Korean:\n\n{doc}"
)
| ChatOpenAI(max_retries=0, model="gpt-4-turbo-preview").bind(
functions=functions, function_call={"name": "hypothetical_questions"}
)
| JsonKeyOutputFunctionsParser(key_name="questions")
)
hypothetical_questions = chain.batch(docs, {"max_concurrency": 5})
가설 질문을 벡터 저장소에 저장
question_docs = [
Document(page_content=s, metadata={id_key: doc_ids[i]})
for i, s in enumerate(hypothetical_questions)
]
retriever.vectorstore.add_documents(question_docs)
retriever.docstore.mset(list(zip(doc_ids, docs)))
검색 및 결과 확인
result_docs = vectorstore.similarity_search("Word2Vec에 대한 정의는 뭐야?")
print(result_docs[0].page_content)
MultiVectorRetriever는 문서를 다양한 방식으로 벡터화하고 저장하여, 검색 성능을 크게 향상시킬 수 있는 도구입니다. 이를 통해 사용자는 더 많은 정보와 문서를 탐색할 수 있으며, 보다 정확한 검색 결과를 얻을 수 있습니다. 이를 통해 문서 검색 및 정보 검색의 효율성을 높일 수 있습니다.
SelfQueryRetriever는 사용자의 자연어 질의를 받아들여, 이 질의를 기반으로 구조화된 쿼리를 자동으로 생성하고, 해당 쿼리를 벡터 저장소에 적용하여 관련된 문서를 검색하는 강력한 도구입니다. 이 과정은 크게 두 단계로 나뉩니다:
이를 통해, SelfQueryRetriever는 보다 정교한 검색을 수행할 수 있으며, 메타데이터를 활용하여 보다 정확한 결과를 제공할 수 있습니다.
먼저, 영화에 대한 간략한 설명과 메타데이터가 포함된 문서 객체 리스트를 생성하고, 이를 기반으로 Chroma 벡터 저장소를 구축합니다.
from langchain_community.vectorstores import Chroma
from langchain_core.documents import Document
from langchain_openai import OpenAIEmbeddings
docs = [
Document(
page_content="A bunch of scientists bring back dinosaurs and mayhem breaks loose",
metadata={"year": 1993, "rating": 7.7, "genre": "science fiction"},
),
Document(
page_content="Leo DiCaprio gets lost in a dream within a dream within a dream within a ...",
metadata={"year": 2010, "director": "Christopher Nolan", "rating": 8.2},
),
Document(
page_content="A psychologist / detective gets lost in a series of dreams within dreams within dreams and Inception reused the idea",
metadata={"year": 2006, "director": "Satoshi Kon", "rating": 8.6},
),
Document(
page_content="A bunch of normal-sized women are supremely wholesome and some men pine after them",
metadata={"year": 2019, "director": "Greta Gerwig", "rating": 8.3},
),
Document(
page_content="Toys come alive and have a blast doing so",
metadata={"year": 1995, "genre": "animated"},
),
Document(
page_content="Three men walk into the Zone, three men walk out of the Zone",
metadata={
"year": 1979,
"director": "Andrei Tarkovsky",
"genre": "thriller",
"rating": 9.9,
},
),
]
vectorstore = Chroma.from_documents(docs, OpenAIEmbeddings())
이제 SelfQueryRetriever를 설정합니다. 여기에는 문서의 메타데이터 필드 정보와 문서 내용에 대한 설명을 제공합니다.
from langchain.chains.query_constructor.base import AttributeInfo
from langchain.retrievers.self_query.base import SelfQueryRetriever
from langchain_openai import ChatOpenAI
metadata_field_info = [
AttributeInfo(
name="genre",
description="The genre of the movie. One of ['science fiction', 'comedy', 'drama', 'thriller', 'romance', 'action', 'animated']",
type="string",
),
AttributeInfo(
name="year",
description="The year the movie was released",
type="integer",
),
AttributeInfo(
name="director",
description="The name of the movie director",
type="string",
),
AttributeInfo(
name="rating", description="A 1-10 rating for the movie", type="float"
),
]
document_content_description = "Brief summary of a movie"
llm = ChatOpenAI(model="gpt-4-turbo-preview", temperature=0)
retriever = SelfQueryRetriever.from_llm(
llm,
vectorstore,
document_content_description,
metadata_field_info,
)
이제 SelfQueryRetriever를 사용하여 다양한 조건의 검색을 수행할 수 있습니다.
예시 1: 평점이 8.5 이상인 영화 검색
retriever.invoke("I want to watch a movie rated higher than 8.5")
예시 2: Greta Gerwig이 감독한 여성에 관한 영화 검색
retriever.invoke("Has Greta Gerwig directed any movies about women")
예시 3: 평점이 8.5 이상인 SF 영화 검색
retriever.invoke("What's a highly rated (above 8.5) science fiction film?")
SelfQueryRetriever에서 k 값을 지정하여 검색 결과의 개수를 제한할 수 있습니다.
retriever = SelfQueryRetriever.from_llm(
llm,
vectorstore,
document_content_description,
metadata_field_info,
enable_limit=True,
search_kwargs={"k": 2},
)
retriever.invoke("What are movies about dinosaurs")
필요에 따라 query_constructor와 structured_query_translator를 사용자 정의하여 더 정교한 검색을 수행할 수 있습니다.
from langchain.retrievers.self_query.chroma import ChromaTranslator
from langchain.chains.query_constructor.base import (
StructuredQueryOutputParser,
get_query_constructor_prompt,
)
prompt = get_query_constructor_prompt(
document_content_description,
metadata_field_info,
)
output_parser = StructuredQueryOutputParser.from_components()
query_constructor = prompt | llm | output_parser
retriever = SelfQueryRetriever(
query_constructor=query_constructor,
vectorstore=vectorstore,
structured_query_translator=ChromaTranslator(),
)
retriever.invoke("What's a movie after 1990 but before 2005 that's all about toys, and preferably is animated")
SelfQueryRetriever는 자연어 질의를 구조화된 쿼리로 변환하여 보다 정밀한 검색을 수행하는 도구입니다. 메타데이터를 활용한 복합 조건 검색이 가능하며, 사용자가 원하는 결과를 더 정확하게 제공할 수 있습니다. 이 기능은 특히 다양한 메타데이터 필드를 가진 문서에서 매우 유용하게 사용될 수 있습니다.
TimeWeightedVectorStoreRetriever는 문서의 "신선함"과 "관련성"을 모두 고려하여 검색 결과를 제공하는 도구입니다. 이는 특정 시점에 문서가 얼마나 자주 접근되었는지를 고려하고, 시간이 지남에 따라 문서의 점수를 감쇠시키는 방식으로 작동합니다. 이 방법을 통해 최신성(Recency)과 유사도(Semantic Similarity) 사이의 균형을 유지하며, 최신 정보일수록 더 높은 점수를 부여하여 검색 결과에 반영합니다.
다음은 TimeWeightedVectorStoreRetriever를 설정하고 사용하는 방법을 단계별로 설명한 예제입니다.
감쇠율이 매우 낮을 때, 문서의 정보는 오랫동안 유지됩니다. 이는 문서가 시간이 지나도 여전히 높은 점수를 유지할 수 있음을 의미합니다.
from datetime import datetime, timedelta
import faiss
from langchain.docstore import InMemoryDocstore
from langchain.retrievers import TimeWeightedVectorStoreRetriever
from langchain_community.vectorstores import FAISS
from langchain_core.documents import Document
from langchain_openai import OpenAIEmbeddings
# 임베딩 모델 설정
embeddings_model = OpenAIEmbeddings()
# 벡터 저장소 초기화
embedding_size = 1536
index = faiss.IndexFlatL2(embedding_size)
vectorstore = FAISS(embeddings_model, index, InMemoryDocstore({}), {})
# TimeWeightedVectorStoreRetriever 초기화 (감쇠율을 0에 가깝게 설정)
retriever = TimeWeightedVectorStoreRetriever(
vectorstore=vectorstore, decay_rate=0.0000000000000000000000001, 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")])
# 검색
retriever.get_relevant_documents("hello world")
이 경우, "hello world" 문서가 가장 먼저 반환됩니다. 이는 낮은 감쇠율로 인해 시간이 지나도 여전히 높은 점수를 유지하기 때문입니다.
감쇠율이 높을 때, 오래된 정보는 빠르게 잊혀집니다. 따라서 최신 정보가 더 높은 점수를 받을 가능성이 높아집니다.
# TimeWeightedVectorStoreRetriever 초기화 (감쇠율을 0.999로 설정)
retriever = TimeWeightedVectorStoreRetriever(
vectorstore=vectorstore, decay_rate=0.999, 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")])
# 검색
retriever.get_relevant_documents("hello world")
이 경우, "hello foo" 문서가 먼저 반환됩니다. 이는 높은 감쇠율로 인해 "hello world" 문서가 대부분 잊혀졌기 때문입니다.
가상의 시간을 설정하여 시간 흐름에 따른 검색 결과 변화를 테스트할 수 있습니다.
import datetime
from langchain.utils import mock_now
# 가상의 시간을 설정합니다.
with mock_now(datetime.datetime(2024, 3, 28, 10, 11)):
# "hello world"와 관련된 문서를 검색하고 출력합니다.
print(retriever.get_relevant_documents("hello world"))
이 설정을 통해, 문서가 마지막으로 접근된 시점을 임의로 설정하고, 그에 따라 문서의 "신선함"을 평가할 수 있습니다.
TimeWeightedVectorStoreRetriever는 최신성과 관련성을 모두 고려한 동적인 검색 결과를 제공합니다. 감쇠율을 조정함으로써, 정보의 신선함을 얼마나 중요시할지를 결정할 수 있으며, 이를 통해 사용자는 보다 적절한 검색 결과를 얻을 수 있습니다.
이 실습에서는 한글 형태소 분석기와 BM25 검색기를 결합하여, 한국어 텍스트에 대해 보다 정확한 유사도 검색을 수행하는 방법을 살펴보겠습니다. 다양한 형태소 분석기(Kiwi, Kkma, Okt)를 활용하여 문서를 처리한 후 BM25 알고리즘으로 유사도를 계산해 보겠습니다.
Kiwi는 한국어 형태소 분석기 중 하나로, 문서를 분석하여 BM25 검색기를 사용해 유사도 검색을 수행할 수 있습니다.
from langchain_community.retrievers import BM25Retriever
from langchain_teddynote.retrievers import KiwiBM25Retriever
sample_texts = [
"금융보험은 장기적인 자산 관리와 위험 대비를 목적으로 고안된 금융 상품입니다.",
"금융저축산물보험은 장기적인 저축 목적과 더불어, 축산물 제공 기능을 갖추고 있는 특별 금융 상품입니다.",
"금융보씨 험한말 좀 하지마시고, 저축이나 좀 하시던가요. 뭐가 그리 급하신지 모르겠네요.",
"금융단폭격보험은 저축은 커녕 위험 대비에 초점을 맞춘 상품입니다. 높은 위험을 감수하고자 하는 고객에게 적합합니다.",
]
# KiwiBM25Retriever 초기화
kiwi = KiwiBM25Retriever.from_texts(sample_texts)
# 유사도 검색 수행
pretty_print(kiwi.invoke("금융보험"))
유사도 검색을 수행한 후, 검색된 결과에 유사도 점수를 계산하여 메타데이터에 추가한 뒤 출력할 수 있습니다.
# 유사도 점수 계산 및 출력
pretty_print(kiwi.search_with_score("금융보험"))
Kiwi를 사용한 BM25 검색기와 일반 BM25 검색기를 비교하여 결과를 확인해 볼 수 있습니다.
bm25 = BM25Retriever.from_texts(sample_texts)
# BM25와 KiwiBM25 비교
print(f'Kiwi: \t {kiwi.invoke("금융보험")[0].page_content}')
print(f'BM25: \t {bm25.invoke("금융보험")[0].page_content}')
KonlPy는 Python에서 제공하는 다양한 한국어 형태소 분석기(Kkma, Okt 등)를 포함한 패키지입니다. 이를 사용하여 BM25 검색기를 설정할 수 있습니다.
from langchain_teddynote.retrievers import KkmaBM25Retriever, OktBM25Retriever
# KkmaBM25Retriever와 OktBM25Retriever 초기화
kkma = KkmaBM25Retriever.from_texts(sample_texts)
okt = OktBM25Retriever.from_texts(sample_texts)
# Kkma + BM25 유사도 검색
pretty_print(kkma.invoke("금융보험"))
pretty_print(kkma.search_with_score("금융보험"))
# Okt + BM25 유사도 검색
pretty_print(okt.invoke("금융보험"))
pretty_print(okt.search_with_score("금융보험"))
위 실습을 통해 다양한 한국어 형태소 분석기를 사용하여 BM25 알고리즘으로 텍스트 유사도를 계산하고, 각 검색기(Kiwi, Kkma, Okt)의 성능을 비교할 수 있습니다. 각 형태소 분석기의 특성과 성능 차이를 비교해 보면서, 특정 도메인이나 데이터셋에 가장 적합한 검색기를 선택할 수 있습니다.
Kiwi, Kkma, Okt와 같은 한국어 형태소 분석기를 활용한 BM25 검색기는 한국어 텍스트의 유사도를 정확하게 계산할 수 있는 효과적인 도구입니다. 이 접근 방식을 통해 보다 정교하고 관련성 높은 검색 결과를 얻을 수 있습니다.