Vector Retrieval Service for Milvus:RAG システムで Milvus BM25 を使用した全文検索とハイブリッド検索

ステップ 1: 依存ライブラリのインストール

pip3 install pymilvus langchain dashscope

ステップ 2: データの準備

このトピックでは、公式の Milvus ドキュメントを例として使用し、LangChain SDK を使用してテキストをチャンク化し、text-embedding-v2 埋め込みモデルで埋め込みを生成し、結果の埋め込みと元のテキストを Milvus に挿入する方法を説明します。

from langchain_community.document_loaders import WebBaseLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.embeddings import DashScopeEmbeddings
from pymilvus import MilvusClient, DataType, Function, FunctionType

dashscope_api_key = "<YOUR_DASHSCOPE_API_KEY>"
milvus_url = "<YOUR_MILVUS_URL>"
user_name = "root"
password = "<YOUR_PASSWORD>"
collection_name = "milvus_overview"
dense_dim = 1536

loader = WebBaseLoader([
    'https://raw.githubusercontent.com/milvus-io/milvus-docs/refs/heads/v2.5.x/site/en/about/overview.md'
])

docs = loader.load()

text_splitter = RecursiveCharacterTextSplitter(chunk_size=1024, chunk_overlap=256)

# LangChain を使用して入力ドキュメントを chunk_size でチャンク化
all_splits = text_splitter.split_documents(docs)

embeddings = DashScopeEmbeddings(
    model="text-embedding-v2", dashscope_api_key=dashscope_api_key
)

text_contents = [doc.page_content for doc in all_splits]

vectors = embeddings.embed_documents(text_contents)


client = MilvusClient(
    uri=f"http://{milvus_url}:19530",
    token=f"{user_name}:{password}",
)

schema = MilvusClient.create_schema(
    enable_dynamic_field=True,
)

analyzer_params = {
    "type": "english"
}

# スキーマにフィールドを追加
schema.add_field(field_name="id", datatype=DataType.INT64, is_primary=True, auto_id=True)
schema.add_field(field_name="text", datatype=DataType.VARCHAR, max_length=65535, enable_analyzer=True, analyzer_params=analyzer_params, enable_match=True)
schema.add_field(field_name="sparse_bm25", datatype=DataType.SPARSE_FLOAT_VECTOR)
schema.add_field(field_name="dense", datatype=DataType.FLOAT_VECTOR, dim=dense_dim)

bm25_function = Function(
   name="bm25",
   function_type=FunctionType.BM25,
   input_field_names=["text"],
   output_field_names="sparse_bm25",
)
schema.add_function(bm25_function)

index_params = client.prepare_index_params()

# インデックスを追加
index_params.add_index(
    field_name="dense",
    index_name="dense_index",
    index_type="IVF_FLAT",
    metric_type="IP",
    params={"nlist": 128},
)

index_params.add_index(
    field_name="sparse_bm25",
    index_name="sparse_bm25_index",
    index_type="SPARSE_WAND",
    metric_type="BM25"
)

# コレクションを作成
client.create_collection(
    collection_name=collection_name,
    schema=schema,
    index_params=index_params
)

data = [
    {"dense": vectors[idx], "text": doc}
    for idx, doc in enumerate(text_contents)
]

# データを挿入
res = client.insert(
    collection_name=collection_name,
    data=data
)

print(f"Generated {len(vectors)} vectors, dimension: {len(vectors[0])}")

この例には、次のパラメーターが含まれています。実際の値に置き換えてください。

この例では、Milvus 2.5 の最新機能を使用しています。bm25_function オブジェクトを作成すると、Milvus はテキスト列を自動的に疎ベクトルに変換します。

同様に、中国語のドキュメントを処理する場合、Milvus 2.5 では対応する中国語アナライザを指定できます。

# アナライザのパラメーターを定義
analyzer_params = {
    "type": "chinese"  # アナライザのタイプを中国語に指定
}

# スキーマにテキストフィールドを追加し、アナライザを有効にする
schema.add_field(
    field_name="text",                      # フィールド名
    datatype=DataType.VARCHAR,              # データ型: 文字列 (VARCHAR)
    max_length=65535,                       # 最大長: 65535 文字
    enable_analyzer=True,                   # アナライザを有効にする
    analyzer_params=analyzer_params         # アナライザのパラメーター
)

ステップ 3: 全文検索の実行

Milvus 2.5 は、最新の全文検索機能を簡単に使用できる API を提供します。次のコードは例を示しています。

from pymilvus import MilvusClient

# Milvus クライアントを作成します。
client = MilvusClient(
    uri="http://c-xxxx.milvus.aliyuncs.com:19530",  # Milvus インスタンスのパブリックエンドポイント。
    token="<yourUsername>:<yourPassword>",  # Milvus インスタンスにログインするためのユーザー名とパスワード。
    db_name="default"  # 接続するデータベースの名前。この例では、デフォルトのデータベースを使用します。
)

search_params = {
    'params': {'drop_ratio_search': 0.2},
}

full_text_search_res = client.search(
    collection_name='milvus_overview',
    data=['what makes milvus so fast?'],
    anns_field='sparse_bm25',
    limit=3,
    search_params=search_params,
    output_fields=["text"],
)

for hits in full_text_search_res:
    for hit in hits:
        print(hit)
        print("\n")

"""
{'id': 456165042536597485, 'distance': 6.128782272338867, 'entity': {'text': '## What Makes Milvus so Fast？\n\nMilvus was designed from day one to be a highly efficient vector database system. In most cases, Milvus outperforms other vector databases by 2-5x (see the VectorDBBench results). This high performance is the result of several key design decisions:\n\n**Hardware-aware Optimization**: To accommodate Milvus in various hardware environments, we have optimized its performance specifically for many hardware architectures and platforms, including AVX512, SIMD, GPUs, and NVMe SSD.\n\n**Advanced Search Algorithms**: Milvus supports a wide range of in-memory and on-disk indexing/search algorithms, including IVF, HNSW, DiskANN, and more, all of which have been deeply optimized. Compared to popular implementations like FAISS and HNSWLib, Milvus delivers 30%-70% better performance.'}}

{'id': 456165042536597487, 'distance': 4.760214805603027, 'entity': {'text': "## What Makes Milvus so Scalable\n\nIn 2022, Milvus supported billion-scale vectors, and in 2023, it scaled up to tens of billions with consistent stability, powering large-scale scenarios for over 300 major enterprises, including Salesforce, PayPal, Shopee, Airbnb, eBay, NVIDIA, IBM, AT&T, LINE, ROBLOX, Inflection, etc.\n\nMilvus's cloud-native and highly decoupled system architecture ensures that the system can continuously expand as data grows:\n\n![Highly decoupled system architecture of Milvus](../../../assets/highly-decoupled-architecture.png)"}}
"""

ステップ 4: キーワードマッチングの実行

キーワードマッチングは Milvus 2.5 の新機能です。ベクトル類似性検索と組み合わせて検索範囲を絞り込み、検索パフォーマンスを向上させることができます。キーワードマッチング機能を使用するには、スキーマを定義する際に enable_analyzer と enable_match の両方を `True` に設定する必要があります。

filter = "TEXT_MATCH(text, 'query') and TEXT_MATCH(text, 'node')"

text_match_res = client.search(
    collection_name="milvus_overview",
    anns_field="dense",
    data=query_embeddings,
    filter=filter,
    search_params={"params": {"nprobe": 10}},
    limit=2,
    output_fields=["text"]
)

filter = "TEXT_MATCH(text, 'scalable fast')"

text_match_res = client.query(
    collection_name="milvus_overview",
    filter=filter,
    output_fields=["text"]
)

ステップ 5: ハイブリッド検索と RAG の使用

ハイブリッド検索は、ベクトル検索と全文検索を組み合わせたものです。相互ランク融合 (RRF) アルゴリズムを使用して、両方の検索結果をマージします。このプロセスは、ソートと重み付けを再最適化して、データの取得率と精度を向上させます。

次のコードは例を示しています。

from pymilvus import MilvusClient
from pymilvus import AnnSearchRequest, RRFRanker
from langchain_community.embeddings import DashScopeEmbeddings
from dashscope import Generation

# Milvus クライアントを作成します。
client = MilvusClient(
    uri="http://c-xxxx.milvus.aliyuncs.com:19530",  # Milvus インスタンスのパブリックエンドポイント。
    token="<yourUsername>:<yourPassword>",  # Milvus インスタンスにログインするためのユーザー名とパスワード。
    db_name="default"  # 接続するデータベースの名前。この例では、デフォルトのデータベースを使用します。
)

collection_name = "milvus_overview"

# DashScope API キーに置き換えてください
dashscope_api_key = "<YOUR_DASHSCOPE_API_KEY>"

# 埋め込みモデルを初期化
embeddings = DashScopeEmbeddings(
    model="text-embedding-v2",  # text-embedding-v2 モデルを使用します。
    dashscope_api_key=dashscope_api_key
)

# クエリを定義
query = "Why does Milvus run so scalable?"

# クエリを埋め込み、対応するベクトル表現を生成
query_embeddings = embeddings.embed_documents([query])

# 上位 K 件の結果数を設定
top_k = 5  # クエリに関連する上位 5 件のドキュメントを取得

# 密ベクトル検索のパラメーターを定義
search_params_dense = {
    "metric_type": "IP",
    "params": {"nprobe": 2}
}

# 密ベクトル検索リクエストを作成
request_dense = AnnSearchRequest([query_embeddings[0]], "dense", search_params_dense, limit=top_k)

# BM25 テキスト検索のパラメーターを定義
search_params_bm25 = {
    "metric_type": "BM25"
}

# BM25 テキスト検索リクエストを作成
request_bm25 = AnnSearchRequest([query], "sparse_bm25", search_params_bm25, limit=top_k)

# 2 つのリクエストを結合
reqs = [request_dense, request_bm25]

# RRF ランキングアルゴリズムを初期化
ranker = RRFRanker(100)

# ハイブリッド検索を実行
hybrid_search_res = client.hybrid_search(
    collection_name=collection_name,
    reqs=reqs,
    ranker=ranker,
    limit=top_k,
    output_fields=["text"]
)

# ハイブリッド検索結果からコンテキストを抽出
context = []
print("Top K Results:")
for hits in hybrid_search_res:  # ここで正しい変数を使用
    for hit in hits:
        context.append(hit['entity']['text'])  # テキストコンテンツをコンテキストリストに抽出
        print(hit['entity']['text'])  # 取得した各ドキュメントを出力


# クエリとコンテキストに基づいて回答を取得する関数を定義
def getAnswer(query, context):
    prompt = f'''Please answer my question based on the content within:
    ```
    {context}
    ```
    My question is: {query}.
    '''
    # 生成モジュールを呼び出して回答を取得
    rsp = Generation.call(model='qwen-turbo', prompt=prompt)
    return rsp.output.text

# 回答を取得
answer = getAnswer(query, context)

print(answer)


# 期待される出力の抜粋
"""
Milvus is highly scalable due to its cloud-native and highly decoupled system architecture. This architecture allows the system to continuously expand as data grows. Additionally, Milvus supports three deployment modes that cover a wide...
"""