Milvus를 사용한 하이브리드 검색

이 튜토리얼에서는 Milvus와 BGE-M3 모델을 사용하여 하이브리드 검색을 수행하는 방법을 보여드리겠습니다. BGE-M3 모델은 텍스트를 조밀하고 희박한 벡터로 변환할 수 있습니다. Milvus는 두 가지 유형의 벡터를 하나의 컬렉션에 저장하여 결과 연관성을 향상시키는 하이브리드 검색을 지원합니다.

Milvus는 밀도, 스파스, 하이브리드 검색 방식을 지원합니다:

• 밀도 검색: 시맨틱 컨텍스트를 활용하여 쿼리 뒤에 숨겨진 의미를 이해합니다.
• 스파스 검색: 키워드 매칭을 강조하여 전체 텍스트 검색과 동일하게 특정 용어에 기반한 결과를 찾습니다.
• 하이브리드 검색: 밀도 검색과 스파스 검색 방식을 모두 결합하여 전체 문맥과 특정 키워드를 파악하여 포괄적인 검색 결과를 제공합니다.

이러한 방법을 통합함으로써 Milvus 하이브리드 검색은 의미론적 유사성과 어휘적 유사성의 균형을 맞춰 검색 결과의 전반적인 관련성을 향상시킵니다. 이 노트북에서는 이러한 검색 전략을 설정하고 사용하는 과정을 안내하며, 다양한 검색 시나리오에서 그 효과를 강조합니다.

종속성 및 환경

$ pip install --upgrade pymilvus "pymilvus[model]"

데이터 세트 다운로드

검색을 시연하려면 문서 말뭉치가 필요합니다. Quora 중복 질문 데이터 집합을 사용하여 로컬 디렉터리에 배치해 보겠습니다.

데이터 집합의 출처: 첫 번째 Quora 데이터 세트 릴리즈: 질문 쌍

# Run this cell to download the dataset
$ wget http://qim.fs.quoracdn.net/quora_duplicate_questions.tsv

데이터 로드 및 준비

데이터 세트를 로드하고 검색을 위한 작은 말뭉치를 준비합니다.

import pandas as pd

file_path = "quora_duplicate_questions.tsv"
df = pd.read_csv(file_path, sep="\t")
questions = set()
for _, row in df.iterrows():
    obj = row.to_dict()
    questions.add(obj["question1"][:512])
    questions.add(obj["question2"][:512])
    if len(questions) > 500:  # Skip this if you want to use the full dataset
        break

docs = list(questions)

# example question
print(docs[0])

What is the strongest Kevlar cord?

임베딩에 BGE-M3 모델 사용

BGE-M3 모델은 텍스트를 고밀도 및 스파스 벡터로 임베드할 수 있습니다.

from milvus_model.hybrid import BGEM3EmbeddingFunction

ef = BGEM3EmbeddingFunction(use_fp16=False, device="cpu")
dense_dim = ef.dim["dense"]

# Generate embeddings using BGE-M3 model
docs_embeddings = ef(docs)

Fetching 30 files: 100%|██████████| 30/30 [00:00<00:00, 302473.85it/s]
Inference Embeddings: 100%|██████████| 32/32 [01:59<00:00,  3.74s/it]

Milvus 컬렉션 및 색인 설정

Milvus 컬렉션을 설정하고 벡터 필드에 대한 인덱스를 생성하겠습니다.

from pymilvus import (
    connections,
    utility,
    FieldSchema,
    CollectionSchema,
    DataType,
    Collection,
)

# Connect to Milvus given URI
connections.connect(uri="./milvus.db")

# Specify the data schema for the new Collection
fields = [
    # Use auto generated id as primary key
    FieldSchema(
        name="pk", dtype=DataType.VARCHAR, is_primary=True, auto_id=True, max_length=100
    ),
    # Store the original text to retrieve based on semantically distance
    FieldSchema(name="text", dtype=DataType.VARCHAR, max_length=512),
    # Milvus now supports both sparse and dense vectors,
    # we can store each in a separate field to conduct hybrid search on both vectors
    FieldSchema(name="sparse_vector", dtype=DataType.SPARSE_FLOAT_VECTOR),
    FieldSchema(name="dense_vector", dtype=DataType.FLOAT_VECTOR, dim=dense_dim),
]
schema = CollectionSchema(fields)

# Create collection (drop the old one if exists)
col_name = "hybrid_demo"
if utility.has_collection(col_name):
    Collection(col_name).drop()
col = Collection(col_name, schema, consistency_level="Strong")

# To make vector search efficient, we need to create indices for the vector fields
sparse_index = {"index_type": "SPARSE_INVERTED_INDEX", "metric_type": "IP"}
col.create_index("sparse_vector", sparse_index)
dense_index = {"index_type": "AUTOINDEX", "metric_type": "IP"}
col.create_index("dense_vector", dense_index)
col.load()

밀버스 컬렉션에 데이터 삽입하기

컬렉션에 문서와 임베딩을 삽입합니다.

# For efficiency, we insert 50 records in each small batch
for i in range(0, len(docs), 50):
    batched_entities = [
        docs[i : i + 50],
        docs_embeddings["sparse"][i : i + 50],
        docs_embeddings["dense"][i : i + 50],
    ]
    col.insert(batched_entities)
print("Number of entities inserted:", col.num_entities)

Number of entities inserted: 502

검색 쿼리 입력

# Enter your search query
query = input("Enter your search query: ")
print(query)

# Generate embeddings for the query
query_embeddings = ef([query])
# print(query_embeddings)

How to start learning programming?

검색 실행

먼저 검색을 실행하는 데 유용한 몇 가지 기능을 준비합니다:

• dense_search밀집 벡터 필드에서만 검색
• sparse_search: 희소 벡터 필드에서만 검색
• hybrid_search: 가중치 재랭커를 사용하여 밀집 벡터 필드와 희소 벡터 필드 모두에서 검색

from pymilvus import (
    AnnSearchRequest,
    WeightedRanker,
)


def dense_search(col, query_dense_embedding, limit=10):
    search_params = {"metric_type": "IP", "params": {}}
    res = col.search(
        [query_dense_embedding],
        anns_field="dense_vector",
        limit=limit,
        output_fields=["text"],
        param=search_params,
    )[0]
    return [hit.get("text") for hit in res]


def sparse_search(col, query_sparse_embedding, limit=10):
    search_params = {
        "metric_type": "IP",
        "params": {},
    }
    res = col.search(
        [query_sparse_embedding],
        anns_field="sparse_vector",
        limit=limit,
        output_fields=["text"],
        param=search_params,
    )[0]
    return [hit.get("text") for hit in res]


def hybrid_search(
    col,
    query_dense_embedding,
    query_sparse_embedding,
    sparse_weight=1.0,
    dense_weight=1.0,
    limit=10,
):
    dense_search_params = {"metric_type": "IP", "params": {}}
    dense_req = AnnSearchRequest(
        [query_dense_embedding], "dense_vector", dense_search_params, limit=limit
    )
    sparse_search_params = {"metric_type": "IP", "params": {}}
    sparse_req = AnnSearchRequest(
        [query_sparse_embedding], "sparse_vector", sparse_search_params, limit=limit
    )
    rerank = WeightedRanker(sparse_weight, dense_weight)
    res = col.hybrid_search(
        [sparse_req, dense_req], rerank=rerank, limit=limit, output_fields=["text"]
    )[0]
    return [hit.get("text") for hit in res]

정의된 함수를 사용하여 세 가지 다른 검색을 실행해 보겠습니다:

dense_results = dense_search(col, query_embeddings["dense"][0])
sparse_results = sparse_search(col, query_embeddings["sparse"]._getrow(0))
hybrid_results = hybrid_search(
    col,
    query_embeddings["dense"][0],
    query_embeddings["sparse"]._getrow(0),
    sparse_weight=0.7,
    dense_weight=1.0,
)

검색 결과 표시

밀도, 스파스, 하이브리드 검색의 결과를 표시하려면 결과 서식을 지정하는 몇 가지 유틸리티가 필요합니다.

def doc_text_formatting(ef, query, docs):
    tokenizer = ef.model.tokenizer
    query_tokens_ids = tokenizer.encode(query, return_offsets_mapping=True)
    query_tokens = tokenizer.convert_ids_to_tokens(query_tokens_ids)
    formatted_texts = []

    for doc in docs:
        ldx = 0
        landmarks = []
        encoding = tokenizer.encode_plus(doc, return_offsets_mapping=True)
        tokens = tokenizer.convert_ids_to_tokens(encoding["input_ids"])[1:-1]
        offsets = encoding["offset_mapping"][1:-1]
        for token, (start, end) in zip(tokens, offsets):
            if token in query_tokens:
                if len(landmarks) != 0 and start == landmarks[-1]:
                    landmarks[-1] = end
                else:
                    landmarks.append(start)
                    landmarks.append(end)
        close = False
        formatted_text = ""
        for i, c in enumerate(doc):
            if ldx == len(landmarks):
                pass
            elif i == landmarks[ldx]:
                if close:
                    formatted_text += "</span>"
                else:
                    formatted_text += "<span style='color:red'>"
                close = not close
                ldx = ldx + 1
            formatted_text += c
        if close is True:
            formatted_text += "</span>"
        formatted_texts.append(formatted_text)
    return formatted_texts

그런 다음 검색 결과를 하이라이트가 있는 텍스트로 표시할 수 있습니다:

from IPython.display import Markdown, display

# Dense search results
display(Markdown("**Dense Search Results:**"))
formatted_results = doc_text_formatting(ef, query, dense_results)
for result in dense_results:
    display(Markdown(result))

# Sparse search results
display(Markdown("\n**Sparse Search Results:**"))
formatted_results = doc_text_formatting(ef, query, sparse_results)
for result in formatted_results:
    display(Markdown(result))

# Hybrid search results
display(Markdown("\n**Hybrid Search Results:**"))
formatted_results = doc_text_formatting(ef, query, hybrid_results)
for result in formatted_results:
    display(Markdown(result))

밀도 검색 결과:

로봇 공학 학습을 시작하는 가장 좋은 방법은 무엇인가요?

자바 같은 컴퓨터 언어를 배우려면 어떻게 해야 하나요?

정보 보안을 배우려면 어떻게 시작해야 하나요?

자바 프로그래밍이란 무엇인가요? 자바 프로그래밍 언어를 배우는 방법?

컴퓨터 보안은 어떻게 배울 수 있나요?

로봇 공학을 시작하는 가장 좋은 방법은 무엇인가요? 작업을 시작할 수 있는 최고의 개발 보드는 무엇인가요?

영어를 유창하게 배우려면 어떻게 해야 하나요?

프랑스어를 배우는 가장 좋은 방법은 무엇인가요?

물리학을 쉽게 배우려면 어떻게 해야 하나요?

UPSC는 어떻게 준비하나요?

스파스 검색 결과:

자바 프로그래밍이란 무엇인가요? 자바 프로그래밍 언어를 배우는 방법?

로봇 공학을 배우는 가장 좋은 방법은 무엇인가요?

머신 러닝의 대안은 무엇인가요?

C 프로그래밍을 사용하여 Linux에서 새 터미널과 새 셸을 만들려면어떻게해야합니까?

C 프로그래밍 (Linux 터미널)을 사용하여 새 터미널에서 새 셸을 만들려면어떻게해야하나요?

하이데라바드에서 어떤 사업을 시작하는 것이 더 낫습니까?

하이데라바드에서 어떤 사업을 시작하는 것이 좋은가요?

로봇 공학을 시작하는 가장 좋은 방법은 무엇인가요? 작업을 시작할 수있는 최고의 개발 보드는 무엇입니까?

초보자가 컴퓨터 프로그래밍 알고리즘을 이해하려면 어떤 수학이 필요하나요? 완전 초보자에게 적합한 알고리즘 관련 서적은 무엇인가요?

어떻게 하면 삶이 자신에게 적합하게 만들고 삶이 정신적, 정서적으로 자신을 학대하는 것을 막을 수 있습니까?

하이브리드 검색 결과:

로봇 공학을 시작하는 가장 좋은 방법은 무엇인가요? 작업을 시작할 수있는 최고의 개발 보드는 무엇입니까?

자바 프로그래밍이란 무엇인가요? 자바 프로그래밍 언어를 배우는 방법?

로봇 공학을 배우는 가장 좋은 방법은 무엇인가요?

UPSC는어떻게 준비하나요?

물리학을 쉽게 배우려면어떻게 해야 하나요?

프랑스어를 배우는 가장 좋은 방법은 무엇인가요?

영어를 유창하게 배우려면어떻게 해야 하나요?

컴퓨터 보안은어떻게 배울 수 있나요?

정보 보안을 배우려면어떻게 시작해야 하나요?

자바 같은 컴퓨터 언어를 배우려면어떻게 해야 하나요?

머신 러닝의 대안은 무엇인가요?

C 프로그래밍을 사용하여 Linux에서 새 터미널과 새 셸을 만들려면어떻게 해야 하나요?

C 프로그래밍 (Linux 터미널)을 사용하여 새 터미널에서 새 셸을 만들려면어떻게해야하나요?

하이데라바드에서 어떤 사업을 시작하는 것이 더 낫습니까?

하이데라바드에서 어떤 사업을 시작하는 것이 좋은가요?

완전한 초보자가 컴퓨터 프로그래밍 알고리즘을 이해하려면 어떤 수학이 필요합니까? 완전한 초보자에게 적합한 알고리즘에 관한 책은 무엇입니까?

어떻게 삶을 자신에게 적합하게 만들고 삶이 정신적, 정서적으로 당신을 학대하는 것을 막을 수 있습니까?

빠른 배포

이 튜토리얼을 통해 온라인 데모를 시작하는 방법에 대해 알아보려면 예제 애플리케이션을 참조하세요.