Vector Retrieval Service for Milvus:個人向けナレッジ Q&A アプリケーションの構築

ソリューション概要

エンドツーエンドのワークフローは 3 つのステージで構成されます。コンソールでナレッジベースを作成してデータをインポートし、バージョンを発行してコンテンツを検索可能にします。その後、ソフトウェア開発キット (SDK) を通じて検索し、結果を LLM に渡して回答を生成します。

データのインポートと検索は SDK を通じて実行できますが、バージョンの発行はコンソールで行う必要があります。対応する OpenAPI 操作はありません。開始する前に、これを踏まえて自動化を計画してください。

一般的なドキュメント 10〜100 件を解析・チャンキングし、データが検索可能になるまでには、通常、合計で 10〜20 分かかります。

前提条件

• Alibaba Cloud アカウント。ナレッジベースは、中国 (杭州)、中国 (北京)、中国 (張家口)、中国 (深圳) の各リージョンで利用可能です。

• 呼び出し元アカウントの RAM 権限。Alibaba Cloud アカウントで RAM コンソールにログインし、RAM ユーザーを作成して Access using a permanent AccessKey を選択し、システムポリシー AliyunMilvusFullAccess をユーザーにアタッチします。

  AccessKey を持つ RAM ユーザーの作成には、セキュリティ認証 (多要素認証 (MFA) コード、SMS 認証コード、または顔スキャン) が必要です。AccessKey シークレットは作成後に一度しか表示されません。ページ上の Save results をクリックしてダウンロードし、保存してください。より厳格な最小権限のためには、milvusknowledgebase:* のみを含む別のカスタムポリシーを作成し、上記のシステムポリシーの代わりに使用してください。

• OpenAI の chat/completions プロトコルをサポートする LLM API。たとえば、Alibaba Cloud Model Studio の OpenAI 互換エンドポイントと API キーなどです。

• ローカルマシンにインストールされた Python 3.9 以降。

  AccessKey ペアと LLM API キーは、ローカルの環境変数にのみ保存してください。ドキュメント、チャットログ、スクリーンショット、またはコードリポジトリには決して書き込まないでください。

ステップ 1：ナレッジベースの作成

1. Alibaba Cloud Milvus コンソールにログインし、上部のリージョンドロップダウンリストからターゲットリージョンを選択し、左側メニューで [Knowledge Base Service] をクリックします。

2. [Knowledge Base] ページで [Create Knowledge Base] をクリックし、次の設定を行います。

   [Data type] と [埋め込みモデル] の設定は、作成後に変更できません。このチュートリアルでは、[Omni-modal knowledge base] と [Built-in model] を使用します。続行する前に、両方の選択内容を確認してください。

   設定項目	説明
   [Name]	2〜64 文字。中国語文字はサポートされていません。名前は同じテナント内で一意である必要があります。
   [Data type]	有効値：[オムニモーダルナレッジベース] と [構造化ナレッジベース] (チャンクはテーブル行ごとに生成され、xls、xlsx、csv のみがサポートされます)。この設定は作成後に変更できません。このチュートリアルでは、mp4、mp3、png、pdf、ppt、txt、markdown、docs、xlsx、csv、jsonl、faq をサポートする [オムニモーダルナレッジベース] を使用します。
   [埋め込みモデル]	有効値：[Built-in model] と [Private/External model]。この設定は作成後に変更できません。このチュートリアルでは、デフォルトで text-embedding-v3 となる [Built-in model] を使用し、[Vector dimensions] には自動的に 1024 が表示されます。[Private/External model] を選択すると、[Vector dimensions] には - が表示され、編集できません。
   [Specification]	必須。現在の最小仕様は 4 CU (4 vCPU / 16 GiB) です。作成後に詳細ページで調整できます。
   [Network settings]	必須。VPC と vSwitch を指定します。その場で作成することもできます。

   チャンキング戦略は作成ページでは設定しません。ナレッジベース詳細ページの [Processing strategy] エリアで、システムが提供する [Default strategy] (最大長 512 文字のインテリジェントな分割) を表示するか、[Create strategy] をクリックしてカスタム戦略を作成できます。SDK を通じてドキュメントを登録する際に、strategy_id で戦略を指定できます。

3. [Create Knowledge Base] をクリックし、ナレッジベースのステータスが [Creating] から [Running] に変わるまで待ちます。

4. ナレッジベースのリストで、[Knowledge base ID] (kd-803ae9b10cc31 などの形式) をメモします。この ID は、後続の SDK 呼び出しに必要です。

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

結果を検証するために、いくつかの Markdown、TXT、PDF、または Word ドキュメントを準備します。ファイル名はドキュメントのトピックを説明するものにしてください。本文には完全なタイトルとコンテキストを含める必要があります。

適切なドキュメントがない場合は、公開されている中国の法律判例データセット LeCaRDv2 を使用できます。法律判例には、特定の事件番号、事実、判決が含まれているため、一般的な知識と比較して、回答が本当にナレッジベースからのものであるかを検証しやすくなります。このデータセットは、検索技術を実証するためにのみ使用され、法的助言を構成するものではありません。

documents という名前のローカルディレクトリを作成し、そこにファイルをコピーします。次のステップのサンプルコードは、./documents からファイルを読み取ります。

ステップ 3：ローカル環境のセットアップ

1. 仮想環境を作成し、依存関係をインストールします。

   python3 -m venv .venv
   source .venv/bin/activate
   pip install alibabacloud_milvusknowledgebase20260604 requests streamlit

2. ランタイムパラメータを環境変数として設定します。KB_ID はステップ 1 で記録したナレッジベース ID、KB_REGION はナレッジベースのリージョンです。

export ALIBABA_CLOUD_ACCESS_KEY_ID="YOUR_ACCESS_KEY_ID"
export ALIBABA_CLOUD_ACCESS_KEY_SECRET="YOUR_ACCESS_KEY_SECRET"
export KB_REGION="cn-hangzhou"
export KB_ID="kd-xxxxxxxxxxxxx"
export KB_VERSION="LATEST_PUBLISHED"

export LLM_BASE_URL="https://your-openai-compatible-api.example.com/v1"
export LLM_API_KEY="YOUR_LLM_TOKEN"
export LLM_MODEL="YOUR_MODEL_NAME"

ステップ 4：サンプルコードの作成

次の内容を kb_demo.py として保存します。この単一のモジュールが、このチュートリアルの完全なサンプルコードです。クライアントの初期化、ドキュメントのアップロード、検索、LLM Q&A の 4 つの部分で構成されています。次のステップでは、ここで定義された関数を呼び出します。

from __future__ import annotations

import os
from pathlib import Path

import requests
from alibabacloud_milvusknowledgebase20260604.client import Client
from alibabacloud_milvusknowledgebase20260604 import models as milvus_kb_models
from alibabacloud_tea_openapi import models as open_api_models

REGION = os.getenv("KB_REGION", "cn-hangzhou")
KB_ID = os.environ["KB_ID"]
KB_VERSION = os.getenv("KB_VERSION", "LATEST_PUBLISHED")
ENDPOINT = f"milvusknowledgebase.{REGION}.aliyuncs.com"

client = Client(open_api_models.Config(
    access_key_id=os.environ["ALIBABA_CLOUD_ACCESS_KEY_ID"],
    access_key_secret=os.environ["ALIBABA_CLOUD_ACCESS_KEY_SECRET"],
    endpoint=ENDPOINT,
    region_id=REGION,
    connect_timeout=10_000,
    read_timeout=60_000,
))

def upload_document(file_path: str) -> dict:
    path = Path(file_path).resolve()
    size = path.stat().st_size

    presigned = client.get_knowledge_base_pre_signed_url(
        KB_ID,
        milvus_kb_models.GetKnowledgeBasePreSignedUrlRequest(
            knowledge_base_id=KB_ID,
            documents=[
                milvus_kb_models.GetKnowledgeBasePreSignedUrlRequestDocuments(
                    path=path.name, name=path.name, size=size,
                )
            ],
            expires_in=3600,
        ),
    )
    upload_url = presigned.body.data.pre_signed_urls[0]

    # 署名付き URL は空の Content-Type で署名されています。PUT リクエストに Content-Type ヘッダーを含めないでください。
    with path.open("rb") as source:
        response = requests.put(upload_url, data=source, timeout=120)
        response.raise_for_status()

    added = client.add_documents(
        KB_ID,
        milvus_kb_models.AddDocumentsRequest(
            knowledge_base_id=KB_ID,
            import_type="LOCAL_UPLOAD",
            documents=[
                milvus_kb_models.AddDocumentsRequestDocuments(
                    path=path.name, name=path.name, size=size,
                )
            ],
            dedup=milvus_kb_models.AddDocumentsRequestDedup(
                doc_name_dedup=True, content_dedup=False,
            ),
        ),
    )
    return added.body

def search_knowledge_base(query: str, page_size: int = 6):
    resp = client.search_knowledge_base(
        KB_ID,
        milvus_kb_models.SearchKnowledgeBaseRequest(
            query=query,
            version=KB_VERSION,
            page_number=1,
            page_size=page_size,
            retrieval_config=milvus_kb_models.SearchKnowledgeBaseRequestRetrievalConfig(
                candidate_count=48,
                min_score=0,
                semantic_weight=0.5,
                enable_query_expansion=False,
            ),
        ),
    )
    return resp.body

def chat(messages: list[dict[str, str]]) -> str:
    base_url = os.environ["LLM_BASE_URL"].rstrip("/")
    response = requests.post(
        f"{base_url}/chat/completions",
        headers={"Authorization": f"Bearer {os.environ['LLM_API_KEY']}"},
        json={
            "model": os.environ["LLM_MODEL"],
            "messages": messages,
            "temperature": 0.1,
        },
        timeout=60,
    )
    response.raise_for_status()
    return response.json()["choices"][0]["message"]["content"].strip()

def ask(question: str) -> tuple[str, object]:
    search_query = chat([
        {
            "role": "system",
            "content": "ユーザーの質問を、ナレッジベース検索に適した自己完結型のクエリに書き換えます。クエリのみを出力してください。",
        },
        {"role": "user", "content": question},
    ])
    search_result = search_knowledge_base(search_query)
    results = search_result.results or []

    context = "\n\n".join(
        f"[Source {i}] {item.document_name or ''}\n{item.content}"
        for i, item in enumerate(results, start=1)
    )
    answer = chat([
        {
            "role": "system",
            "content": "提供された資料にのみ基づいて回答してください。[Source N] のように資料を引用してください。資料が不十分な場合は、その旨を明記してください。",
        },
        {"role": "user", "content": f"Question: {question}\n\nMaterials:\n{context}"},
    ])
    return answer, search_result

ステップ 5：データのアップロード

アップロードには、署名付き URL の取得、Object Storage Service (OSS) へのファイルの書き込み、ドキュメントの登録という 3 つのステップがあります。upload_document() は、このプロセス全体をラップします。

単一ドキュメントのアップロード：

from kb_demo import upload_document

result = upload_document("./documents/example.md")
print(result.request_id)

バッチアップロードの場合、10 ドキュメントのグループから始めます：

from pathlib import Path
from kb_demo import upload_document

for path in sorted(Path("./documents").glob("*.md")):
    upload_document(str(path))
    print("Submitted:", path.name)

署名付き URL は空の Content-Type で署名されています。PUT リクエストには Content-Type リクエストヘッダーを含めてはいけません。含めると、OSS は 403 SignatureDoesNotMatch を返します。ネットワークの問題で PUT が失敗した場合は、直接再試行してください。

OSS にファイルを置くだけでは、ナレッジベースには追加されません。登録を完了するには、AddDocuments を呼び出す必要があります。登録が成功すると、レスポンスの run は RUNNING となり、解析とチャンキングが進行中であることを意味します。一般的なドキュメント 10〜100 件の場合、この処理は合計で通常 10〜20 分かかります。バージョンを発行する前に、ステータスが [Processed] になるまで待ってください。

ドキュメントを登録する際、path と name の両方に、ファイル名のみを指定します。署名付き URL のオブジェクトパスは使用しないでください。direct_upload/… のような内部パスを指定すると、400 Path must not use an internal storage prefix. が返されます。path を空にすると 400 Path can't be empty. が返されます。登録成功のレスポンスでは、data.documents は空の配列である可能性があります。これは登録が失敗したことを意味するものではありません。ナレッジベース詳細ページの [Documents] タブや [Chunks] タブ、または後続の検索結果で結果を検証してください。

アップロードが完了したら、コンソールの [Data management] ページでデータステータスとチャンク数を確認してください。ステータスが [Processed] になった後にのみ、バージョンを発行してください。[View chunks] をクリックしてチャンキング結果をプレビューしてください。

ステップ 6：バージョンの発行

インポートされたデータは未発行のままであり、バージョンが発行されるまで検索できません。

1. ナレッジベース詳細ページで、[Version management] タブをクリックします。ページに保留中の変更が表示されていることを確認し、[Publish version] をクリックします。

2. [Confirm changes] ステップで、変更ログを確認し、[Next] をクリックします。

3. [Enter description] ステップで、発行の説明 (最大 200 文字) を入力し、[Publish] をクリックします。[Version published] というメッセージが表示されるまで待ちます。

4. [Version history] で、バージョン番号を確認します。最初に発行されたバージョンは v1 で、ステータスは [Published] です。

   発行後、ローカルの環境変数を更新してください：

export KB_VERSION="v1"

ステップ 7：データの検索

search_knowledge_base() を呼び出して、発行されたバージョンを検索します：

from kb_demo import search_knowledge_base

result = search_knowledge_base("契約を解除するにはどのような条件を満たす必要がありますか？")

for item in result.results or []:
    print(item.document_name, item.score)
    print(item.content[:300])

version には、v1 や v2 などの明示的なバージョン番号を使用します。LATEST_PUBLISHED を使用して、最新の発行済みバージョンを検索することもできます。外部に安定した Q&A を提供するために DRAFT は使用しないでください。

この例では、min_score=0 は関連性の低いチャンクも返します。本番環境では、実際の結果に基づいて min_score を上げるか page_size を減らして、関連性の低いコンテンツが LLM のコンテキストに含まれないようにします。

また、コンソールの [Search validation] ページで TopK、セマンティック検索の重み、リランキングモデル、フィルタースコープを調整して、検索結果を迅速に比較することもできます。

ステップ 8：Q&A のための LLM 接続

ask() は、まず LLM に質問を検索に適したクエリに書き換えさせ、次に SearchKnowledgeBase を呼び出し、最後に返されたチャンクのみに基づいて回答を要約するように LLM に依頼します。

from kb_demo import ask

answer, search_result = ask("これらのドキュメントは契約解除について何と述べていますか？")
print(answer)
print("Request ID:", search_result.request_id)

ステップ 9：Web アプリケーションの構築

1. 次の内容を app.py として保存します。

   import streamlit as st
   from kb_demo import ask
   
   st.set_page_config(page_title="ナレッジベース Q&A")
   st.title("ナレッジベース Q&A")
   
   question = st.chat_input("質問を入力してください")
   if question:
       with st.chat_message("user"):
           st.write(question)
       with st.chat_message("assistant"):
           with st.spinner("ナレッジベースを検索中..."):
               answer, result = ask(question)
           st.write(answer)
           with st.expander("ソースを表示"):
               for item in result.results or []:
                   st.markdown(f"**{item.document_name or '無題のドキュメント'}**")
                   st.caption(f"score: {item.score}  ·  Request ID: {result.request_id}")
                   st.write(item.content)

2. アプリケーションを起動します。

streamlit run app.py

ブラウザが自動的にローカルのナレッジベース Q&A ページを開きます。

よくある質問

公開前のチェックリスト

• キーローテーションをサポートする最小権限を持つ専用の RAM ユーザーを使用してください。Alibaba Cloud アカウントの AccessKey ペアを長期間使用しないでください。

• 処理する権限のあるデータのみをアップロードしてください。

• すべての回答は、展開されたソースチャンクまで追跡可能でなければなりません。

• ページを公開する場合は、認証と HTTPS を追加し、アクセスログ内の機密情報をマスクしてください。

• API の可用性と権限要件については、製品ドキュメントと実際のコンソールの動作を信頼できる唯一の情報源として扱ってください。