LangChain 및 Milvus에서 전체 텍스트 검색 사용하기

전체 텍스트 검색은 텍스트 내에서 키워드를 직접 일치시켜 특정 용어나 구문이 포함된 문서를 검색하는 전통적인 방법입니다. 일반적으로 용어 빈도 및 근접성과 같은 요소에 의해 결정되는 관련성에 따라 결과의 순위를 매깁니다. 시맨틱 검색은 의도와 문맥을 이해하는 데 탁월하지만, 전체 텍스트 검색은 정확한 키워드 매칭을 위한 정밀도를 제공하므로 상호 보완적인 도구로 유용합니다. BM25 알고리즘은 전체 텍스트 검색에 널리 사용되는 순위 지정 방법으로, 특히 검색 증강 생성(RAG)에 유용합니다.

Milvus 2.5부터는 BM25 알고리즘을 스파스 벡터로 표현하는 스파스-BM25 접근 방식을 통해 전체 텍스트 검색이 기본적으로 지원됩니다. Milvus는 원시 텍스트를 입력으로 받아 지정된 필드에 저장된 스파스 벡터로 자동 변환하므로 수동으로 스파스 임베딩을 생성할 필요가 없습니다.

LangChain과 Milvus의 통합으로 이 기능도 도입되어 전체 텍스트 검색을 RAG 애플리케이션에 통합하는 프로세스가 간소화되었습니다. 전체 텍스트 검색과 밀도 벡터를 사용한 시맨틱 검색을 결합함으로써 밀도 임베딩의 시맨틱 컨텍스트와 단어 매칭의 정확한 키워드 관련성을 모두 활용하는 하이브리드 접근 방식을 달성할 수 있습니다. 이러한 통합은 검색 시스템의 정확성, 연관성 및 사용자 경험을 향상시킵니다.

이 튜토리얼에서는 애플리케이션에서 전체 텍스트 검색을 구현하기 위해 LangChain과 Milvus를 사용하는 방법을 보여드립니다.

전제 조건

이 노트북을 실행하기 전에 다음 종속성이 설치되어 있는지 확인하세요:

$ pip install --upgrade --quiet  langchain langchain-core langchain-community langchain-text-splitters langchain-milvus langchain-openai bs4 #langchain-voyageai

OpenAI의 모델을 사용하겠습니다. OpenAI에서 환경 변수 OPENAI_API_KEY 를 준비해야 합니다.

import os

os.environ["OPENAI_API_KEY"] = "sk-***********"

Milvus 서버 URI (및 선택적으로 TOKEN)를 지정합니다. Milvus 서버를 설치하고 시작하는 방법은 이 가이드를 따르세요.

URI = "http://localhost:19530"
# TOKEN = ...

몇 가지 예제 문서를 준비합니다:

from langchain_core.documents import Document

docs = [
    Document(page_content="I like this apple", metadata={"category": "fruit"}),
    Document(page_content="I like swimming", metadata={"category": "sport"}),
    Document(page_content="I like dogs", metadata={"category": "pets"}),
]

BM25 함수를 사용한 초기화

하이브리드 검색

전체 텍스트 검색을 위해 Milvus VectorStore는 builtin_function 파라미터를 허용합니다. 이 매개변수를 통해 BM25BuiltInFunction 의 인스턴스를 전달할 수 있습니다. 이는 일반적으로 밀도가 높은 임베딩을 VectorStore 으로 전달하는 시맨틱 검색과는 다릅니다,

다음은 시맨틱 검색을 위한 OpenAI 고밀도 임베딩과 전체 텍스트 검색을 위한 BM25를 사용한 Milvus의 간단한 하이브리드 검색 예시입니다:

from langchain_milvus import Milvus, BM25BuiltInFunction
from langchain_openai import OpenAIEmbeddings


vectorstore = Milvus.from_documents(
    documents=docs,
    embedding=OpenAIEmbeddings(),
    builtin_function=BM25BuiltInFunction(),
    # `dense` is for OpenAI embeddings, `sparse` is the output field of BM25 function
    vector_field=["dense", "sparse"],
    connection_args={
        "uri": URI,
    },
    consistency_level="Strong",
    drop_old=True,
)

위의 코드에서는 BM25BuiltInFunction 의 인스턴스를 정의하고 Milvus 객체에 전달합니다. BM25BuiltInFunction 는 Milvus의 경량 래퍼 클래스입니다. Function 의 경량 래퍼 클래스입니다.

BM25BuiltInFunction 의 매개변수에서 이 함수에 대한 입력 및 출력 필드를 지정할 수 있습니다:

• input_field_names (str): 입력 필드의 이름, 기본값은 text 입니다. 이 함수가 입력으로 읽을 필드를 나타냅니다.
• output_field_names (str): 출력 필드의 이름, 기본값은 sparse 입니다. 이 함수가 계산된 결과를 출력할 필드를 나타냅니다.

위에서 언급한 Milvus 초기화 매개변수에서는 vector_field=["dense", "sparse"] 도 지정합니다. sparse 필드는 BM25BuiltInFunction 에 정의된 출력 필드로 간주되므로 다른 dense 필드는 OpenAIEmbedding의 출력 필드에 자동으로 할당됩니다.

실제로, 특히 여러 임베딩이나 함수를 결합할 때는 모호성을 피하기 위해 각 함수에 대한 입력 및 출력 필드를 명시적으로 지정하는 것이 좋습니다.

다음 예시에서는 BM25BuiltInFunction 의 입력 및 출력 필드를 명시적으로 지정하여 내장 함수가 어떤 필드를 위한 것인지 명확히 합니다.

# from langchain_voyageai import VoyageAIEmbeddings

embedding1 = OpenAIEmbeddings(model="text-embedding-ada-002")
embedding2 = OpenAIEmbeddings(model="text-embedding-3-large")
# embedding2 = VoyageAIEmbeddings(model="voyage-3")  # You can also use embedding from other embedding model providers, e.g VoyageAIEmbeddings


vectorstore = Milvus.from_documents(
    documents=docs,
    embedding=[embedding1, embedding2],
    builtin_function=BM25BuiltInFunction(
        input_field_names="text", output_field_names="sparse"
    ),
    text_field="text",  # `text` is the input field name of BM25BuiltInFunction
    # `sparse` is the output field name of BM25BuiltInFunction, and `dense1` and `dense2` are the output field names of embedding1 and embedding2
    vector_field=["dense1", "dense2", "sparse"],
    connection_args={
        "uri": URI,
    },
    consistency_level="Strong",
    drop_old=True,
)

vectorstore.vector_fields

['dense1', 'dense2', 'sparse']

이 예제에는 세 개의 벡터 필드가 있습니다. 이 중 sparse 은 BM25BuiltInFunction 의 출력 필드로 사용되며, 나머지 두 개의 dense1 과 dense2 은 두 개의 OpenAIEmbeddings 모델의 출력 필드로 자동 할당됩니다(순서에 따라).

이러한 방식으로 여러 개의 벡터 필드를 정의하고 다양한 임베딩 또는 함수 조합을 할당하여 하이브리드 검색을 구현할 수 있습니다.

하이브리드 검색을 수행할 때는 쿼리 텍스트를 전달하고 선택적으로 topK 및 재랭커 매개변수를 설정하기만 하면 됩니다. vectorstore 인스턴스는 벡터 임베딩과 내장 함수를 자동으로 처리하고 마지막으로 재랭커를 사용하여 결과를 구체화합니다. 검색 프로세스의 기본 구현 세부 사항은 사용자에게 숨겨져 있습니다.

vectorstore.similarity_search(
    "Do I like apples?", k=1
)  # , ranker_type="weighted", ranker_params={"weights":[0.3, 0.3, 0.4]})

[Document(metadata={'category': 'fruit', 'pk': 454646931479251897}, page_content='I like this apple')]

하이브리드 검색에 대한 자세한 내용은 하이브리드 검색 소개 및 이 LangChain Milvus 하이브리드 검색 튜토리얼을 참조하세요.

임베딩 없이 BM25 검색

임베딩 기반의 시맨틱 검색을 사용하지 않고 BM25 함수로만 전체 텍스트 검색을 수행하려면 임베딩 파라미터를 None 로 설정하고, BM25 함수 인스턴스로 지정된 builtin_function 만 유지하면 됩니다. 벡터 필드에는 "스파스" 필드만 있습니다. 예를 들어

vectorstore = Milvus.from_documents(
    documents=docs,
    embedding=None,
    builtin_function=BM25BuiltInFunction(
        output_field_names="sparse",
    ),
    vector_field="sparse",
    connection_args={
        "uri": URI,
    },
    consistency_level="Strong",
    drop_old=True,
)

vectorstore.vector_fields

['sparse']

분석기 사용자 지정

분석기는 문장을 토큰으로 나누고 어간 및 중지 단어 제거와 같은 어휘 분석을 수행하여 전체 텍스트 검색에 필수적입니다. 분석기는 일반적으로 언어별로 다릅니다. 이 가이드를 참조하여 Milvus의 분석기에 대해 자세히 알아보세요.

Milvus는 두 가지 유형의 분석기를 지원합니다: 기본 제공 분석 기와 사용자 지정 분석기입니다. 기본적으로 BM25BuiltInFunction 에서는 구두점으로 텍스트를 토큰화하는 가장 기본적인 분석기인 표준 내장 분석기를 사용합니다.

다른 분석기를 사용하거나 분석기를 사용자 정의하려면 BM25BuiltInFunction 초기화에서 analyzer_params 매개 변수를 전달하면 됩니다.

analyzer_params_custom = {
    "tokenizer": "standard",
    "filter": [
        "lowercase",  # Built-in filter
        {"type": "length", "max": 40},  # Custom filter
        {"type": "stop", "stop_words": ["of", "to"]},  # Custom filter
    ],
}


vectorstore = Milvus.from_documents(
    documents=docs,
    embedding=OpenAIEmbeddings(),
    builtin_function=BM25BuiltInFunction(
        output_field_names="sparse",
        enable_match=True,
        analyzer_params=analyzer_params_custom,
    ),
    vector_field=["dense", "sparse"],
    connection_args={
        "uri": URI,
    },
    consistency_level="Strong",
    drop_old=True,
)

Milvus 컬렉션의 스키마를 살펴보고 사용자 정의된 분석기가 올바르게 설정되었는지 확인할 수 있습니다.

vectorstore.col.schema

{'auto_id': True, 'description': '', 'fields': [{'name': 'text', 'description': '', 'type': <DataType.VARCHAR: 21>, 'params': {'max_length': 65535, 'enable_match': True, 'enable_analyzer': True, 'analyzer_params': {'tokenizer': 'standard', 'filter': ['lowercase', {'type': 'length', 'max': 40}, {'type': 'stop', 'stop_words': ['of', 'to']}]}}}, {'name': 'pk', 'description': '', 'type': <DataType.INT64: 5>, 'is_primary': True, 'auto_id': True}, {'name': 'dense', 'description': '', 'type': <DataType.FLOAT_VECTOR: 101>, 'params': {'dim': 1536}}, {'name': 'sparse', 'description': '', 'type': <DataType.SPARSE_FLOAT_VECTOR: 104>, 'is_function_output': True}, {'name': 'category', 'description': '', 'type': <DataType.VARCHAR: 21>, 'params': {'max_length': 65535}}], 'enable_dynamic_field': False, 'functions': [{'name': 'bm25_function_de368e79', 'description': '', 'type': <FunctionType.BM25: 1>, 'input_field_names': ['text'], 'output_field_names': ['sparse'], 'params': {}}]}

더 자세한 개념 설명은 analyzer, tokenizer, filter, enable_match, analyzer_params, 분석기 설명서를 참조하세요.

RAG에서 하이브리드 검색 및 재랭크 사용하기

지금까지 랭체인과 밀버스에서 기본 BM25 빌트인 기능을 사용하는 방법을 알아보았습니다. 이번에는 하이브리드 검색과 재랭킹을 통해 최적화된 RAG 구현을 소개하겠습니다.

이 다이어그램은 키워드 매칭을 위한 BM25와 시맨틱 검색을 위한 벡터 검색을 결합한 하이브리드 검색 및 재랭크 프로세스를 보여줍니다. 두 방법의 결과를 병합하고 순위를 재조정하여 LLM으로 전달하여 최종 답변을 생성합니다.

하이브리드 검색은 정확도와 의미론적 이해의 균형을 유지하여 다양한 쿼리에 대한 정확도와 견고성을 향상시킵니다. BM25 전체 텍스트 검색과 벡터 검색으로 후보를 검색하여 의미론적, 문맥 인식, 정확한 검색을 모두 보장합니다.

예제를 통해 시작해 보겠습니다.

데이터 준비

Langchain WebBaseLoader를 사용해 웹 소스에서 문서를 로드하고 RecursiveCharacterTextSplitter를 사용해 청크로 분할합니다.

import bs4
from langchain_community.document_loaders import WebBaseLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter

# Create a WebBaseLoader instance to load documents from web sources
loader = WebBaseLoader(
    web_paths=(
        "https://lilianweng.github.io/posts/2023-06-23-agent/",
        "https://lilianweng.github.io/posts/2023-03-15-prompt-engineering/",
    ),
    bs_kwargs=dict(
        parse_only=bs4.SoupStrainer(
            class_=("post-content", "post-title", "post-header")
        )
    ),
)
# Load documents from web sources using the loader
documents = loader.load()
# Initialize a RecursiveCharacterTextSplitter for splitting text into chunks
text_splitter = RecursiveCharacterTextSplitter(chunk_size=2000, chunk_overlap=200)

# Split the documents into chunks using the text_splitter
docs = text_splitter.split_documents(documents)

# Let's take a look at the first document
docs[1]

Document(metadata={'source': 'https://lilianweng.github.io/posts/2023-06-23-agent/'}, page_content='Fig. 1. Overview of a LLM-powered autonomous agent system.\nComponent One: Planning#\nA complicated task usually involves many steps. An agent needs to know what they are and plan ahead.\nTask Decomposition#\nChain of thought (CoT; Wei et al. 2022) has become a standard prompting technique for enhancing model performance on complex tasks. The model is instructed to “think step by step” to utilize more test-time computation to decompose hard tasks into smaller and simpler steps. CoT transforms big tasks into multiple manageable tasks and shed lights into an interpretation of the model’s thinking process.\nTree of Thoughts (Yao et al. 2023) extends CoT by exploring multiple reasoning possibilities at each step. It first decomposes the problem into multiple thought steps and generates multiple thoughts per step, creating a tree structure. The search process can be BFS (breadth-first search) or DFS (depth-first search) with each state evaluated by a classifier (via a prompt) or majority vote.\nTask decomposition can be done (1) by LLM with simple prompting like "Steps for XYZ.\\n1.", "What are the subgoals for achieving XYZ?", (2) by using task-specific instructions; e.g. "Write a story outline." for writing a novel, or (3) with human inputs.\nAnother quite distinct approach, LLM+P (Liu et al. 2023), involves relying on an external classical planner to do long-horizon planning. This approach utilizes the Planning Domain Definition Language (PDDL) as an intermediate interface to describe the planning problem. In this process, LLM (1) translates the problem into “Problem PDDL”, then (2) requests a classical planner to generate a PDDL plan based on an existing “Domain PDDL”, and finally (3) translates the PDDL plan back into natural language. Essentially, the planning step is outsourced to an external tool, assuming the availability of domain-specific PDDL and a suitable planner which is common in certain robotic setups but not in many other domains.\nSelf-Reflection#')

밀버스 벡터 저장소에 문서 로드

위에서 소개한 것처럼 준비된 문서를 초기화하여 Milvus 벡터 스토어에 로드합니다. dense 은 OpenAI 임베딩용, sparse 은 BM25 함수용 벡터 필드 두 개가 포함되어 있습니다.

vectorstore = Milvus.from_documents(
    documents=docs,
    embedding=OpenAIEmbeddings(),
    builtin_function=BM25BuiltInFunction(),
    vector_field=["dense", "sparse"],
    connection_args={
        "uri": URI,
    },
    consistency_level="Strong",
    drop_old=True,
)

RAG 체인 구축

LLM 인스턴스와 프롬프트를 준비한 다음 LangChain 표현 언어를 사용하여 RAG 파이프라인으로 결합합니다.

from langchain_core.runnables import RunnablePassthrough
from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI

# Initialize the OpenAI language model for response generation
llm = ChatOpenAI(model_name="gpt-4o", temperature=0)

# Define the prompt template for generating AI responses
PROMPT_TEMPLATE = """
Human: You are an AI assistant, and provides answers to questions by using fact based and statistical information when possible.
Use the following pieces of information to provide a concise answer to the question enclosed in <question> tags.
If you don't know the answer, just say that you don't know, don't try to make up an answer.
<context>
{context}
</context>

<question>
{question}
</question>

The response should be specific and use statistics or numbers when possible.

Assistant:"""

# Create a PromptTemplate instance with the defined template and input variables
prompt = PromptTemplate(
    template=PROMPT_TEMPLATE, input_variables=["context", "question"]
)
# Convert the vector store to a retriever
retriever = vectorstore.as_retriever()


# Define a function to format the retrieved documents
def format_docs(docs):
    return "\n\n".join(doc.page_content for doc in docs)

LCEL(LangChain 표현식 언어)을 사용하여 RAG 체인을 빌드합니다.

# Define the RAG (Retrieval-Augmented Generation) chain for AI response generation
rag_chain = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | prompt
    | llm
    | StrOutputParser()
)

# rag_chain.get_graph().print_ascii()

특정 질문으로 RAG 체인을 호출하고 응답을 검색합니다.

query = "What is PAL and PoT?"
res = rag_chain.invoke(query)
res

'PAL (Program-aided Language models) and PoT (Program of Thoughts prompting) are approaches that involve using language models to generate programming language statements to solve natural language reasoning problems. This method offloads the solution step to a runtime, such as a Python interpreter, allowing for complex computation and reasoning to be handled externally. PAL and PoT rely on language models with strong coding skills to effectively generate and execute these programming statements.'

축하합니다! Milvus와 LangChain으로 구동되는 하이브리드(밀도 벡터 + 스파스 bm25 함수) 검색 RAG 체인을 구축하셨습니다.